# 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
        log.debug("Finished assignment for group %s: %s",
                  self.group_id, assignments)

        # `set_topics()` will not trigger a metadata update if we only
        # removed some topics (client has all needed metadata already),
        # so we force a snapshot update.
        self._metadata_snapshot = self._get_metadata_snapshot()

        group_assignment = {}
        for member_id, assignment in assignments.items():
            group_assignment[member_id] = assignment
        return group_assignment

    async def _on_join_complete(
        self, generation, member_id, protocol,
        member_assignment_bytes
    ):
        assignor = self._lookup_assignor(protocol)
        assert assignor, 'invalid assignment protocol: %s' % protocol

        assignment = ConsumerProtocol.ASSIGNMENT.decode(
            member_assignment_bytes)

        # update partition assignment
        self._subscription.assign_from_subscribed(assignment.partitions())
        # The await bellow can change subscription, remember the ongoing one.
        subscription = self._subscription.subscription

        # give the assignor a chance to update internal state
        # based on the received assignment

# 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

        log.debug("Finished assignment for group %s: %s",
                  self.group_id, assignments)

        # `set_topics()` will not trigger a metadata update if we only
        # removed some topics (client has all needed metadata already),
        # so we force a snapshot update.
        self._metadata_snapshot = self._get_metadata_snapshot()

        group_assignment = {}
        for member_id, assignment in assignments.items():
            group_assignment[member_id] = assignment
        return group_assignment

    async def _on_join_complete(
        self, generation, member_id, protocol,
        member_assignment_bytes
    ):
        assignor = self._lookup_assignor(protocol)
        assert assignor, 'invalid assignment protocol: %s' % protocol

        assignment = ConsumerProtocol.ASSIGNMENT.decode(
            member_assignment_bytes)

        # update partition assignment
        self._subscription.assign_from_subscribed(assignment.partitions())
        # The await bellow can change subscription, remember the ongoing one.
        subscription = self._subscription.subscription

        # give the assignor a chance to update internal state
        # based on the received assignment

        assignor.on_assignment(assignment)