Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: remove new exceptions #326

Merged
merged 6 commits into from
Mar 21, 2024
Merged

fix: remove new exceptions #326

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
f97139a
fix: remove new exceptions
wey-gu Mar 21, 2024
d9f8550
fix: opt-in the the execution error retry
wey-gu Mar 21, 2024
57efb3e
add logging for session.execute when retry execution enabled
wey-gu Mar 21, 2024
3bfa12f
address comments from nicco & optmizations
wey-gu Mar 21, 2024
714dd75
remove unnecessarily use space
wey-gu Mar 21, 2024
e66b007
fix: should return when session failed to be newed(but removed from
wey-gu Mar 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 2 additions & 10 deletions nebula3/gclient/net/Connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,6 @@
AuthFailedException,
IOErrorException,
ClientServerIncompatibleException,
SessionException,
ExecutionErrorException,
)

from nebula3.gclient.net.AuthResult import AuthResult
Expand Down Expand Up @@ -198,12 +196,6 @@ def execute_parameter(self, session_id, stmt, params):
"""
try:
resp = self._connection.executeWithParameter(session_id, stmt, params)
if resp.error_code == ErrorCode.E_SESSION_INVALID:
raise SessionException(resp.error_code, resp.error_msg)
if resp.error_code == ErrorCode.E_SESSION_TIMEOUT:
raise SessionException(resp.error_code, resp.error_msg)
if resp.error_code == ErrorCode.E_EXECUTION_ERROR:
raise ExecutionErrorException(resp.error_msg)
return resp
except Exception as te:
if isinstance(te, TTransportException):
Expand Down Expand Up @@ -274,15 +266,15 @@ def close(self):
self._connection._iprot.trans.close()
except Exception as e:
logger.error(
'Close connection to {}:{} failed:{}'.format(self._ip, self._port, e)
"Close connection to {}:{} failed:{}".format(self._ip, self._port, e)
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please ignore formatter's change, sorry

)

def ping(self):
"""check the connection if ok
:return: True or False
"""
try:
resp = self._connection.execute(0, 'YIELD 1;')
resp = self._connection.execute(0, "YIELD 1;")
return True
except Exception:
return False
Expand Down
76 changes: 31 additions & 45 deletions nebula3/gclient/net/Session.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,8 @@
from nebula3.Exception import (
IOErrorException,
NotValidConnectionException,
ExecutionErrorException,
)

from nebula3.common.ttypes import ErrorCode
from nebula3.data.ResultSet import ResultSet
from nebula3.gclient.net.AuthResult import AuthResult
from nebula3.logger import logger
Expand Down Expand Up @@ -57,11 +56,23 @@ def execute_parameter(self, stmt, params):
:return: ResultSet
"""
if self._connection is None:
raise RuntimeError('The session has been released')
raise RuntimeError("The session has been released")
try:
start_time = time.time()
resp = self._connection.execute_parameter(self._session_id, stmt, params)
end_time = time.time()

if resp.error_code == ErrorCode.E_EXECUTION_ERROR:
retry_count = 0
while retry_count < self._retry_times:
time.sleep(self._retry_interval_sec)
resp = self._connection.execute_parameter(
self._session_id, stmt, params
)
if resp.error_code != ErrorCode.E_EXECUTION_ERROR:
break
retry_count += 1

return ResultSet(
resp,
all_latency=int((end_time - start_time) * 1000000),
Expand All @@ -72,7 +83,7 @@ def execute_parameter(self, stmt, params):
self._pool.update_servers_status()
if self._retry_connect:
if not self._reconnect():
logger.warning('Retry connect failed')
logger.warning("Retry connect failed")
raise IOErrorException(
IOErrorException.E_ALL_BROKEN, ie.message
)
Expand All @@ -86,27 +97,6 @@ def execute_parameter(self, stmt, params):
timezone_offset=self._timezone_offset,
)
raise
except ExecutionErrorException as eee:
retry_count = 0
while retry_count < self._retry_times:
try:
# TODO: add exponential backoff
time.sleep(self._retry_interval_sec)
resp = self._connection.execute_parameter(
self._session_id, stmt, params
)
end_time = time.time()
return ResultSet(
resp,
all_latency=int((end_time - start_time) * 1000000),
timezone_offset=self._timezone_offset,
)
except ExecutionErrorException:
if retry_count >= self._retry_times - 1:
raise eee
else:
retry_count += 1
continue
except Exception:
raise

Expand Down Expand Up @@ -244,18 +234,30 @@ def execute_json_with_parameter(self, stmt, params):
:return: JSON string
"""
if self._connection is None:
raise RuntimeError('The session has been released')
raise RuntimeError("The session has been released")
try:
resp_json = self._connection.execute_json_with_parameter(
self._session_id, stmt, params
)
retry_count = 0
while (
retry_count < self._retry_times
and resp_json.get("errors", [{}])[0].get("code")
== ErrorCode.E_EXECUTION_ERROR
):
time.sleep(self._retry_interval_sec)
resp_json = self._connection.execute_json_with_parameter(
self._session_id, stmt, params
)
retry_count += 1
return resp_json

except IOErrorException as ie:
if ie.type == IOErrorException.E_CONNECT_BROKEN:
self._pool.update_servers_status()
if self._retry_connect:
if not self._reconnect():
logger.warning('Retry connect failed')
logger.warning("Retry connect failed")
raise IOErrorException(
IOErrorException.E_ALL_BROKEN, ie.message
)
Expand All @@ -264,22 +266,6 @@ def execute_json_with_parameter(self, stmt, params):
)
return resp_json
raise
except ExecutionErrorException as eee:
retry_count = 0
while retry_count < self._retry_times:
try:
# TODO: add exponential backoff
time.sleep(self._retry_interval_sec)
resp = self._connection.execute_json_with_parameter(
self._session_id, stmt, params
)
return resp
except ExecutionErrorException:
if retry_count >= self._retry_times - 1:
raise eee
else:
retry_count += 1
continue
except Exception:
raise

Expand Down Expand Up @@ -310,7 +296,7 @@ def ping_session(self):
return True
else:
logger.error(
'failed to ping the session: error code:{}, error message:{}'.format(
"failed to ping the session: error code:{}, error message:{}".format(
resp.error_code, resp.error_msg
)
)
Expand Down Expand Up @@ -342,5 +328,5 @@ def _idle_time(self):
def _sign_out(self):
"""sign out the session"""
if self._connection is None:
raise RuntimeError('The session has been released')
raise RuntimeError("The session has been released")
self._connection.signout(self._session_id)
Loading
Loading