o
    `jX                     @   s   d dl mZmZ d dlmZ d dlmZ d dlmZ d dl	Z	d dl
Z
d dlZd dlZd dlZd dlmZ d dlmZ d dlmZ d dlmZmZ e	eZG d	d
 d
eZG dd dZG dd dZG dd deZG dd deZdS )    )ABCabstractmethod)OrderedDict)Sequence)IntEnumN)OffsetResetStrategy)OffsetAndMetadata)ensure_valid_topic_namesynchronizedc                   @   s   e Zd ZdZdZdZdZdS )SubscriptionTyper            N)__name__
__module____qualname__NONEAUTO_TOPICSAUTO_PATTERNUSER_ASSIGNED r   r   _/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/consumer/subscription_state.pyr      s
    r   c                   @   s"  e Zd ZdZdZd_ddZdd Zed`d
dZedd Z	edd Z
edd Zedd Zedd ZdaddZedd Zedd Zedd Zedd  Zed!d" Zed#d$ Zed%d& Zed'd( Zedbd)d*Zed+d, Zed-d. Zed/d0 Zed1d2 Zed3d4 Zed5d6 Zed7d8 Zed9d: Zed;d< Z ed=d> Z!ed?d@ Z"edAdB Z#edCdD Z$edEdF Z%edGdH Z&edbdIdJZ'edKdL Z(edMdN Z)edOdP Z*edQdR Z+edSdT Z,edUdV Z-edWdX Z.edYdZ Z/ed[d\ Z0ed]d^ Z1d	S )cSubscriptionStatea  
    A class for tracking the topics, partitions, and offsets for the consumer.
    A partition is "assigned" either directly with assign_from_user() (manual
    assignment) or with assign_from_subscribed() (automatic assignment from
    subscription).

    Once assigned, the partition is not considered "fetchable" until its initial
    position has been set with seek(). Fetchable partitions track a fetch
    position which is used to set the offset of the next fetch, and a consumed
    position which is the last offset that has been returned to the user. You
    can suspend fetching from a partition through pause() without affecting the
    fetched/consumed offsets. The partition will remain unfetchable until the
    resume() is used. You can also query the pause state independently with
    is_paused().

    Note that pause state as well as fetch/consumed positions are not preserved
    when partition assignment is changed whether directly by the user or
    through a group rebalance.
    zYou must choose only one way to configure your consumer: (1) subscribe to specific topics by name, (2) subscribe to topics matching a regex pattern, (3) assign itself specific topic-partitions.earliestc                 C   s   z	t t| }W n ty   td tj}Y nw || _d| _t	j| _
d| _t | _t | _t | _d| _g | _t | _dS )a  Initialize a SubscriptionState instance

        Keyword Arguments:
            offset_reset_strategy: 'earliest' or 'latest', otherwise
                exception will be raised when fetching an offset that is no
                longer available. Default: 'earliest'
        z.Unrecognized offset_reset_strategy, using NONEN)getattrr   upperAttributeErrorlogwarningr   _default_offset_reset_strategysubscriptionr   subscription_typesubscribed_patternset_group_subscription_user_assignmentr   
assignmentrebalance_listener	listeners	threadingRLock_lock)selfoffset_reset_strategyr   r   r   __init__4   s$   

zSubscriptionState.__init__c                 C   sB   t |ts	td| jtjkr|| _d S | j|krt| jd S )NzSubscriptionType enum required)
isinstancer   
ValueErrorr!   r   ErrorsIllegalStateError_SUBSCRIPTION_EXCEPTION_MESSAGE)r,   r!   r   r   r   _set_subscription_typeN   s   


