Apache Kafka is a distributed, partitioned, replicated commit log. Producers append records to topics and consumers read them at their own pace by tracking an offset. This page explains the moving parts behind the short producer and consumer scripts below, then adds a production-grade example with keys, acknowledgements and manual offset commits.
from kafka import KafkaProducer
import json
# Initialize the Kafka producer
producer = KafkaProducer(
bootstrap_servers=['localhost:9092'],
value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
# Send a message to the Kafka topic 'test_topic'
producer.send('test_topic', {'key': 'value'})
# Ensure all messages are sent before closing the producer
producer.flush()
producer.close()
KafkaProducer: Initializes the producer with a list of Kafka brokers. Here, it connects to
localhost:9092.
value_serializer: Serializes the message to JSON format before sending it to Kafka.
send: Sends the message to the specified topic (test_topic).
flush: Ensures all messages are sent.
close: Closes the producer after sending messages.
from kafka import KafkaConsumer
import json
# Initialize the Kafka consumer
consumer = KafkaConsumer(
'test_topic',
bootstrap_servers=['localhost:9092'],
auto_offset_reset='earliest',
enable_auto_commit=True,
value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)
# Read and print messages from the Kafka topic
for message in consumer:
print(f"Received message: {message.value}")
# Close the consumer when done
consumer.close()
KafkaConsumer: Initializes the consumer to connect to localhost:9092 and reads messages from
the test_topic.
auto_offset_reset: Sets the consumer to start reading from the earliest message.
enable_auto_commit: Automatically commits the offset of the messages.
value_deserializer: Deserializes the JSON messages received from Kafka.
for message in consumer: Continuously reads messages from the topic and prints them.
A producer serializes each record, picks a partition (by hashing the key, or round-robin when there is no key), batches records per partition and sends them to that partition's leader broker. The leader appends the batch to its log segment and followers in the in-sync replica set (ISR) copy it. With acks=all the producer receives an acknowledgement only after every in-sync replica has the data. Consumers pull from the leader, process records, and periodically commit the offset of the next record they expect to read; that committed offset is stored in the internal __consumer_offsets topic.
Partitions are the unit of parallelism and ordering. Kafka guarantees order only within a partition, so records that must be processed in order (all events for one customer, for example) should share a key. A consumer group spreads partitions across its members; adding consumers beyond the partition count leaves the extra consumers idle. When a member joins or leaves, the group rebalances and partitions are reassigned, which is why consumers should commit offsets before giving up a partition.
import json
from kafka import KafkaProducer, KafkaConsumer
producer = KafkaProducer(
bootstrap_servers=["localhost:9092"],
key_serializer=str.encode,
value_serializer=lambda v: json.dumps(v).encode("utf-8"),
acks="all", # wait for all in-sync replicas
linger_ms=10, # small batching delay
retries=5,
)
for i in range(10):
order = {"order_id": i, "customer": f"c{i % 3}", "amount": 10 * i}
producer.send("orders", key=order["customer"], value=order)
producer.flush()
consumer = KafkaConsumer(
"orders",
bootstrap_servers=["localhost:9092"],
group_id="billing",
enable_auto_commit=False,
auto_offset_reset="earliest",
value_deserializer=lambda b: json.loads(b.decode("utf-8")),
)
for msg in consumer:
print(msg.partition, msg.offset, msg.key, msg.value)
# ... write to a database here ...
consumer.commit() # commit only after the side effect succeededKeying by customer sends every order for one customer to the same partition, preserving their order. Disabling auto-commit and committing after processing gives at-least-once delivery: if the process crashes between the database write and the commit, the record is redelivered, so the write should be idempotent (an upsert keyed on order_id).
| Goal | Producer | Consumer |
|---|---|---|
| At most once | acks=0 or 1 | commit before processing |
| At least once | acks=all, retries | commit after processing |
| Exactly once (Kafka to Kafka) | idempotent + transactional producer | isolation.level=read_committed |
Exactly-once semantics in Kafka cover read-process-write cycles that stay inside Kafka. When the sink is an external database, the practical approach is at-least-once delivery plus idempotent writes.
Unlike a traditional message queue, Kafka does not delete a record when it is consumed. Records stay in the log until a retention limit is reached, either by time (retention.ms, seven days by default) or by size, or until compaction removes older values for the same key on a compacted topic. This is what lets a new consumer group read the full history from the earliest offset, and what lets an existing group rewind with kafka-consumer-groups --reset-offsets after a bug fix. It also means storage, not consumption, determines how long data is available, so retention should be sized for the longest replay you expect to need.
Compacted topics keep at least the latest value for every key indefinitely, which makes them suitable for changelogs and lookup tables such as the current address of each customer. Producing a record with a null value (a tombstone) marks the key for deletion during compaction.
flush() or close() in short scripts: buffered records are lost on exit.max.poll.interval.ms and triggers a rebalance.