o `jh@sdZddlmZddlZddlmZddlmZddlm Z ddlm Z ddl m Z mZmZmZmZmZmZddlmZmZmZmZdd lmZmZdd lmZerXdd lmZe e!Z"Gd d d Z#GdddZ$dS)zYPartition management mixin for KafkaAdminClient. Also defines NewPartitions data class. ) annotationsN) defaultdict) TYPE_CHECKING)UnknownTopicOrPartitionError)"AlterPartitionReassignmentsRequestCreatePartitionsRequestDeleteRecordsRequestDescribeTopicPartitionsRequestElectLeadersRequest ElectionType!ListPartitionReassignmentsRequest)ListOffsetsRequestIsolationLevel OffsetSpecOffsetTimestamp)TopicPartitionOffsetAndTimestamp)Timer)KafkaConnectionManagerc@seZdZUdZded<ded<eddZd.d d Zd dZd/ddZ d/ddZ d0ddZ ddZ d1ddZ eddZd0ddZd/ddZd/dd Zd!d"Zd2d$d%Zed&d'Zed(d)Zejdfd*d+Zejdfd,d-ZdS)3PartitionAdminMixinz8Mixin providing partition and record management methods.r_managerdictconfigc stj}tjjg}|D]K\}}t|tr ||||dq t|tr;||||dfdd|dDdq |jdurKfdd|jD}nd}||||j |dq |S)N)namecountrcg|]}|dqS) broker_ids.0r _AssignmentrU/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/admin/_partitions.py 6zHPartitionAdminMixin._process_create_partitions_input.. assignments)rrr&crrrrr!rr#r$;r%) rCreatePartitionsTopicCreatePartitionsAssignmentitems isinstanceintappendrnew_assignments total_count)topic_partitions_Topictopicstopicrr-rr!r# _process_create_partitions_input)s:     z4PartitionAdminMixin._process_create_partitions_inputNFTcCs:||}t||||d}dd}|j|j|||S)aLCreate additional partitions for an existing topic. Arguments: topic_partitions: A dict of topic name strings to total partition count (int), or a dict of {topic_name: {count: int, assignments: [[broker_ids]]}} if manual assignment is desired. dict of {topic_name: NewPartitions} is deprecated. Keyword Arguments: timeout_ms (numeric, optional): Milliseconds to wait for new partitions to be created before the broker returns. validate_only (bool, optional): If True, don't actually create new partitions. Default: False raise_errors (bool, optional): Whether to raise errors as exceptions. Default True. Returns: Appropriate version of CreatePartitionsResponse class. )r1 timeout_ms validate_onlycss |jD] }t|jVqdSN)resultsErrorsfor_code error_code)rresultrrr#response_errors_s z>PartitionAdminMixin.create_partitions..response_errors)_validate_timeoutrr3rrun_send_request_to_controller)selfr/r4r5 raise_errorsrequestr=rrr#create_partitionsFs z%PartitionAdminMixin.create_partitionsc st|}tdd|D}||IdH}tt}t}|ddD]'}|ddD]}t|d|dd }||vrI||d |||q+q#||kra||} td d d d| D|S)zgszGPartitionAdminMixin._async_get_leader_for_partitions..Nr1r partitionsrpartition_index)r2 partition leader_idz*The following partitions are not known: %s, css|]}t|VqdSr6)str)r xrrr#rGxs)set_get_cluster_metadatargetraddrjoin) rArHr1metadataleader2partitionsvalid_partitionsr2rJt2punknownrrr# _async_get_leader_for_partitionsds, z4PartitionAdminMixin._async_get_leader_for_partitionscs ||}t|}|jdd}t}i}i}|r|dur(||IdH} n|t|i} g} | D]@\} } tt} | D] }| |j |q>t fdd| D|j dur^t |j n|d}|j j|| dIdH}| |q4t}| D]A}|dD]:}|dD]3}t|d |d }|d }|tjjkr|dur|js||q|||<|||d kr|||<qqqz|sntd t||}|jt|td|jpdIdH|s|rt|dkrtt |\}}t!|d|j|j"ft#dd$dd|D|S)Nretry_backoff_mscs&g|]\}}|fdd|DfqS)csg|] }|j|fqSr)rJrErecords_to_deleterr#r$szHPartitionAdminMixin._async_delete_records...rr r2partsr\rr#r$sz=PartitionAdminMixin._async_delete_records..)r1r4node_idr1rHrrIr:rz_delete_records: NotLeaderForPartitionError on %d partition(s); refreshing metadata and retryingz1Error deleting records from topic %s partition %sz.)%r>rrrOrYr)rlistr2r,rr4r+rsendto_dictrr8NotLeaderForPartitionErrorerrnoexpiredrRdiscardlogdebuglen_netsleepminmax timeout_secsnextiterr9rJBrokerResponseErrorrS)rAr]r4partition_leader_idtimer backoff_secspendingpartition2resultpartition2errorrU responsesleaderrHtopic2partitionsrJrCresponseretry_partitionsr2rFerr_codekeyrerr\r#_async_delete_records|s        $/   z)PartitionAdminMixin._async_delete_recordscC|j|j|||S)aDelete records whose offset is smaller than the given offset of the corresponding partition. Partitions whose response is :class:`~kafka.errors.NotLeaderForPartitionError` are retried with refreshed metadata, bounded by ``timeout_ms`` (or the admin client's ``request_timeout_ms`` when ``None``). When ``partition_leader_id`` is supplied no retry is attempted; the caller is asserting routing and any error is reported as-is. Arguments: records_to_delete ({TopicPartition: int}): The earliest available offsets for the given partitions. Keyword Arguments: timeout_ms (numeric, optional): Timeout in milliseconds. Also caps the total time spent retrying NotLeaderForPartitionError. partition_leader_id (node_id / int, optional): If specified, all deletion requests will be sent to this node. Returns: dict {topicPartition -> metadata} )rr?r)rAr]r4rxrrr#delete_recordssz"PartitionAdminMixin.delete_recordscCsdd||DS)NcSs&g|]}|ddd|dDfqS)rcSsg|]}|dqS)rIrr prrr#r$szLPartitionAdminMixin._get_all_topic_partitions...rHr)r r2rrr#r$s zAPartitionAdminMixin._get_all_topic_partitions..)describe_topics)rAr1rrr#_get_all_topic_partitionssz-PartitionAdminMixin._get_all_topic_partitionscCst|tr |S||Sr6)r*rr)r)rAr/rrr#_get_topic_partitionss  z)PartitionAdminMixin._get_topic_partitionscCsH||}tt||||d}dd}tjf}|j|j||||S)a"Trigger leader election for the specified topic partitions. Arguments: election_type: Type of election to attempt. 0 for Preferred, 1 for Unclean Keyword Arguments: topic_partitions (dict, list, optional): Either: dict of {topic_name: [partition ids]}. Or: list of [topic_name], and election will run on all partitions for topic. Or: None, and election runs against all topics / all partitions. Default: None timeout_ms (num, optional): Milliseconds to wait for the leader election process. raise_errors (bool, optional): Whether to raise errors as exceptions. Default True. Returns: Appropriate version of ElectLeadersResponse class. ) election_typer/r4cssD|jdkr t|jV|jD]}|jD] }t|jVqqdS)Nrc) API_VERSIONr8r9r:replica_election_resultspartition_result)r;r<rJrrr#r=s   z:PartitionAdminMixin.elect_leaders..response_errors) r>r r rr8ElectionNotNeededErrorrr?r@)rArr/r4rBrCr= ignore_errorsrrr# elect_leaderss z!PartitionAdminMixin.elect_leaderscstjj}tt}|D]2\}}|dur3t|}|s#td|ftdd|Ds3td|f||j ||j |dqfdd|DS)NzIReplica list for %s must be non-empty; use None to cancel a reassignment.css|]}t|tVqdSr6)r*r+)r itemrrr#rGszSPartitionAdminMixin._process_alter_partition_reassignments_input..z+Replica list for %s must be int broker_ids.)rIreplicascg|] \}}||dqS)rrHrr^r0rr#r$!szTPartitionAdminMixin._process_alter_partition_reassignments_input..) rReassignableTopicReassignablePartitionrrfr) ValueErrorallr2r,rJ) reassignments _PartitionrrFrrrr#,_process_alter_partition_reassignments_inputs*  z@PartitionAdminMixin._process_alter_partition_reassignments_inputc Cs||}t|||d}dd}|j|j||}i}|jD] }|jD]}t|j |j } t |j } | t jur;| nd|| <q%q |S)aAlter the replica sets for the given partitions. Arguments: reassignments (dict): A dict mapping :class:`~kafka.TopicPartition` to a list of broker IDs for the new replica set, or ``None`` to cancel a pending reassignment for that partition. Keyword Arguments: timeout_ms (numeric, optional): The time in ms to wait for the request to complete. Raises: top-level failures that prevents processing request. Does not raise partition-specific errors. Returns: dict: A dict mapping each :class:`~kafka.TopicPartition` that the broker acknowledged to the error class for that partition, or ``None`` if the reassignment was accepted. Partitions the broker did not report on are absent from the dict. r4r1cst|jVdSr6r8r9r:r;rrr#top_level_errorAzJPartitionAdminMixin.alter_partition_reassignments..top_level_errorN)r>rrrr?r@r~rHrrrIr8r9r:NoError) rArr4rCrrr7r2rJrFerrrrr#alter_partition_reassignments#s"    z1PartitionAdminMixin.alter_partition_reassignmentsc s||}|dur d}n1tjt|tr!fdd|D}ntt}|D] }||j |j q'fdd|D}t||d}dd}| ||IdH}i} |j D]} | j D]} t| jt| jt| jd| t| j| j<q[qV| S)Ncs g|] \}}|t|dqS)rpartition_indexes)rfr r2rHrrr#r$VszKPartitionAdminMixin._async_list_partition_reassignments..crrrrrrr#r$^s rcsrr6rrrrr#rhrzPPartitionAdminMixin._async_list_partition_reassignments..top_level_error)radding_replicasremoving_replicas)r>r ListPartitionReassignmentsTopicsr*rr)rrfr2r,rJr@r1rHrrrrrrI) rAr/r4 topics_fieldrrFrCrrretr2rJrrr##_async_list_partition_reassignmentsNs<      z7PartitionAdminMixin._async_list_partition_reassignmentscCs|j|j||S)aList the current ongoing partition reassignments. Arguments: topic_partitions (dict, list, optional): Either: a dict of ``{topic_name: [partition_ids]}``, or a list of :class:`~kafka.TopicPartition`, or ``None`` to list ongoing reassignments for all partitions. Default: None. Keyword Arguments: timeout_ms (numeric, optional): The time in ms to wait for the request to complete. Returns: dict: A dict mapping :class:`~kafka.TopicPartition` to a dict with keys ``'replicas'``, ``'adding_replicas'``, and ``'removing_replicas'`` (each a list of broker IDs). )rr?r)rAr/r4rrr#list_partition_reassignmentsvsz0PartitionAdminMixin.list_partition_reassignmentsc stjtj}|dur||d|dd}nd}tfdd|D||d}|j|IdH}g}|jD]} | j| j| j| j dd| j D| j d} | | q4d} |j dur`|j j|j jd} || d S) N topic_namerI)rrIcr))rr)r trrr#r$szHPartitionAdminMixin._async_describe_topic_partitions..)r1response_partition_limitcursorc Ss^g|]+}|j|j|j|jt|jt|j|jrt|jnd|jr%t|jndt|j d qS)N) r:rIrK leader_epoch replica_nodes isr_nodeseligible_leader_replicaslast_known_elroffline_replicas) r:rIrKrrfrrrrrrrrr#r$s )r:rtopic_id is_internalrHtopic_authorized_operations)r1 next_cursor)r TopicRequestCursorrrgr1r:rrrrHrr,rrrI) rAr1rr_Cursor cursor_fieldrCrr<r2 topic_dictrrrr# _async_describe_topic_partitionssB     z4PartitionAdminMixin._async_describe_topic_partitionscCr)aDescribe topics with fine-grained partition-level control (KIP-966). Unlike :meth:`describe_topics`, this uses the DescribeTopicPartitions API (apiKey 75, broker 3.7+) which supports pagination via a cursor and partition-level ELR (Eligible Leader Replicas) information. Arguments: topics ([str]): A list of topic names. Keyword Arguments: response_partition_limit (int, optional): Maximum number of partitions to include in the response. Default: 2000. cursor (dict, optional): Dict with ``'topic_name'`` and ``'partition_index'`` keys to start pagination from. Default: None. Returns: dict: ``{'topics': [...], 'next_cursor': None | {...}}``. ``topics`` is a list of dicts (one per topic) with keys ``error_code``, ``name``, ``topic_id``, ``is_internal``, ``partitions``, and ``topic_authorized_operations``. ``next_cursor`` is None if pagination is complete, otherwise a dict with the next page's ``topic_name`` and ``partition_index``. )rr?r)rAr1rrrrr#describe_topic_partitionss z-PartitionAdminMixin.describe_topic_partitionscstt|d}tjj}tt}|D]0\}}t|t t fs+t dt |dt |dkr9tt ||}||j||j|dqtd|fdd|D|dS) NrzUnsupported ts type z, expected int or OffsetSpec)rI timestampcrrr)r rr_rrr#r$szGPartitionAdminMixin._list_partition_offsets_request..) replica_idisolation_levelr1 min_version)rsr min_version_for_isolation_levelListOffsetsTopicListOffsetsPartitionrrfr)r*r+r TypeErrortypemin_version_for_timestampr2r,rJ)partition_timestampsisolation_level_intrrrrFtsrrr#_list_partition_offsets_requests(    z3PartitionAdminMixin._list_partition_offsets_requestcCsi}|jD]:}|jD]4}t|j|j}t|j}|tjur'|d||j ft |dd}t |j |j |dkr8|ndd||<q q|S)Nz$ListOffsetsRequest failed for %s: %srrr)offsetrr)r1rHrrrIr8r9r:rrdgetattrrrr)rr7r2rJrFrrrrr#(_list_partition_offsets_process_responses"        zszEPartitionAdminMixin._async_list_partition_offsets..r`zWlist_partition_offsets timed out retrying NotLeaderForPartitionError for partitions: %szglist_partition_offsets: NotLeaderForPartitionError on %d partition(s); refreshing metadata and retryingrb)r build_fromrr>rrOkeysrYr)rvaluerrgupdaterr8rirksortedrmrnrorprqrrrsrt) rAtopic_partition_specsrr4ryrzr7r/rUrrCrrrr#_async_list_partition_offsetssD   $z1PartitionAdminMixin._async_list_partition_offsetscCr)a8Look up offsets for the given partitions by spec. Partitions are routed to their respective leader brokers via cluster metadata; one ``ListOffsetsRequest`` is sent per leader. Partitions that return :class:`~kafka.errors.NotLeaderForPartitionError` are retried with refreshed metadata, bounded by ``timeout_ms`` (or the admin client's ``request_timeout_ms`` when ``None``). Arguments: topic_partition_specs: dict mapping :class:`~kafka.TopicPartition` to :class:`OffsetSpec` (or a raw integer timestamp / wire-level sentinel). Keyword Arguments: isolation_level (IsolationLevel, optional): Requires broker support for ListOffsets v2+. Default: IsolationLevel.READ_UNCOMMITTED. timeout_ms (int, optional): Maximum time to spend retrying NotLeaderForPartitionError. Default: ``request_timeout_ms``. Returns: dict: A dict mapping :class:`~kafka.TopicPartition` to :class:`~kafka.structs.OffsetAndTimestamp` Raises: KafkaError: If any partition response carries an error code. NotLeaderForPartitionError: If NotLeaderForPartitionError retries do not converge within ``timeout_ms``. UnknownTopicOrPartitionError: If a requested partition is not known to the cluster. UnsupportedVersionError: If the broker does not support a version of ListOffsetsRequest compatible with the requested specs. )rr?r)rArrr4rrr#list_partition_offsets#s! z*PartitionAdminMixin.list_partition_offsets)NFT)NNr6)NNT)rN)rd __module__ __qualname____doc____annotations__ staticmethodr3rDrYrrrrrrrrrrrrrrREAD_UNCOMMITTEDrrrrrr#r$s4     J  !   + ( 5  rc@seZdZdZdddZdS) NewPartitionsa!DEPRECATED: A class for new partition creation on existing topics. Note that the length of new_assignments, if specified, must be the difference between the new total number of partitions and the existing number of partitions. Arguments: total_count (int): the total number of partitions that should exist on the topic new_assignments ([[int]]): an array of arrays of replica assignments for new partitions. If not set, broker assigns replicas per an internal algorithm. NcCs||_||_dSr6)r.r-)rAr.r-rrr#__init__Vs zNewPartitions.__init__r6)rdrrrrrrrr#rHs r)%r __future__rlogging collectionsrtypingr kafka.errorserrorsr8rkafka.protocol.adminrrrr r r r kafka.protocol.consumerr rrr kafka.structsrr kafka.utilrkafka.net.managerr getLoggerrdrmrrrrrr#s(     $    (