z(SubscriptionState._set_subscription_typer   Nc                 C   s   |s|st d|r|rt| j|r+| tj td| t	 | _
t|| _nt|ts5t|ts9td| tj | | |rQt|ttfsQtd|| _dS )a"  Subscribe to a list of topics, or a topic regex pattern.

        Partitions will be dynamically assigned via a group coordinator.
        Topic subscriptions are not incremental: this list will replace the
        current assignment (if there is one).

        This method is incompatible with assign_from_user()

        Arguments:
            topics (list): List of topics for subscription.
            pattern (str): Pattern to match available topics. You must provide
                either topics or pattern, but not both.
            listener (ConsumerRebalanceListener): Optionally include listener
                callback, which will be called before and after each rebalance
                operation.

                As part of group management, the consumer will keep track of the
                list of consumers that belong to a particular group and will
                trigger a rebalance operation if one of the following events
                trigger:

                * Number of partitions change for any of the subscribed topics
                * Topic is created or deleted
                * An existing member of the consumer group dies
                * A new member is added to the consumer group

                When any of these events are triggered, the provided listener
                will be invoked first to indicate that the consumer's assignment
                has been revoked, and then again when the new assignment has
                been received. Note that this listener will immediately override
                any listener set in a previous call to subscribe. It is
                guaranteed, however, that the partitions revoked/assigned
                through this interface are from topics subscribed in this call.

        Raises:
            ValueError: if neither topics nor pattern provided.
            IllegalStateError: if both topics and pattern provided.
            TypeError: if topics is not a list/sequence, or listener is not
                a AsyncConsumerRebalanceListener or ConsumerRebalanceListener.
        zMust provide topics or patternzSubscribing to pattern: /%s/z+Topics must be a list (or non-str sequence)zNlistener must be a ConsumerRebalanceListener or AsyncConsumerRebalanceListenerN)r0   r1   r2   r3   r4   r   r   r   infor#   r    recompiler"   r/   strr   	TypeErrorr   change_subscriptionConsumerRebalanceListenerAsyncConsumerRebalanceListenerr'   )r,   topicspatternlistenerr   r   r   	subscribeV   s(   *

zSubscriptionState.subscribec                 C   sz   |   s
t| jt|tr|g}| jt|kr!t	d| dS |D ]}t
| q#td| t|| _| j| dS )a  Change the topic subscription.

        Arguments:
            topics (list of str): topics for subscription

        Raises:
            IllegalStateError: if assign_from_user has been used already
            TypeError: if a topic is None or a non-str
            ValueError: if a topic is an empty string or
                        - a topic name is '.' or '..' or
                        - a topic name does not consist of ASCII-characters/'-'/'_'/'.'
        z1subscription unchanged by change_subscription(%s)Nz!Updating subscribed topics to: %s)partitions_auto_assignedr1   r2   r3   r/   r8   r    r#   r   r   r	   r5   r$   update)r,   r=   tr   r   r   r:      s   


z%SubscriptionState.change_subscriptionc                 C   s$   |   s
t| j| j| dS )a0  Add topics to the current group subscription.

        This is used by the group leader to ensure that it receives metadata
        updates for all topics that any member of the group is subscribed to.

        Arguments:
            topics (list of str): topics to add to the group subscription
        N)rA   r1   r2   r3   r$   rB   )r,   r=   r   r   r   group_subscribe   s   
z!SubscriptionState.group_subscribec                 C   s:   |   s
t| j| jdu rtd| j| j dS )zRReset the group's subscription to only contain topics subscribed by this consumer.NzSubscription required)rA   r1   r2   r3   r    r$   intersection_updater,   r   r   r   reset_group_subscription   s
   

z*SubscriptionState.reset_group_subscriptionc                 C   s6   |  tj | jt|krdS t|| _| | dS )ab  Manually assign a list of TopicPartitions to this consumer.

        The new assignment replaces the previous one (this is not an
        incremental-add API), but ``TopicPartitionState`` is preserved
        for any partition that's present in both the old and new
        assignment. Manual topic assignment through this method does
        not use the consumer's group management functionality. As
        such, there will be no rebalance operation triggered when
        group membership or cluster and topic metadata change. Note
        that it is not possible to use both manual partition
        assignment with assign() and group assignment with subscribe().

        Arguments:
            partitions (list of TopicPartition): assignment for this instance.

        Raises:
            IllegalStateError: if consumer has already called subscribe()
        N)r4   r   r   r%   r#   _apply_assignment)r,   
partitionsr   r   r   assign_from_user   s
   
