Skip to content

client_async: Allow throwing an exception upon socket error during wakeup #134

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

Merged
merged 3 commits into from
Mar 10, 2024
Merged

client_async: Allow throwing an exception upon socket error during wakeup #134

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ef4721c
client_async: Allow throwing an exception upon socket error during
shimonturjeman Jun 26, 2023
e9f5b9b
Merge branch 'master' into shimon.enable_throwing_an_exception_upon_s…
wbarnha Mar 8, 2024
176c8f1
Merge branch 'master' into shimon.enable_throwing_an_exception_upon_s…
wbarnha Mar 10, 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
11 changes: 9 additions & 2 deletions kafka/client_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,8 @@ class KafkaClient(object):
sasl mechanism handshake. Default: one of bootstrap servers
sasl_oauth_token_provider (AbstractTokenProvider): OAuthBearer token provider
instance. (See kafka.oauth.abstract). Default: None
raise_upon_socket_err_during_wakeup (bool): If set to True, raise an exception
upon socket error during wakeup(). Default: False
"""

DEFAULT_CONFIG = {
Expand Down Expand Up @@ -192,7 +194,8 @@ class KafkaClient(object):
'sasl_plain_password': None,
'sasl_kerberos_service_name': 'kafka',
'sasl_kerberos_domain_name': None,
'sasl_oauth_token_provider': None
'sasl_oauth_token_provider': None,
'raise_upon_socket_err_during_wakeup': False
}

def __init__(self, **configs):
Expand Down Expand Up @@ -243,6 +246,8 @@ def __init__(self, **configs):
check_timeout = self.config['api_version_auto_timeout_ms'] / 1000
self.config['api_version'] = self.check_version(timeout=check_timeout)

self._raise_upon_socket_err_during_wakeup = self.config['raise_upon_socket_err_during_wakeup']

def _can_bootstrap(self):
effective_failures = self._bootstrap_fails // self._num_bootstrap_hosts
backoff_factor = 2 ** effective_failures
Expand Down Expand Up @@ -936,8 +941,10 @@ def wakeup(self):
except socket.timeout:
log.warning('Timeout to send to wakeup socket!')
raise Errors.KafkaTimeoutError()
except socket.error:
except socket.error as e:
log.warning('Unable to send to wakeup socket!')
if self._raise_upon_socket_err_during_wakeup:
raise e

def _clear_wake_fd(self):
# reading from wake socket should only happen in a single thread
Expand Down