# 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)
# 
# def get_running_loop() -> asyncio.AbstractEventLoop:
#     loop = asyncio.get_event_loop()
#     if not loop.is_running():
#         raise RuntimeError(
#             "The object should be created within an async function or "
#             "provide loop directly."
#         )
#     return loop
# 
# async def wait_for(fut: Awaitable[T], timeout: Union[None, int, float] = None) -> T:
#     # A replacement for buggy (since 3.8.6) `asyncio.wait_for()`
#     # https://bugs.python.org/issue42130
#     async with async_timeout.timeout(timeout):
#         return await fut
# 
# Path: aiokafka/abc.py
# class AbstractTokenProvider(abc.ABC):
#     """
#     A Token Provider must be used for the `SASL OAuthBearer`_ protocol.
# 
#     The implementation should ensure token reuse so that multiple
#     calls at connect time do not create multiple tokens.
#     The implementation should also periodically refresh the token in order to
#     guarantee that each call returns an unexpired token.
# 
#     A timeout error should be returned after a short period of inactivity so
#     that the broker can log debugging info and retry.
# 
#     Token Providers MUST implement the :meth:`token` method
# 
#     .. _SASL OAuthBearer:
#         https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_oauth.html
#     """
# 
#     def __init__(self, **config):
#         pass
# 
#     @abc.abstractmethod
#     async def token(self):
#         """
#         An async callback returning a :class:`str` ID/Access Token to be sent to
#         the Kafka client. In case where a synchoronous callback is needed,
#         implementations like following can be used:
# 
#         .. code-block:: python
# 
#             from aiokafka.abc import AbstractTokenProvider
# 
#             class CustomTokenProvider(AbstractTokenProvider):
#                 async def token(self):
#                     return asyncio.get_running_loop().run_in_executor(
#                         None, self._token)
# 
#                 def _token(self):
#                     # The actual synchoronous token callback.
#         """
#         pass
# 
#     def extensions(self):
#         """
#         This is an OPTIONAL method that may be implemented.
# 
#         Returns a map of key-value pairs that can be sent with the
#         SASL/OAUTHBEARER initial client request. If not implemented, the values
#         are ignored
# 
#         This feature is only available in Kafka >= 2.1.0.
#         """
#         return {}


import asyncio
import collections
import base64
import functools
import hashlib
import hmac
import logging
import struct
import sys
import time
import traceback
import uuid
import warnings
import weakref
import async_timeout
import aiokafka.errors as Errors
    import gssapi
from kafka.protocol.api import RequestHeader
from kafka.protocol.admin import (
    SaslHandShakeRequest, SaslAuthenticateRequest, ApiVersionRequest
)
from kafka.protocol.commit import (
    GroupCoordinatorResponse_v0 as GroupCoordinatorResponse)
from aiokafka.util import create_future, create_task, get_running_loop, wait_for
from aiokafka.abc import AbstractTokenProvider
    def pick_best(self, request_versions):
        api_key = request_versions[0].API_KEY
        supported_versions = self._versions.get(api_key)
        if supported_versions is None:
            return request_versions[0]
        else:
            for req_klass in reversed(request_versions):
                if supported_versions[0] <= req_klass.API_VERSION and \
                        req_klass.API_VERSION <= supported_versions[1]:
                    return req_klass
        raise Errors.KafkaError(
            f"Could not pick a version for API_KEY={api_key} from {supported_versions}."
        )