z"SubscriptionState.assign_from_userc                 C   sV   |   s
t| j|D ]}|j| jvrtd|f q| j|dd t	d| dS )a  Update the assignment to the specified partitions.

        This method is called by the coordinator to dynamically assign
        partitions based on the consumer's topic subscription. Differs
        from :meth:`assign_from_user` which directly sets the assignment
        from a user-supplied TopicPartition list.

        Preserves ``TopicPartitionState`` (position, paused flag,
        preferred read replica, fetch buffers tied to the partition)
        for any partition present in both the prior and new assignments.

        Validation raises ``ValueError`` BEFORE any mutation if a
        partition's topic isn't subscribed.

        Arguments:
            assignments (list of TopicPartition): the *full* new
                assignment (not a diff). Partitions present in both
                the old and new assignment retain their state;
                revoked partitions are dropped; new partitions get
                fresh state.
        z/Assigned partition %s for non-subscribed topic.T)	randomizez Updated partition assignment: %sN)
rA   r1   r2   r3   topicr    r0   rH   r   r5   )r,   assignmentstpr   r   r   assign_from_subscribed   s   z(SubscriptionState.assign_from_subscribedFc                    sx   t |}t j D ]}||vr j|= q j| d q fdd|D }|r/t| |D ]}t  j|< q1dS )a  Mutate ``self.assignment`` in place to contain exactly
        ``partitions``, preserving the existing ``TopicPartitionState``
        for any partition present in both the old and new sets.

        Shared between :meth:`assign_from_user` and
        :meth:`assign_from_subscribed` - the algorithm is identical;
        only the validation around it differs.

        When ``randomize=True``, the *newly-added* partitions are
        inserted in random order. Kept partitions retain their
        existing position regardless. This is intended for the
        coordinator-driven path (assign_from_subscribed) - see the
        comment at that call site for rationale.
        Fc                    s   g | ]	}| j vr|qS r   r&   ).0rN   rF   r   r   
<listcomp>+  s    z7SubscriptionState._apply_assignment.<locals>.<listcomp>N)r#   listr&   keysmark_pending_revocationrandomshuffleTopicPartitionState)r,   rI   rK   new_setrN   new_partitionsr   rF   r   rH     s   

z#SubscriptionState._apply_assignmentc                 C   s,   d| _ | j  | j  d| _tj| _dS )z7Clear all topic subscriptions and partition assignmentsN)r    r%   clearr&   r"   r   r   r!   rF   r   r   r   unsubscribe1  s
   

