class ConsumerRebalanceListener(BaseConsumerRebalanceListener):
    """
    A callback interface that the user can implement to trigger custom actions
    when the set of partitions assigned to the consumer changes.

    This is applicable when the consumer is having Kafka auto-manage group
    membership. If the consumer's directly assign partitions, those
    partitions will never be reassigned and this callback is not applicable.

    When Kafka is managing the group membership, a partition re-assignment will
    be triggered any time the members of the group changes or the subscription
    of the members changes. This can occur when processes die, new process
    instances are added or old instances come back to life after failure.
    Rebalances can also be triggered by changes affecting the subscribed
    topics (e.g. when then number of partitions is administratively adjusted).

    There are many uses for this functionality. One common use is saving
    offsets in a custom store. By saving offsets in the
    :meth:`on_partitions_revoked`, call we can ensure that any time partition
    assignment changes the offset gets saved.

    Another use is flushing out any kind of cache of intermediate results the
    consumer may be keeping. For example, consider a case where the consumer is
    subscribed to a topic containing user page views, and the goal is to count
    the number of page views per users for each five minute window.  Let's say
    the topic is partitioned by the user id so that all events for a particular
    user will go to a single consumer instance. The consumer can keep in memory
    a running tally of actions per user and only flush these out to a remote
    data store when its cache gets too big. However if a partition is
    reassigned it may want to automatically trigger a flush of this cache,
    before the new owner takes over consumption.

    This callback will execute during the rebalance process, and Consumer will
    wait for callbacks to finish before proceeding with group join.

    It is guaranteed that all consumer processes will invoke
    :meth:`on_partitions_revoked` prior to any process invoking
    :meth:`on_partitions_assigned`. So if offsets or other state is saved in the
    :meth:`on_partitions_revoked` call, it should be saved by the time the process
    taking over that partition has their :meth:`on_partitions_assigned` callback
    called to load the state.
    """

    @abc.abstractmethod
    def on_partitions_revoked(self, revoked):
        """
        A coroutine or function the user can implement to provide cleanup or
        custom state save on the start of a rebalance operation.

        This method will be called *before* a rebalance operation starts and
        *after* the consumer stops fetching data.

        If you are using manual commit you have to commit all consumed offsets
        here, to avoid duplicate message delivery after rebalance is finished.

        .. note:: This method is only called before rebalances. It is not
            called prior to :meth:`.AIOKafkaConsumer.stop`

        Arguments:
            revoked (list(TopicPartition)): the partitions that were assigned
                to the consumer on the last rebalance
        """
        pass

    @abc.abstractmethod
    def on_partitions_assigned(self, assigned):
        """
        A coroutine or function the user can implement to provide load of
        custom consumer state or cache warmup on completion of a successful
        partition re-assignment.

        This method will be called *after* partition re-assignment completes
        and *before* the consumer starts fetching data again.

        It is guaranteed that all the processes in a consumer group will
        execute their :meth:`on_partitions_revoked` callback before any instance
        executes its :meth:`on_partitions_assigned` callback.

        Arguments:
            assigned (list(TopicPartition)): the partitions assigned to the
                consumer (may include partitions that were previously assigned)
        """
        pass class AIOKafkaConsumer:
    """
    A client that consumes records from a Kafka cluster.

    The consumer will transparently handle the failure of servers in the Kafka
    cluster, and adapt as topic-partitions are created or migrate between
    brokers.

    It also interacts with the assigned Kafka Group Coordinator node to allow
    multiple consumers to load balance consumption of topics (feature of Kafka
    >= 0.9.0.0).

    .. _kip-62:
        https://cwiki.apache.org/confluence/display/KAFKA/KIP-62%3A+Allow+consumer+to+send+heartbeats+from+a+background+thread

    Arguments:
        *topics (list(str)): optional list of topics to subscribe to. If not set,
            call :meth:`.subscribe` or :meth:`.assign` before consuming records.
            Passing topics directly is same as calling :meth:`.subscribe` API.
        bootstrap_servers (str, list(str)): a ``host[:port]`` string (or list of
            ``host[:port]`` strings) that the consumer should contact to bootstrap
            initial cluster metadata.

            This does not have to be the full node list.
            It just needs to have at least one broker that will respond to a
            Metadata API Request. Default port is 9092. If no servers are
            specified, will default to ``localhost:9092``.
        client_id (str): a name for this client. This string is passed in
            each request to servers and can be used to identify specific
            server-side log entries that correspond to this client. Also
            submitted to :class:`~.consumer.group_coordinator.GroupCoordinator`
            for logging with respect to consumer group administration. Default:
            ``aiokafka-{version}``
        group_id (str or None): name of the consumer group to join for dynamic
            partition assignment (if enabled), and to use for fetching and
            committing offsets. If None, auto-partition assignment (via
            group coordinator) and offset commits are disabled.
            Default: None
        key_deserializer (Callable): Any callable that takes a
            raw message key and returns a deserialized key.
        value_deserializer (Callable, Optional): Any callable that takes a
            raw message value and returns a deserialized value.
        fetch_min_bytes (int): Minimum amount of data the server should
            return for a fetch request, otherwise wait up to
            `fetch_max_wait_ms` for more data to accumulate. Default: 1.
        fetch_max_bytes (int): The maximum amount of data the server should
            return for a fetch request. This is not an absolute maximum, if
            the first message in the first non-empty partition of the fetch
            is larger than this value, the message will still be returned
            to ensure that the consumer can make progress. NOTE: consumer
            performs fetches to multiple brokers in parallel so memory
            usage will depend on the number of brokers containing
            partitions for the topic.
            Supported Kafka version >= 0.10.1.0. Default: 52428800 (50 Mb).
        fetch_max_wait_ms (int): The maximum amount of time in milliseconds
            the server will block before answering the fetch request if
            there isn't sufficient data to immediately satisfy the
            requirement given by fetch_min_bytes. Default: 500.
        max_partition_fetch_bytes (int): The maximum amount of data
            per-partition the server will return. The maximum total memory
            used for a request ``= #partitions * max_partition_fetch_bytes``.
            This size must be at least as large as the maximum message size
            the server allows or else it is possible for the producer to
            send messages larger than the consumer can fetch. If that
            happens, the consumer can get stuck trying to fetch a large
            message on a certain partition. Default: 1048576.
        max_poll_records (int): The maximum number of records returned in a
            single call to :meth:`.getmany`. Defaults ``None``, no limit.
        request_timeout_ms (int): Client request timeout in milliseconds.
            Default: 40000.
        retry_backoff_ms (int): Milliseconds to backoff when retrying on
            errors. Default: 100.
        auto_offset_reset (str): A policy for resetting offsets on
            :exc:`.OffsetOutOfRangeError` errors: ``earliest`` will move to the oldest
            available message, ``latest`` will move to the most recent, and
            ``none`` will raise an exception so you can handle this case.
            Default: ``latest``.
        enable_auto_commit (bool): If true the consumer's offset will be
            periodically committed in the background. Default: True.
        auto_commit_interval_ms (int): milliseconds between automatic
            offset commits, if enable_auto_commit is True. Default: 5000.
        check_crcs (bool): Automatically check the CRC32 of the records
            consumed. This ensures no on-the-wire or on-disk corruption to
            the messages occurred. This check adds some overhead, so it may
            be disabled in cases seeking extreme performance. Default: True
        metadata_max_age_ms (int): The period of time in milliseconds after
            which we force a refresh of metadata even if we haven't seen any
            partition leadership changes to proactively discover any new
            brokers or partitions. Default: 300000
        partition_assignment_strategy (list): List of objects to use to
            distribute partition ownership amongst consumer instances when
            group management is used. This preference is implicit in the order
            of the strategies in the list. When assignment strategy changes:
            to support a change to the assignment strategy, new versions must
            enable support both for the old assignment strategy and the new
            one. The coordinator will choose the old assignment strategy until
            all members have been updated. Then it will choose the new
            strategy. Default: [:class:`.RoundRobinPartitionAssignor`]

        max_poll_interval_ms (int): Maximum allowed time between calls to
            consume messages (e.g., :meth:`.getmany`). If this interval
            is exceeded the consumer is considered failed and the group will
            rebalance in order to reassign the partitions to another consumer
            group member. If API methods block waiting for messages, that time
            does not count against this timeout. See `KIP-62`_ for more
            information. Default 300000
        rebalance_timeout_ms (int): The maximum time server will wait for this
            consumer to rejoin the group in a case of rebalance. In Java client
            this behaviour is bound to `max.poll.interval.ms` configuration,
            but as ``aiokafka`` will rejoin the group in the background, we
            decouple this setting to allow finer tuning by users that use
            :class:`.ConsumerRebalanceListener` to delay rebalacing. Defaults
            to ``session_timeout_ms``
        session_timeout_ms (int): Client group session and failure detection
            timeout. The consumer sends periodic heartbeats
            (`heartbeat.interval.ms`) to indicate its liveness to the broker.
            If no hearts are received by the broker for a group member within
            the session timeout, the broker will remove the consumer from the
            group and trigger a rebalance. The allowed range is configured with
            the **broker** configuration properties
            `group.min.session.timeout.ms` and `group.max.session.timeout.ms`.
            Default: 10000
        heartbeat_interval_ms (int): The expected time in milliseconds
            between heartbeats to the consumer coordinator when using
            Kafka's group management feature. Heartbeats are used to ensure
            that the consumer's session stays active and to facilitate
            rebalancing when new consumers join or leave the group. The
            value must be set lower than `session_timeout_ms`, but typically
            should be set no higher than 1/3 of that value. It can be
            adjusted even lower to control the expected time for normal
            rebalances. Default: 3000

        consumer_timeout_ms (int): maximum wait timeout for background fetching
            routine. Mostly defines how fast the system will see rebalance and
            request new data for new partitions. Default: 200
        api_version (str): specify which kafka API version to use.
            :class:`AIOKafkaConsumer` supports Kafka API versions >=0.9 only.
            If set to ``auto``, will attempt to infer the broker version by
            probing various APIs. Default: ``auto``
        security_protocol (str): Protocol used to communicate with brokers.
            Valid values are: ``PLAINTEXT``, ``SSL``. Default: ``PLAINTEXT``.
        ssl_context (ssl.SSLContext): pre-configured :class:`~ssl.SSLContext`
            for wrapping socket connections. Directly passed into asyncio's
            :meth:`~asyncio.loop.create_connection`. For more information see
            :ref:`ssl_auth`. Default: None.
        exclude_internal_topics (bool): Whether records from internal topics
            (such as offsets) should be exposed to the consumer. If set to True
            the only way to receive records from an internal topic is
            subscribing to it. Requires 0.10+ Default: True
        connections_max_idle_ms (int): Close idle connections after the number
            of milliseconds specified by this config. Specifying `None` will
            disable idle checks. Default: 540000 (9 minutes).
        isolation_level (str): Controls how to read messages written
            transactionally.

            If set to ``read_committed``, :meth:`.getmany` will only return
            transactional messages which have been committed.
            If set to ``read_uncommitted`` (the default), :meth:`.getmany` will
            return all messages, even transactional messages which have been
            aborted.

            Non-transactional messages will be returned unconditionally in
            either mode.

            Messages will always be returned in offset order. Hence, in
            `read_committed` mode, :meth:`.getmany` will only return
            messages up to the last stable offset (LSO), which is the one less
            than the offset of the first open transaction. In particular any
            messages appearing after messages belonging to ongoing transactions
            will be withheld until the relevant transaction has been completed.
            As a result, `read_committed` consumers will not be able to read up
            to the high watermark when there are in flight transactions.
            Further, when in `read_committed` the seek_to_end method will
            return the LSO. See method docs below. Default: ``read_uncommitted``

        sasl_mechanism (str): Authentication mechanism when security_protocol
            is configured for ``SASL_PLAINTEXT`` or ``SASL_SSL``. Valid values are:
            ``PLAIN``, ``GSSAPI``, ``SCRAM-SHA-256``, ``SCRAM-SHA-512``,
            ``OAUTHBEARER``.
            Default: ``PLAIN``
        sasl_plain_username (str): username for SASL ``PLAIN`` authentication.
            Default: None
        sasl_plain_password (str): password for SASL ``PLAIN`` authentication.
            Default: None
        sasl_oauth_token_provider (~aiokafka.abc.AbstractTokenProvider):
            OAuthBearer token provider instance. (See :mod:`kafka.oauth.abstract`).
            Default: None

    Note:
        Many configuration parameters are taken from Java Client:
        https://kafka.apache.org/documentation.html#newconsumerconfigs

    """

    _closed = None  # Serves as an uninitialized flag for __del__
    _source_traceback = None

    def __init__(self, *topics, loop=None,
                 bootstrap_servers='localhost',
                 client_id='aiokafka-' + __version__,
                 group_id=None,
                 key_deserializer=None, value_deserializer=None,
                 fetch_max_wait_ms=500,
                 fetch_max_bytes=52428800,
                 fetch_min_bytes=1,
                 max_partition_fetch_bytes=1 * 1024 * 1024,
                 request_timeout_ms=40 * 1000,
                 retry_backoff_ms=100,
                 auto_offset_reset='latest',
                 enable_auto_commit=True,
                 auto_commit_interval_ms=5000,
                 check_crcs=True,
                 metadata_max_age_ms=5 * 60 * 1000,
                 partition_assignment_strategy=(RoundRobinPartitionAssignor,),
                 max_poll_interval_ms=300000,
                 rebalance_timeout_ms=None,
                 session_timeout_ms=10000,
                 heartbeat_interval_ms=3000,
                 consumer_timeout_ms=200,
                 max_poll_records=None,
                 ssl_context=None,
                 security_protocol='PLAINTEXT',
                 api_version='auto',
                 exclude_internal_topics=True,
                 connections_max_idle_ms=540000,
                 isolation_level="read_uncommitted",
                 sasl_mechanism="PLAIN",
                 sasl_plain_password=None,
                 sasl_plain_username=None,
                 sasl_kerberos_service_name='kafka',
                 sasl_kerberos_domain_name=None,
                 sasl_oauth_token_provider=None):
        if loop is None:
            loop = get_running_loop()
        else:
            warnings.warn("The loop argument is deprecated since 0.7.1, "
                          "and scheduled for removal in 0.8.0",
                          DeprecationWarning, stacklevel=2)

        if max_poll_records is not None and (
                not isinstance(max_poll_records, int) or max_poll_records < 1):
            raise ValueError("`max_poll_records` should be positive Integer")

        if rebalance_timeout_ms is None:
            rebalance_timeout_ms = session_timeout_ms

        self._client = AIOKafkaClient(
            loop=loop, bootstrap_servers=bootstrap_servers,
            client_id=client_id, metadata_max_age_ms=metadata_max_age_ms,
            request_timeout_ms=request_timeout_ms,
            retry_backoff_ms=retry_backoff_ms,
            api_version=api_version,
            ssl_context=ssl_context,
            security_protocol=security_protocol,
            connections_max_idle_ms=connections_max_idle_ms,
            sasl_mechanism=sasl_mechanism,
            sasl_plain_username=sasl_plain_username,
            sasl_plain_password=sasl_plain_password,
            sasl_kerberos_service_name=sasl_kerberos_service_name,
            sasl_kerberos_domain_name=sasl_kerberos_domain_name,
            sasl_oauth_token_provider=sasl_oauth_token_provider)

        self._group_id = group_id
        self._heartbeat_interval_ms = heartbeat_interval_ms
        self._session_timeout_ms = session_timeout_ms
        self._retry_backoff_ms = retry_backoff_ms
        self._auto_offset_reset = auto_offset_reset
        self._request_timeout_ms = request_timeout_ms
        self._enable_auto_commit = enable_auto_commit
        self._auto_commit_interval_ms = auto_commit_interval_ms
        self._partition_assignment_strategy = partition_assignment_strategy
        self._key_deserializer = key_deserializer
        self._value_deserializer = value_deserializer
        self._fetch_min_bytes = fetch_min_bytes
        self._fetch_max_bytes = fetch_max_bytes
        self._fetch_max_wait_ms = fetch_max_wait_ms
        self._max_partition_fetch_bytes = max_partition_fetch_bytes
        self._exclude_internal_topics = exclude_internal_topics
        self._max_poll_records = max_poll_records
        self._consumer_timeout = consumer_timeout_ms / 1000
        self._isolation_level = isolation_level
        self._rebalance_timeout_ms = rebalance_timeout_ms
        self._max_poll_interval_ms = max_poll_interval_ms

        self._check_crcs = check_crcs
        self._subscription = SubscriptionState(loop=loop)
        self._fetcher = None
        self._coordinator = None
        self._loop = loop

        if loop.get_debug():
            self._source_traceback = traceback.extract_stack(sys._getframe(1))
        self._closed = False

        if topics:
            topics = self._validate_topics(topics)
            self._client.set_topics(topics)
            self._subscription.subscribe(topics=topics)

    def __del__(self, _warnings=warnings):
        if self._closed is False:
            _warnings.warn(f"Unclosed AIOKafkaConsumer {self!r}",
                           ResourceWarning,
                           source=self)
            context = {'consumer': self,
                       'message': 'Unclosed AIOKafkaConsumer'}
            if self._source_traceback is not None:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)

    async def start(self):
        """ Connect to Kafka cluster. This will:

            * Load metadata for all cluster nodes and partition allocation
            * Wait for possible topic autocreation
            * Join group if ``group_id`` provided
        """
        assert self._loop is get_running_loop(), (
            "Please create objects with the same loop as running with"
        )
        assert self._fetcher is None, "Did you call `start` twice?"
        await self._client.bootstrap()
        await self._wait_topics()

        if self._client.api_version < (0, 9):
            raise ValueError(
                f"Unsupported Kafka version: {self._client.api_version}"
            )

        if self._isolation_level == "read_committed" and \
                self._client.api_version < (0, 11):
            raise UnsupportedVersionError(
                "`read_committed` isolation_level available only for Brokers "
                "0.11 and above")

        self._fetcher = Fetcher(
            self._client, self._subscription,
            key_deserializer=self._key_deserializer,
            value_deserializer=self._value_deserializer,
            fetch_min_bytes=self._fetch_min_bytes,
            fetch_max_bytes=self._fetch_max_bytes,
            fetch_max_wait_ms=self._fetch_max_wait_ms,
            max_partition_fetch_bytes=self._max_partition_fetch_bytes,
            check_crcs=self._check_crcs,
            fetcher_timeout=self._consumer_timeout,
            retry_backoff_ms=self._retry_backoff_ms,
            auto_offset_reset=self._auto_offset_reset,
            isolation_level=self._isolation_level)

        if self._group_id is not None:
            # using group coordinator for automatic partitions assignment
            self._coordinator = GroupCoordinator(
                self._client, self._subscription,
                group_id=self._group_id,
                heartbeat_interval_ms=self._heartbeat_interval_ms,
                session_timeout_ms=self._session_timeout_ms,
                retry_backoff_ms=self._retry_backoff_ms,
                enable_auto_commit=self._enable_auto_commit,
                auto_commit_interval_ms=self._auto_commit_interval_ms,
                assignors=self._partition_assignment_strategy,
                exclude_internal_topics=self._exclude_internal_topics,
                rebalance_timeout_ms=self._rebalance_timeout_ms,
                max_poll_interval_ms=self._max_poll_interval_ms
            )
            if self._subscription.subscription is not None:
                if self._subscription.partitions_auto_assigned():
                    # Either we passed `topics` to constructor or `subscribe`
                    # was called before `start`
                    await self._subscription.wait_for_assignment()
                else:
                    # `assign` was called before `start`. We did not start
                    # this task on that call, as coordinator was yet to be
                    # created
                    self._coordinator.start_commit_offsets_refresh_task(
                        self._subscription.subscription.assignment)
        else:
            # Using a simple assignment coordinator for reassignment on
            # metadata changes
            self._coordinator = NoGroupCoordinator(
                self._client, self._subscription,
                exclude_internal_topics=self._exclude_internal_topics)

            if self._subscription.subscription is not None:
                if self._subscription.partitions_auto_assigned():
                    # Either we passed `topics` to constructor or `subscribe`
                    # was called before `start`
                    await self._client.force_metadata_update()
                    self._coordinator.assign_all_partitions(check_unknown=True)

    async def _wait_topics(self):
        if self._subscription.subscription is not None:
            for topic in self._subscription.subscription.topics:
                await self._client._wait_on_metadata(topic)

    def _validate_topics(self, topics):
        if not isinstance(topics, (tuple, set, list)):
            raise ValueError("Topics should be list of strings")
        return set(topics)

    def assign(self, partitions):
        """Manually assign a list of :class:`.TopicPartition` to this consumer.

        This interface does not support incremental assignment and will
        replace the previous assignment (if there was one).

        Arguments:
            partitions (list(TopicPartition)): assignment for this instance.

        Raises:
            IllegalStateError: if consumer has already called :meth:`subscribe`

        Warning:
            It is not possible to use both manual partition assignment with
            :meth:`assign` and group assignment with :meth:`subscribe`.

        Note:
            Manual topic assignment through this method does not use the
            consumer's group management functionality. As such, there will be
            **no rebalance operation triggered** when group membership or
            cluster and topic metadata change.
        """
        self._subscription.assign_from_user(partitions)
        self._client.set_topics([tp.topic for tp in partitions])

        # If called before `start` we will delegate this to `start` call
        if self._coordinator is not None:
            if self._group_id is not None:
                # refresh commit positions for all assigned partitions
                assignment = self._subscription.subscription.assignment
                self._coordinator.start_commit_offsets_refresh_task(assignment)

    def assignment(self):
        """ Get the set of partitions currently assigned to this consumer.

        If partitions were directly assigned using :meth:`assign`, then this will
        simply return the same partitions that were previously assigned.

        If topics were subscribed using :meth:`subscribe`, then this will give
        the set of topic partitions currently assigned to the consumer (which
        may be empty if the assignment hasn't happened yet or if the partitions
        are in the process of being reassigned).

        Returns:
            set(TopicPartition): the set of partitions currently assigned to
            this consumer
        """
        return self._subscription.assigned_partitions()

    async def stop(self):
        """ Close the consumer, while waiting for finalizers:

            * Commit last consumed message if autocommit enabled
            * Leave group if used Consumer Groups
        """
        if self._closed:
            return
        log.debug("Closing the KafkaConsumer.")
        self._closed = True
        if self._coordinator:
            await self._coordinator.close()
        if self._fetcher:
            await self._fetcher.close()
        await self._client.close()
        log.debug("The KafkaConsumer has closed.")

    async def commit(self, offsets=None):
        """ Commit offsets to Kafka.

        This commits offsets only to Kafka. The offsets committed using this
        API will be used on the first fetch after every rebalance and also on
        startup. As such, if you need to store offsets in anything other than
        Kafka, this API should not be used.

        Currently only supports kafka-topic offset storage (not Zookeeper)

        When explicitly passing `offsets` use either offset of next record,
        or tuple of offset and metadata::

            tp = TopicPartition(msg.topic, msg.partition)
            metadata = "Some utf-8 metadata"
            # Either
            await consumer.commit({tp: msg.offset + 1})
            # Or position directly
            await consumer.commit({tp: (msg.offset + 1, metadata)})

        .. note:: If you want *fire and forget* commit, like
            :meth:`~kafka.KafkaConsumer.commit_async` in `kafka-python`_, just
            run it in a task. Something like::

                fut = loop.create_task(consumer.commit())
                fut.add_done_callback(on_commit_done)

        Arguments:
            offsets (dict, Optional): A mapping from :class:`.TopicPartition` to
              ``(offset, metadata)`` to commit with the configured ``group_id``.
              Defaults to current consumed offsets for all subscribed partitions.
        Raises:
            ~aiokafka.errors.CommitFailedError: If membership already changed on broker.
            ~aiokafka.errors.IllegalOperation: If used with ``group_id == None``.
            ~aiokafka.errors.IllegalStateError: If partitions not assigned.
            ~aiokafka.errors.KafkaError: If commit failed on broker side. This
                could be due to invalid offset, too long metadata, authorization
                failure, etc.
            ValueError: If offsets is of wrong format.

        .. versionchanged:: 0.4.0

            Changed :exc:`AssertionError` to
            :exc:`~aiokafka.errors.IllegalStateError` in case of unassigned
            partition.

        .. versionchanged:: 0.4.0

            Will now raise :exc:`~aiokafka.errors.CommitFailedError` in case
            membership changed, as (possibly) this partition is handled by
            another consumer.
        """
        if self._group_id is None:
            raise IllegalOperation("Requires group_id")

        subscription = self._subscription.subscription
        if subscription is None:
            raise IllegalStateError("Not subscribed to any topics")
        assignment = subscription.assignment
        if assignment is None:
            raise IllegalStateError("No partitions assigned")

        if offsets is None:
            offsets = assignment.all_consumed_offsets()
        else:
            offsets = commit_structure_validate(offsets)
            for tp in offsets:
                if tp not in assignment.tps:
                    raise IllegalStateError(
                        f"Partition {tp} is not assigned")

        await self._coordinator.commit_offsets(assignment, offsets)

    async def committed(self, partition):
        """ Get the last committed offset for the given partition. (whether the
        commit happened by this process or another).

        This offset will be used as the position for the consumer in the event
        of a failure.

        This call will block to do a remote call to get the latest offset, as
        those are not cached by consumer (Transactional Producer can change
        them without Consumer knowledge as of Kafka 0.11.0)

        Arguments:
            partition (TopicPartition): the partition to check

        Returns:
            The last committed offset, or None if there was no prior commit.

        Raises:
            IllegalOperation: If used with ``group_id == None``
        """
        if self._group_id is None:
            raise IllegalOperation("Requires group_id")

        commit_map = await self._coordinator.fetch_committed_offsets(
            [partition])
        if partition in commit_map:
            committed = commit_map[partition].offset
            if committed == -1:
                committed = None
        else:
            committed = None
        return committed

    async def topics(self):
        """ Get all topics the user is authorized to view.

        Returns:
            set: topics
        """
        cluster = await self._client.fetch_all_metadata()
        return cluster.topics()

    def partitions_for_topic(self, topic):
        """ Get metadata about the partitions for a given topic.

        This method will return `None` if Consumer does not already have
        metadata for this topic.

        Arguments:
            topic (str): topic to check

        Returns:
            set: partition ids
        """
        return self._client.cluster.partitions_for_topic(topic)

    async def position(self, partition):
        """ Get the offset of the *next record* that will be fetched (if a
        record with that offset exists on broker).

        Arguments:
            partition (TopicPartition): partition to check

        Returns:
            int: offset

        Raises:
            IllegalStateError: partition is not assigned

        .. versionchanged:: 0.4.0

            Changed :exc:`AssertionError` to
            :exc:`~aiokafka.errors.IllegalStateError` in case of unassigned
            partition
        """
        while True:
            if not self._subscription.is_assigned(partition):
                raise IllegalStateError(
                    f'Partition {partition} is not assigned')

            assignment = self._subscription.subscription.assignment
            tp_state = assignment.state_value(partition)
            if not tp_state.has_valid_position:
                self._coordinator.check_errors()
                await asyncio.wait(
                    [tp_state.wait_for_position(),
                     assignment.unassign_future],
                    timeout=self._request_timeout_ms / 1000,
                    return_when=asyncio.FIRST_COMPLETED
                )
                if not tp_state.has_valid_position:
                    if self._subscription.subscription is None:
                        raise IllegalStateError(
                            f'Partition {partition} is not assigned')
                    if self._subscription.subscription.assignment is None:
                        self._coordinator.check_errors()
                        await self._subscription.wait_for_assignment()
                    continue
            return tp_state.position

    def highwater(self, partition):
        """ Last known highwater offset for a partition.

        A highwater offset is the offset that will be assigned to the next
        message that is produced. It may be useful for calculating lag, by
        comparing with the reported position. Note that both position and
        highwater refer to the *next* offset – i.e., highwater offset is one
        greater than the newest available message.

        Highwater offsets are returned as part of ``FetchResponse``, so will
        not be available if messages for this partition were not requested yet.

        Arguments:
            partition (TopicPartition): partition to check

        Returns:
            int or None: offset if available
        """
        assert self._subscription.is_assigned(partition), \
            'Partition is not assigned'
        assignment = self._subscription.subscription.assignment
        return assignment.state_value(partition).highwater

    def last_stable_offset(self, partition):
        """ Returns the Last Stable Offset of a topic. It will be the last
        offset up to which point all transactions were completed. Only
        available in with isolation_level `read_committed`, in
        `read_uncommitted` will always return -1. Will return None for older
        Brokers.

        As with :meth:`highwater` will not be available until some messages are
        consumed.

        Arguments:
            partition (TopicPartition): partition to check

        Returns:
            int or None: offset if available
        """
        assert self._subscription.is_assigned(partition), \
            'Partition is not assigned'
        assignment = self._subscription.subscription.assignment
        return assignment.state_value(partition).lso

    def last_poll_timestamp(self, partition):
        """ Returns the timestamp of the last poll of this partition (in ms).
        It is the last time :meth:`highwater` and :meth:`last_stable_offset` were
        updated. However it does not mean that new messages were received.

        As with :meth:`highwater` will not be available until some messages are
        consumed.

        Arguments:
            partition (TopicPartition): partition to check

        Returns:
            int or None: timestamp if available
        """
        assert self._subscription.is_assigned(partition), \
            'Partition is not assigned'
        assignment = self._subscription.subscription.assignment
        return assignment.state_value(partition).timestamp

    def seek(self, partition, offset):
        """ Manually specify the fetch offset for a :class:`.TopicPartition`.

        Overrides the fetch offsets that the consumer will use on the next
        :meth:`getmany`/:meth:`getone` call. If this API is invoked for the same
        partition more than once, the latest offset will be used on the next
        fetch.

        Note:
            You may lose data if this API is arbitrarily used in the middle
            of consumption to reset the fetch offsets. Use it either on
            rebalance listeners or after all pending messages are processed.

        Arguments:
            partition (TopicPartition): partition for seek operation
            offset (int): message offset in partition

        Raises:
            ValueError: if offset is not a positive integer
            IllegalStateError: partition is not currently assigned

        .. versionchanged:: 0.4.0

            Changed :exc:`AssertionError` to
            :exc:`~aiokafka.errors.IllegalStateError` and :exc:`ValueError` in
            respective cases.
        """
        if not isinstance(offset, int) or offset < 0:
            raise ValueError("Offset must be a positive integer")
        log.debug("Seeking to offset %s for partition %s", offset, partition)
        self._fetcher.seek_to(partition, offset)

    async def seek_to_beginning(self, *partitions):
        """ Seek to the oldest available offset for partitions.

        Arguments:
            *partitions: Optionally provide specific :class:`.TopicPartition`,
                otherwise default to all assigned partitions.

        Raises:
            IllegalStateError: If any partition is not currently assigned
            TypeError: If partitions are not instances of :class:`.TopicPartition`

        .. versionadded:: 0.3.0

        """
        if not all([isinstance(p, TopicPartition) for p in partitions]):
            raise TypeError('partitions must be TopicPartition instances')

        if not partitions:
            partitions = self._subscription.assigned_partitions()
            assert partitions, 'No partitions are currently assigned'
        else:
            not_assigned = (
                set(partitions) - self._subscription.assigned_partitions()
            )
            if not_assigned:
                raise IllegalStateError(
                    f"Partitions {not_assigned} are not assigned")

        for tp in partitions:
            log.debug("Seeking to beginning of partition %s", tp)

        fut = self._fetcher.request_offset_reset(
            partitions, OffsetResetStrategy.EARLIEST)
        assignment = self._subscription.subscription.assignment
        await asyncio.wait(
            [fut, assignment.unassign_future],
            timeout=self._request_timeout_ms / 1000,
            return_when=asyncio.FIRST_COMPLETED
        )
        self._coordinator.check_errors()
        return fut.done()

    async def seek_to_end(self, *partitions):
        """Seek to the most recent available offset for partitions.

        Arguments:
            *partitions: Optionally provide specific :class:`.TopicPartition`,
                otherwise default to all assigned partitions.

        Raises:
            IllegalStateError: If any partition is not currently assigned
            TypeError: If partitions are not instances of :class:`.TopicPartition`

        .. versionadded:: 0.3.0

        """
        if not all([isinstance(p, TopicPartition) for p in partitions]):
            raise TypeError('partitions must be TopicPartition instances')

        if not partitions:
            partitions = self._subscription.assigned_partitions()
            assert partitions, 'No partitions are currently assigned'
        else:
            not_assigned = (
                set(partitions) - self._subscription.assigned_partitions()
            )
            if not_assigned:
                raise IllegalStateError(
                    f"Partitions {not_assigned} are not assigned")

        for tp in partitions:
            log.debug("Seeking to end of partition %s", tp)
        fut = self._fetcher.request_offset_reset(
            partitions, OffsetResetStrategy.LATEST)
        assignment = self._subscription.subscription.assignment
        await asyncio.wait(
            [fut, assignment.unassign_future],
            timeout=self._request_timeout_ms / 1000,
            return_when=asyncio.FIRST_COMPLETED
        )
        self._coordinator.check_errors()
        return fut.done()

    async def seek_to_committed(self, *partitions):
        """ Seek to the committed offset for partitions.

        Arguments:
            *partitions: Optionally provide specific :class:`.TopicPartition`,
                otherwise default to all assigned partitions.

        Returns:
            dict(TopicPartition, int): mapping
            of the currently committed offsets.

        Raises:
            IllegalStateError: If any partition is not currently assigned
            IllegalOperation: If used with ``group_id == None``

        .. versionchanged:: 0.3.0

            Changed :exc:`AssertionError` to
            :exc:`~aiokafka.errors.IllegalStateError` in case of unassigned
            partition
        """
        if not all([isinstance(p, TopicPartition) for p in partitions]):
            raise TypeError('partitions must be TopicPartition instances')

        if not partitions:
            partitions = self._subscription.assigned_partitions()
            assert partitions, 'No partitions are currently assigned'
        else:
            not_assigned = (
                set(partitions) - self._subscription.assigned_partitions()
            )
            if not_assigned:
                raise IllegalStateError(
                    f"Partitions {not_assigned} are not assigned")

        committed_offsets = {}
        for tp in partitions:
            offset = await self.committed(tp)
            committed_offsets[tp] = offset
            log.debug("Seeking to committed of partition %s %s", tp, offset)
            if offset and offset > 0:
                self._fetcher.seek_to(tp, offset)
        return committed_offsets

    async def offsets_for_times(self, timestamps):
        """
        Look up the offsets for the given partitions by timestamp. The returned
        offset for each partition is the earliest offset whose timestamp is
        greater than or equal to the given timestamp in the corresponding
        partition.

        The consumer does not have to be assigned the partitions.

        If the message format version in a partition is before 0.10.0, i.e.
        the messages do not have timestamps, ``None`` will be returned for that
        partition.

        Note:
            This method may block indefinitely if the partition does not exist.

        Arguments:
            timestamps (dict(TopicPartition, int)): mapping from partition
                to the timestamp to look up. Unit should be milliseconds since
                beginning of the epoch (midnight Jan 1, 1970 (UTC))

        Returns:
            dict(TopicPartition, OffsetAndTimestamp): mapping from
            partition to the timestamp and offset of the first message with
            timestamp greater than or equal to the target timestamp.

        Raises:
            ValueError: If the target timestamp is negative
            UnsupportedVersionError: If the broker does not support looking
                up the offsets by timestamp.
            KafkaTimeoutError: If fetch failed in `request_timeout_ms`

        .. versionadded:: 0.3.0

        """
        if self._client.api_version <= (0, 10, 0):
            raise UnsupportedVersionError(
                "offsets_for_times API not supported"
                f" for cluster version {self._client.api_version}"
            )
        for tp, ts in timestamps.items():
            timestamps[tp] = int(ts)
            if ts < 0:
                raise ValueError(
                    f"The target time for partition {tp} is {ts}."
                    " The target time cannot be negative."
                )
        offsets = await self._fetcher.get_offsets_by_times(
            timestamps, self._request_timeout_ms)
        return offsets

    async def beginning_offsets(self, partitions):
        """ Get the first offset for the given partitions.

        This method does not change the current consumer position of the
        partitions.

        Note:
            This method may block indefinitely if the partition does not exist.

        Arguments:
            partitions (list[TopicPartition]): List of :class:`.TopicPartition`
                instances to fetch offsets for.

        Returns:
            dict [TopicPartition, int]: mapping of partition to  earliest
            available offset.

        Raises:
            UnsupportedVersionError: If the broker does not support looking
                up the offsets by timestamp.
            KafkaTimeoutError: If fetch failed in `request_timeout_ms`.

        .. versionadded:: 0.3.0

        """
        if self._client.api_version <= (0, 10, 0):
            raise UnsupportedVersionError(
                "offsets_for_times API not supported"
                f" for cluster version {self._client.api_version}"
            )
        offsets = await self._fetcher.beginning_offsets(
            partitions, self._request_timeout_ms)
        return offsets

    async def end_offsets(self, partitions):
        """ Get the last offset for the given partitions. The last offset of a
        partition is the offset of the upcoming message, i.e. the offset of the
        last available message + 1.

        This method does not change the current consumer position of the
        partitions.

        Note:
            This method may block indefinitely if the partition does not exist.

        Arguments:
            partitions (list[TopicPartition]): List of :class:`.TopicPartition`
                instances to fetch offsets for.

        Returns:
            dict [TopicPartition, int]: mapping of partition to last
            available offset + 1.

        Raises:
            UnsupportedVersionError: If the broker does not support looking
                up the offsets by timestamp.
            KafkaTimeoutError: If fetch failed in ``request_timeout_ms``

        .. versionadded:: 0.3.0

        """
        if self._client.api_version <= (0, 10, 0):
            raise UnsupportedVersionError(
                "offsets_for_times API not supported"
                f" for cluster version {self._client.api_version}"
            )
        offsets = await self._fetcher.end_offsets(
            partitions, self._request_timeout_ms)
        return offsets

    def subscribe(self, topics=(), pattern=None, listener=None):
        """ Subscribe to a list of topics, or a topic regex pattern.

        Partitions will be dynamically assigned via a group coordinator.
        Topic subscriptions are not incremental: this list will replace the
        current assignment (if there is one).

        This method is incompatible with :meth:`assign`.

        Arguments:
           topics (list): List of topics for subscription.
           pattern (str): Pattern to match available topics. You must provide
               either topics or pattern, but not both.
           listener (ConsumerRebalanceListener): Optionally include listener
               callback, which will be called before and after each rebalance
               operation.
               As part of group management, the consumer will keep track of
               the list of consumers that belong to a particular group and
               will trigger a rebalance operation if one of the following
               events trigger:

               * Number of partitions change for any of the subscribed topics
               * Topic is created or deleted
               * An existing member of the consumer group dies
               * A new member is added to the consumer group

               When any of these events are triggered, the provided listener
               will be invoked first to indicate that the consumer's
               assignment has been revoked, and then again when the new
               assignment has been received. Note that this listener will
               immediately override any listener set in a previous call
               to subscribe. It is guaranteed, however, that the partitions
               revoked/assigned
               through this interface are from topics subscribed in this call.
        Raises:
            IllegalStateError: if called after previously calling :meth:`assign`
            ValueError: if neither topics or pattern is provided or both
               are provided
            TypeError: if listener is not a :class:`.ConsumerRebalanceListener`
        """
        if not (topics or pattern):
            raise ValueError(
                "You should provide either `topics` or `pattern`")
        if topics and pattern:
            raise ValueError(
                "You can't provide both `topics` and `pattern`")
        if listener is not None and \
                not isinstance(listener, ConsumerRebalanceListener):
            raise TypeError(
                "listener should be an instance of ConsumerRebalanceListener")
        if pattern is not None:
            try:
                pattern = re.compile(pattern)
            except re.error as err:
                raise ValueError(
                    f"{pattern!r} is not a valid pattern: {err}")
            self._subscription.subscribe_pattern(
                pattern=pattern, listener=listener)
            # NOTE: set_topics will trigger a rebalance, so the coordinator
            # will get the initial subscription shortly by ``metadata_changed``
            # handler.
            self._client.set_topics([])
            log.info("Subscribed to topic pattern: %s", pattern)
        elif topics:
            topics = self._validate_topics(topics)
            self._subscription.subscribe(
                topics=topics, listener=listener)
            self._client.set_topics(self._subscription.subscription.topics)
            if self._group_id is None:
                # We have reset the assignment, but client.set_topics will
                # not always do a metadata update. We force it to do it even
                # if metadata did not change. This will trigger a reassignment
                # on NoGroupCoordinator, but only if snapshot did not change,
                # thus we reset it too.
                self._client.force_metadata_update()
                if self._coordinator is not None:
                    self._coordinator._metadata_snapshot = {}
            log.info("Subscribed to topic(s): %s", topics)

    def subscription(self):
        """ Get the current topics subscription.

        Returns:
            frozenset(str): a set of topics
        """
        return self._subscription.topics

    def unsubscribe(self):
        """ Unsubscribe from all topics and clear all assigned partitions. """
        self._subscription.unsubscribe()
        if self._group_id is not None:
            self._coordinator.maybe_leave_group()
        self._client.set_topics([])
        log.info(
            "Unsubscribed all topics or patterns and assigned partitions")

    async def getone(self, *partitions):
        """
        Get one message from Kafka.
        If no new messages prefetched, this method will wait for it.

        Arguments:
            partitions (list(TopicPartition)): Optional list of partitions to
                return from. If no partitions specified then returned message
                will be from any partition, which consumer is subscribed to.

        Returns:
            ~aiokafka.structs.ConsumerRecord: the message

        Will return instance of

        .. code:: python

            collections.namedtuple(
                "ConsumerRecord",
                ["topic", "partition", "offset", "key", "value"])

        Example usage:

        .. code:: python

            while True:
                message = await consumer.getone()
                topic = message.topic
                partition = message.partition
                # Process message
                print(message.offset, message.key, message.value)

        """
        assert all(map(lambda k: isinstance(k, TopicPartition), partitions))
        if self._closed:
            raise ConsumerStoppedError()

        # Raise coordination errors if any
        self._coordinator.check_errors()

        with self._subscription.fetch_context():
            msg = await self._fetcher.next_record(partitions)
        return msg

    async def getmany(self, *partitions, timeout_ms=0, max_records=None):
        """Get messages from assigned topics / partitions.

        Prefetched messages are returned in batches by topic-partition.
        If messages is not available in the prefetched buffer this method waits
        `timeout_ms` milliseconds.

        Arguments:
            partitions (list[TopicPartition]): The partitions that need
                fetching message. If no one partition specified then all
                subscribed partitions will be used
            timeout_ms (int, Optional): milliseconds spent waiting if
                data is not available in the buffer. If 0, returns immediately
                with any records that are available currently in the buffer,
                else returns empty. Must not be negative. Default: 0
        Returns:
            dict(TopicPartition, list[ConsumerRecord]): topic to list of
            records since the last fetch for the subscribed list of topics and
            partitions

        Example usage:


        .. code:: python

            data = await consumer.getmany()
            for tp, messages in data.items():
                topic = tp.topic
                partition = tp.partition
                for message in messages:
                    # Process message
                    print(message.offset, message.key, message.value)

        """
        assert all(map(lambda k: isinstance(k, TopicPartition), partitions))
        if self._closed:
            raise ConsumerStoppedError()

        if max_records is not None and (
                not isinstance(max_records, int) or max_records < 1):
            raise ValueError("`max_records` must be a positive Integer")

        # Raise coordination errors if any
        self._coordinator.check_errors()

        timeout = timeout_ms / 1000
        with self._subscription.fetch_context():
            records = await self._fetcher.fetched_records(
                partitions, timeout,
                max_records=max_records or self._max_poll_records)
        return records

    def pause(self, *partitions):
        """Suspend fetching from the requested partitions.

        Future calls to :meth:`.getmany` will not return any records from these
        partitions until they have been resumed using :meth:`.resume`.

        Note: This method does not affect partition subscription.
        In particular, it does not cause a group rebalance when automatic
        assignment is used.

        Arguments:
            *partitions (list[TopicPartition]): Partitions to pause.
        """
        if not all([isinstance(p, TopicPartition) for p in partitions]):
            raise TypeError('partitions must be TopicPartition namedtuples')

        for partition in partitions:
            log.debug("Pausing partition %s", partition)
            self._subscription.pause(partition)

    def paused(self):
        """Get the partitions that were previously paused using
        :meth:`.pause`.

        Returns:
            set[TopicPartition]: partitions
        """
        return self._subscription.paused_partitions()

    def resume(self, *partitions):
        """Resume fetching from the specified (paused) partitions.

        Arguments:
            *partitions (list[TopicPartition]): Partitions to resume.
        """
        if not all([isinstance(p, TopicPartition) for p in partitions]):
            raise TypeError('partitions must be TopicPartition namedtuples')

        for partition in partitions:
            log.debug("Resuming partition %s", partition)
            self._subscription.resume(partition)

    def __aiter__(self):
        if self._closed:
            raise ConsumerStoppedError()
        return self

    async def __anext__(self):
        """Asyncio iterator interface for consumer

        Note:
            TopicAuthorizationFailedError and OffsetOutOfRangeError
            exceptions can be raised in iterator.
            All other KafkaError exceptions will be logged and not raised
        """
        while True:
            try:
                return (await self.getone())
            except ConsumerStoppedError:
                raise StopAsyncIteration  # noqa: F821
            except (TopicAuthorizationFailedError,
                    OffsetOutOfRangeError,
                    NoOffsetForPartitionError) as err:
                raise err
            except RecordTooLargeError:
                log.exception("error in consumer iterator: %s")

    async def __aenter__(self):
        await self.start()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.stop() UNKNOWN_OFFSET = -1
