# kafka.coordinator package

## Subpackages

- kafka.coordinator.assignors package
  - Submodules
  - kafka.coordinator.assignors.abstract module
  - kafka.coordinator.assignors.roundrobin module
  - Module contents

## Submodules

## kafka.coordinator.base module

class `kafka.coordinator.base.``BaseCoordinator`(client, metrics, **configs)

  Bases: `object`

  BaseCoordinator implements group management for a single group member by interacting with a designated Kafka broker (the coordinator). Group semantics are provided by extending this class. See ConsumerCoordinator for example usage.

  From a high level, Kafka’s group management protocol consists of the following sequence of actions:

  1. Group Registration: Group members register with the coordinator providing their own metadata (such as the set of topics they are interested in).
  2. Group/Leader Selection: The coordinator select the members of the group and chooses one member as the leader.
  3. State Assignment: The leader collects the metadata from all the members of the group and assigns state.
  4. Group Stabilization: Each member receives the state assigned by the leader and begins processing.

  To leverage this protocol, an implementation must define the format of metadata provided by each member for group registration in `group_protocols()` and the format of the state assignment provided by the leader in `_perform_assignment()` and which becomes available to members in `_on_join_complete()`.

  Note on locking: this class shares state between the caller and a background thread which is used for sending heartbeats after the client has joined the group. All mutable state as well as state transitions are protected with the class’s monitor. Generally this means acquiring the lock before reading or writing the state of the group (e.g. generation, member_id) and holding the lock when sending a request that affects the state of the group (e.g. JoinGroup, LeaveGroup).

  `DEFAULT_CONFIG` = {'api_version': (0, 10, 1), 'group_id': 'kafka-python-default-group', 'heartbeat_interval_ms': 3000, 'max_poll_interval_ms': 300000, 'metric_group_prefix': '', 'retry_backoff_ms': 100, 'session_timeout_ms': 10000}
  `close`()

    Close the coordinator, leave the current group, and reset local generation / member_id

  `coordinator`()

    Get the current coordinator

    Returns: the current coordinator id or None if it is unknown

  `coordinator_dead`(error)

    Mark the current coordinator as dead.

  `coordinator_unknown`()

    Check if we know who the coordinator is and have an active connection

    Side-effect: reset coordinator_id to None if connection failed

    Returns:
      bool: True if the coordinator is unknown
  `ensure_active_group`()

    Ensure that the group is active (i.e. joined and synced)

  `ensure_coordinator_ready`()

    Block until the coordinator for this group is known (and we have an active connection – java client uses unsent queue).

  `generation`()

    Get the current generation state if the group is stable.

    Returns: the current generation or None if the group is unjoined/rebalancing

  `group_protocols`()

    Return the list of supported group protocols and metadata.

    This list is submitted by each group member via a JoinGroupRequest. The order of the protocols in the list indicates the preference of the protocol (the first entry is the most preferred). The coordinator takes this preference into account when selecting the generation protocol (generally more preferred protocols will be selected as long as all members support them and there is no disagreement on the preference).

    Note: metadata must be type bytes or support an encode() method

    Returns:
      list: [(protocol, metadata), …]
  `lookup_coordinator`()
  `maybe_leave_group`()

    Leave the current group and reset local generation/memberId.

  `need_rejoin`()

    Check whether the group should be rejoined (e.g. if metadata changes)

    Returns:
      bool: True if it should, False otherwise
  `poll_heartbeat`()

    Check the status of the heartbeat thread (if it is active) and indicate the liveness of the client. This must be called periodically after joining with `ensure_active_group()` to ensure that the member stays in the group. If an interval of time longer than the provided rebalance timeout (max_poll_interval_ms) expires without calling this method, then the client will proactively leave the group.

    Raises: RuntimeError for unexpected errors raised from the heartbeat thread

  `protocol_type`()

    Unique identifier for the class of supported protocols (e.g. “consumer” or “connect”).

    Returns:
      str: protocol type name
  `request_rejoin`()
  `reset_generation`()

    Reset the generation and memberId because we have fallen out of the group.

  `time_to_next_heartbeat`()

class `kafka.coordinator.base.``Generation`(generation_id, member_id, protocol)

  Bases: `object`

  `NO_GENERATION` = <kafka.coordinator.base.Generation object>

class `kafka.coordinator.base.``GroupCoordinatorMetrics`(heartbeat, metrics, prefix, tags=None)

  Bases: `object`

