o `jG @sdZddlZddlZddlZddlZddlZddlmZddlm Z m Z ddl m Z m Z ddlmZddlmZmZmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd l m!Z!ddl"m#Z#ddl$m%Z%e&e'Z(Gdddeeeeee!e#e% Z)dS)z;KafkaAdminClient - high-level Kafka cluster administration.N)KafkaConfigurationErrorUnrecognizedBrokerVersion) MetricConfigMetrics)KafkaNetClient)MetadataRequestFindCoordinatorRequestCoordinatorType) __version__) ACLAdminMixin)ClusterAdminMixin)ConfigAdminMixin)GroupAdminMixin)PartitionAdminMixin)TopicAdminMixin)TransactionsAdminMixin)UserAdminMixinc@sfeZdZdZiddddedddd d d d dd ddddddddejejdfgdddddddddddd id!dd"dd#dd$dd%dd&dd'dd(ej d)dd*dd+dd,dd-d.d/dd0dd1dd2dgd3de d4Z d5d6Z d7d8Z d9d:Zd;d<Zd=d>ZdJd?d@ZejfdAdBZejfdCdDZdEdFd dGfdHdIZdS)KKafkaAdminClientao!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 bootstrap_servers localhost client_idz kafka-python-request_timeout_ms0uconnections_max_idle_msi`=reconnect_backoff_ms2reconnect_backoff_max_ms%max_in_flight_requests_per_connectionreceive_message_max_bytesireceive_buffer_bytesNsend_buffer_bytessocket_optionsretry_backoff_msdmetadata_max_age_msiclient_dns_lookupuse_all_dns_ipssecurity_protocol PLAINTEXT ssl_contextssl_check_hostnameT ssl_cafile ssl_certfile ssl_keyfile ssl_password ssl_crlfile api_versionbootstrap_timeout_msselectorsasl_mechanismsasl_plain_usernamesasl_plain_passwordsasl_kerberos_namesasl_kerberos_service_namekafkasasl_kerberos_domain_namesasl_oauth_token_provider proxy_url socks5_proxy)metric_reportersmetrics_num_samplesmetrics_sample_window_ms kafka_clientcKstd|t||j}|rtd|t|j|_|j |d|jdi}t |jd|jd|d}dd |jd D}t |||_ |jd d|j d d |j|_ |j j|_|jj|_|j|j|jdd|_d|_tddS)Nz0Starting KafkaAdminClient with configuration: %szUnrecognized configs: {}z client-idrrArB)samplestime_window_mstagscSsg|]}|qSrG).0reporterrGrGP/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/admin/client.py sz-KafkaAdminClient.__init__..r@rCadmin)metricsmetric_group_prefixr3FzKafkaAdminClient started.rG)logdebugset differenceDEFAULT_CONFIGrformatcopyconfigupdaterr_metrics_client_manager_netstart bootstrap_closed_controller_id)selfconfigs extra_configs metrics_tags metric_config reportersrGrGrJ__init__s4        zKafkaAdminClient.__init__cCs|SNrGr`rGrGrJ __enter__szKafkaAdminClient.__enter__cCs |dSrg)close)r`exc_typeexc_valexc_tbrGrGrJ__exit__s zKafkaAdminClient.__exit__cCsPt|dr|jrtddSd|_|j|j|jt ddS)z:Close the KafkaAdminClient connection to the Kafka broker.r^z KafkaAdminClient already closed.NTzKafkaAdminClient is now closed.) hasattrr^rOinforXrjrYr[stoprPrhrGrGrJrjs    zKafkaAdminClient.closecCs|p|jdS)z=Validate the timeout is set or use the configuration default.r)rV)r` timeout_msrGrGrJ_validate_timeout sz"KafkaAdminClient._validate_timeoutcs|jjdkr tdt}t|d}t|kr=|j|IdH}|j}|dkr;t d|j dIdHq|St d) z'Determine the Kafka cluster controller.)r zIKafka Admin Client controller requests requires broker version >= (0, 10)iNz#Controller ID not available, got -1r# controller)rZbroker_versionrrtime monotonicsend controller_idrOwarningr[sleepErrorsNodeNotReadyError)r`rrrequest timeout_atresponser{rGrGrJ_refresh_controller_ids     z'KafkaAdminClient._refresh_controller_idc st|}|jj}i}g}|D]}|||}|dur!|||<q||q|s+|S|jjtdkr\t|j |dd}|j |IdH} | j D]} |j | || j dd} | || j <qH|S|D] }t||j dd}|j |IdH} |j | ||dd} | ||<q^|S)a(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. N)key_typecoordinator_keys min_versionF)synthesize_node_id)keyr max_version)r build_fromrZclusterget_coordinatorappendbroker_version_datar2rvaluerz coordinatorsadd_coordinatorr) r`keysrrresultunknownrcachedrr coordinatornode_idrGrGrJ_find_coordinator_ids&sH        z&KafkaAdminClient._find_coordinator_idscs |j|g|dIdH}||S)z.Single-key wrapper for _find_coordinator_ids())rN)r)r`rridsrGrGrJ_find_coordinator_idXsz%KafkaAdminClient._find_coordinator_idcCsdS)NrGrG)rrGrGrJ]szKafkaAdminClient.rGcs|jdus |jdkr|IdH|_|jj||jdIdH}tj||vr:|IdH|_|jj||jdIdH}||D]}|tjurFq>|tjurOtd|r]||vr]|d||q>|S)zSend a Kafka protocol message to the cluster controller. Retries once on NotControllerError after refreshing the controller id. Nru)rz$Failed to find active controller id!z'Request '{}' failed with response '{}'.) r_rrZrzr~NotControllerErrorNoError RuntimeErrorrT)r`r get_errors_fn raise_errors ignore_errorsr error_typerGrGrJ_send_request_to_controller]s&    z,KafkaAdminClient._send_request_to_controller)r)__name__ __module__ __qualname____doc__r socket IPPROTO_TCP TCP_NODELAY selectorsDefaultSelectorrrSrfrirnrjrsrr GROUPrrrrGrGrGrJrs       !"#&,$ 2r)*rrUloggingrrrx kafka.errorserrorsr~rr kafka.metricsrrkafka.net.compatrkafka.protocol.metadatarrr kafka.versionr kafka.admin._aclsr kafka.admin._clusterr kafka.admin._configsr kafka.admin._groupsrkafka.admin._partitionsrkafka.admin._topicsrkafka.admin._transactionsrkafka.admin._usersr getLoggerrrOrrGrGrGrJs>