First message (produce and consume) in Kafka - Step-by-Step Execution

Process Flow - First message (produce and consume)

Start Producer

↓

Create message

↓

Send message to Kafka topic

↓

Message stored in topic partition

↓

Start Consumer

↓

Poll topic for messages

↓

Receive first message

↓

Process and print message

↓

End

The producer creates and sends a message to a Kafka topic. The consumer then polls the topic, receives the first message, and processes it.

Execution Sample

Kafka

from kafka import KafkaProducer, KafkaConsumer

producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('my_topic', b'Hello Kafka')
producer.flush()

consumer = KafkaConsumer('my_topic', bootstrap_servers='localhost:9092', auto_offset_reset='earliest', consumer_timeout_ms=1000)
for msg in consumer:
    print(msg.value.decode())
    break

This code sends a single message 'Hello Kafka' to 'my_topic' and then consumes the first message from that topic, printing it.

Process Table

Step	Action	Evaluation	Result
1	Start Producer	Producer connected to localhost:9092	Ready to send messages
2	Create message	Message content: b'Hello Kafka'	Message prepared
3	Send message	Send to topic 'my_topic'	Message sent to Kafka broker
4	Flush producer	Ensure message is sent	Message committed to topic partition
5	Start Consumer	Consumer connected to localhost:9092	Ready to poll messages
6	Poll topic 'my_topic'	Check for messages from earliest offset	Message received: b'Hello Kafka'
7	Process message	Decode bytes to string	Output: 'Hello Kafka'
8	Break loop	Stop after first message	Consumer stops polling
9	End	No more messages or stopped	Program ends

💡 Consumer stops after receiving and processing the first message.

Status Tracker

Variable	Start	After Step 2	After Step 3	After Step 6	Final
producer	None	KafkaProducer instance	KafkaProducer instance	KafkaProducer instance	KafkaProducer instance
message	None	b'Hello Kafka'	b'Hello Kafka'	b'Hello Kafka'	b'Hello Kafka'
consumer	None	None	None	KafkaConsumer instance	KafkaConsumer instance
msg.value	None	None	None	b'Hello Kafka'	b'Hello Kafka'
output	None	None	None	None	'Hello Kafka'

Key Moments - 3 Insights

Why does the consumer use 'auto_offset_reset=earliest'?

Because the consumer needs to read the first message from the beginning of the topic. Without this, it might start reading only new messages sent after it starts, missing the first message (see execution_table step 6).

Why do we call 'producer.flush()' after sending the message?

Flush ensures the message is actually sent to Kafka before the program continues. Without flush, the message might stay in a buffer and not be sent immediately (see execution_table step 4).

Why does the consumer break after processing one message?

To stop the loop after receiving the first message. Otherwise, the consumer would keep waiting for more messages indefinitely (see execution_table step 8).

Visual Quiz - 3 Questions

Look at the execution table, what is the content of 'msg.value' at step 6?

A None

B b'Hello Kafka'

C String 'Hello Kafka'

D Empty bytes b''

💡 Hint

Check the 'msg.value' variable in variable_tracker after step 6.

At which step does the producer ensure the message is sent to Kafka?

A Step 2

B Step 3

C Step 4

D Step 5

💡 Hint

Look at the action 'Flush producer' in execution_table.

If we remove the 'break' in the consumer loop, what happens?

A Consumer keeps polling for more messages

B Consumer stops after first message anyway

C Consumer crashes immediately

D Producer sends more messages automatically

💡 Hint

See execution_table step 8 about breaking the loop.

Concept Snapshot

Kafka first message flow:
- Producer creates and sends message to topic
- Call producer.flush() to ensure send
- Consumer connects with auto_offset_reset='earliest'
- Consumer polls topic and receives messages
- Break loop after first message to stop
- Decode message bytes to string for use

Full Transcript

This example shows how a Kafka producer sends a first message to a topic and how a consumer reads that first message. The producer connects to the Kafka broker, creates a message 'Hello Kafka', sends it to the topic 'my_topic', and flushes to ensure delivery. The consumer connects to the same broker, subscribes to 'my_topic' with 'auto_offset_reset' set to 'earliest' so it reads from the beginning. It polls the topic, receives the first message, decodes it from bytes to string, prints it, and then breaks the loop to stop. Key points include flushing the producer to send messages immediately, setting the consumer to read from earliest offset, and breaking the consumer loop after the first message to avoid waiting forever.

Practice

1. What is the main purpose of producing a message in Kafka?

easy

A. To send data to a Kafka topic

B. To read data from a Kafka topic

C. To delete messages from a Kafka topic

D. To create a new Kafka topic

Solution

1. Step 1: Understand Kafka producing

   Producing means sending messages to a Kafka topic as bytes.

2. Step 2: Differentiate producing from consuming

   Consuming reads messages, but producing sends them.

3. Final Answer:

   To send data to a Kafka topic -> Option A

4. Quick Check:

   Producing = Sending messages [OK]

Hint: Producing means sending messages to Kafka topics [OK]

