o `j`~@sddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl mZddlmZddlmZmZmZddlmZddlmZeeZGdd d Zdd d Zd dZddZdZ ddZ!dS)N)errors)Future)WakeupNotifier)MetadataRequestMetadataResponseCoordinatorType)TopicPartition)ensure_valid_topic_namec@s\eZdZdZddgdddZddZed d Zd d Zd dZ ddZ ddZ dRddZ ddZ ddZddZddZddZd d!Zd"d#Zd$d%Zd&d'Zd(d)Zd*d+Zd,d-Zd.d/Zd0d1Zd2d3Zejfd4d5Zd6d7Zd8d9Z d:d;Z!edd?Z#d@dAZ$dBdCZ%dDdEZ&dFdGZ'dHdIZ(dJdKZ)dLdMZ*dSdNdOZ+dPdQZ,dS)TClusterMetadataa* A class to manage kafka cluster metadata. Keyword Arguments: retry_backoff_ms (int): Milliseconds to backoff when retrying on errors. Default: 100. 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 bootstrap_servers: 'host[:port]' string (or list of 'host[:port]' strings) that the client 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. allow_auto_create_topics (bool): Enable/disable auto topic creation on metadata request. Only available with api_version >= (0, 11). Default: True diTuse_all_dns_ips)retry_backoff_msmetadata_max_age_msbootstrap_serversallow_auto_create_topicsclient_dns_lookupcKsd|_t|_i|_i|_tt|_i|_i|_ i|_ d|_ d|_ d|_ d|_t|_t|_d|_t|_t|_d|_d|_d|_d|_d|_d|_t|j|_|jD] }||vre|||j|<qX||_ i|_!dS)NrTF)"_managerset_topics_brokers _partitions collections defaultdict_broker_partitions _topic_ids_topic_names_by_id _coordinators_last_refresh_ms_last_successful_refresh_ms _need_update_future _listeners threadingLock_lockneed_all_topic_metadataunauthorized_topicsinternal_topics controller cluster_id_refresh_loop_future_refresh_future_wakeupclosedcopyDEFAULT_CONFIGconfig_generate_bootstrap_brokers_bootstrap_brokers_coordinator_brokers)selfconfigskeyr7K/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/cluster.py__init__3s<     zClusterMetadata.__init__cCs|jduo |jj S)z True if a refresh is mid-flight.N)r+is_doner4r7r7r8metadata_refresh_in_progressUsz,ClusterMetadata.metadata_refresh_in_progresscCst||_t|jj|_dS)ajWire this cluster to its connection manager. Construction is split from attach so ClusterMetadata can be built standalone (tests, snapshots) without a live manager. The reference is held via weakref.proxy so that manager <-> cluster does not form a GC cycle; manager.close() still calls cluster.close() to clear eagerly. N)weakrefproxyrr_netr,)r4managerr7r7r8attachZs zClusterMetadata.attachcCsd|_|jdS)NT)r-r,notifyr;r7r7r8closeeszClusterMetadata.closecCs4|jdur td|jdurdS|j|j|_dS)zOSpawn the periodic refresh coroutine. Idempotent. Triggers bootstrap if needed.N*start_refresh_loop requires prior attach())r RuntimeErrorr* call_soon _refresh_loopr;r7r7r8start_refresh_loopis  z"ClusterMetadata.start_refresh_loopc s|jdur td|jjstd|jIdHtd|js|jr,|j IdH| }|dkrPz | IdHWnt j yNtjdddYnwq ztd |d ||d IdHWnty|}z td |WYd}~nd}~ww|jr#td dS) zLAwaits ttl() then triggers refresh_metadata(); request_update() wakes early.NrDz(Metadata refresh loop needs bootstrap...zStarting metadata refresh looprzMetadata refresh failedT)exc_infoz%Sleeping %s for next Metadata refreshzMetadata refresh loop error: %szStopping metadata refresh loop)rrE bootstrappedlogdebugbootstrap_asyncinfor-r<r+ttlrefresh_metadataErrors KafkaErrorr, Exceptionerror)r4ttl_msexcr7r7r8rGqs6    zClusterMetadata._refresh_loopNc s|jdur td|jrtd|jIdHdSt|_z ||IdHWnty;}z|j |d}~ww|j ddS)aSend one MetadataRequest and apply the response. Concurrent callers share a single in-flight request: if a refresh is already underway, additional callers await the same Future and see the same outcome (success or exception). This avoids duplicate broker requests when bootstrap and the refresh loop race, or when external callers invoke refresh while the loop is mid-flight. Nz(refresh_metadata requires prior attach()z5Metadata refresh already in flight; awaiting existing) rrEr<rLrMr+r_do_refresh_metadatarTfailuresuccess)r4node_idrWr7r7r8rQs     z ClusterMetadata.refresh_metadatac s|dur |jn|}|dur td|jdtd|jdz|}t d|||j ||IdH}Wnt yU}z t d|| |d}~wwt d|d||dS)Nz6No node available for metadata refresh - backoff/retrymetadataz&Sending metadata request %s to node %szMetadata refresh: failed %sz#Metadata refresh: success (node_id=))rleast_loaded_noderLwarningupdate_backoffrRNodeNotReadyError reset_backoffmetadata_requestrMsendrTrU failed_updateupdate_metadata)r4r[requestresponserWr7r7r8rXs&      z$ClusterMetadata._do_refresh_metadatacCs`t|jd}|jddkrt|}i}t|D]\}\}}}d|}t|||d||<q|S)Nrr(resolve_canonical_bootstrap_servers_onlyz bootstrap-%s) collect_hostsr0#expand_to_canonical_bootstrap_hosts enumeraterMetadataResponseBroker)r4bootstrap_hostsbrokersihostport_r[r7r7r8r1sz+ClusterMetadata._generate_bootstrap_brokerscCs ||jvSN)r2)r4r[r7r7r8 is_bootstraps zClusterMetadata.is_bootstrapcCs@|D]}t|qt||jst|St||_|S)zSet specific topics to track for metadata. Arguments: topics (list of str): topics to check for metadata Returns: Future: resolves after metadata request/response )r r differencerrrZrequest_update)r4topicstopicr7r7r8 set_topicss    zClusterMetadata.set_topicscCs2t|||jvrt|S|j||S)apAdd a topic to the list of topics tracked via metadata. Arguments: topic (str): topic to track Returns: Future: resolves after metadata request/response Raises: TypeError: if topic is not a string ValueError: if topic is invalid: must be chars (a-zA-Z0-9._-), and less than 250 length )r rrrZaddrwr4ryr7r7r8 add_topics    zClusterMetadata.add_topiccCt|jS)ziGet all MetadataResponseBroker Returns: list: [MetadataResponseBroker, ...] )listrvaluesr;r7r7r8roszClusterMetadata.brokerscCr~)zGet bootstrap brokers only, extracted from the bootstrap_servers config option. Node ids are synthesized as 'bootstrap-0' etc. Returns: list: [MetadataResponseBroker, ...] )rr2rr;r7r7r8bootstrap_brokerssz!ClusterMetadata.bootstrap_brokerscCs$|j|p|j|p|j|S)zGet MetadataResponseBroker Arguments: broker_id (int or str): node_id for a broker to check Returns: MetadataResponseBroker or None if not found )rgetr2r3r4 broker_idr7r7r8broker_metadatas  zClusterMetadata.broker_metadatacCs ||jvrdSt|j|S)zReturn set of all partitions for topic (whether available or not) Arguments: topic (str): topic to check for partitions Returns: set: {partition (int), ...} None if topic not found. N)rrkeysr|r7r7r8partitions_for_topic s z$ClusterMetadata.partitions_for_topiccCs*||jvrdStdd|j|DS)zReturn set of partitions with known leaders Arguments: topic (str): topic to check for partitions Returns: set: {partition (int), ...} None if topic not found. NcSsg|] \}}|jdkr|qS)) leader_id).0 partitionr\r7r7r8 &s zBClusterMetadata.available_partitions_for_topic..)rritemsr|r7r7r8available_partitions_for_topics  z.ClusterMetadata.available_partitions_for_topiccC:|j|jvrdS|j|j|jvrdS|j|j|jjS)z:Return node_id of leader, -1 unavailable, None if unknown.N)ryrrrr4rr7r7r8leader_for_partition* z$ClusterMetadata.leader_for_partitioncCsX||}|dur dS|j|jvrdS|j|j|j}|dur#dS||jvr*dS|S)aReturn MetadataResponseBroker for ``node_id`` only when it is known AND still listed as a replica of ``partition`` (KIP-392). Used by the consumer's preferred-read-replica routing to avoid sending fetches to a broker that has been demoted out of the partition's replica set even though it still exists as a node. Arguments: partition (TopicPartition): topic / partition to look up. node_id (int): broker id to validate. Returns: MetadataResponseBroker if the node exists in cluster metadata and is currently listed as a replica of ``partition``; otherwise None. N)rryrrr replica_nodes)r4rr[brokerpartition_datar7r7r8is_replica_node2s   zClusterMetadata.is_replica_nodecCr)zIReturn leader_epoch for partition, or None if topic/partition is unknown.N)ryrr leader_epochrr7r7r8leader_epoch_for_partitionOrz*ClusterMetadata.leader_epoch_for_partitioncCs|j[|j|ji|j}|dur WddS|dks&||jkr/ WddS|j}||_||_||krU||jvrI|j|||dkrU|j| | WddS1sawYdS)aApply a KIP-951 current-leader hint from a Fetch/Produce response. The cached leader id and epoch for ``partition`` are replaced only when ``leader_epoch`` is strictly newer than the cached value (and non-negative). When the leader id moves, ``_broker_partitions`` is rewired so leader-based routing follows immediately. Arguments: partition (TopicPartition): topic / partition the hint is about. leader_id (int): broker id named as the new leader. leader_epoch (int): epoch of that new leader. Returns: bool: True iff cached state was changed. NFrrT) r$rrryrrrrdiscardr{)r4rrrp_data old_leaderr7r7r8update_partition_leaderWs$ $z'ClusterMetadata.update_partition_leadercC |j|S)aReturn TopicPartitions for which the broker is a leader. Arguments: broker_id (int or str): node id for a broker Returns: set: {TopicPartition, ...} None if the broker either has no partitions or does not exist. )rrrr7r7r8partitions_for_brokerws z%ClusterMetadata.partitions_for_brokercCs|j||fS)aXReturn node_id of group coordinator from cache. Arguments: key (str): name of consumer group or transaction_id key_type (CoordinatorType, optional): Default GROUP Returns: node_id (int or str) for coordinator, -1 if coordinator unknown None if the group does not exist. )rr)r4r6key_typer7r7r8get_coordinators zClusterMetadata.get_coordinatorcCsztd}|jdur|jdr|jddS|jrd}n ||j}|jd|}||j}|jd|}t||dS)z/Milliseconds until metadata should be refreshedrJNr\rrr ) time monotonicrconnection_delayrrr0rmax)r4nowrP metadata_age retry_age next_retryr7r7r8rPs    zClusterMetadata.ttlcCs |jdS)zDReturn milliseconds to wait before attempting to retry after failurer )r0r;r7r7r8refresh_backoffs zClusterMetadata.refresh_backoffcCsv|j.d|_|jr|jjrt|_|j}|jr)||jWd|SWd|S1s4wY|S)aFlags metadata for update, return Future() Actual update must be handled separately. This method will only change the reported ttl() Returns: kafka.future.Future (value will be the cluster object after update) TN) r$rr r:rrrHr,rB)r4retr7r7r8rws    zClusterMetadata.request_updatecCs|jSrt)rr;r7r7r8 need_updateszClusterMetadata.need_updatecCs t|j}|r||jS|S)aGet set of known topics. Arguments: exclude_internal_topics (bool): Whether records from internal topics (such as offsets) should be exposed to the consumer. If set to True the only way to receive records from an internal topic is subscribing to it. Default True Returns: set: {topic (str), ...} )rrrr')r4exclude_internal_topicsrxr7r7r8rxs  zClusterMetadata.topicscCsB|jrtj}n|jstj}ndd|jD}t||jddddS)NcSsg|]}tj|dqS))name)rMetadataRequestTopic)rryr7r7r8rsz4ClusterMetadata.metadata_request..rF)rxallow_auto_topic_creation%include_cluster_authorized_operations#include_topic_authorized_operations)r%r ALL_TOPICSr NO_TOPICSr0)r4rxr7r7r8rcsz ClusterMetadata.metadata_requestcCr)zReturn the topic UUID for ``topic_name``, or None if unknown. Populated from MetadataResponse v10+ (Kafka 2.8+, KIP-516). Older responses leave this empty. )rr)r4 topic_namer7r7r8topic_id zClusterMetadata.topic_idcCr)zReturn the topic name for ``topic_id`` (uuid.UUID), or None. Reverse lookup of :meth:`topic_id`. Populated from MetadataResponse v10+ (KIP-516). )rr)r4rr7r7r8topic_name_for_idrz!ClusterMetadata.topic_name_for_idcCs`d}|j|jr|j}d|_Wdn1swYtd|_|r.||dSdS)z4Update cluster state given a failed MetadataRequest.NrJ)r$r rrrrY)r4 exceptionfr7r7r8reszClusterMetadata.failed_updatec Cs|jstd|t|Si}|jD]"}|jdkr$|\}}}d}n|\}}}}||t ||||iq|jdkr@d}n| |j }|jdkrNd} n|j } i} t t} t} t} i}i}t}|jdk}|jD]}|j}|jrz| |t|j}|j}d}|dur|dur|j |}|dur||krtd|||d}|||<|||<|tjuri| |<|jD]"}|j}|r|rd |_|| ||<|jd kr| |jt||qqm||jvr9t|tj r|||tj!urtd |qm|tj"urt#d |qm|tj$urt#d || |qm|tj%ur2t#d |||jvr1|j&|qmt#d||qm|j'<||_(||_)| |_ | |_*| |_+| |_,| |_-|jdkr_||_||_.t/|dk|_0d}|j1ro|j1}d|_1Wdn 1s}wYt23d}||_4||_5|rt/|jdkr|jddtjj6kr|jddd\}}t||}|7|n|8|t9d||j:D]}||qdS)zUpdate cluster state given a MetadataResponse. Arguments: metadata (MetadataResponse): broker response to a metadata request Returns: None z9No broker metadata found in MetadataResponse -- ignoring.rN Fz\Topic %s topic_id changed from %s to %s -- likely recreated; resetting cached leader epochs.Trz;Topic %s is not available during auto-create initializationz&Topic %s not found in cluster metadataz*Topic %s is not authorized for this clientz'%s' is not a valid topic namez(Error fetching metadata for topic %s: %s rJzUpdated metadata: %s);rorLr_rerRMetadataEmptyBrokerList API_VERSIONupdaterrmr controller_idr)rrrrxr is_internalr{for_code error_coderrNoError partitionspartition_indexrrrr issubclassRetriableErrorLeaderNotAvailableErrorUnknownTopicOrPartitionErrorrUTopicAuthorizationFailedErrorInvalidTopicErrorremover$rr(rrr&r'rlenrr rrrrerrnorYrZrOr!)r4r\ _new_brokersrr[rqrrrack_new_controller_new_cluster_id_new_partitions_new_broker_partitions_new_unauthorized_topics_new_internal_topics_new_topic_ids_new_topic_names_by_id _retry_topicsepoch_reliabletry error_type new_topic_id recreatedpriorrrrrrrUlistenerr7r7r8rfs                               (     zClusterMetadata.update_metadatacCs|j|dS)zs8       ,