READ_UNCOMMITTED = 0
READ_COMMITTED = 1
    LATEST = -1
    EARLIEST = -2
    NONE = 0
class OffsetResetStrategy:
class FetchResult:
class FetchError:
class PartitionRecords:
class Fetcher:
    def from_str(cls, name):
    def to_str(cls, value):
    def __init__(
            self, tp, *, assignment, partition_records, backoff):
    def calculate_backoff(self):
    def check_assignment(self, tp):
    def _update_position(self):
    def getone(self):
    def getall(self, max_records=None):
    def has_more(self):
    def __repr__(self):
    def __init__(self, *, error, backoff):
    def calculate_backoff(self):
    def check_raise(self):
    def __repr__(self):
    def __init__(
            self, tp, records, aborted_transactions, fetch_offset,
            key_deserializer, value_deserializer, check_crcs, isolation_level):
    def __iter__(self):
    def __next__(self):
    def _unpack_records(self):
    def _consume_aborted_up_to(self, batch_offset):
    def _contains_abort_marker(self, next_batch):
    def _consumer_record(self, tp, record):
    def __init__(
            self, client, subscriptions, *,
            key_deserializer=None,
            value_deserializer=None,
            fetch_min_bytes=1,
            fetch_max_bytes=52428800,
            fetch_max_wait_ms=500,
            max_partition_fetch_bytes=1048576,
            check_crcs=True,
            fetcher_timeout=0.2,
            prefetch_backoff=0.1,
            retry_backoff_ms=100,
            auto_offset_reset='latest',
            isolation_level="read_uncommitted"):
    async def close(self):
    def _notify(self, future):
    def _create_fetch_waiter(self):
    def error_future(self):
    async def _fetch_requests_routine(self):
            def start_pending_task(coro, node_id, self=self):
                def on_done(fut, self=self):
    def _get_actions_per_node(self, assignment):
    async def _proc_fetch_request(self, assignment, node_id, request):
    def _set_error(self, tp, error):
    async def _update_fetch_positions(self, assignment, node_id, tps):
    async def _retrieve_offsets(self, timestamps, timeout_ms=None):
    async def _proc_offset_requests(self, timestamps):
    async def _proc_offset_request(self, node_id, topic_data):
    async def next_record(self, partitions):
    async def fetched_records(self, partitions, timeout=0, max_records=None):
    async def get_offsets_by_times(self, timestamps, timeout_ms):
    async def beginning_offsets(self, partitions, timeout_ms):
    async def end_offsets(self, partitions, timeout_ms):
    def request_offset_reset(self, tps, strategy):
    def seek_to(self, tp, offset): UNKNOWN_OFFSET = -1
