# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")
# 
# Path: aiokafka/client.py
# class ConnectionGroup:
# 
#     DEFAULT = 0
#     COORDINATION = 1
# 
# class CoordinationType:
# 
#     GROUP = 0
#     TRANSACTION = 1
# 
# Path: aiokafka/util.py
# def create_future(loop: AbstractEventLoop = None) -> "asyncio.Future[T]":
#     if loop is None:
#         loop = get_running_loop()
#     return loop.create_future()
# 
# def create_task(coro: Awaitable[T]) -> "asyncio.Task[T]":
#     loop = get_running_loop()
#     return loop.create_task(coro)


import asyncio
import collections
import logging
import copy
import time
import aiokafka.errors as Errors
from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
from kafka.coordinator.protocol import ConsumerProtocol
from kafka.protocol.commit import (
    OffsetCommitRequest_v2 as OffsetCommitRequest,
    OffsetFetchRequest_v1 as OffsetFetchRequest)
from kafka.protocol.group import (
    HeartbeatRequest, JoinGroupRequest, LeaveGroupRequest, SyncGroupRequest)
from aiokafka.structs import OffsetAndMetadata, TopicPartition
from aiokafka.client import ConnectionGroup, CoordinationType
from aiokafka.util import create_future, create_task
            except Exception:
                log.exception("User provided subscription listener %s"
                              " for group %s failed on_partitions_revoked",
                              self._subscription.listener, self.group_id)

    async def _perform_assignment(
        self, leader_id, assignment_strategy, members
    ):
        assignor = self._lookup_assignor(assignment_strategy)
        assert assignor, \
            'Invalid assignment protocol: %s' % assignment_strategy
        member_metadata = {}
        all_subscribed_topics = set()
        for member_id, metadata_bytes in members:
            metadata = ConsumerProtocol.METADATA.decode(metadata_bytes)
            member_metadata[member_id] = metadata
            all_subscribed_topics.update(metadata.subscription)

        # the leader will begin watching for changes to any of the topics
        # the group is interested in, which ensures that all metadata changes
        # will eventually be seen
        self._group_subscription = all_subscribed_topics
        if not self._subscription.subscribed_pattern:
            self._client.set_topics(self._group_subscription)
        # If somewhere we forced a metadata update (like in some `set_topics`
        # call) we should wait for it before performing assignment
        await self._client._maybe_wait_metadata()

        log.debug("Performing assignment for group %s using strategy %s"
                  " with subscriptions %s", self.group_id, assignor.name,

# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")
# 
# Path: aiokafka/client.py
# class ConnectionGroup:
# 
#     DEFAULT = 0
#     COORDINATION = 1
# 
# class CoordinationType:
# 
#     GROUP = 0
#     TRANSACTION = 1
# 
# Path: aiokafka/util.py
# def create_future(loop: AbstractEventLoop = None) -> "asyncio.Future[T]":
#     if loop is None:
#         loop = get_running_loop()
#     return loop.create_future()
# 
# def create_task(coro: Awaitable[T]) -> "asyncio.Task[T]":
#     loop = get_running_loop()
#     return loop.create_task(coro)


# Path: aiokafka/consumer/group_coordinator.py
import asyncio
import collections
import logging
import copy
import time
import aiokafka.errors as Errors
from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
from kafka.coordinator.protocol import ConsumerProtocol
from kafka.protocol.commit import (
    OffsetCommitRequest_v2 as OffsetCommitRequest,
    OffsetFetchRequest_v1 as OffsetFetchRequest)
from kafka.protocol.group import (
    HeartbeatRequest, JoinGroupRequest, LeaveGroupRequest, SyncGroupRequest)
from aiokafka.structs import OffsetAndMetadata, TopicPartition
from aiokafka.client import ConnectionGroup, CoordinationType
from aiokafka.util import create_future, create_task

            except Exception:
                log.exception("User provided subscription listener %s"
                              " for group %s failed on_partitions_revoked",
                              self._subscription.listener, self.group_id)

    async def _perform_assignment(
        self, leader_id, assignment_strategy, members
    ):
        assignor = self._lookup_assignor(assignment_strategy)
        assert assignor, \
            'Invalid assignment protocol: %s' % assignment_strategy
        member_metadata = {}
        all_subscribed_topics = set()
        for member_id, metadata_bytes in members:
            metadata = ConsumerProtocol.METADATA.decode(metadata_bytes)
            member_metadata[member_id] = metadata
            all_subscribed_topics.update(metadata.subscription)

        # the leader will begin watching for changes to any of the topics
        # the group is interested in, which ensures that all metadata changes
        # will eventually be seen
        self._group_subscription = all_subscribed_topics
        if not self._subscription.subscribed_pattern:
            self._client.set_topics(self._group_subscription)
        # If somewhere we forced a metadata update (like in some `set_topics`
        # call) we should wait for it before performing assignment
        await self._client._maybe_wait_metadata()

        log.debug("Performing assignment for group %s using strategy %s"
                  " with subscriptions %s", self.group_id, assignor.name,

                  member_metadata)