class `kafka.coordinator.base.``HeartbeatThread`(coordinator)

  Bases: `threading.Thread`

  `close`()
  `disable`()
  `enable`()
  `run`()

    Method representing the thread’s activity.

    You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

class `kafka.coordinator.base.``MemberState`

  Bases: `object`

  `REBALANCING` = '<rebalancing>'
  `STABLE` = '<stable>'
  `UNJOINED` = '<unjoined>'

exception `kafka.coordinator.base.``UnjoinedGroupException`

  Bases: `kafka.errors.KafkaError`

  `retriable` = True

## kafka.coordinator.consumer module

class `kafka.coordinator.consumer.``ConsumerCoordinator`(client, subscription, metrics, **configs)

  Bases: `kafka.coordinator.base.BaseCoordinator`

  This class manages the coordination process with the consumer coordinator.

  `DEFAULT_CONFIG` = {'api_version': (0, 10, 1), 'assignors': (<class 'kafka.coordinator.assignors.range.RangePartitionAssignor'>, <class 'kafka.coordinator.assignors.roundrobin.RoundRobinPartitionAssignor'>), 'auto_commit_interval_ms': 5000, 'default_offset_commit_callback': None, 'enable_auto_commit': True, 'exclude_internal_topics': True, 'group_id': 'kafka-python-default-group', 'heartbeat_interval_ms': 3000, 'max_poll_interval_ms': 300000, 'metric_group_prefix': 'consumer', 'retry_backoff_ms': 100, 'session_timeout_ms': 10000}
  `close`(autocommit=True)

    Close the coordinator, leave the current group, and reset local generation / member_id.

    Keyword Arguments:
      autocommit (bool): If auto-commit is configured for this consumer,
        this optional flag causes the consumer to attempt to commit any pending consumed offsets prior to close. Default: True
  `commit_offsets_async`(offsets, callback=None)

    Commit specific offsets asynchronously.

    Arguments:

      offsets (dict {TopicPartition: OffsetAndMetadata}): what to commit callback (callable, optional): called as callback(offsets, response)

      > response will be either an Exception or a OffsetCommitResponse struct. This callback can be used to trigger custom actions when a commit request completes.

    Returns:
      kafka.future.Future
  `commit_offsets_sync`(offsets)

    Commit specific offsets synchronously.

    This method will retry until the commit completes successfully or an unrecoverable error is encountered.

    Arguments:
      offsets (dict {TopicPartition: OffsetAndMetadata}): what to commit

    Raises error on failure

  `fetch_committed_offsets`(partitions)

    Fetch the current committed offsets for specified partitions

    Arguments:
      partitions (list of TopicPartition): partitions to fetch
    Returns:
      dict: {TopicPartition: OffsetAndMetadata}
  `group_protocols`()

    Returns list of preferred (protocols, metadata)

  `need_rejoin`()

    Check whether the group should be rejoined

    Returns:
      bool: True if consumer should rejoin group, False otherwise
  `poll`()

    Poll for coordinator events. Only applicable if group_id is set, and broker version supports GroupCoordinators. This ensures that the coordinator is known, and if using automatic partition assignment, ensures that the consumer has joined the group. This also handles periodic offset commits if they are enabled.

  `protocol_type`()

    Unique identifier for the class of supported protocols (e.g. “consumer” or “connect”).

    Returns:
      str: protocol type name
  `refresh_committed_offsets_if_needed`()

    Fetch committed offsets for assigned partitions.

  `time_to_next_poll`()

    Return seconds (float) remaining until `poll()` should be called again

class `kafka.coordinator.consumer.``ConsumerCoordinatorMetrics`(metrics, metric_group_prefix, subscription)

  Bases: `object`

## kafka.coordinator.heartbeat module

class `kafka.coordinator.heartbeat.``Heartbeat`(**configs)

  Bases: `object`

  `DEFAULT_CONFIG` = {'group_id': None, 'heartbeat_interval_ms': 3000, 'max_poll_interval_ms': 300000, 'retry_backoff_ms': 100, 'session_timeout_ms': 10000}
  `fail_heartbeat`()
  `poll`()
  `poll_timeout_expired`()
  `received_heartbeat`()
  `reset_timeouts`()
  `sent_heartbeat`()
  `session_timeout_expired`()
  `should_heartbeat`()
  `time_to_next_heartbeat`()

    Returns seconds (float) remaining before next heartbeat should be sent

## Module contents