READ_UNCOMMITTED = 0
READ_COMMITTED = 1
    LATEST = -1
    EARLIEST = -2
    NONE = 0
class OffsetResetStrategy:
class FetchResult:
class FetchError:
class PartitionRecords:
class Fetcher:
    def from_str(cls, name):
    def to_str(cls, value):
    def __init__(
            self, tp, *, assignment, partition_records, backoff):
    def calculate_backoff(self):
    def check_assignment(self, tp):
    def _update_position(self):
    def getone(self):
    def getall(self, max_records=None):
    def has_more(self):
    def __repr__(self):
    def __init__(self, *, error, backoff):
    def calculate_backoff(self):
    def check_raise(self):
    def __repr__(self):
    def __init__(
            self, tp, records, aborted_transactions, fetch_offset,
            key_deserializer, value_deserializer, check_crcs, isolation_level):
    def __iter__(self):
    def __next__(self):
    def _unpack_records(self):
    def _consume_aborted_up_to(self, batch_offset):
    def _contains_abort_marker(self, next_batch):
    def _consumer_record(self, tp, record):
    def __init__(
            self, client, subscriptions, *,
            key_deserializer=None,
            value_deserializer=None,
            fetch_min_bytes=1,
            fetch_max_bytes=52428800,
            fetch_max_wait_ms=500,
            max_partition_fetch_bytes=1048576,
            check_crcs=True,
            fetcher_timeout=0.2,
            prefetch_backoff=0.1,
            retry_backoff_ms=100,
            auto_offset_reset='latest',
            isolation_level="read_uncommitted"):
    async def close(self):
    def _notify(self, future):
    def _create_fetch_waiter(self):
    def error_future(self):
    async def _fetch_requests_routine(self):
            def start_pending_task(coro, node_id, self=self):
                def on_done(fut, self=self):
    def _get_actions_per_node(self, assignment):
    async def _proc_fetch_request(self, assignment, node_id, request):
    def _set_error(self, tp, error):
    async def _update_fetch_positions(self, assignment, node_id, tps):
    async def _retrieve_offsets(self, timestamps, timeout_ms=None):
    async def _proc_offset_requests(self, timestamps):
    async def _proc_offset_request(self, node_id, topic_data):
    async def next_record(self, partitions):
    async def fetched_records(self, partitions, timeout=0, max_records=None):
    async def get_offsets_by_times(self, timestamps, timeout_ms):
    async def beginning_offsets(self, partitions, timeout_ms):
    async def end_offsets(self, partitions, timeout_ms):
    def request_offset_reset(self, tps, strategy):
    def seek_to(self, tp, offset): class AIOKafkaProducer:
    """A Kafka client that publishes records to the Kafka cluster.

    The producer consists of a pool of buffer space that holds records that
    haven't yet been transmitted to the server as well as a background task
    that is responsible for turning these records into requests and
    transmitting them to the cluster.

    The :meth:`send` method is asynchronous. When called it adds the record to a
    buffer of pending record sends and immediately returns. This allows the
    producer to batch together individual records for efficiency.

    The `acks` config controls the criteria under which requests are considered
    complete. The ``all`` setting will result in waiting for all replicas to
    respond, the slowest but most durable setting.

    The `key_serializer` and `value_serializer` instruct how to turn the key and
    value objects the user provides into :class:`bytes`.

    Arguments:
        bootstrap_servers (str, list(str)): a ``host[:port]`` string or list of
            ``host[:port]`` strings that the producer should contact to
            bootstrap initial cluster metadata. This does not have to be the
            full node list.  It just needs to have at least one broker that will
            respond to a Metadata API Request. Default port is 9092. If no
            servers are specified, will default to ``localhost:9092``.
        client_id (str): a name for this client. This string is passed in
            each request to servers and can be used to identify specific
            server-side log entries that correspond to this client.
            Default: ``aiokafka-producer-#`` (appended with a unique number
            per instance)
        key_serializer (Callable): used to convert user-supplied keys to bytes
            If not :data:`None`, called as ``f(key),`` should return
            :class:`bytes`.
            Default: :data:`None`.
        value_serializer (Callable): used to convert user-supplied message
            values to :class:`bytes`. If not :data:`None`, called as
            ``f(value)``, should return :class:`bytes`.
            Default: :data:`None`.
        acks (Any): one of ``0``, ``1``, ``all``. The number of acknowledgments
            the producer requires the leader to have received before considering a
            request complete. This controls the durability of records that are
            sent. The following settings are common:

            * ``0``: Producer will not wait for any acknowledgment from the server
              at all. The message will immediately be added to the socket
              buffer and considered sent. No guarantee can be made that the
              server has received the record in this case, and the retries
              configuration will not take effect (as the client won't
              generally know of any failures). The offset given back for each
              record will always be set to -1.
            * ``1``: The broker leader will write the record to its local log but
              will respond without awaiting full acknowledgement from all
              followers. In this case should the leader fail immediately
              after acknowledging the record but before the followers have
              replicated it then the record will be lost.
            * ``all``: The broker leader will wait for the full set of in-sync
              replicas to acknowledge the record. This guarantees that the
              record will not be lost as long as at least one in-sync replica
              remains alive. This is the strongest available guarantee.

            If unset, defaults to ``acks=1``. If `enable_idempotence` is
            :data:`True` defaults to ``acks=all``
        compression_type (str): The compression type for all data generated by
            the producer. Valid values are ``gzip``, ``snappy``, ``lz4``, ``zstd``
            or :data:`None`.
            Compression is of full batches of data, so the efficacy of batching
            will also impact the compression ratio (more batching means better
            compression). Default: :data:`None`.
        max_batch_size (int): Maximum size of buffered data per partition.
            After this amount :meth:`send` coroutine will block until batch is
            drained.
            Default: 16384
        linger_ms (int): The producer groups together any records that arrive
            in between request transmissions into a single batched request.
            Normally this occurs only under load when records arrive faster
            than they can be sent out. However in some circumstances the client
            may want to reduce the number of requests even under moderate load.
            This setting accomplishes this by adding a small amount of
            artificial delay; that is, if first request is processed faster,
            than `linger_ms`, producer will wait ``linger_ms - process_time``.
            Default: 0 (i.e. no delay).
        partitioner (Callable): Callable used to determine which partition
            each message is assigned to. Called (after key serialization):
            ``partitioner(key_bytes, all_partitions, available_partitions)``.
            The default partitioner implementation hashes each non-None key
            using the same murmur2 algorithm as the Java client so that
            messages with the same key are assigned to the same partition.
            When a key is :data:`None`, the message is delivered to a random partition
            (filtered to partitions with available leaders only, if possible).
        max_request_size (int): The maximum size of a request. This is also
            effectively a cap on the maximum record size. Note that the server
            has its own cap on record size which may be different from this.
            This setting will limit the number of record batches the producer
            will send in a single request to avoid sending huge requests.
            Default: 1048576.
        metadata_max_age_ms (int): The period of time in milliseconds after
            which we force a refresh of metadata even if we haven't seen any
            partition leadership changes to proactively discover any new
            brokers or partitions. Default: 300000
        request_timeout_ms (int): Produce request timeout in milliseconds.
            As it's sent as part of
            :class:`~kafka.protocol.produce.ProduceRequest` (it's a blocking
            call), maximum waiting time can be up to ``2 *
            request_timeout_ms``.
            Default: 40000.
        retry_backoff_ms (int): Milliseconds to backoff when retrying on
            errors. Default: 100.
        api_version (str): specify which kafka API version to use.
            If set to ``auto``, will attempt to infer the broker version by
            probing various APIs. Default: ``auto``
        security_protocol (str): Protocol used to communicate with brokers.
            Valid values are: ``PLAINTEXT``, ``SSL``. Default: ``PLAINTEXT``.
            Default: ``PLAINTEXT``.
        ssl_context (ssl.SSLContext): pre-configured :class:`~ssl.SSLContext`
            for wrapping socket connections. Directly passed into asyncio's
            :meth:`~asyncio.loop.create_connection`. For more
            information see :ref:`ssl_auth`.
            Default: :data:`None`
        connections_max_idle_ms (int): Close idle connections after the number
            of milliseconds specified by this config. Specifying :data:`None` will
            disable idle checks. Default: 540000 (9 minutes).
        enable_idempotence (bool): When set to :data:`True`, the producer will
            ensure that exactly one copy of each message is written in the
            stream. If :data:`False`, producer retries due to broker failures,
            etc., may write duplicates of the retried message in the stream.
            Note that enabling idempotence acks to set to ``all``. If it is not
            explicitly set by the user it will be chosen. If incompatible
            values are set, a :exc:`ValueError` will be thrown.
            New in version 0.5.0.
        sasl_mechanism (str): Authentication mechanism when security_protocol
            is configured for ``SASL_PLAINTEXT`` or ``SASL_SSL``. Valid values
            are: ``PLAIN``, ``GSSAPI``, ``SCRAM-SHA-256``, ``SCRAM-SHA-512``,
            ``OAUTHBEARER``.
            Default: ``PLAIN``
        sasl_plain_username (str): username for SASL ``PLAIN`` authentication.
            Default: :data:`None`
        sasl_plain_password (str): password for SASL ``PLAIN`` authentication.
            Default: :data:`None`
        sasl_oauth_token_provider (:class:`~aiokafka.abc.AbstractTokenProvider`):
            OAuthBearer token provider instance. (See
            :mod:`kafka.oauth.abstract`).
            Default: :data:`None`

    Note:
        Many configuration parameters are taken from the Java client:
        https://kafka.apache.org/documentation.html#producerconfigs
    """
    _PRODUCER_CLIENT_ID_SEQUENCE = 0

    _COMPRESSORS = {
        'gzip': (has_gzip, DefaultRecordBatch.CODEC_GZIP),
        'snappy': (has_snappy, DefaultRecordBatch.CODEC_SNAPPY),
        'lz4': (has_lz4, DefaultRecordBatch.CODEC_LZ4),
        'zstd': (has_zstd, DefaultRecordBatch.CODEC_ZSTD),
    }

    _closed = None  # Serves as an uninitialized flag for __del__
    _source_traceback = None

    def __init__(self, *, loop=None, bootstrap_servers='localhost',
                 client_id=None,
                 metadata_max_age_ms=300000, request_timeout_ms=40000,
                 api_version='auto', acks=_missing,
                 key_serializer=None, value_serializer=None,
                 compression_type=None, max_batch_size=16384,
                 partitioner=DefaultPartitioner(), max_request_size=1048576,
                 linger_ms=0, send_backoff_ms=100,
                 retry_backoff_ms=100, security_protocol="PLAINTEXT",
                 ssl_context=None, connections_max_idle_ms=540000,
                 enable_idempotence=False, transactional_id=None,
                 transaction_timeout_ms=60000, sasl_mechanism="PLAIN",
                 sasl_plain_password=None, sasl_plain_username=None,
                 sasl_kerberos_service_name='kafka',
                 sasl_kerberos_domain_name=None,
                 sasl_oauth_token_provider=None):
        if loop is None:
            loop = get_running_loop()
        else:
            warnings.warn("The loop argument is deprecated since 0.7.1, "
                          "and scheduled for removal in 0.8.0",
                          DeprecationWarning, stacklevel=2)
        if loop.get_debug():
            self._source_traceback = traceback.extract_stack(sys._getframe(1))
        self._loop = loop

        if acks not in (0, 1, -1, 'all', _missing):
            raise ValueError("Invalid ACKS parameter")
        if compression_type not in ('gzip', 'snappy', 'lz4', 'zstd', None):
            raise ValueError("Invalid compression type!")
        if compression_type:
            checker, compression_attrs = self._COMPRESSORS[compression_type]
            if not checker():
                raise RuntimeError("Compression library for {} not found"
                                   .format(compression_type))
        else:
            compression_attrs = 0

        if transactional_id is not None:
            enable_idempotence = True
        else:
            transaction_timeout_ms = INTEGER_MAX_VALUE

        if enable_idempotence:
            if acks is _missing:
                acks = -1
            elif acks not in ('all', -1):
                raise ValueError(
                    "acks={} not supported if enable_idempotence=True"
                    .format(acks))
            self._txn_manager = TransactionManager(
                transactional_id, transaction_timeout_ms)
        else:
            self._txn_manager = None

        if acks is _missing:
            acks = 1
        elif acks == 'all':
            acks = -1

        AIOKafkaProducer._PRODUCER_CLIENT_ID_SEQUENCE += 1
        if client_id is None:
            client_id = 'aiokafka-producer-%s' % \
                AIOKafkaProducer._PRODUCER_CLIENT_ID_SEQUENCE

        self._key_serializer = key_serializer
        self._value_serializer = value_serializer
        self._compression_type = compression_type
        self._partitioner = partitioner
        self._max_request_size = max_request_size
        self._request_timeout_ms = request_timeout_ms

        self.client = AIOKafkaClient(
            loop=loop, bootstrap_servers=bootstrap_servers,
            client_id=client_id, metadata_max_age_ms=metadata_max_age_ms,
            request_timeout_ms=request_timeout_ms,
            retry_backoff_ms=retry_backoff_ms,
            api_version=api_version, security_protocol=security_protocol,
            ssl_context=ssl_context,
            connections_max_idle_ms=connections_max_idle_ms,
            sasl_mechanism=sasl_mechanism,
            sasl_plain_username=sasl_plain_username,
            sasl_plain_password=sasl_plain_password,
            sasl_kerberos_service_name=sasl_kerberos_service_name,
            sasl_kerberos_domain_name=sasl_kerberos_domain_name,
            sasl_oauth_token_provider=sasl_oauth_token_provider)
        self._metadata = self.client.cluster
        self._message_accumulator = MessageAccumulator(
            self._metadata, max_batch_size, compression_attrs,
            self._request_timeout_ms / 1000, txn_manager=self._txn_manager,
            loop=loop)
        self._sender = Sender(
            self.client, acks=acks, txn_manager=self._txn_manager,
            retry_backoff_ms=retry_backoff_ms, linger_ms=linger_ms,
            message_accumulator=self._message_accumulator,
            request_timeout_ms=request_timeout_ms)

        self._closed = False

    # Warn if producer was not closed properly
    # We don't attempt to close the Consumer, as __del__ is synchronous
    def __del__(self, _warnings=warnings):
        if self._closed is False:
            _warnings.warn(f"Unclosed AIOKafkaProducer {self!r}",
                           ResourceWarning,
                           source=self)
            context = {'producer': self,
                       'message': 'Unclosed AIOKafkaProducer'}
            if self._source_traceback is not None:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)

    async def start(self):
        """Connect to Kafka cluster and check server version"""
        assert self._loop is get_running_loop(), (
            "Please create objects with the same loop as running with"
        )
        log.debug("Starting the Kafka producer")  # trace
        await self.client.bootstrap()

        if self._compression_type == 'lz4':
            assert self.client.api_version >= (0, 8, 2), \
                'LZ4 Requires >= Kafka 0.8.2 Brokers'
        elif self._compression_type == 'zstd':
            assert self.client.api_version >= (2, 1, 0), \
                'Zstd Requires >= Kafka 2.1.0 Brokers'

        if self._txn_manager is not None and self.client.api_version < (0, 11):
            raise UnsupportedVersionError(
                "Idempotent producer available only for Broker version 0.11"
                " and above")

        await self._sender.start()
        self._message_accumulator.set_api_version(self.client.api_version)
        self._producer_magic = 0 if self.client.api_version < (0, 10) else 1
        log.debug("Kafka producer started")

    async def flush(self):
        """Wait until all batches are Delivered and futures resolved"""
        await self._message_accumulator.flush()

    async def stop(self):
        """Flush all pending data and close all connections to kafka cluster"""
        if self._closed:
            return
        self._closed = True

        # If the sender task is down there is no way for accumulator to flush
        if self._sender is not None and self._sender.sender_task is not None:
            await asyncio.wait([
                create_task(self._message_accumulator.close()),
                self._sender.sender_task],
                return_when=asyncio.FIRST_COMPLETED)

            await self._sender.close()

        await self.client.close()
        log.debug("The Kafka producer has closed.")

    async def partitions_for(self, topic):
        """Returns set of all known partitions for the topic."""
        return (await self.client._wait_on_metadata(topic))

    def _serialize(self, topic, key, value):
        if self._key_serializer:
            serialized_key = self._key_serializer(key)
        else:
            serialized_key = key
        if self._value_serializer:
            serialized_value = self._value_serializer(value)
        else:
            serialized_value = value

        message_size = LegacyRecordBatchBuilder.record_overhead(
            self._producer_magic)
        if serialized_key is not None:
            message_size += len(serialized_key)
        if serialized_value is not None:
            message_size += len(serialized_value)
        if message_size > self._max_request_size:
            raise MessageSizeTooLargeError(
                "The message is %d bytes when serialized which is larger than"
                " the maximum request size you have configured with the"
                " max_request_size configuration" % message_size)

        return serialized_key, serialized_value

    def _partition(self, topic, partition, key, value,
                   serialized_key, serialized_value):
        if partition is not None:
            assert partition >= 0
            assert partition in self._metadata.partitions_for_topic(topic), \
                'Unrecognized partition'
            return partition

        all_partitions = list(self._metadata.partitions_for_topic(topic))
        available = list(self._metadata.available_partitions_for_topic(topic))
        return self._partitioner(
            serialized_key, all_partitions, available)

    async def send(
        self, topic, value=None, key=None, partition=None,
        timestamp_ms=None, headers=None
    ):
        """Publish a message to a topic.

        Arguments:
            topic (str): topic where the message will be published
            value (Optional): message value. Must be type :class:`bytes`, or be
                serializable to :class:`bytes` via configured `value_serializer`. If
                value is :data:`None`, key is required and message acts as a
                ``delete``.

                See `Kafka compaction documentation
                <https://kafka.apache.org/documentation.html#compaction>`__ for
                more details. (compaction requires kafka >= 0.8.1)
            partition (int, Optional): optionally specify a partition. If not
                set, the partition will be selected using the configured
                `partitioner`.
            key (Optional): a key to associate with the message. Can be used to
                determine which partition to send the message to. If partition
                is :data:`None` (and producer's partitioner config is left as default),
                then messages with the same key will be delivered to the same
                partition (but if key is :data:`None`, partition is chosen randomly).
                Must be type :class:`bytes`, or be serializable to bytes via configured
                `key_serializer`.
            timestamp_ms (int, Optional): epoch milliseconds (from Jan 1 1970
                UTC) to use as the message timestamp. Defaults to current time.
            headers (Optional): Kafka headers to be included in the message using
                the format ``[("key", b"value")]``. Iterable of tuples where key
                is a normal string and value is a byte string.

        Returns:
            asyncio.Future: object that will be set when message is
            processed

        Raises:
            ~aiokafka.errors.KafkaTimeoutError: if we can't schedule this record
                (pending buffer is full) in up to `request_timeout_ms`
                milliseconds.

        Note:
            The returned future will wait based on `request_timeout_ms`
            setting. Cancelling the returned future **will not** stop event
            from being sent, but cancelling the :meth:`send` coroutine itself
            **will**.
        """
        assert value is not None or self.client.api_version >= (0, 8, 1), (
            'Null messages require kafka >= 0.8.1')
        assert not (value is None and key is None), \
            'Need at least one: key or value'

        # first make sure the metadata for the topic is available
        await self.client._wait_on_metadata(topic)

        # Ensure transaction is started and not committing
        if self._txn_manager is not None:
            txn_manager = self._txn_manager
            if txn_manager.transactional_id is not None and \
                    not self._txn_manager.is_in_transaction():
                raise IllegalOperation(
                    "Can't send messages while not in transaction")

        if headers is not None:
            if self.client.api_version < (0, 11):
                raise UnsupportedVersionError(
                    "Headers not supported before Kafka 0.11")
        else:
            # Record parser/builder support only list type, no explicit None
            headers = []

        key_bytes, value_bytes = self._serialize(topic, key, value)
        partition = self._partition(topic, partition, key, value,
                                    key_bytes, value_bytes)

        tp = TopicPartition(topic, partition)
        log.debug("Sending (key=%s value=%s) to %s", key, value, tp)

        fut = await self._message_accumulator.add_message(
            tp, key_bytes, value_bytes, self._request_timeout_ms / 1000,
            timestamp_ms=timestamp_ms, headers=headers)
        return fut

    async def send_and_wait(
        self, topic, value=None, key=None, partition=None,
        timestamp_ms=None, headers=None
    ):
        """Publish a message to a topic and wait the result"""
        future = await self.send(
            topic, value, key, partition, timestamp_ms, headers)
        return (await future)

    def create_batch(self):
        """Create and return an empty :class:`.BatchBuilder`.

        The batch is not queued for send until submission to :meth:`send_batch`.

        Returns:
            BatchBuilder: empty batch to be filled and submitted by the caller.
        """
        return self._message_accumulator.create_builder()

    async def send_batch(self, batch, topic, *, partition):
        """Submit a BatchBuilder for publication.

        Arguments:
            batch (BatchBuilder): batch object to be published.
            topic (str): topic where the batch will be published.
            partition (int): partition where this batch will be published.

        Returns:
            asyncio.Future: object that will be set when the batch is
                delivered.
        """
        # first make sure the metadata for the topic is available
        await self.client._wait_on_metadata(topic)
        # We only validate we have the partition in the metadata here
        partition = self._partition(topic, partition, None, None, None, None)

        # Ensure transaction is started and not committing
        if self._txn_manager is not None:
            txn_manager = self._txn_manager
            if txn_manager.transactional_id is not None and \
                    not self._txn_manager.is_in_transaction():
                raise IllegalOperation(
                    "Can't send messages while not in transaction")

        tp = TopicPartition(topic, partition)
        log.debug("Sending batch to %s", tp)
        future = await self._message_accumulator.add_batch(
            batch, tp, self._request_timeout_ms / 1000)
        return future

    def _ensure_transactional(self):
        if self._txn_manager is None or \
                self._txn_manager.transactional_id is None:
            raise IllegalOperation(
                "You need to configure transaction_id to use transactions")

    async def begin_transaction(self):
        self._ensure_transactional()
        log.debug(
            "Beginning a new transaction for id %s",
            self._txn_manager.transactional_id)
        await asyncio.shield(
            self._txn_manager.wait_for_pid()
        )
        self._txn_manager.begin_transaction()

    async def commit_transaction(self):
        self._ensure_transactional()
        log.debug(
            "Committing transaction for id %s",
            self._txn_manager.transactional_id)
        self._txn_manager.committing_transaction()
        await asyncio.shield(
            self._txn_manager.wait_for_transaction_end(),
        )

    async def abort_transaction(self):
        self._ensure_transactional()
        log.debug(
            "Aborting transaction for id %s",
            self._txn_manager.transactional_id)
        self._txn_manager.aborting_transaction()
        await asyncio.shield(
            self._txn_manager.wait_for_transaction_end(),
        )

    def transaction(self):
        """Start a transaction context"""

        return TransactionContext(self)

    async def send_offsets_to_transaction(self, offsets, group_id):
        self._ensure_transactional()

        if not self._txn_manager.is_in_transaction():
            raise IllegalOperation("Not in the middle of a transaction")

        if not group_id or not isinstance(group_id, str):
            raise ValueError(group_id)

        # validate `offsets` structure
        formatted_offsets = commit_structure_validate(offsets)

        log.debug(
            "Begin adding offsets %s for consumer group %s to transaction",
            formatted_offsets, group_id)
        fut = self._txn_manager.add_offsets_to_txn(formatted_offsets, group_id)
        await asyncio.shield(fut)

    async def __aenter__(self):
        await self.start()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await self.stop() class _MemoryRecordsPy:
    LENGTH_OFFSET = struct.calcsize(">q")
    LOG_OVERHEAD = struct.calcsize(">qi")
    MAGIC_OFFSET = struct.calcsize(">qii")
    MIN_SLICE = LOG_OVERHEAD + LegacyRecordBatch.RECORD_OVERHEAD_V0
    def __init__(self, bytes_data):
    def size_in_bytes(self):
    def _cache_next(self, len_offset=LENGTH_OFFSET, log_overhead=LOG_OVERHEAD):
    def has_next(self):
    def next_batch(self, _min_slice=MIN_SLICE,
                   _magic_offset=MAGIC_OFFSET): class AIOKafkaClient:
    """Initialize an asynchronous kafka client

    Keyword Arguments:
        bootstrap_servers: 'host[:port]' string (or list of 'host[:port]'
            strings) that the consumer should contact to bootstrap initial
            cluster metadata. This does not have to be the full node list.
            It just needs to have at least one broker that will respond to
            Metadata API Request. Default port is 9092. If no servers are
            specified, will default to localhost:9092.
        client_id (str): a name for this client. This string is passed in
            each request to servers and can be used to identify specific
            server-side log entries that correspond to this client. Also
            submitted to GroupCoordinator for logging with respect to
            consumer group administration. Default: 'aiokafka-{ver}'
        request_timeout_ms (int): Client request timeout in milliseconds.
            Default: 40000.
        metadata_max_age_ms (int): The period of time in milliseconds after
            which we force a refresh of metadata even if we haven't seen
            any partition leadership changes to proactively discover any
            new brokers or partitions. Default: 300000
        retry_backoff_ms (int): Milliseconds to backoff when retrying on
            errors. Default: 100.
        api_version (str): specify which kafka API version to use.
            AIOKafka supports Kafka API versions >=0.9 only.
            If set to 'auto', will attempt to infer the broker version by
            probing various APIs. Default: auto
        security_protocol (str): Protocol used to communicate with brokers.
            Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.
            Default: PLAINTEXT.
        ssl_context (ssl.SSLContext): pre-configured SSLContext for wrapping
            socket connections. For more information see :ref:`ssl_auth`.
            Default: None.
        connections_max_idle_ms (int): Close idle connections after the number
            of milliseconds specified by this config. Specifying `None` will
            disable idle checks. Default: 540000 (9 minutes).
    """

    def __init__(self, *, loop=None, bootstrap_servers='localhost',
                 client_id='aiokafka-' + __version__,
                 metadata_max_age_ms=300000,
                 request_timeout_ms=40000,
                 retry_backoff_ms=100,
                 ssl_context=None,
                 security_protocol='PLAINTEXT',
                 api_version='auto',
                 connections_max_idle_ms=540000,
                 sasl_mechanism='PLAIN',
                 sasl_plain_username=None,
                 sasl_plain_password=None,
                 sasl_kerberos_service_name='kafka',
                 sasl_kerberos_domain_name=None,
                 sasl_oauth_token_provider=None
                 ):
        if loop is None:
            loop = get_running_loop()

        if security_protocol not in (
                'SSL', 'PLAINTEXT', 'SASL_PLAINTEXT', 'SASL_SSL'):
            raise ValueError("`security_protocol` should be SSL or PLAINTEXT")
        if security_protocol in ["SSL", "SASL_SSL"] and ssl_context is None:
            raise ValueError(
                "`ssl_context` is mandatory if security_protocol=='SSL'")
        if security_protocol in ["SASL_SSL", "SASL_PLAINTEXT"]:
            if sasl_mechanism not in (
                    "PLAIN", "GSSAPI", "SCRAM-SHA-256", "SCRAM-SHA-512",
                    "OAUTHBEARER"):
                raise ValueError(
                    "only `PLAIN`, `GSSAPI`, `SCRAM-SHA-256`, "
                    "`SCRAM-SHA-512` and `OAUTHBEARER`"
                    "sasl_mechanism are supported "
                    "at the moment")
            if sasl_mechanism == "PLAIN" and \
               (sasl_plain_username is None or sasl_plain_password is None):
                raise ValueError(
                    "sasl_plain_username and sasl_plain_password required for "
                    "PLAIN sasl")

        self._bootstrap_servers = bootstrap_servers
        self._client_id = client_id
        self._metadata_max_age_ms = metadata_max_age_ms
        self._request_timeout_ms = request_timeout_ms
        if api_version != "auto":
            api_version = parse_kafka_version(api_version)
        self._api_version = api_version
        self._security_protocol = security_protocol
        self._ssl_context = ssl_context
        self._retry_backoff = retry_backoff_ms / 1000
        self._connections_max_idle_ms = connections_max_idle_ms
        self._sasl_mechanism = sasl_mechanism
        self._sasl_plain_username = sasl_plain_username
        self._sasl_plain_password = sasl_plain_password
        self._sasl_kerberos_service_name = sasl_kerberos_service_name
        self._sasl_kerberos_domain_name = sasl_kerberos_domain_name
        self._sasl_oauth_token_provider = sasl_oauth_token_provider

        self.cluster = ClusterMetadata(metadata_max_age_ms=metadata_max_age_ms)

        self._topics = set()  # empty set will fetch all topic metadata
        self._conns = {}
        self._loop = loop
        self._sync_task = None

        self._md_update_fut = None
        self._md_update_waiter = loop.create_future()
        self._get_conn_lock_value = None

    @property
    def _get_conn_lock(self):
        if self._get_conn_lock_value is None:
            self._get_conn_lock_value = asyncio.Lock()
        return self._get_conn_lock_value

    def __repr__(self):
        return '<AIOKafkaClient client_id=%s>' % self._client_id

    @property
    def api_version(self):
        if type(self._api_version) is tuple:
            return self._api_version
        # unknown api version, return minimal supported version
        return (0, 9, 0)

    @property
    def hosts(self):
        return collect_hosts(self._bootstrap_servers)

    async def close(self):
        if self._sync_task:
            self._sync_task.cancel()
            try:
                await self._sync_task
            except asyncio.CancelledError:
                pass
            self._sync_task = None
        # Be careful to wait for graceful closure of all connections, so we
        # process all pending buffers.
        futs = []
        for conn in self._conns.values():
            futs.append(conn.close(reason=CloseReason.SHUTDOWN))
        if futs:
            await asyncio.gather(*futs)

    async def bootstrap(self):
        """Try to to bootstrap initial cluster metadata"""
        assert self._loop is get_running_loop(), (
            "Please create objects with the same loop as running with"
        )
        # using request v0 for bootstrap if not sure v1 is available
        if self._api_version == "auto" or self._api_version < (0, 10):
            metadata_request = MetadataRequest[0]([])
        else:
            metadata_request = MetadataRequest[1]([])

        version_hint = None
        if self._api_version != "auto":
            version_hint = self._api_version

        for host, port, _ in self.hosts:
            log.debug("Attempting to bootstrap via node at %s:%s", host, port)

            try:
                bootstrap_conn = await create_conn(
                    host, port, client_id=self._client_id,
                    request_timeout_ms=self._request_timeout_ms,
                    ssl_context=self._ssl_context,
                    security_protocol=self._security_protocol,
                    max_idle_ms=self._connections_max_idle_ms,
                    sasl_mechanism=self._sasl_mechanism,
                    sasl_plain_username=self._sasl_plain_username,
                    sasl_plain_password=self._sasl_plain_password,
                    sasl_kerberos_service_name=self._sasl_kerberos_service_name,  # noqa: ignore=E501
                    sasl_kerberos_domain_name=self._sasl_kerberos_domain_name,
                    sasl_oauth_token_provider=self._sasl_oauth_token_provider,
                    version_hint=version_hint)
            except (OSError, asyncio.TimeoutError) as err:
                log.error('Unable connect to "%s:%s": %s', host, port, err)
                continue

            try:
                metadata = await bootstrap_conn.send(metadata_request)
            except (KafkaError, asyncio.TimeoutError) as err:
                log.warning('Unable to request metadata from "%s:%s": %s',
                            host, port, err)
                bootstrap_conn.close()
                continue

            self.cluster.update_metadata(metadata)

            # A cluster with no topics can return no broker metadata...
            # In that case, we should keep the bootstrap connection till
            # we get a normal cluster layout.
            if not len(self.cluster.brokers()):
                bootstrap_id = ('bootstrap', ConnectionGroup.DEFAULT)
                self._conns[bootstrap_id] = bootstrap_conn
            else:
                bootstrap_conn.close()

            log.debug('Received cluster metadata: %s', self.cluster)
            break
        else:
            raise KafkaConnectionError(
                f'Unable to bootstrap from {self.hosts}')

        # detect api version if need
        if self._api_version == 'auto':
            self._api_version = await self.check_version()

        if self._sync_task is None:
            # starting metadata synchronizer task
            self._sync_task = create_task(self._md_synchronizer())

    async def _md_synchronizer(self):
        """routine (async task) for synchronize cluster metadata every
        `metadata_max_age_ms` milliseconds"""
        while True:
            await asyncio.wait(
                [self._md_update_waiter],
                timeout=self._metadata_max_age_ms / 1000)

            topics = self._topics
            if self._md_update_fut is None:
                self._md_update_fut = create_future()
            ret = await self._metadata_update(self.cluster, topics)
            # If list of topics changed during metadata update we must update
            # it again right away.
            if topics != self._topics:
                continue
            # Earlier this waiter was set before sending metadata_request,
            # but that was to avoid topic list changes being unnoticed, which
            # is handled explicitly now.
            self._md_update_waiter = create_future()

            self._md_update_fut.set_result(ret)
            self._md_update_fut = None

    def get_random_node(self):
        """choice random node from known cluster brokers

        Returns:
            nodeId - identifier of broker
        """
        nodeids = [b.nodeId for b in self.cluster.brokers()]
        if not nodeids:
            return None
        return random.choice(nodeids)

    async def _metadata_update(self, cluster_metadata, topics):
        assert isinstance(cluster_metadata, ClusterMetadata)
        topics = list(topics)
        version_id = 0 if self.api_version < (0, 10) else 1
        if version_id == 1 and not topics:
            topics = None
        metadata_request = MetadataRequest[version_id](topics)
        nodeids = [b.nodeId for b in self.cluster.brokers()]
        bootstrap_id = ('bootstrap', ConnectionGroup.DEFAULT)
        if bootstrap_id in self._conns:
            nodeids.append('bootstrap')
        random.shuffle(nodeids)
        for node_id in nodeids:
            conn = await self._get_conn(node_id)

            if conn is None:
                continue
            log.debug("Sending metadata request %s to node %s",
                      metadata_request, node_id)

            try:
                metadata = await conn.send(metadata_request)
            except (KafkaError, asyncio.TimeoutError) as err:
                log.error(
                    'Unable to request metadata from node with id %s: %r',
                    node_id, err)
                continue

            # don't update the cluster if there are no valid nodes...the topic
            # we want may still be in the process of being created which means
            # we will get errors and no nodes until it exists
            if not metadata.brokers:
                return False

            cluster_metadata.update_metadata(metadata)

            # We only keep bootstrap connection to update metadata until
            # proper cluster layout is available.
            if bootstrap_id in self._conns and len(self.cluster.brokers()):
                conn = self._conns.pop(bootstrap_id)
                conn.close()

            break
        else:
            log.error('Unable to update metadata from %s', nodeids)
            cluster_metadata.failed_update(None)
            return False
        return True

    def force_metadata_update(self):
        """Update cluster metadata

        Returns:
            True/False - metadata updated or not
        """
        if self._md_update_fut is None:
            # Wake up the `_md_synchronizer` task
            if not self._md_update_waiter.done():
                self._md_update_waiter.set_result(None)
            self._md_update_fut = self._loop.create_future()
        # Metadata will be updated in the background by syncronizer
        return asyncio.shield(self._md_update_fut)

    async def fetch_all_metadata(self):
        cluster_md = ClusterMetadata(
            metadata_max_age_ms=self._metadata_max_age_ms)
        updated = await self._metadata_update(cluster_md, [])
        if not updated:
            raise KafkaError(
                'Unable to get cluster metadata over all known brokers')
        return cluster_md

    def add_topic(self, topic):
        """Add a topic to the list of topics tracked via metadata.

        Arguments:
            topic (str): topic to track
        """
        if topic in self._topics:
            res = self._loop.create_future()
            res.set_result(True)
        else:
            res = self.force_metadata_update()
        self._topics.add(topic)
        return res

    def set_topics(self, topics):
        """Set specific topics to track for metadata.

        Arguments:
            topics (list of str): topics to track
        """
        assert not isinstance(topics, str)
        if not topics or set(topics).difference(self._topics):
            res = self.force_metadata_update()
        else:
            res = self._loop.create_future()
            res.set_result(True)
        self._topics = set(topics)
        return res

    def _on_connection_closed(self, conn, reason):
        """ Callback called when connection is closed
        """
        # Connection failures imply that our metadata is stale, so let's
        # refresh
        if reason == CloseReason.CONNECTION_BROKEN or \
                reason == CloseReason.CONNECTION_TIMEOUT:
            self.force_metadata_update()

    async def _get_conn(
        self, node_id, *, group=ConnectionGroup.DEFAULT,
        no_hint=False
    ):
        "Get or create a connection to a broker using host and port"
        conn_id = (node_id, group)
        if conn_id in self._conns:
            conn = self._conns[conn_id]
            if not conn.connected():
                del self._conns[conn_id]
            else:
                return conn

        try:
            if group == ConnectionGroup.DEFAULT:
                broker = self.cluster.broker_metadata(node_id)
                # XXX: earlier we only did an assert here, but it seems it's
                # possible to get a leader that is for some reason not in
                # metadata.
                # I think requerying metadata should solve this problem
                if broker is None:
                    raise StaleMetadata(
                        'Broker id %s not in current metadata' % node_id)
            else:
                broker = self.cluster.coordinator_metadata(node_id)
                assert broker is not None

            log.debug("Initiating connection to node %s at %s:%s",
                      node_id, broker.host, broker.port)

            async with self._get_conn_lock:
                if conn_id in self._conns:
                    return self._conns[conn_id]

                version_hint = self._api_version
                if version_hint == "auto" or no_hint:
                    version_hint = None

                self._conns[conn_id] = await create_conn(
                    broker.host, broker.port,
                    client_id=self._client_id,
                    request_timeout_ms=self._request_timeout_ms,
                    ssl_context=self._ssl_context,
                    security_protocol=self._security_protocol,
                    on_close=self._on_connection_closed,
                    max_idle_ms=self._connections_max_idle_ms,
                    sasl_mechanism=self._sasl_mechanism,
                    sasl_plain_username=self._sasl_plain_username,
                    sasl_plain_password=self._sasl_plain_password,
                    sasl_kerberos_service_name=self._sasl_kerberos_service_name,  # noqa: ignore=E501
                    sasl_kerberos_domain_name=self._sasl_kerberos_domain_name,
                    sasl_oauth_token_provider=self._sasl_oauth_token_provider,
                    version_hint=version_hint
                )
        except (OSError, asyncio.TimeoutError, KafkaError) as err:
            log.error('Unable connect to node with id %s: %s', node_id, err)
            if group == ConnectionGroup.DEFAULT:
                # Connection failures imply that our metadata is stale, so
                # let's refresh
                self.force_metadata_update()
            return None
        else:
            return self._conns[conn_id]

    async def ready(self, node_id, *, group=ConnectionGroup.DEFAULT):
        conn = await self._get_conn(node_id, group=group)
        if conn is None:
            return False
        return True

    async def send(self, node_id, request, *, group=ConnectionGroup.DEFAULT):
        """Send a request to a specific node.

        Arguments:
            node_id (int): destination node
            request (Struct): request object (not-encoded)

        Raises:
            kafka.errors.RequestTimedOutError
            kafka.errors.NodeNotReadyError
            kafka.errors.KafkaConnectionError
            kafka.errors.CorrelationIdError

        Returns:
            Future: resolves to Response struct
        """
        if not (await self.ready(node_id, group=group)):
            raise NodeNotReadyError(
                "Attempt to send a request to node"
                " which is not ready (node id {}).".format(node_id))

        # Every request gets a response, except one special case:
        expect_response = True
        if isinstance(request, tuple(ProduceRequest)) and \
                request.required_acks == 0:
            expect_response = False

        future = self._conns[(node_id, group)].send(
            request, expect_response=expect_response)
        try:
            result = await future
        except asyncio.TimeoutError:
            # close connection so it is renewed in next request
            self._conns[(node_id, group)].close(
                reason=CloseReason.CONNECTION_TIMEOUT)
            raise RequestTimedOutError()
        else:
            return result

    async def check_version(self, node_id=None):
        """Attempt to guess the broker version"""
        if node_id is None:
            default_group_conns = [
                n_id for (n_id, group) in self._conns.keys()
                if group == ConnectionGroup.DEFAULT
            ]
            if default_group_conns:
                node_id = default_group_conns[0]
            else:
                assert self.cluster.brokers(), 'no brokers in metadata'
                node_id = list(self.cluster.brokers())[0].nodeId

        from kafka.protocol.admin import (
            ListGroupsRequest_v0, ApiVersionRequest_v0)
        from kafka.protocol.commit import (
            OffsetFetchRequest_v0, GroupCoordinatorRequest_v0)
        from kafka.protocol.metadata import MetadataRequest_v0
        test_cases = [
            ((0, 10), ApiVersionRequest_v0()),
            ((0, 9), ListGroupsRequest_v0()),
            ((0, 8, 2), GroupCoordinatorRequest_v0('aiokafka-default-group')),
            ((0, 8, 1), OffsetFetchRequest_v0('aiokafka-default-group', [])),
            ((0, 8, 0), MetadataRequest_v0([])),
        ]

        # kafka kills the connection when it does not recognize an API request
        # so we can send a test request and then follow immediately with a
        # vanilla MetadataRequest. If the server did not recognize the first
        # request, both will be failed with a ConnectionError that wraps
        # socket.error (32, 54, or 104)
        conn = await self._get_conn(node_id, no_hint=True)
        if conn is None:
            raise KafkaConnectionError(
                f"No connection to node with id {node_id}")
        for version, request in test_cases:
            try:
                if not conn.connected():
                    await conn.connect()
                assert conn, f'no connection to node with id {node_id}'
                # request can be ignored by Kafka broker,
                # so we send metadata request and wait response
                task = create_task(conn.send(request))
                await asyncio.wait([task], timeout=0.1)
                try:
                    await conn.send(MetadataRequest_v0([]))
                except KafkaError:
                    # metadata request can be cancelled in case
                    # of invalid correlationIds order
                    pass
                response = await task
            except KafkaError:
                continue
            else:
                # To avoid having a connection in undefined state
                if node_id != "bootstrap" and conn.connected():
                    conn.close()
                if isinstance(request, ApiVersionRequest_v0):
                    # Starting from 0.10 kafka broker we determine version
                    # by looking at ApiVersionResponse
                    version = self._check_api_version_response(response)
                return version

        raise UnrecognizedBrokerVersion()

    def _check_api_version_response(self, response):
        # The logic here is to check the list of supported request versions
        # in descending order. As soon as we find one that works, return it
        test_cases = [
            # format (<broker version>, <needed struct>)
            # TODO Requires unreleased version of python-kafka
            # ((2, 6, 0), DescribeClientQuotasRequest[0]),
            ((2, 5, 0), DescribeAclsRequest_v2),
            ((2, 4, 0), ProduceRequest[8]),
            ((2, 3, 0), FetchRequest[11]),
            ((2, 2, 0), OffsetRequest[5]),
            ((2, 1, 0), FetchRequest[10]),
            ((2, 0, 0), FetchRequest[8]),
            ((1, 1, 0), FetchRequest[7]),
            ((1, 0, 0), MetadataRequest[5]),
            ((0, 11, 0), MetadataRequest[4]),
            ((0, 10, 2), OffsetFetchRequest[2]),
            ((0, 10, 1), MetadataRequest[2]),
        ]

        error_type = Errors.for_code(response.error_code)
        assert error_type is Errors.NoError, "API version check failed"
        max_versions = {
            api_key: max_version
            for api_key, _, max_version in response.api_versions
        }
        # Get the best match of test cases
        for broker_version, struct in test_cases:
            if max_versions.get(struct.API_KEY, -1) >= struct.API_VERSION:
                return broker_version

        # We know that ApiVersionResponse is only supported in 0.10+
        # so if all else fails, choose that
        return (0, 10, 0)

    async def _wait_on_metadata(self, topic):
        """
        Wait for cluster metadata including partitions for the given topic to
        be available.

        Arguments:
            topic (str): topic we want metadata for

        Returns:
            set: partition ids for the topic

        Raises:
            UnknownTopicOrPartitionError: if no topic or partitions found
                in cluster metadata
        """
        if topic in self.cluster.topics():
            return self.cluster.partitions_for_topic(topic)

        # add topic to metadata topic list if it is not there already.
        self.add_topic(topic)

        t0 = time.monotonic()
        while True:
            await self.force_metadata_update()
            if topic in self.cluster.topics():
                break
            if (time.monotonic() - t0) > (self._request_timeout_ms / 1000):
                raise UnknownTopicOrPartitionError()
            if topic in self.cluster.unauthorized_topics:
                raise Errors.TopicAuthorizationFailedError(topic)
            await asyncio.sleep(self._retry_backoff)

        return self.cluster.partitions_for_topic(topic)

    async def _maybe_wait_metadata(self):
        if self._md_update_fut is not None:
            await asyncio.shield(self._md_update_fut)

    async def coordinator_lookup(self, coordinator_type, coordinator_key):
        """ Lookup which node in the cluster is the coordinator for a certain
        role (Transaction coordinator or Group coordinator atm.)
        NOTE: Client keeps track of all coordination nodes separately, as they
        all have different sockets and ids.
        """

        node_id = self.get_random_node()
        assert node_id is not None, "Did we not perform bootstrap?"

        log.debug(
            "Sending FindCoordinator request for key %s to broker %s",
            coordinator_key, node_id)

        if self.api_version > (0, 11):
            request = FindCoordinatorRequest[1](
                coordinator_key, coordinator_type)
        else:
            # Group coordination only
            assert coordinator_type == CoordinationType.GROUP, \
                "No transactions for older brokers"
            request = FindCoordinatorRequest[0](coordinator_key)

        resp = await self.send(node_id, request)
        log.debug("Received group coordinator response %s", resp)
        error_type = Errors.for_code(resp.error_code)
        if error_type is not Errors.NoError:
            err = error_type()
            raise err
        self.cluster.add_coordinator(
            resp.coordinator_id, resp.host, resp.port, rack=None,
            purpose=(coordinator_type, coordinator_key))
        return resp.coordinator_id 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 class RecordMetadata(NamedTuple):