zSubscriptionState.unsubscribec                 C      | j S )a  Get the topic subscription for the group.

        For the leader, this will include the union of all member subscriptions.
        For followers, it is the member's subscription only.

        This is used when querying topic metadata to detect metadata changes
        that would require rebalancing (the leader fetches metadata for all
        topics in the group so that it can do partition assignment).

        Returns:
            set: topics
        )r$   rF   r   r   r   group_subscription:  s   z$SubscriptionState.group_subscriptionc                 C   s*   t |ttfstd| j| | dS )a?  Manually specify the fetch offset for a TopicPartition.

        Overrides the fetch offsets that the consumer will use on the next
        poll(). If this API is invoked for the same partition more than once,
        the latest offset will be used on the next poll(). Note that you may
        lose data if this API is arbitrarily used in the middle of consumption,
        to reset the fetch offsets.

        Arguments:
            partition (TopicPartition): partition for seek operation
            offset (int or OffsetAndMetadata): message offset in partition
        z+offset must be type in or OffsetAndMetadataN)r/   intr   r9   r&   seek)r,   	partitionoffsetr   r   r   r`   J  s   zSubscriptionState.seekc                 C   s   t | j S )z4Return set of TopicPartitions in current assignment.)r#   r&   rT   rF   r   r   r   assigned_partitions\  s   z%SubscriptionState.assigned_partitionsc                    s   t  fdd jD S )z-Return current set of paused TopicPartitions.c                 3   s    | ]
}  |r|V  qd S N)	is_paused)rQ   ra   rF   r   r   	<genexpr>d  s    z6SubscriptionState.paused_partitions.<locals>.<genexpr>)r#   r&   rF   r   rF   r   paused_partitionsa  s   z#SubscriptionState.paused_partitionsc                 C   0   t  }| j D ]\}}| r|| q|S )z>Return ordered list of TopicPartitions that should be Fetched.)rS   r&   itemsis_fetchableappend)r,   	fetchablera   stater   r   r   fetchable_partitionsg  s   
z&SubscriptionState.fetchable_partitionsc                 C   s   | j tjtjfv S )z5Return True unless user supplied partitions manually.)r!   r   r   r   rF   r   r   r   rA   p  s   z*SubscriptionState.partitions_auto_assignedc                 C   s,   i }| j  D ]\}}|jr|j||< q|S )z?Returns consumed offsets as {TopicPartition: OffsetAndMetadata})r&   ri   has_valid_positionposition)r,   all_consumedra   rm   r   r   r   all_consumed_offsetsu  s   
z&SubscriptionState.all_consumed_offsetsc                 C   s"   |du r| j }| j| | dS )zMark partition for offset reset using specified or default strategy.

        Arguments:
            partition (TopicPartition): partition to mark
            offset_reset_strategy (OffsetResetStrategy, optional)
        N)r   r&   reset)r,   ra   r-   r   r   r   request_offset_reset~  s   z&SubscriptionState.request_offset_resetc                 C      |D ]
}| j | | qd S rd   )r&   set_reset_pending)r,   rI   next_allowed_reset_timera   r   r   r   rv        z#SubscriptionState.set_reset_pendingc                 C   s   | j tjkS )z@Return True if default offset reset policy is Earliest or Latest)r   r   r   rF   r   r   r   has_default_offset_reset_policy  s   z1SubscriptionState.has_default_offset_reset_policyc                 C      | j | jS rd   )r&   awaiting_resetr,   ra   r   r   r   is_offset_reset_needed     z(SubscriptionState.is_offset_reset_neededc                 C   s    | j  D ]}|js dS qdS )NFT)r&   valuesro   )r,   rm   r   r   r   has_all_fetch_positions  s
   z)SubscriptionState.has_all_fetch_positionsc                 C   rh   rd   )r#   r&   ri   is_missing_positionadd)r,   missingra   rm   r   r   r   missing_fetch_positions  s   
z)SubscriptionState.missing_fetch_positionsc                 C      || j v o
| j | jS rd   )r&   ro   r|   r   r   r   ro        z$SubscriptionState.has_valid_positionc                 C   sX   t  }| j D ]\}}| r"| jtjkr|| q|| j q|r*t	
|d S rd   )r#   r&   ri   r   r   r   r   r   rs   r1   NoOffsetForPartitionError)r,   partitions_with_no_offsetsrN   rm   r   r   r   reset_missing_positions  s   
z)SubscriptionState.reset_missing_positionsc                 C   6   t  }| j D ]\}}|jr| r|| q|S rd   )r#   r&   ri   r{   is_reset_allowedr   r,   rI   rN   rm   r   r   r   partitions_needing_reset     
z*SubscriptionState.partitions_needing_resetc                 C   $   dd | j  D }|rt|S d S )Nc                 S   "   g | ]}|j r|jd ur|jqS rd   )r{   next_allowed_retry_timerQ   rm   r   r   r   rR         
zBSubscriptionState.next_offset_reset_retry_time.<locals>.<listcomp>r&   r   minr,   timesr   r   r   next_offset_reset_retry_time     z.SubscriptionState.next_offset_reset_retry_timec                 C   s   || j vrdS | j | |S NF)r&   maybe_validate_position)r,   ra   current_leader_epochr   r   r   *maybe_validate_position_for_current_leader  s   
z<SubscriptionState.maybe_validate_position_for_current_leaderc                 C   s   || j vrdS | j |  S r   )r&   request_position_validationr|   r   r   r   r     s   
z-SubscriptionState.request_position_validationc                 C   r   rd   )r#   r&   ri   awaiting_validationis_validation_allowedr   r   r   r   r   partitions_needing_validation  r   z/SubscriptionState.partitions_needing_validationc                 C   r   )Nc                 S   r   rd   )r   r   r   r   r   r   rR     r   zGSubscriptionState.next_offset_validation_retry_time.<locals>.<listcomp>r   r   r   r   r   !next_offset_validation_retry_time  r   z3SubscriptionState.next_offset_validation_retry_timec                 C   ru   rd   )r&   set_validation_pendingr,   rI   r   ra   r   r   r   r     rx   z(SubscriptionState.set_validation_pendingc                 C   ru   rd   )r&   validation_failedr   r   r   r   r     rx   z#SubscriptionState.validation_failedc                 C   s"   || j v r| j | | d S d S rd   )r&   complete_validation)r,   ra   validated_positionr   r   r   r     s   
z%SubscriptionState.complete_validationc                 C   r   rd   )r&   r   r|   r   r   r   is_offset_validation_needed  r   z-SubscriptionState.is_offset_validation_neededc                 C   s
   || j v S rd   rP   r|   r   r   r   is_assigned     
zSubscriptionState.is_assignedc                 C   r   rd   )r&   pausedr|   r   r   r   re     r   zSubscriptionState.is_pausedc                 C   s   || j v o| j |  S rd   )r&   rj   r|   r   r   r   rj     s   zSubscriptionState.is_fetchablec                 C      | j |   d S rd   )r&   pauser|   r   r   r   r        zSubscriptionState.pausec                 C   r   rd   )r&   resumer|   r   r   r   r     r   zSubscriptionState.resumec                 C   s&   |D ]}|| j v r| j |   qdS )a  KIP-429: gate ``is_fetchable()`` for each partition's state
        so the fetcher would skip them while an on_partitions_revoked /
        on_partitions_lost listener runs. Called immediately before
        invoking the listener. The flag is single-shot - the
        surrounding ``assign_from_subscribed`` drops the
        ``TopicPartitionState`` for revoked partitions when the
        listener returns.

        Currently a no-op while the user thread is blocked in ``_net.run``
        during rebalance and so the only path that calls ``send_fetches``
        cannot fire. Kept as a defensive gate in case this changes in
        the future.N)r&   rU   )r,   rI   rN   r   r   r   rU   	  s
   
z)SubscriptionState.mark_pending_revocationc                 C   ru   rd   )r&   reset_failed)r,   rI   next_retry_timera   r   r   r   r     rx   zSubscriptionState.reset_failedc                 C   sL   || j v r$z	| j | W d S  ty#   | j |}|| j |< Y d S w d S rd   )r&   move_to_endr   pop)r,   ra   rm   r   r   r   move_partition_to_end   s   
z'SubscriptionState.move_partition_to_endc                 C   rz   rd   )r&   rp   r|   r   r   r   rp   )  r~   zSubscriptionState.position)r   )r   NN)Frd   )2r   r   r   __doc__r3   r.   r4   r
   r@   r:   rD   rG   rJ   rO   rH   r\   r^   r`   rc   rg   rn   rA   rr   rt   rv   ry   r}   r   r   ro   r   r   r   r   r   r   r   r   r   r   r   r   re   rj   r   r   rU   r   r   rp   r   r   r   r   r      s    
