import asyncio import logging import re import sys import traceback import warnings from aiokafka import __version__ from aiokafka.abc import ConsumerRebalanceListener from aiokafka.client import AIOKafkaClient from aiokafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor from aiokafka.errors import ( ConsumerStoppedError, IllegalOperation, IllegalStateError, RecordTooLargeError, UnsupportedVersionError, ) from aiokafka.structs import ConsumerRecord, TopicPartition from aiokafka.util import commit_structure_validate, get_running_loop from .fetcher import Fetcher, OffsetResetStrategy from .group_coordinator import GroupCoordinator, NoGroupCoordinator from .subscription_state import SubscriptionState log = logging.getLogger(__name__) 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 group_instance_id (str or None): name of the group instance ID used for static membership (KIP-345) 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``, ``SASL_PLAINTEXT``, ``SASL_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. 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, group_instance_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.9.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._group_instance_id = group_instance_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) # fmt: skip ): 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, group_instance_id=self._group_instance_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 and 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 TypeError("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 and 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. .. _kafka-python: https://github.com/dpkp/kafka-python """ 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 TypeError("You should provide either `topics` or `pattern`") if topics and pattern: raise TypeError("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}") from 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) -> ConsumerRecord: """ 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(isinstance(k, TopicPartition) for k in 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 ) -> dict[TopicPartition, list[ConsumerRecord]]: """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(isinstance(k, TopicPartition) for k in 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) -> ConsumerRecord: """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: # noqa: PERF203 raise StopAsyncIteration from None 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()