class ConsumerRecord(Generic[KT, VT]):
class OffsetAndTimestamp(NamedTuple):
KT = TypeVar("KT")
VT = TypeVar("VT") class CoordinatorNotAvailableError(GroupCoordinatorNotAvailableError):
class NotCoordinatorError(NotCoordinatorForGroupError):
class CoordinatorLoadInProgressError(GroupLoadInProgressError):
class ConsumerStoppedError(Exception):
class IllegalOperation(Exception):
class NoOffsetForPartitionError(KafkaError):
class RecordTooLargeError(KafkaError):
class ProducerClosed(KafkaError):
class ProducerFenced(KafkaError):
class OutOfOrderSequenceNumber(BrokerResponseError):
class DuplicateSequenceNumber(BrokerResponseError):
class InvalidProducerEpoch(BrokerResponseError):
class InvalidTxnState(BrokerResponseError):
class InvalidProducerIdMapping(BrokerResponseError):
class InvalidTransactionTimeout(BrokerResponseError):
class ConcurrentTransactions(BrokerResponseError):
class TransactionCoordinatorFenced(BrokerResponseError):
class TransactionalIdAuthorizationFailed(BrokerResponseError):
class SecurityDisabled(BrokerResponseError):
class OperationNotAttempted(BrokerResponseError):
class KafkaStorageError(BrokerResponseError):
class LogDirNotFound(BrokerResponseError):
class SaslAuthenticationFailed(BrokerResponseError):
class UnknownProducerId(BrokerResponseError):
class ReassignmentInProgress(BrokerResponseError):
class DelegationTokenAuthDisabled(BrokerResponseError):
class DelegationTokenNotFound(BrokerResponseError):
class DelegationTokenOwnerMismatch(BrokerResponseError):
class DelegationTokenRequestNotAllowed(BrokerResponseError):
class DelegationTokenAuthorizationFailed(BrokerResponseError):
class DelegationTokenExpired(BrokerResponseError):
class InvalidPrincipalType(BrokerResponseError):
class NonEmptyGroup(BrokerResponseError):
class GroupIdNotFound(BrokerResponseError):
class FetchSessionIdNotFound(BrokerResponseError):
class InvalidFetchSessionEpoch(BrokerResponseError):
class ListenerNotFound(BrokerResponseError):
    def __init__(
        self,
        msg="There is a newer producer using the same transactional_id or"
            "transaction timeout occurred (check that processing time is "
            "below transaction_timeout_ms)"
    ):
