aiokafka Documentation

Kafka uses Consumer Groups to allow a pool of processes to divide the work of consuming and processing records. All AIOKafkaConsumer instances sharing the same group_id are part of the same group.

Kafka balances the partitions between all members in the group so that each partition is assigned to exactly one consumer in the group. This provides scalability and fault tolerance:

• Rebalancing: If a process fails, its partitions are reassigned to other consumers. If a new consumer joins, partitions are moved from existing consumers to the new one.
• Scalability: If a topic has four partitions and a group has two processes, each process consumes from two partitions.

Conceptually, a Consumer Group acts as a single logical subscriber made up of multiple processes.

# Process 1
consumer = AIOKafkaConsumer(
    "my_topic", bootstrap_servers='localhost:9092',
    group_id="MyGreatConsumerGroup"  # This enables Consumer Groups
)
await consumer.start()
async for msg in consumer:
    print("Process %s consumed msg from partition %s" % (os.getpid(), msg.partition))

# Process 2
consumer2 = AIOKafkaConsumer(
    "my_topic", bootstrap_servers='localhost:9092',
    group_id="MyGreatConsumerGroup"  # Part of the same group
)
await consumer2.start()
async for msg in consumer2:
    print("Process %s consumed msg from partition %s" % (os.getpid(), msg.partition))

In aiokafka, heartbeating and group rebalancing are delegated to a background Task. This task sends heartbeats to the Coordinator as long as the event loop is running, similar to the Java client. This means rebalancing is not directly affected by the time taken between getmany() calls.

Because rebalancing and heartbeating are decoupled from the main processing loop, aiokafka provides two distinct configuration settings that are often conflated in the Java client:

• rebalance_timeout_ms: Controls the rebalance timeout. If you use a ConsumerRebalanceListener to control rebalance start/end moments, set this to the maximum time your application might spend waiting in the callback.
• max_poll_interval_ms: Controls the consumer processing timeout.

Note: If you use a ConsumerRebalanceListener that waits for the last getmany() result to be processed, it is safe to set rebalance_timeout_ms equal to max_poll_interval_ms.

When you call send(), messages are not sent immediately but are added to a buffer space. A background task then sends batches of messages to the cluster.

Key behaviors:

• Per-partition batches: Batches are created per partition with a maximum size defined by max_batch_size.
• Ordering: Messages in a batch are strictly in append order. Only one batch per partition is sent at a time, guaranteeing strict message order within a partition.
• Linger time: By default, a new batch is sent immediately after the previous one. You can set linger_ms to a non-zero value to introduce a delay, allowing more messages to accumulate in a single batch for higher throughput and better compression.
• Async sending: Using send() returns a future. You can await this future to get the RecordMetadata once the message is delivered.

# Will add the message to 1st partition's batch. If this method times out, 
# we can say for sure that message will never be sent.
fut = await producer.send("my_topic", b"Super message", partition=1)

# Message will either be delivered or an unrecoverable error will occur.
# Cancelling this future will not cancel the send.
msg = await fut

The AIOKafkaProducer is a high-level, asynchronous message producer.

To use it:

1. Initialize AIOKafkaProducer with bootstrap_servers.
2. Call await producer.start() to get cluster layout and initial topic/partition leadership information.
3. Use await producer.send_and_wait(topic, value) to produce a message and wait for delivery.
4. Call await producer.stop() to ensure all pending messages are delivered or expire before shutting down.

from aiokafka import AIOKafkaProducer
import asyncio

async def send_one():
    producer = AIOKafkaProducer(
        bootstrap_servers='localhost:9092')
    # Get cluster layout and initial topic/partition leadership information
    await producer.start()
    try:
        # Produce message
        await producer.send_and_wait("my_topic", b"Super message")
    finally:
        # Wait for all pending messages to be delivered or expire.
        await producer.stop()

asyncio.run(send_one())

Transactional producers allow sending messages to one or more topics such that they only become visible to consumers after the transaction is committed. To use this, you must set a transactional_id.

Requirements and Behavior:

• Setting transactional_id automatically enables idempotence.
• For durability, topics used in transactions should have replication.factor $\ge 3$ and min.insync.replicas $\ge 2$.
• Consumers must be configured to read only committed messages.
• transactional_id enables recovery across multiple sessions. It should be unique to each producer instance. If a new instance uses the same ID, the previous instance will raise a non-retriable ProducerFenced error.
• You can also commit consumer offsets as part of the same transaction using send_offsets_to_transaction.

# Basic Transaction
producer = aiokafka.AIOKafkaProducer(
    bootstrap_servers='localhost:9092',
    transactional_id="transactional_test")
await producer.start()
try:
    async with producer.transaction():
        res = await producer.send_and_wait(
            "test-topic", b"Super transactional message")
finally:
    await producer.stop()

# Committing offsets within a transaction
async with producer.transaction():
    commit_offsets = {
        TopicPartition("some-topic", 0): 100
    }
    await producer.send_offsets_to_transaction(
        commit_offsets, "some-consumer-group")

When group reassignment (rebalancing) occurs, you can use a ConsumerRebalanceListener to perform application-level logic like state cleanup or manual offset commits.

Warning: Avoid Deadlocks The consumer awaits the listener's handlers and will block subsequent calls to getmany() or getone(). If your listener attempts to acquire a lock that is already held by the main loop (which is currently awaiting the consumer), a deadlock will occur. Always ensure calls like consumer.getmany() are outside the lock used within the listener.

lock = asyncio.Lock()
consumer = AIOKafkaConsumer(...)

class MyRebalancer(aiokafka.ConsumerRebalanceListener):
    async def on_partitions_revoked(self, revoked):
        async with lock:
            pass  # Perform cleanup

    async def on_partitions_assigned(self, assigned):
        pass

async def main():
    consumer.subscribe("topic", listener=MyRebalancer())
    while True:
        # IMPORTANT: The consumer call must be outside the lock
        async with lock:
            # process other shared state
            pass
        msgs = await consumer.getmany(timeout_ms=1000)
        # process messages