o `j@s4dZddlZddlZddlmZGdddeZdS)aCKIP-480 sticky partitioner. Records with a non-None key are hashed to a partition just like :class:`~kafka.partitioner.default.DefaultPartitioner`. Records with a None key go to a *sticky* partition - i.e. the same partition is reused for every null-key record on a topic until KafkaProducer signals that a batch has been completed (via :meth:`StickyPartitioner.on_new_batch`), at which point a different partition is picked. The goal is to give the RecordAccumulator larger, denser batches for null-key sends so per-batch overhead (CRC, compression, broker round-trip) is amortized across more records. Java's benchmark in KIP-480 reported substantial throughput/latency improvements over the default-random behavior, though kafka-python is unlikely to see similar improvements while predominantly CPU-bound on per-record overhead. N)DefaultPartitionercs:eZdZdZddZfddZddZd d d ZZS) StickyPartitioneraPartitioner that sticks null-key records to one partition per topic until ``on_new_batch`` rotates it. Thread-safety: ``_sticky`` mutations are protected by ``_lock`` so concurrent ``send()`` callers can't observe a torn read-modify-write. cCsi|_t|_dSN)_sticky threadingLock_lock)selfr V/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/partitioner/sticky.py__init__ szStickyPartitioner.__init__c s||vr td|f|durt||||||S|jC|j|}|durUt||}t | |} | rH|| vrG|WdSn ||vrU|WdS| ||WdS1sewYdS)aJChoose a partition for the next record. Arguments: topic (str): topic to partition on. key (any): Unserialized key. serialized_key (bytes or None): partitioning key. value (any): Unserialized value. serialized_value (bytes or None): serialized value. cluster (ClusterMetadata): metadata for cluster; provides all and available partitions for topic. Raises: ValueError: if topic is not in ClusterMetadata Returns: int: chosen partition ID. z%Topic %s not found in ClusterMetadataN) topics ValueErrorsuper partitionr rgetsortedpartitions_for_topiclistavailable_partitions_for_topic_pick_sticky_locked) r topickeyserialized_keyvalueserialized_valueclusterrall_partitions available __class__r r r$s&   $zStickyPartitioner.partitioncCs^|j"|j||kr WddS|j|||dWddS1s(wYdS)a'Hook called by ``KafkaProducer`` on the abort-for-new-batch retry path: rotate the sticky for ``topic`` so the next null-key record lands on a different partition. Stale events (where another thread already rotated us off ``prev_partition``) are no-ops. Navoid)r rrr)r rrprev_partitionr r r on_new_batchHs "zStickyPartitioner.on_new_batchNcs||}|s dSt|}t||pd}|rdsz9StickyPartitioner._pick_sticky_locked..)rrrrlenrandomchoicer)r rrr#rrr candidatesr r"r rVs      z%StickyPartitioner._pick_sticky_lockedr) __name__ __module__ __qualname____doc__r rr%r __classcell__r r r r rs  $r)r0r*rdefaultrrr r r r s