Skip to content

Commit

Permalink
Avoid FD spike after retrying KafkaAdminClient
Browse files Browse the repository at this point in the history
A caller might call kafka.KafkaAdminClient repeatedly and handle
kafka.errors.NoBrokersAvailable if the broker is not available.

However, each retry will cause 3 extra FD being used. Depends on how
long the caller wait before retry, the FD usage can reach 300~700 before
Python garbage collector collecting those FD.

This commit close those FD early.
  • Loading branch information
orange-kao committed Aug 12, 2024
1 parent 31a6b92 commit f67bc1e
Showing 1 changed file with 14 additions and 8 deletions.
22 changes: 14 additions & 8 deletions kafka/admin/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -209,14 +209,20 @@ def __init__(self, **configs):
metric_group_prefix='admin',
**self.config
)
self._client.check_version(timeout=(self.config['api_version_auto_timeout_ms'] / 1000))

# Get auto-discovered version from client if necessary
if self.config['api_version'] is None:
self.config['api_version'] = self._client.config['api_version']

self._closed = False
self._refresh_controller_id()
try:
self._client.check_version(timeout=(self.config['api_version_auto_timeout_ms'] / 1000))

# Get auto-discovered version from client if necessary
if self.config['api_version'] is None:
self.config['api_version'] = self._client.config['api_version']

self._closed = False
self._refresh_controller_id()
except Exception:
self._metrics.close()
self._client.close() # prevent FD leak
self._closed = True
raise
log.debug("KafkaAdminClient started.")

def close(self):
Expand Down

0 comments on commit f67bc1e

Please sign in to comment.