def _iter_broker_errors():
def for_code(error_code): class KafkaIntegrationTestCase(unittest.TestCase):

    topic = None

    @contextmanager
    def silence_loop_exception_handler(self):
        if hasattr(self.loop, "get_exception_handler"):
            orig_handler = self.loop.get_exception_handler()
        else:
            orig_handler = None  # Will set default handler
        self.loop.set_exception_handler(lambda loop, ctx: None)
        yield
        self.loop.set_exception_handler(orig_handler)

    def random_topic_name(self):
        return "topic-{}-{}".format(
            self.id()[self.id().rindex(".") + 1:],
            random_string(10).decode('utf-8')
        )

    def setUp(self):
        super().setUp()
        self._messages = {}
        if not self.topic:
            self.topic = self.random_topic_name()
        self._cleanup = []

    def tearDown(self):
        super().tearDown()
        for coro, args, kw in reversed(self._cleanup):
            task = asyncio.wait_for(coro(*args, **kw), 30)
            self.loop.run_until_complete(task)

    def add_cleanup(self, cb_or_coro, *args, **kw):
        self._cleanup.append((cb_or_coro, args, kw))

    async def wait_topic(self, client, topic):
        client.add_topic(topic)
        for i in range(5):
            ok = await client.force_metadata_update()
            if ok:
                ok = topic in client.cluster.topics()
            if not ok:
                await asyncio.sleep(1)
            else:
                return
        raise AssertionError(f'No topic "{topic}" exists')

    async def send_messages(
        self, partition, messages, *, topic=None,
        timestamp_ms=None, return_inst=False, headers=None
    ):
        topic = topic or self.topic
        ret = []
        producer = AIOKafkaProducer(bootstrap_servers=self.hosts)
        await producer.start()
        try:
            await self.wait_topic(producer.client, topic)

            for msg in messages:
                if isinstance(msg, str):
                    msg = msg.encode()
                elif isinstance(msg, int):
                    msg = str(msg).encode()
                future = await producer.send(
                    topic, msg, partition=partition,
                    timestamp_ms=timestamp_ms, headers=headers)
                resp = await future
                self.assertEqual(resp.topic, topic)
                self.assertEqual(resp.partition, partition)
                if return_inst:
                    ret.append(resp)
                else:
                    ret.append(msg)
        finally:
            await producer.stop()
        return ret

    def assert_message_count(self, messages, num_messages):
        # Make sure we got them all
        self.assertEqual(len(messages), num_messages)

        # Make sure there are no duplicates
        self.assertTrue(all(messages.count(m) == 1 for m in messages))

    def create_ssl_context(self):
        context = create_ssl_context(
            cafile=str(self.ssl_folder / "ca-cert"),
            certfile=str(self.ssl_folder / "cl_client.pem"),
            keyfile=str(self.ssl_folder / "cl_client.key"),
            password="abcdefgh")
        context.check_hostname = False
        return context class StubRebalanceListener(ConsumerRebalanceListener):

    def __init__(self):
        self.assigns = asyncio.Queue()
        self.revokes = asyncio.Queue()
        self.assigned = None
        self.revoked = None

    async def wait_assign(self):
        return (await self.assigns.get())

    def reset(self):
        while not self.assigns.empty():
            self.assigns.get_nowait()
        while not self.revokes.empty():
            self.revokes.get_nowait()

    def on_partitions_revoked(self, revoked):
        self.revokes.put_nowait(revoked)

    def on_partitions_assigned(self, assigned):
        self.assigns.put_nowait(assigned) def run_until_complete(fun):
    assert inspect.iscoroutinefunction(fun), (
        "Can not decorate ordinary function, only async ones"
    )

    @wraps(fun)
    def wrapper(test, *args, **kw):
        loop = test.loop
        timeout = getattr(test, "TEST_TIMEOUT", 120)
        ret = loop.run_until_complete(
            asyncio.wait_for(fun(test, *args, **kw), timeout))
        return ret
    return wrapper def run_in_thread(fun):

    @wraps(fun)
    def wrapper(test, *args, **kw):
        timeout = getattr(test, "TEST_TIMEOUT", 120)
        with futures.ThreadPoolExecutor() as executor:
            fut = executor.submit(fun, test, *args, **kw)
            fut.result(timeout=timeout)

    return wrapper def random_string(length):
    s = "".join(random.choice(string.ascii_letters) for _ in range(length))
    return s.encode('utf-8') def kafka_versions(*versions):
    # Took from kafka-python

    def version_str_to_list(s):
        return list(map(int, s.split('.')))  # e.g., [0, 8, 1, 1]

    def construct_lambda(s):
        if s[0].isdigit():
            op_str = '='
            v_str = s
        elif s[1].isdigit():
            op_str = s[0]  # ! < > =
            v_str = s[1:]
        elif s[2].isdigit():
            op_str = s[0:2]  # >= <=
            v_str = s[2:]
        else:
            raise ValueError('Unrecognized kafka version / operator: %s' % s)

        op_map = {
            '=': operator.eq,
            '!': operator.ne,
            '>': operator.gt,
            '<': operator.lt,
            '>=': operator.ge,
            '<=': operator.le
        }
        op = op_map[op_str]
        version = version_str_to_list(v_str)
        return lambda a: op(version_str_to_list(a), version)

    validators = map(construct_lambda, versions)

    def kafka_versions(func):
        @wraps(func)
        def wrapper(self, *args, **kw):
            kafka_version = self.kafka_version

            if not kafka_version:
                self.skipTest(
                    "no kafka version found. Is this an integration test?")

            for f in validators:
                if not f(kafka_version):
                    self.skipTest("unsupported kafka version")

            return func(self, *args, **kw)
        return wrapper
    return kafka_versions
