# 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
                        " with offset %s: %s", self.group_id, tp, offset,
                        error_type.__name__)
                    errored[tp] = error_type()

        if errored:
            first_error = list(errored.values())[0]
            raise first_error
        if unauthorized_topics:
            log.error("OffsetCommit failed for unauthorized topics %s",
                      unauthorized_topics)
            raise Errors.TopicAuthorizationFailedError(unauthorized_topics)

    async def _maybe_refresh_commit_offsets(self, assignment):
        need_update = assignment.requesting_committed()
        if need_update:
            try:
                offsets = await self._do_fetch_commit_offsets(
                    need_update)
            except Errors.KafkaError as err:
                if not err.retriable:
                    raise
                else:
                    log.debug("Failed to fetch committed offsets: %r", err)
                return False
            for tp in need_update:
                tp_state = assignment.state_value(tp)
                if tp in offsets:
                    tp_state.update_committed(offsets[tp])
                else:
                    tp_state.update_committed(

# 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

                        " with offset %s: %s", self.group_id, tp, offset,
                        error_type.__name__)
                    errored[tp] = error_type()

        if errored:
            first_error = list(errored.values())[0]
            raise first_error
        if unauthorized_topics:
            log.error("OffsetCommit failed for unauthorized topics %s",
                      unauthorized_topics)
            raise Errors.TopicAuthorizationFailedError(unauthorized_topics)

    async def _maybe_refresh_commit_offsets(self, assignment):
        need_update = assignment.requesting_committed()
        if need_update:
            try:
                offsets = await self._do_fetch_commit_offsets(
                    need_update)
            except Errors.KafkaError as err:
                if not err.retriable:
                    raise
                else:
                    log.debug("Failed to fetch committed offsets: %r", err)
                return False
            for tp in need_update:
                tp_state = assignment.state_value(tp)
                if tp in offsets:
                    tp_state.update_committed(offsets[tp])
                else:
                    tp_state.update_committed(

                        OffsetAndMetadata(UNKNOWN_OFFSET, ""))