?





(!































r   c                   @   s   e Zd Zdd Zdd Zdd ZeeeddZd	d
 Zdd Z	edd Z
dd Zdd Zedd Zdd Zdd Zdd Zdd Zd6ddZd d! Zd"d# Zd$d% Zd&d' Zed(d) Zd*d+ Zd,d- Zd.d/ Zd0d1 Zd2d3 Zd7d4d5ZdS )8rX   c                 C   sF   d| _ d | _d | _d | _d| _d | _d| _d| _d | _d | _	d| _
d S )NF)r   reset_strategy	_position	highwaterdrop_pending_record_batchr   _awaiting_validation_current_leader_epoch_preferred_read_replica"_preferred_read_replica_expiration_pending_revocationrF   r   r   r   r.   /  s   
zTopicPartitionState.__init__c                 C   s,   | j stdt|tstd|| _d S )NzValid position requiredz offset must be OffsetAndMetadata)ro   r1   r2   r/   r   r9   r   r,   rb   r   r   r   _set_positionT  s
   


z!TopicPartitionState._set_positionc                 C   r]   rd   r   rF   r   r   r   _get_position[  s   z!TopicPartitionState._get_positionNzlast positionc                 C   s:   |d u rt d|| _d | _d | _d| _d| _|   d S )Nzstrategy cannot be NoneFr   )r0   r   r   r   r   r   clear_preferred_read_replica)r,   strategyr   r   r   rs   `  s   zTopicPartitionState.resetc                 C      | j d u p| j t k S rd   r   time	monotonicrF   r   r   r   r   j     z$TopicPartitionState.is_reset_allowedc                 C   
   | j d uS rd   )r   rF   r   r   r   r{   m  r   z"TopicPartitionState.awaiting_resetc                 C   
   || _ d S rd   r   r,   r   r   r   r   rv   q     
z%TopicPartitionState.set_reset_pendingc                 C   r   rd   r   r   r   r   r   r   t  r   z TopicPartitionState.reset_failedc                 C   r   rd   r   rF   r   r   r   ro   w  r   z&TopicPartitionState.has_valid_positionc                 C   s   | j  o| j S rd   )ro   r{   rF   r   r   r   r   {  s   z'TopicPartitionState.is_missing_positionc                 C   sF   t |tr|nt|dd| _d | _d| _d | _d| _d| _|   d S )N r   TF)	r/   r   r   r   r   r   r   r   r   r   r   r   r   r`   ~  s   zTopicPartitionState.seekc                 C   
   d| _ d S )NTr   rF   r   r   r   r     r   zTopicPartitionState.pausec                 C   r   r   r   rF   r   r   r   r     r   zTopicPartitionState.resumeTc                 C   s
   || _ dS )zKIP-429: gate fetches while an on_partitions_revoked /
        on_partitions_lost listener is in progress for this partition.
        Single-shot: the surrounding ``assign_from_subscribed`` drops
        the state object once the listener returns.N)r   )r,   valr   r   r   rU     s   
