# 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
            raise Errors.KafkaConnectionError(
                f"Connection at {self._host}:{self._port} broken: {err}"
            )

        self.log.debug(
            '%s Request %d: %s', self, correlation_id, request)

        if not expect_response:
            return self._writer.drain()
        fut = self._loop.create_future()
        self._requests.append((correlation_id, request.RESPONSE_TYPE, fut))
        return wait_for(fut, self._request_timeout)

    def _send_sasl_token(self, payload, expect_response=True):
        if self._writer is None:
            raise Errors.KafkaConnectionError(
                f"No connection to broker at {self._host}:{self._port}"
            )
        size = struct.pack(">i", len(payload))
        try:
            self._writer.write(size + payload)
        except OSError as err:
            self.close(reason=CloseReason.CONNECTION_BROKEN)
            raise Errors.KafkaConnectionError(
                f"Connection at {self._host}:{self._port} broken: {err}"
            )

        if not expect_response:
            return self._writer.drain()


# 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

            raise Errors.KafkaConnectionError(
                f"Connection at {self._host}:{self._port} broken: {err}"
            )

        self.log.debug(
            '%s Request %d: %s', self, correlation_id, request)

        if not expect_response:
            return self._writer.drain()
        fut = self._loop.create_future()
        self._requests.append((correlation_id, request.RESPONSE_TYPE, fut))
        return wait_for(fut, self._request_timeout)

    def _send_sasl_token(self, payload, expect_response=True):
        if self._writer is None:
            raise Errors.KafkaConnectionError(
                f"No connection to broker at {self._host}:{self._port}"
            )
        size = struct.pack(">i", len(payload))
        try:
            self._writer.write(size + payload)
        except OSError as err:
            self.close(reason=CloseReason.CONNECTION_BROKEN)
            raise Errors.KafkaConnectionError(
                f"Connection at {self._host}:{self._port} broken: {err}"
            )

        if not expect_response:
            return self._writer.drain()


        fut = self._loop.create_future()