# 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

        async with self._coordinator_lookup_lock:
            retry_backoff = self._retry_backoff_ms / 1000
            while self.coordinator_id is None and not self._closing.done():
                try:
                    coordinator_id = (
                        await self._client.coordinator_lookup(
                            CoordinationType.GROUP, self.group_id)
                    )
                except Errors.GroupAuthorizationFailedError:
                    err = Errors.GroupAuthorizationFailedError(self.group_id)
                    raise err
                except Errors.KafkaError as err:
                    log.error("Group Coordinator Request failed: %s", err)
                    if err.retriable:
                        await self._client.force_metadata_update()
                        await asyncio.sleep(retry_backoff)
                        continue
                    else:
                        raise

                # Try to connect to confirm that the connection can be
                # established.
                ready = await self._client.ready(
                    coordinator_id, group=ConnectionGroup.COORDINATION)
                if not ready:
                    await asyncio.sleep(retry_backoff)
                    continue

                self.coordinator_id = coordinator_id

# 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


        async with self._coordinator_lookup_lock:
            retry_backoff = self._retry_backoff_ms / 1000
            while self.coordinator_id is None and not self._closing.done():
                try:
                    coordinator_id = (
                        await self._client.coordinator_lookup(
                            CoordinationType.GROUP, self.group_id)
                    )
                except Errors.GroupAuthorizationFailedError:
                    err = Errors.GroupAuthorizationFailedError(self.group_id)
                    raise err
                except Errors.KafkaError as err:
                    log.error("Group Coordinator Request failed: %s", err)
                    if err.retriable:
                        await self._client.force_metadata_update()
                        await asyncio.sleep(retry_backoff)
                        continue
                    else:
                        raise

                # Try to connect to confirm that the connection can be
                # established.
                ready = await self._client.ready(
                    coordinator_id, group=ConnectionGroup.COORDINATION)
                if not ready:
                    await asyncio.sleep(retry_backoff)
                    continue

                self.coordinator_id = coordinator_id

                self._coordinator_dead_fut = create_future()