z+TopicPartitionState.mark_pending_revocationc                 C   s   | j  o| j o| jo| j S rd   )r   r   ro   r   rF   r   r   r   rj     s   z TopicPartitionState.is_fetchablec                 C   s8   | j du rdS | jdurt | jkr|   dS | j S )zReturn the currently-cached preferred read replica (KIP-392),
        or None if unset/expired. Lazily clears the cache on expiry.N)r   r   r   r   r   rF   r   r   r   preferred_read_replica  s   

z*TopicPartitionState.preferred_read_replicac                 C   sD   |du s|dk r| j du}|   |S || j krdS || _ || _dS )zCache the broker's chosen preferred read replica until ``expiration_time``
        (monotonic). ``node_id == -1`` (or None) clears the cache.

        Returns True if the cached replica actually changed (caller can log).
        Nr   FT)r   r   r   )r,   node_idexpiration_timechangedr   r   r   update_preferred_read_replica  s   

z1TopicPartitionState.update_preferred_read_replicac                 C   s   | j }d| _ d| _|S )zClear the cached preferred read replica. Returns the previously-
        cached node_id (or None) so the caller can log the eviction.N)r   r   )r,   previousr   r   r   r     s   z0TopicPartitionState.clear_preferred_read_replicac                 C   r]   rd   )r   rF   r   r   r   r     s   z'TopicPartitionState.awaiting_validationc                 C   sv   | j rdS | jdu rdS |du s|dk rdS t| j| jj}||kr$dS || _| jjdk r/dS |   d| _d| _dS )zMark for validation if the current leader has advanced beyond the
        leader epoch this position was last reconciled against.

        Returns True if the partition is now awaiting validation.
        FNr   T)r{   r   maxr   leader_epochr   r   r   )r,   r   reconciled_epochr   r   r   r     s    