Common Mistakes:

• Confusing producing with consuming
• Thinking producing deletes messages
• Assuming producing creates topics

2. Which of the following is the correct way to create a Kafka producer in Python using the kafka-python library?

easy

A. producer = KafkaConsumer(topic='mytopic')

B. producer = KafkaConsumer(bootstrap_servers='localhost:9092')

C. producer = KafkaProducer(topic='mytopic')

D. producer = KafkaProducer(bootstrap_servers='localhost:9092')

Solution

1. Step 1: Identify Kafka producer creation syntax

   KafkaProducer is created with bootstrap_servers parameter specifying Kafka server address.

2. Step 2: Eliminate consumer and wrong parameters

   KafkaConsumer is for consuming, not producing. Topic is not a parameter for producer creation.

3. Final Answer:

   producer = KafkaProducer(bootstrap_servers='localhost:9092') -> Option D

4. Quick Check:

   Producer uses KafkaProducer with bootstrap_servers [OK]

Hint: Producer uses KafkaProducer with bootstrap_servers [OK]

Common Mistakes:

• Using KafkaConsumer instead of KafkaProducer
• Passing topic during producer creation
• Missing bootstrap_servers parameter

3. Consider this Python code snippet using kafka-python:

from kafka import KafkaProducer, KafkaConsumer

producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('test-topic', b'Hello Kafka')
producer.flush()

consumer = KafkaConsumer('test-topic', bootstrap_servers='localhost:9092', auto_offset_reset='earliest', consumer_timeout_ms=1000)
for msg in consumer:
    print(msg.value.decode())

What will be the output when this code runs?

medium

A. Error: topic not found

B. Hello Kafka

C. No output, consumer times out

D. b'Hello Kafka'

Solution

1. Step 1: Analyze producer sending message

   Producer sends bytes message b'Hello Kafka' to 'test-topic' and flushes to ensure delivery.

2. Step 2: Analyze consumer reading message

   Consumer reads from 'test-topic' from earliest offset and decodes bytes to string before printing.

3. Final Answer:

   Hello Kafka -> Option B

4. Quick Check:

   Consumer prints decoded message string [OK]

Hint: Consumer decodes bytes to string before printing [OK]

Common Mistakes:

• Expecting raw bytes output
• Assuming consumer misses message due to timeout
• Thinking topic does not exist

4. You wrote this code to consume messages but it raises an error:

consumer = KafkaConsumer('my-topic')
for message in consumer:
    print(message.value.decode())

What is the most likely cause of the error?

medium

A. Using decode() on message.value which is already a string

B. Not calling consumer.poll() before iterating

C. Missing bootstrap_servers parameter when creating KafkaConsumer

D. Topic name should be a list, not a string

Solution

1. Step 1: Check KafkaConsumer creation requirements

   KafkaConsumer requires bootstrap_servers parameter to connect to Kafka cluster.

2. Step 2: Validate other options

   decode() is needed because message.value is bytes; consumer.poll() is optional; topic can be string or list.

3. Final Answer:

   Missing bootstrap_servers parameter when creating KafkaConsumer -> Option C

4. Quick Check:

   KafkaConsumer needs bootstrap_servers [OK]

Hint: Always specify bootstrap_servers when creating KafkaConsumer [OK]

Common Mistakes:

• Omitting bootstrap_servers parameter
• Assuming message.value is string without decoding
• Thinking poll() is mandatory before iteration

5. You want to produce and consume the first message in a Kafka topic named 'orders'. Which sequence of steps correctly achieves this in Python using kafka-python?

hard

A. Create KafkaProducer with bootstrap_servers, send message as bytes, flush; create KafkaConsumer with bootstrap_servers and auto_offset_reset='earliest', then read and decode messages

B. Create KafkaConsumer first, then KafkaProducer; send message as string; consumer reads without decoding

C. Create KafkaProducer without bootstrap_servers, send string message; create KafkaConsumer with bootstrap_servers, read messages as bytes

D. Create KafkaProducer with topic parameter, send message; create KafkaConsumer without bootstrap_servers, read and decode messages

Solution

1. Step 1: Properly create KafkaProducer and send message

   KafkaProducer must be created with bootstrap_servers; messages sent as bytes and flushed to ensure delivery.

2. Step 2: Properly create KafkaConsumer and read messages

   KafkaConsumer needs bootstrap_servers and auto_offset_reset='earliest' to read from the first message; decode bytes to string when reading.

3. Final Answer:

   Create KafkaProducer with bootstrap_servers, send message as bytes, flush; create KafkaConsumer with bootstrap_servers and auto_offset_reset='earliest', then read and decode messages -> Option A

4. Quick Check:

   Correct producer and consumer setup with bytes and decoding [OK]

Hint: Producer sends bytes and flushes; consumer reads earliest and decodes [OK]

Common Mistakes:

• Creating producer or consumer without bootstrap_servers
• Sending string instead of bytes
• Not setting auto_offset_reset to 'earliest'
• Not decoding message bytes before printing