o `j<@sdZddlmZddlmZddlmZddlZddlm Z ddl Z ddl m Z ddlmZddlmZmZdd lmZmZmZmZmZdd lmZdd lmZe rXdd lmZee Z!Gd ddZ"GdddeeZ#dS)z,Cluster metadata mixin for KafkaAdminClient.) annotations defaultdict)IntEnumN) TYPE_CHECKINGApiKey)ApiVersionsRequestMetadataRequest)AlterReplicaLogDirsRequestDescribeClusterRequestDescribeLogDirsRequestDescribeQuorumRequestUpdateFeaturesRequest)TopicPartitionReplica) EnumHelper)KafkaConnectionManagerc@seZdZUdZded<ddZddZdd Zd,d d Zd-ddZ e ddZ ddZ ddZ ddZddZddZddZddZd.d!d"Zd.d#d$Ze d%d&Zd/d(d)Zd/d*d+Zd S)0ClusterAdminMixinz@Mixin providing cluster management methods for KafkaAdminClient.r_managercsxtj|dur d}n fdd|D}t|dddd}|j|IdH}|}|||dD]}||q2|S)ztopics = [] for no topics, None for all. Items may be topic-name strings or :class:`uuid.UUID` topic ids (KIP-516, requires broker >= 2.8 / MetadataRequest v12+).Ncs.g|]}t|tjrd|dn|dqS)N)nametopic_id)r) isinstanceuuidUUID).0t_TopicR/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/admin/_cluster.py *s z;ClusterAdminMixin._get_cluster_metadata..FT)topicsallow_auto_topic_creation%include_cluster_authorized_operations#include_topic_authorized_operationsr!)r MetadataRequestTopicrsendto_dict_process_acl_operations)selfr!request_topicsrequestresponsemetadatatopicrrr_get_cluster_metadata"s&    z'ClusterAdminMixin._get_cluster_metadatacsz@tddd}|j|IdH}t|j}|tjur!||j|}| d| d| d| dd| ||WStj yn| gIdH}| d| dd|dD] }| d |d <q`|YSw) NT)r#include_fenced_brokers error_code error_message endpoint_typethrottle_time_msr!brokersnode_id broker_id) r rr&Errorsfor_coder1NoErrorr2r'popr(IncompatibleBrokerVersionr/)r)r+r, error_typer-brokerrrr_describe_cluster<s2           z#ClusterAdminMixin._describe_clustercCs|j|jS)zFetch cluster-wide metadata such as the list of brokers, the controller ID, and the cluster ID. Returns: A dict with cluster-wide metadata, excluding topic details. )rrunr?r)rrrdescribe_clusterWsz"ClusterAdminMixin.describe_clusterrNcsnt|d}g}|durdd|jjD}|D]}|jj||dIdH}||dd|jDdq|S)Nr!cSsg|]}|jqSrr6)rr>rrrr dsz>ClusterAdminMixin._async_describe_log_dirs..rDcSsg|]}|qSr)r')rresultrrrr gs)r>log_dirs)r rclusterr5r&appendresults)r)topic_partitionsr5r+ responsesr6r,rrr_async_describe_log_dirs`s z*ClusterAdminMixin._async_describe_log_dirscCs||}|j|j||S)acFetch broker log directory and topic/partition stats Keyword Arguments: topic_partitions (dict, list, optional): Either: dict of {topic_name: [partition ids]}. Or: list of [topic_name], to query all partitions for topic. Or: None, to query all topics / all partitions. Default: None brokers (list, optional): List of [node_id] for brokers to query. If None, query is sent to all brokers. Default: None Returns: list of dicts, containing per-broker log-dir data )_get_topic_partitionsrr@rL)r)rJr5rrrdescribe_log_dirsjs z#ClusterAdminMixin.describe_log_dirscsptjjtdd}|D]\}}t|tst|}||j||j |j qfdd|DS)NcSs tddS)NcSsttSN)rlistrrrrszVClusterAdminMixin._alter_replica_log_dirs_requests....rrrrrrQs zDClusterAdminMixin._alter_replica_log_dirs_requests..cs0i|]\}}|tfdd|DdqS)cs.g|]\}}|fdd|DdqS)csg|] \}}||dqS))r partitionsr)rr.partsrrrr s z\ClusterAdminMixin._alter_replica_log_dirs_requests....)pathr!)items)rrTr!_Dirrrrr s zQClusterAdminMixin._alter_replica_log_dirs_requests...)dirs)r rU)rr7rXrVrr s zFClusterAdminMixin._alter_replica_log_dirs_requests..) r AlterReplicaLogDirAlterReplicaLogDirTopicrrUrrr7r.rH partition)replica_assignmentsbroker_to_dirstprlog_dirrrVr _alter_replica_log_dirs_requests|s   z2ClusterAdminMixin._alter_replica_log_dirs_requestsc s~|siS||}i}|D],\}}|jj||dIdH}|jD]}|jD]}t|j|j|d} t |j || <q'q"q|S)NrD)r.r\r7) rarUrr&rIrRr topic_namepartition_indexr8r9r1) r)r]broker_requestsrEr7r+r,r.r\r_rrr_async_alter_replica_log_dirss$   z/ClusterAdminMixin._async_alter_replica_log_dirscC|j|j|S)aMMove replicas between log directories on their hosting brokers. Each entry instructs the targeted broker to move (or place) the replica for a given partition into the specified absolute log directory path. Requests are sent to each broker in parallel; a broker will only act on replicas it currently hosts. Arguments: replica_assignments: A dict mapping :class:`~kafka.TopicPartitionReplica` (``topic``, ``partition``, ``broker_id``) to the destination log directory path (absolute string). Tuples of ``(topic, partition, broker_id)`` are also accepted. Returns: dict mapping :class:`~kafka.TopicPartitionReplica` to the corresponding error class (``kafka.errors.NoError`` on success). )rr@re)r)r]rrralter_replica_log_dirssz(ClusterAdminMixin.alter_replica_log_dirsc stj}|j}t||||dgdgd}|j|IdH}t|j}|tjur0||j p.d| }| dd| dd| dd|dD]'}|d D] }t| d| d} t | tjslt | |d <qPd|d <qPqJ|S) N)rc)rbrRrCr4r1r2r!rRerror)r TopicData PartitionDatarr&r8r9r1r:r2r'r;rstr) r)r.r\r _Partitionr+r, top_errorrErirrr_async_describe_quorums,         z(ClusterAdminMixin._async_describe_quorumcCs|j|jddS)aDescribe the KRaft quorum state for the cluster metadata log. Returns quorum info for the ``__cluster_metadata`` topic (partition 0), including the current leader, leader epoch, high watermark, voters, and observers. On broker version >= 3.8 (KIP-853), the response also reports controller node endpoints in ``nodes``. Requires a KRaft cluster. Returns: dict matching the DescribeQuorumResponse shape. __cluster_metadatar)rr@rorArrrdescribe_metadata_quorums z*ClusterAdminMixin.describe_metadata_quorumcs|j|IdH}|jSrO)rget_connectionbroker_version_data)r)r7connrrr_async_get_broker_version_datasz0ClusterAdminMixin._async_get_broker_version_datacCrf)z.Return BrokerVersionData for a specific broker)rr@ru)r)r7rrrget_broker_version_datasz)ClusterAdminMixin.get_broker_version_datacCs|jjj}dd|DS)NcSsi|] \}}t||qSrr)rkvrrrrYsz2ClusterAdminMixin.api_versions..)rrs api_versionsrU)r)ryrrrrys zClusterAdminMixin.api_versionsFcst|jjd|jjddd}|r||IdH}n |j|IdH}t|j}|tjur6|d|t t }|j }|dusE|dkrGd}|j pKgD] }|j |jf||jd<qL|jp^gD]}|j|jf||jd<|||jd <q_t |S) Nclient_software_nameclient_software_version)rzr{ min_versionzApiVersionsRequest failed: r supported finalizedfinalized_epoch)r rconfig_send_request_to_controllerr&r8r9r1r:rdictfinalized_features_epochsupported_featuresr} max_versionrfinalized_featuresmin_version_levelmax_version_level)r)send_request_to_controllerr+r,r=rEepochfeaturerrr_async_describe_featuress,    z*ClusterAdminMixin._async_describe_featurescCrf)aFetch the cluster's supported and finalized feature flags. Features are broker-level capabilities (e.g. ``metadata.version``) that can be finalized cluster-wide via ``update_features`` (KIP-584). Requires broker >= 2.4. Keyword Arguments: send_request_to_controller (bool, optional): If True, route the request to the active controller. By default the request is sent to any available broker. Default: False. Returns: dict with keys: - ``supported_features``: dict of ``{feature_name: (min_version, max_version)}`` - ``finalized_features``: dict of ``{feature_name: (min_version_level, max_version_level)}`` - ``finalized_features_epoch``: int, or None if unknown (broker did not report an epoch, or reported -1) )rr@r)r)rrrrdescribe_featuressz#ClusterAdminMixin.describe_featuresc Cst|ts tdtj}g}|D]/\}}t|tr |\}}ntj}|}t |}|tj j tj j fv}| ||t|||dq|S)Nzxfeature_updates must be a dict of {feature_name: (max_version_level, upgrade_type)} or {feature_name: max_version_level})rrallow_downgrade upgrade_type)rr TypeErrorrFeatureUpdateKeyrUtupleUpdateFeatureTypeUPGRADE value_forSAFE_DOWNGRADEvalueUNSAFE_DOWNGRADErHint) feature_updates_FeatureUpdateKeyupdatesrspecrr upgrade_code downgraderrr_build_feature_updatess*     z(ClusterAdminMixin._build_feature_updates`c s|rdnd}t|||||d}|j|dddIdH}i}|jp$gD]}|jdkr2d||j<q%tt|j|j ||j<q%|D]} | | dqC|S)Nr) timeout_msr validate_onlyr}cSst|jgSrO)r8r9r1)rrrrrQ8sz:ClusterAdminMixin._async_update_features..) get_errors_fnOK) rrrrIr1rrlr8r9r2 setdefault) r)rrrr}r+r,retrErrrr_async_update_features.s(    z(ClusterAdminMixin._async_update_featurescCs|j|j|||S)aUpdate cluster-wide finalized feature flags. Finalize cluster-wide feature capabilities (e.g. ``metadata.version``). The request is always routed to the active controller. See KIP-584. Requires broker >= 2.7. Arguments: feature_updates: A dict of ``{feature_name: (upgrade_type, max_version_level)}`` or ``{feature_name: max_version_level}`` (implicit UPGRADE). ``upgrade_type`` may be a :class:`UpdateFeatureType`, its name, or int value. A ``max_version_level < 1`` requests deletion of the finalized feature. Keyword Arguments: validate_only (bool, optional): If True, validate the request but do not apply it. Default: False. timeout_ms (int, optional): Broker-side timeout in milliseconds. Default: 60000. Returns: dict of {feature_name: 'OK' | error message} )rr@r)r)rrrrrrupdate_featuresGs z!ClusterAdminMixin.update_features)rN)NN)F)Fr)__name__ __module__ __qualname____doc____annotations__r/r?rBrLrN staticmethodrarergrorqrurvryrrrrrrrrrrs.       rc@seZdZdZdZdZdZdS)rrrr|N)rrrUNKNOWNrrrrrrrrcs r)$r __future__r collectionsrenumrloggingtypingrr kafka.errorserrorsr8kafka.protocol.api_keyrkafka.protocol.metadatar r kafka.protocol.adminr r r rr kafka.structsr kafka.utilrkafka.net.managerr getLoggerrlogrrrrrrs(          G