import asyncio
import gc
import time
import json
import pytest
from unittest import mock
from contextlib import contextmanager
from aiokafka.abc import ConsumerRebalanceListener
from aiokafka.consumer import AIOKafkaConsumer
from aiokafka.consumer.fetcher import RecordTooLargeError, FetchRequest
from aiokafka.consumer import fetcher
from aiokafka.producer import AIOKafkaProducer
from aiokafka.record import MemoryRecords
from aiokafka.client import AIOKafkaClient
from aiokafka.util import create_task, get_running_loop
from aiokafka.structs import (
    OffsetAndTimestamp, TopicPartition, OffsetAndMetadata
)
from aiokafka.errors import (
    IllegalStateError, OffsetOutOfRangeError, UnsupportedVersionError,
    KafkaTimeoutError, NoOffsetForPartitionError, ConsumerStoppedError,
    IllegalOperation, UnknownError, KafkaError, InvalidSessionTimeoutError,
    CorruptRecordException
)
from ._testutil import (
    KafkaIntegrationTestCase, StubRebalanceListener,
    run_until_complete, run_in_thread, random_string, kafka_versions)
        consumer = AIOKafkaConsumer(
            topic, bootstrap_servers=self.hosts)
        await consumer.start()
        self.add_cleanup(consumer.stop)
        consume_task = create_task(consumer.getone())
        # just to be sure getone does not fail (before produce)
        await asyncio.sleep(0.5)
        self.assertFalse(consume_task.done())

        producer = AIOKafkaProducer(
            bootstrap_servers=self.hosts)
        await producer.start()
        await producer.send(topic, b'test msg')
        await producer.stop()

        data = await consume_task
        self.assertEqual(data.value, b'test msg')

    @run_until_complete
    async def test_consumer_subscribe_pattern_with_autocreate(self):
        pattern = "^some-autocreate-pattern-.*$"
        consumer = AIOKafkaConsumer(
            bootstrap_servers=self.hosts,
            metadata_max_age_ms=200, group_id="some_group",
            fetch_max_wait_ms=50,
            auto_offset_reset="earliest")
        self.add_cleanup(consumer.stop)
        await consumer.start()
        consumer.subscribe(pattern=pattern)
        # Start getter for the topics. Should not create any topics
        consume_task = create_task(consumer.getone())
        await asyncio.sleep(0.3)
        self.assertFalse(consume_task.done())
        self.assertEqual(consumer.subscription(), set())

        # Now lets autocreate the topic by fetching metadata for it.
        producer = AIOKafkaProducer(
            bootstrap_servers=self.hosts)
        self.add_cleanup(producer.stop)
        await producer.start()
        my_topic = "some-autocreate-pattern-1"
        await producer.client._wait_on_metadata(my_topic)
        # Wait for consumer to refresh metadata with new topic
        await asyncio.sleep(0.3)
        self.assertFalse(consume_task.done())
        self.assertTrue(consumer._client.cluster.topics() >= {my_topic})
        self.assertEqual(consumer.subscription(), {my_topic})

        # Add another topic
        my_topic2 = "some-autocreate-pattern-2"
        await producer.client._wait_on_metadata(my_topic2)
        # Wait for consumer to refresh metadata with new topic
        await asyncio.sleep(0.3)
        self.assertFalse(consume_task.done())
        self.assertTrue(consumer._client.cluster.topics() >=
                        {my_topic, my_topic2})
        self.assertEqual(consumer.subscription(), {my_topic, my_topic2})

        # Now lets actually produce some data and verify that it is consumed
        await producer.send(my_topic, b'test msg')
        data = await asyncio.wait_for(
            consume_task, timeout=2)
        self.assertEqual(data.value, b'test msg')

    @run_until_complete
    async def test_consumer_subscribe_pattern_autocreate_no_group_id(self):
        pattern = "^no-group-pattern-.*$"
        consumer = AIOKafkaConsumer(
            bootstrap_servers=self.hosts,
            metadata_max_age_ms=200, group_id=None,
            fetch_max_wait_ms=50,
            auto_offset_reset="earliest")
        self.add_cleanup(consumer.stop)
        await consumer.start()
        consumer.subscribe(pattern=pattern)
        # Start getter for the topics. Should not create any topics
        consume_task = create_task(consumer.getone())
        await asyncio.sleep(0.3)
        self.assertFalse(consume_task.done())
        self.assertEqual(consumer.subscription(), set())

        # Now lets autocreate the topic by fetching metadata for it.
        producer = AIOKafkaProducer(
            bootstrap_servers=self.hosts)
        self.add_cleanup(producer.stop)
        await producer.start()
        my_topic = "no-group-pattern-1"
        await producer.client._wait_on_metadata(my_topic)
        # Wait for consumer to refresh metadata with new topic
        await asyncio.sleep(0.3)
        self.assertFalse(consume_task.done())
        self.assertTrue(consumer._client.cluster.topics() >= {my_topic})
        self.assertEqual(consumer.subscription(), {my_topic})

        # Add another topic
        my_topic2 = "no-group-pattern-2"
        await producer.client._wait_on_metadata(my_topic2)
        # Wait for consumer to refresh metadata with new topic
        await asyncio.sleep(0.3)
        self.assertFalse(consume_task.done())
        self.assertTrue(consumer._client.cluster.topics() >=
                        {my_topic, my_topic2})
        self.assertEqual(consumer.subscription(), {my_topic, my_topic2})

        # Now lets actually produce some data and verify that it is consumed
        await producer.send(my_topic, b'test msg')
        data = await asyncio.wait_for(
            consume_task, timeout=2)
        self.assertEqual(data.value, b'test msg')

    @run_until_complete
    async def test_consumer_rebalance_on_new_topic(self):
        # Test will create a consumer group and check if adding new topic
        # will trigger a group rebalance and assign partitions
        pattern = "^another-autocreate-pattern-.*$"
        client = AIOKafkaClient(
            bootstrap_servers=self.hosts,
            client_id="test_autocreate")
        await client.bootstrap()
        listener1 = StubRebalanceListener()

        listener2 = StubRebalanceListener()