async def create_conn(
    host, port, *, client_id='aiokafka',
    request_timeout_ms=40000, api_version=(0, 8, 2),
    ssl_context=None, security_protocol="PLAINTEXT",
    max_idle_ms=None, on_close=None,
    sasl_mechanism=None,
    sasl_plain_username=None,
    sasl_plain_password=None,
    sasl_kerberos_service_name='kafka',
    sasl_kerberos_domain_name=None,
    sasl_oauth_token_provider=None,
    version_hint=None
):
    conn = AIOKafkaConnection(
        host, port, client_id=client_id,

# 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)
# 
# def get_running_loop() -> asyncio.AbstractEventLoop:
#     loop = asyncio.get_event_loop()
#     if not loop.is_running():
#         raise RuntimeError(
#             "The object should be created within an async function or "
#             "provide loop directly."
#         )
#     return loop
# 
# async def wait_for(fut: Awaitable[T], timeout: Union[None, int, float] = None) -> T:
#     # A replacement for buggy (since 3.8.6) `asyncio.wait_for()`
#     # https://bugs.python.org/issue42130
#     async with async_timeout.timeout(timeout):
#         return await fut
# 
# Path: aiokafka/abc.py
# class AbstractTokenProvider(abc.ABC):
#     """
#     A Token Provider must be used for the `SASL OAuthBearer`_ protocol.
# 
#     The implementation should ensure token reuse so that multiple
#     calls at connect time do not create multiple tokens.
#     The implementation should also periodically refresh the token in order to
#     guarantee that each call returns an unexpired token.
# 
#     A timeout error should be returned after a short period of inactivity so
#     that the broker can log debugging info and retry.
# 
#     Token Providers MUST implement the :meth:`token` method
# 
#     .. _SASL OAuthBearer:
#         https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_oauth.html
#     """
# 
#     def __init__(self, **config):
#         pass
# 
#     @abc.abstractmethod
#     async def token(self):
#         """
#         An async callback returning a :class:`str` ID/Access Token to be sent to
#         the Kafka client. In case where a synchoronous callback is needed,
#         implementations like following can be used:
# 
#         .. code-block:: python
# 
#             from aiokafka.abc import AbstractTokenProvider
# 
#             class CustomTokenProvider(AbstractTokenProvider):
#                 async def token(self):
#                     return asyncio.get_running_loop().run_in_executor(
#                         None, self._token)
# 
#                 def _token(self):
#                     # The actual synchoronous token callback.
#         """
#         pass
# 
#     def extensions(self):
#         """
#         This is an OPTIONAL method that may be implemented.
# 
#         Returns a map of key-value pairs that can be sent with the
#         SASL/OAUTHBEARER initial client request. If not implemented, the values
#         are ignored
# 
#         This feature is only available in Kafka >= 2.1.0.
#         """
#         return {}


# Path: aiokafka/conn.py
import asyncio
import collections
import base64
import functools
import hashlib
import hmac
import logging
import struct
import sys
import time
import traceback
import uuid
import warnings
import weakref
import async_timeout
import aiokafka.errors as Errors
    import gssapi
from kafka.protocol.api import RequestHeader
from kafka.protocol.admin import (
    SaslHandShakeRequest, SaslAuthenticateRequest, ApiVersionRequest
)
from kafka.protocol.commit import (
    GroupCoordinatorResponse_v0 as GroupCoordinatorResponse)
from aiokafka.util import create_future, create_task, get_running_loop, wait_for
from aiokafka.abc import AbstractTokenProvider

    def pick_best(self, request_versions):
        api_key = request_versions[0].API_KEY
        supported_versions = self._versions.get(api_key)
        if supported_versions is None:
            return request_versions[0]
        else:
            for req_klass in reversed(request_versions):
                if supported_versions[0] <= req_klass.API_VERSION and \
                        req_klass.API_VERSION <= supported_versions[1]:
                    return req_klass
        raise Errors.KafkaError(
            f"Could not pick a version for API_KEY={api_key} from {supported_versions}."
        )


async def create_conn(
    host, port, *, client_id='aiokafka',
    request_timeout_ms=40000, api_version=(0, 8, 2),
    ssl_context=None, security_protocol="PLAINTEXT",
    max_idle_ms=None, on_close=None,
    sasl_mechanism=None,
    sasl_plain_username=None,
    sasl_plain_password=None,
    sasl_kerberos_service_name='kafka',
    sasl_kerberos_domain_name=None,
    sasl_oauth_token_provider=None,
    version_hint=None
):
    conn = AIOKafkaConnection(
        host, port, client_id=client_id,

        request_timeout_ms=request_timeout_ms,