o `jt0@sdZddlmZddlZddlZddlmZddlZddlm Z ddlm Z ddl m Z mZddlmZerMixin providing topic management methods for KafkaAdminClient.r _managerdictconfigcCs"|j|jd}dd|dDS)zwRetrieve a list of all topic names in the cluster. Returns: A list of topic name strings. NcSsg|]}|dqSname.0trrQ/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/admin/_topics.py $sz/TopicAdminMixin.list_topics..topicsr run_get_cluster_metadata)selfmetadatarrr list_topicsszTopicAdminMixin.list_topicsNcCs|j|j|}|dS)a!Fetch metadata for the specified topics or all topics if None. Keyword Arguments: topics (list, optional): A list of topic names or :class:`uuid.UUID` topic ids (KIP-516). Strings and UUIDs may be mixed. Describing by id requires broker >= 2.8 (MetadataRequest v12+); name-based describe works on any broker. If None, metadata for all topics is retrieved. Returns: A list of dicts describing each topic (including partition info). rr)rrrrrrdescribe_topics&s zTopicAdminMixin.describe_topicsc sFtj}|j|jg}t|trN|D]8\}}|di}||||dd|ddfdd|diDfdd|diDd qnJt d d |Drg|D] }|||ddd qYn1t d d |Dr|D]%}|||j |j |j fdd|j Dfdd|jDd qr|std||S)Nconfigsnum_partitionsreplication_factorcg|] \}}||dqS)partition_index broker_idsrr partition_idreplicas _AssignmentrrrDz@TopicAdminMixin._process_create_topics_input.. assignmentscr")rvaluerr config_key config_value_ConfigrrrFr+)rrr!r,rcs|]}t|tVqdSN) isinstancestrrvrrr Iz?TopicAdminMixin._process_create_topics_input..)rrr!csr4r5)r6NewTopicr8rrrr:Mr;cr"r#rr&r)rrrSr+cr"r-rr/r2rrrUr+z%No valid topics found in new_topics: )rCreatableTopicCreatableReplicaAssignmentCreatableTopicConfigr6r itemsgetappendallrrr!replica_assignments topic_configs ValueError) new_topics_Topicrtopicdatar new_topicr)r*r3r_process_create_topics_input6sN            z,TopicAdminMixin._process_create_topics_inputFTcCs|r|rtd||}|r|jjdkrtd|jj||}|jjdkr;tdd|Dr;td|jjt|||d}d d }|j |j |||} |r]| d d |j D| } | d d| dD] } | dd} | ri} | D] }||tj}|| |<qy| | d<qk| S)aLCreate new topics in the cluster. Arguments: new_topics: A list of topic names, or a dict mapping each topic name to a dict of options (all keys optional):: {topic_name: {num_partitions: int (default -1), replication_factor: int (default -1), assignments: {partition_id: [broker_ids]}, configs: {key: value}}} List of NewTopic objects is deprecated. Note: for brokers < 2.4, num_partitions and replication_factor are required and must be provided via dict or [NewTopic]. Keyword Arguments: timeout_ms (numeric, optional): Milliseconds to wait for new topics to be created before the broker returns. validate_only (bool, optional): If True, don't actually create new topics. Not supported by all versions. Default: False raise_errors (bool, optional): Whether to raise errors as exceptions. Default True. wait_for_metadata (bool, optional): If True, block until each new topic is visible in broker metadata with a leader assigned for every partition. Default: False Returns: dict of CreateTopicsResponse key/vals. z:validate_only and wait_for_metadata are mutually exclusive)r zUvalidate_only requires CreateTopicsRequest >= v1, which is not supported by Kafka {}.)rNcss$|] }|jdkp |jdkVqdS)r N)rr!rrIrrrr:s"z0TopicAdminMixin.create_topics..zIBroker version {} requires explicit num_partitions and replication_factor)r timeout_ms validate_onlycs |jD] }t|jVqdSr5)rErrorsfor_code error_code)rrIrrrresponse_errors z6TopicAdminMixin.create_topics..response_errorscSsg|]}|jqSrr)rrKrrrrz1TopicAdminMixin.create_topics..throttle_time_msNrr)rF_validate_timeoutr broker_versionrformatrLanyrr_send_request_to_controllerwait_for_topicsrto_dictpop_process_configrTOPIC)rrGrQrR raise_errorswait_for_metadatarrequestrXresponseresultrIrprocessed_configsr rrrr create_topics\sH         zTopicAdminMixin.create_topics'c Cs|sdSt|}t|d}dd|D} z|j|d}Wnty:}ztd|g}WYd}~nd}~wwdd|D}i}|D]}|||} | durX| ||<qF|s]dSt|krlt d ||ft d q) aBlock until each of the given topics is ready to use. CreateTopicsResponse only confirms that the broker accepted the create request; propagating the new topics into the broker's metadata cache -- and electing a leader for every partition -- can lag behind, especially on KRaft clusters. This method polls :meth:`describe_topics` at a fixed interval until every requested topic both: - is returned with ``error_code == 0``, and - has ``error_code == 0`` and a leader assigned (``leader_id >= 0``) for every partition. Arguments: topic_names ([str]): Topic names to wait for. Keyword Arguments: timeout_ms (numeric, optional): Maximum milliseconds to wait. Default: 10000. Raises: KafkaTimeoutError: if any topic is still not ready when the deadline expires. Ng@@cSsi|]}|dqS)znot yet queriedr)rrrrr rZz3TopicAdminMixin.wait_for_topics..T)rz=describe_topics failed while waiting for topic visibility: %scSsi|]}|d|qSr)rArrrrrnszTopics not ready after %sms: %sg?) listtime monotonicr Exceptionlogdebug_topic_not_ready_reasonrArTKafkaTimeoutErrorsleep) r topic_namesrQdeadlinependingrexcby_namerreasonrrrras8    zTopicAdminMixin.wait_for_topicscCs|durdS|dd}|dkrt|jS|dpg}|s!dSg}|D].}|dd}|d}|dkrD|d|t|jfq%|d d dkrS|d |q%|r[d |SdS) z@Return a string reason if ``topic_info`` isn't ready, else None.Nzmissing from metadata responserVr partitionszno partitions reportedr$zp%s=%s leader_idr z p%s=no leader,)rArTrU__name__rBjoin) topic_inforVr~badpp_erridxrrrrus*     z'TopicAdminMixin._topic_not_ready_reasoncsl||}tjtfdd|D|d}dd}|j|j|||}|}|dd|d|d <|S) aDelete topics from the cluster. Arguments: topics ([str]): A list of topic name strings or uuid.UUID ids. Keyword Arguments: timeout_ms (numeric, optional): Milliseconds to wait for topics to be deleted before the broker returns. raise_errors (bool, optional): Whether to raise errors as exceptions. Default True. Returns: dict of DeleteTopicsResponse key/vals (version-dependent) cs,g|]}t|tjr|dn|dqS))topic_idr)r6uuidUUIDrPrHrrrs$z1TopicAdminMixin.delete_topics..)rrQcsrSr5) responsesrTrUrV)rWrirrrrXrYz6TopicAdminMixin.delete_topics..response_errorsr[Nrr)r\rDeleteTopicStater rr`rbrc)rrrQrfrhrXrirjrrr delete_topicss   zTopicAdminMixin.delete_topicsr5)NFTF)rm)NT) r __module__ __qualname____doc____annotations__rr staticmethodrLrlrarurrrrrr s   % C0 r c@seZdZdZ  dddZdS)r<a(DEPRECATED: A class for new topic creation. Arguments: name (string): name of the topic num_partitions (int): number of partitions, or -1 if replica_assignment has been specified replication_factor (int): replication factor, or -1 if replica assignment is specified replica_assignments (dict of int: [int]): A mapping containing partition id and replicas to assign to it. topic_configs (dict of str: str): A mapping of config key and value for the topic. r NcCs*||_||_||_|p i|_|pi|_dSr5)rrr!rDrE)rrrr!rDrErrr__init__s  zNewTopic.__init__)r r NN)rrrrrrrrrr<s  r<)r __future__rloggingrptypingrr kafka.errorserrorsrTrkafka.protocol.adminrr_configsrkafka.net.managerr getLoggerrrsr r<rrrrs        n