"""KafkaAdminClient - high-level Kafka cluster administration.""" import copy import logging import selectors import socket import time import kafka.errors as Errors from kafka.errors import KafkaConfigurationError, UnrecognizedBrokerVersion from kafka.metrics import MetricConfig, Metrics from kafka.net.compat import KafkaNetClient from kafka.protocol.metadata import MetadataRequest, FindCoordinatorRequest, CoordinatorType from kafka.version import __version__ from kafka.admin._acls import ACLAdminMixin from kafka.admin._cluster import ClusterAdminMixin from kafka.admin._configs import ConfigAdminMixin from kafka.admin._groups import GroupAdminMixin from kafka.admin._partitions import PartitionAdminMixin from kafka.admin._topics import TopicAdminMixin from kafka.admin._transactions import TransactionsAdminMixin from kafka.admin._users import UserAdminMixin log = logging.getLogger(__name__) class KafkaAdminClient( ACLAdminMixin, ClusterAdminMixin, ConfigAdminMixin, GroupAdminMixin, PartitionAdminMixin, TopicAdminMixin, TransactionsAdminMixin, UserAdminMixin, ): """A class for administering the Kafka cluster. Keyword Arguments: bootstrap_servers: 'host[:port]' string (or list of 'host[:port]' strings) that the consumer should contact to bootstrap initial cluster metadata. This does not have to be the full node list. It just needs to have at least one broker that will respond to a Metadata API Request. Default port is 9092. If no servers are specified, will default to localhost:9092. client_id (str): a name for this client. This string is passed in each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to GroupCoordinator for logging with respect to consumer group administration. Default: 'kafka-python-{version}' reconnect_backoff_ms (int): The amount of time in milliseconds to wait before attempting to reconnect to a given host. Default: 50. reconnect_backoff_max_ms (int): The maximum amount of time in milliseconds to backoff/wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. Once the maximum is reached, reconnection attempts will continue periodically with this fixed rate. To avoid connection storms, a randomization factor of 0.2 will be applied to the backoff resulting in a random range between 20% below and 20% above the computed value. Default: 30000. request_timeout_ms (int): Client request timeout in milliseconds. Default: 30000. connections_max_idle_ms: Close idle connections after the number of milliseconds specified by this config. The broker closes idle connections after connections.max.idle.ms, so this avoids hitting unexpected socket disconnected errors on the client. Default: 540000 retry_backoff_ms (int): Milliseconds to backoff when retrying on errors. Default: 100. max_in_flight_requests_per_connection (int): Requests are pipelined to kafka brokers up to this number of maximum requests per broker connection. Default: 5. receive_message_max_bytes (int): Maximum allowed network frame size. Used to avoid OOM when decoding malformed network message header. Default: 100_000_000. receive_buffer_bytes (int): The size of the TCP receive buffer (SO_RCVBUF) to use when reading data. Default: None (relies on system defaults). Java client defaults to 32768. send_buffer_bytes (int): The size of the TCP send buffer (SO_SNDBUF) to use when sending data. Default: None (relies on system defaults). Java client defaults to 131072. socket_options (list): List of tuple-arguments to socket.setsockopt to apply to broker connection sockets. Default: [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)] metadata_max_age_ms (int): The period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions. Default: 300000 security_protocol (str): Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. Default: PLAINTEXT. ssl_context (ssl.SSLContext): Pre-configured SSLContext for wrapping socket connections. If provided, all other ssl_* configurations will be ignored. Default: None. ssl_check_hostname (bool): Flag to configure whether SSL handshake should verify that the certificate matches the broker's hostname. Default: True. ssl_cafile (str): Optional filename of CA file to use in certificate verification. Default: None. ssl_certfile (str): Optional filename of file in PEM format containing the client certificate, as well as any CA certificates needed to establish the certificate's authenticity. Default: None. ssl_keyfile (str): Optional filename containing the client private key. Default: None. ssl_password (str): Optional password to be used when loading the certificate chain. Default: None. ssl_crlfile (str): Optional filename containing the CRL to check for certificate expiration. By default, no CRL check is done. When providing a file, only the leaf certificate will be checked against this CRL. Default: None. api_version (tuple): Specify which Kafka API version to use. If set to None, the client will infer the broker version from the results of ApiVersionsRequest API. For brokers earlier than 0.10, which do not support the ApiVersionsRequest API, api_version is required. Note: Dynamic version checking is performed eagerly during __init__ and can raise KafkaTimeoutError if no connection can be made before timeout (see bootstrap_timeout_ms below). Different versions enable different functionality. Examples:: (4, 3) most recent broker release, enable all supported features (2, 7) support SCRAM user credential apis (0, 11) enables message format v2 (internal) (0, 10, 0) enables sasl authentication and message format v1 (0, 9) enables full group coordination features with automatic partition assignment and rebalancing, (0, 8, 2) enables kafka-storage offset commits with manual partition assignment only, (0, 8, 1) enables zookeeper-storage offset commits with manual partition assignment only, (0, 8, 0) enables basic functionality but requires manual partition assignment and offset management. Default: None bootstrap_timeout_ms (int): number of milliseconds to wait for first successful cluster bootstrap. If provided, an attempt to bootstrap will raise KafkaTimeoutError if it is unable to fetch cluster metadata before the configured timeout. Note that bootstrap is called eagerly from __init__(). Default: 30000 selector (selectors.BaseSelector): Provide a specific selector implementation to use for I/O multiplexing. Default: selectors.DefaultSelector metrics (kafka.metrics.Metrics): Optionally provide a metrics instance for capturing network IO stats. Default: None. metric_group_prefix (str): Prefix for metric names. Default: '' sasl_mechanism (str): Authentication mechanism when security_protocol is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are: PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512. sasl_plain_username (str): username for sasl PLAIN and SCRAM authentication. Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms. sasl_plain_password (str): password for sasl PLAIN and SCRAM authentication. Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms. sasl_kerberos_name (str or gssapi.Name): Constructed gssapi.Name for use with sasl mechanism handshake. If provided, sasl_kerberos_service_name and sasl_kerberos_domain name are ignored. Default: None. sasl_kerberos_service_name (str): Service name to include in GSSAPI sasl mechanism handshake. Default: 'kafka' sasl_kerberos_domain_name (str): kerberos domain name to use in GSSAPI sasl mechanism handshake. Default: one of bootstrap servers sasl_oauth_token_provider (kafka.net.sasl.oauth.AbstractTokenProvider): OAuthBearer token provider instance. Default: None proxy_url (str): URL to proxy socket connections through. Supports SOCKS5 only. Requires scheme:// (e.g., socks5://foo.bar/). Default: None kafka_client (callable): Custom class / callable for creating KafkaNetClient instances """ DEFAULT_CONFIG = { # client configs 'bootstrap_servers': 'localhost', 'client_id': 'kafka-python-' + __version__, 'request_timeout_ms': 30000, 'connections_max_idle_ms': 9 * 60 * 1000, 'reconnect_backoff_ms': 50, 'reconnect_backoff_max_ms': 30000, 'max_in_flight_requests_per_connection': 5, 'receive_message_max_bytes': 100_000_000, 'receive_buffer_bytes': None, 'send_buffer_bytes': None, 'socket_options': [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)], 'retry_backoff_ms': 100, 'metadata_max_age_ms': 300000, 'client_dns_lookup': 'use_all_dns_ips', 'security_protocol': 'PLAINTEXT', 'ssl_context': None, 'ssl_check_hostname': True, 'ssl_cafile': None, 'ssl_certfile': None, 'ssl_keyfile': None, 'ssl_password': None, 'ssl_crlfile': None, 'api_version': None, 'bootstrap_timeout_ms': 30000, 'selector': selectors.DefaultSelector, 'sasl_mechanism': None, 'sasl_plain_username': None, 'sasl_plain_password': None, 'sasl_kerberos_name': None, 'sasl_kerberos_service_name': 'kafka', 'sasl_kerberos_domain_name': None, 'sasl_oauth_token_provider': None, 'proxy_url': None, 'socks5_proxy': None, # deprecated # metrics configs 'metric_reporters': [], 'metrics_num_samples': 2, 'metrics_sample_window_ms': 30000, 'kafka_client': KafkaNetClient, } def __init__(self, **configs): log.debug("Starting KafkaAdminClient with configuration: %s", configs) extra_configs = set(configs).difference(self.DEFAULT_CONFIG) if extra_configs: raise KafkaConfigurationError("Unrecognized configs: {}".format(extra_configs)) self.config = copy.copy(self.DEFAULT_CONFIG) self.config.update(configs) # Configure metrics metrics_tags = {'client-id': self.config['client_id']} metric_config = MetricConfig(samples=self.config['metrics_num_samples'], time_window_ms=self.config['metrics_sample_window_ms'], tags=metrics_tags) reporters = [reporter() for reporter in self.config['metric_reporters']] self._metrics = Metrics(metric_config, reporters) self._client = self.config['kafka_client']( metrics=self._metrics, metric_group_prefix='admin', **self.config ) # Goal: migrate all self._client calls -> self._manager (skipping compat layer) self._manager = self._client._manager self._net = self._manager._net # Run all IO on a dedicated background thread; public admin methods # block on cross-thread Events via self._manager.run(...). self._net.start() # Bootstrap on __init__ self._manager.bootstrap(self.config['bootstrap_timeout_ms']) self._closed = False self._controller_id = None log.debug("KafkaAdminClient started.") def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def close(self): """Close the KafkaAdminClient connection to the Kafka broker.""" if not hasattr(self, '_closed') or self._closed: log.info("KafkaAdminClient already closed.") return self._closed = True self._metrics.close() self._client.close() self._net.stop() log.debug("KafkaAdminClient is now closed.") def _validate_timeout(self, timeout_ms): """Validate the timeout is set or use the configuration default.""" return timeout_ms or self.config['request_timeout_ms'] # -- Routing primitives (used by mixins) ---------------------------------- async def _refresh_controller_id(self, timeout_ms=30000): """Determine the Kafka cluster controller.""" if self._manager.broker_version < (0, 10): raise UnrecognizedBrokerVersion( "Kafka Admin Client controller requests requires broker version >= (0, 10)") request = MetadataRequest() timeout_at = time.monotonic() + timeout_ms / 1000 while time.monotonic() < timeout_at: response = await self._manager.send(request) controller_id = response.controller_id if controller_id == -1: log.warning("Controller ID not available, got -1") await self._net.sleep(1) continue return controller_id else: raise Errors.NodeNotReadyError('controller') async def _find_coordinator_ids(self, keys, key_type=CoordinatorType.GROUP): """Find broker node_ids of the coordinators for a set of keys. ``key_type`` is the CoordinatorType enum (GROUP=0, TRANSACTION=1, SHARE=2). Results are cached in the shared ``ClusterMetadata._coordinators`` map; only keys not already in the cache hit the network. On brokers supporting FindCoordinator v4+ (KIP-699, Apache Kafka 3.0+), all unknown keys are resolved in a single batched request; older brokers fall back to one request per key. Returns a dict mapping key -> node_id. """ key_type = CoordinatorType.build_from(key_type) cluster = self._manager.cluster result = {} unknown = [] for key in keys: cached = cluster.get_coordinator(key, key_type) if cached is not None: result[key] = cached else: unknown.append(key) if not unknown: return result if self._manager.broker_version_data.api_version(FindCoordinatorRequest) >= 4: request = FindCoordinatorRequest(key_type=key_type.value, coordinator_keys=unknown, min_version=4) response = await self._manager.send(request) # v4+ Coordinator rows are duck-compatible with v0-v3 response # shape: both have error_code/error_message/node_id/host/port. for coordinator in response.coordinators: node_id = cluster.add_coordinator( coordinator, key_type, coordinator.key, synthesize_node_id=False) result[coordinator.key] = node_id else: # Broker does not support batch api; fan-out request per-key for key in unknown: request = FindCoordinatorRequest(key=key, key_type=key_type.value, max_version=3) response = await self._manager.send(request) node_id = cluster.add_coordinator( response, key_type, key, synthesize_node_id=False) result[key] = node_id return result async def _find_coordinator_id(self, key, key_type=CoordinatorType.GROUP): """Single-key wrapper for _find_coordinator_ids()""" ids = await self._find_coordinator_ids([key], key_type=key_type) return ids[key] async def _send_request_to_controller(self, request, get_errors_fn=lambda r: (), raise_errors=True, ignore_errors=()): """Send a Kafka protocol message to the cluster controller. Retries once on NotControllerError after refreshing the controller id. """ if self._controller_id is None or self._controller_id == -1: self._controller_id = await self._refresh_controller_id() response = await self._manager.send(request, node_id=self._controller_id) # Refresh controller and retry on NotControllerError if Errors.NotControllerError in get_errors_fn(response): self._controller_id = await self._refresh_controller_id() response = await self._manager.send(request, node_id=self._controller_id) for error_type in get_errors_fn(response): if error_type is Errors.NoError: continue elif error_type is Errors.NotControllerError: raise RuntimeError("Failed to find active controller id!") elif raise_errors and error_type not in ignore_errors: raise error_type( "Request '{}' failed with response '{}'." .format(request, response)) return response