z+TopicPartitionState.maybe_validate_positionc                 C   s*   | j du s| j jdk rdS d| _d| _dS )zKForce validation (e.g., after FENCED/UNKNOWN epoch errors from the broker).Nr   FT)r   r   r   r   rF   r   r   r   r     s
   z/TopicPartitionState.request_position_validationc                 C   r   rd   r   rF   r   r   r   r     r   z)TopicPartitionState.is_validation_allowedc                 C   r   rd   r   r   r   r   r   r     r   z*TopicPartitionState.set_validation_pendingc                 C   r   rd   r   r   r   r   r   r     r   z%TopicPartitionState.validation_failedc                 C   s"   d| _ d | _|d ur|| _d S d S r   )r   r   r   )r,   r   r   r   r   r     s
   
z'TopicPartitionState.complete_validation)Trd   )r   r   r   r.   r   r   propertyrp   rs   r   r{   rv   r   ro   r   r`   r   r   rU   rj   r   r   r   r   r   r   r   r   r   r   r   r   r   r   rX   .  s<    %




&rX   c                   @   0   e Zd ZdZedd Zedd Zdd ZdS )	r;   a
  
    A callback interface that the user can implement to trigger custom actions
    when the set of partitions assigned to the consumer changes.

    This is applicable when the consumer is having Kafka auto-manage group
    membership. If the consumer's directly assign partitions, those
    partitions will never be reassigned and this callback is not applicable.

    When Kafka is managing the group membership, a partition re-assignment will
    be triggered any time the members of the group changes or the subscription
    of the members changes. This can occur when processes die, new process
    instances are added or old instances come back to life after failure.
    Rebalances can also be triggered by changes affecting the subscribed
    topics (e.g. when then number of partitions is administratively adjusted).

    There are many uses for this functionality. One common use is saving offsets
    in a custom store. By saving offsets in the on_partitions_revoked(), call we
    can ensure that any time partition assignment changes the offset gets saved.

    Another use is flushing out any kind of cache of intermediate results the
    consumer may be keeping. For example, consider a case where the consumer is
    subscribed to a topic containing user page views, and the goal is to count
    the number of page views per users for each five minute window.  Let's say
    the topic is partitioned by the user id so that all events for a particular
    user will go to a single consumer instance. The consumer can keep in memory
    a running tally of actions per user and only flush these out to a remote
    data store when its cache gets too big. However if a partition is reassigned
    it may want to automatically trigger a flush of this cache, before the new
    owner takes over consumption.

    Threading: callbacks run on the consumer's IO event loop, the same loop
    that drives heartbeats. Sync listener methods must return promptly --
    blocking IO inside a sync listener will block heartbeats for the duration
    and can cause the consumer to be kicked from the group if the listener
    runs longer than ``session_timeout_ms``. For listeners that need to do
    blocking work (e.g. flushing state to a database), prefer
    :class:`AsyncConsumerRebalanceListener`, which lets you ``await`` while
    keeping the loop responsive, or wrap the blocking call in your own
    worker thread.

    It is guaranteed that all consumer processes will invoke
    on_partitions_revoked() prior to any process invoking
    on_partitions_assigned(). So if offsets or other state is saved in the
    on_partitions_revoked() call, it should be saved by the time the process
    taking over that partition has their on_partitions_assigned() callback
    called to load the state.
    c                 C      dS )a  
        A callback method the user can implement to provide handling of offset
        commits to a customized store on the start of a rebalance operation.
        This method will be called before a rebalance operation starts and
        after the consumer stops fetching data. It is recommended that offsets
        should be committed in this callback to either Kafka or a custom offset
        store to prevent duplicate data.

        NOTE: This method is called before each rebalance and also when the
        consumer is closing (KafkaConsumer.close()), so that offsets / state
        can be committed before the partitions are given up. If the group
        membership has already been lost (forced eviction),
        on_partitions_lost() is called instead.

        Arguments:
            revoked (list of TopicPartition): the partitions that were assigned
                to the consumer on the last rebalance
        Nr   r,   revokedr   r   r   on_partitions_revoked3  s   z/ConsumerRebalanceListener.on_partitions_revokedc                 C   r   )a  
        A callback method the user can implement to provide handling of
        customized offsets on completion of a successful partition
        re-assignment. This method will be called after an offset re-assignment
        completes and before the consumer starts fetching data.

        It is guaranteed that all the processes in a consumer group will execute
        their on_partitions_revoked() callback before any instance executes its
        on_partitions_assigned() callback.

        Arguments:
            assigned (list of TopicPartition): the partitions assigned to the
                consumer (may include partitions that were previously assigned)
        Nr   r,   assignedr   r   r   on_partitions_assignedI  s   z0ConsumerRebalanceListener.on_partitions_assignedc                 C   s
   |  |S )a  KIP-429: called when the consumer has been forcibly removed
        from the group (heartbeat session expiry, ``UnknownMemberIdError``,
        ``IllegalGenerationError``, ``FencedInstanceIdError``) and the
        partitions cannot be cleanly committed. ``on_partitions_revoked``
        implies the user *can* still commit; ``on_partitions_lost`` makes
        explicit that the member has been booted and any in-flight state
        for these partitions should be discarded.

        Default behaviour is to delegate to ``on_partitions_revoked`` so
        listeners written before KIP-429 keep working unchanged. Override
        for cleanup that is specific to the forced-eviction case (e.g.
        skipping a commit attempt that will fail anyway).

        Arguments:
            lost (set of TopicPartition): the partitions that were
                assigned but have been lost due to forced eviction.
        r   r,   lostr   r   r   on_partitions_lost[  s   
z,ConsumerRebalanceListener.on_partitions_lostNr   r   r   r   r   r   r   r   r   r   r   r   r;     s    /

r;   c                   @   r   )	r<   a  
    Async variant of :class:`ConsumerRebalanceListener`.

    Implement this when your rebalance hooks need to perform IO that would
    otherwise block the consumer's event loop -- e.g. flushing state to a
    database, calling an external service, or coordinating with other async
    code. The coordinator detects coroutine functions and ``await`` s them
    instead of calling inline, so other tasks on the loop (notably the
    heartbeat coroutine) continue to run while your listener is suspended.

    Same lifecycle and ordering guarantees as the sync listener: all
    consumers in the group invoke ``on_partitions_revoked`` before any
    invokes ``on_partitions_assigned``. Both methods must be defined as
    ``async def``; otherwise use :class:`ConsumerRebalanceListener`.
    c                       dS )a  Async-callback for the start of a rebalance operation.

        See :meth:`ConsumerRebalanceListener.on_partitions_revoked` for
        semantics. The coordinator awaits this method, so non-blocking IO
        via ``await`` keeps the heartbeat loop responsive during the call.

        Arguments:
            revoked (set of TopicPartition): the partitions that were
                assigned to the consumer on the last rebalance.
        Nr   r   r   r   r   r        z4AsyncConsumerRebalanceListener.on_partitions_revokedc                    r   )ag  Async-callback for the completion of a partition re-assignment.

        See :meth:`ConsumerRebalanceListener.on_partitions_assigned` for
        semantics.

        Arguments:
            assigned (set of TopicPartition): the partitions assigned to
                the consumer (may include partitions that were previously
                assigned).
        Nr   r   r   r   r   r     r   z5AsyncConsumerRebalanceListener.on_partitions_assignedc                    s   |  |I dH  dS )a|  Async variant of
        :meth:`ConsumerRebalanceListener.on_partitions_lost`. Default
        implementation awaits ``on_partitions_revoked`` for backward
        compatibility with listeners written before KIP-429.

        Arguments:
            lost (set of TopicPartition): the partitions that were
                assigned but have been lost due to forced eviction.
        Nr   r   r   r   r   r     s   
z1AsyncConsumerRebalanceListener.on_partitions_lostNr   r   r   r   r   r<   p  s    

r<   ) abcr   r   collectionsr   collections.abcr   enumr   loggingrV   r6   r)   r   kafka.errorserrorsr1   kafka.protocol.consumerr   kafka.structsr   
kafka.utilr	   r
   	getLoggerr   r   r   r   rX   r;   r<   r   r   r   r   <module>   s0    
     Vm