Subscribing to topics in Kafka - Step-by-Step Execution

Process Flow - Subscribing to topics

Create Kafka Consumer

↓

Configure Consumer Settings

↓

Subscribe to Topic(s)

↓

Poll for Messages

↓

Process Messages

↓

Repeat Polling or Exit

This flow shows how a Kafka consumer is created, configured, subscribes to topics, polls messages, and processes them repeatedly.

Execution Sample

Kafka

from kafka import KafkaConsumer
consumer = KafkaConsumer('my_topic', bootstrap_servers=['localhost:9092'])
for message in consumer:
    print(message.value.decode('utf-8'))

This code creates a Kafka consumer that subscribes to 'my_topic' and prints each message received.

Process Table

Step	Action	Evaluation	Result
1	Create KafkaConsumer object	KafkaConsumer('my_topic', ...)	Consumer created, ready to subscribe
2	Subscribe to 'my_topic'	Subscription set	Consumer listens to 'my_topic'
3	Poll for messages	Check if messages available	No messages yet, wait
4	Receive message	Message received from 'my_topic'	Message object with value bytes
5	Decode message.value	message.value.decode('utf-8')	String message content
6	Print message content	print output	Message content shown on screen
7	Repeat polling	Loop continues	Wait for next message or exit

💡 Consumer stops when program ends or interrupted

Status Tracker

Variable	Start	After Step 1	After Step 4	After Step 5	Final
consumer	None	KafkaConsumer object	KafkaConsumer with message ready	KafkaConsumer object	KafkaConsumer looping for messages
message	None	None	Message object with bytes	Message object with bytes	Next message or None

Key Moments - 3 Insights

Why do we decode message.value before printing?

Kafka messages are bytes, not strings. Decoding converts bytes to readable text, as shown in step 5 of the execution_table.

What happens if no messages are available when polling?

The consumer waits (blocks) until messages arrive, as seen in step 3 where polling checks but finds no messages yet.

How does the consumer know which topic to listen to?

The topic is specified when creating the consumer or by calling subscribe(), as in step 2 where 'my_topic' is set.

Visual Quiz - 3 Questions

Look at the execution_table, what is the state of 'message' after step 4?

A Decoded string message

B Message object with bytes

C None

D KafkaConsumer object

💡 Hint

Check the 'message' variable in variable_tracker after step 4 and execution_table row 4

At which step does the consumer start listening to 'my_topic'?

A Step 2

B Step 3

C Step 1

D Step 5

💡 Hint

Look at execution_table row 2 where subscription is set

If the topic name changes, which step in the execution_table would be affected?

A Step 3

B Step 5

C Step 1

D Step 6

💡 Hint

Topic is specified when creating the consumer object in step 1

Concept Snapshot

Kafka Consumer Subscription:
- Create KafkaConsumer with topic(s)
- Consumer subscribes to topics automatically
- Poll messages in a loop
- Messages are bytes, decode before use
- Process messages as they arrive
- Loop continues until stopped

Full Transcript

This visual execution shows how a Kafka consumer subscribes to topics and processes messages. First, a KafkaConsumer object is created with the topic name. Then the consumer subscribes to that topic and waits for messages. When messages arrive, they are received as bytes and decoded to strings before printing. The consumer keeps polling and processing messages until the program stops. Key points include decoding bytes to strings and understanding the subscription step.

Practice

1. What is the main purpose of the subscribe() method in Kafka consumers?

easy

A. To close the Kafka consumer connection

B. To send messages to Kafka topics

C. To listen to one or more Kafka topics and receive messages automatically

D. To create new Kafka topics

Solution

1. Step 1: Understand the role of subscribe()

   The subscribe() method is used by Kafka consumers to specify which topics they want to listen to.

2. Step 2: Identify what subscribe() enables

   Once subscribed, the consumer automatically receives messages from those topics when calling poll().

3. Final Answer:

   To listen to one or more Kafka topics and receive messages automatically -> Option C

4. Quick Check:

   subscribe() = listen to topics [OK]

Hint: subscribe() means start listening to topics [OK]

Common Mistakes:

• Confusing subscribe() with sending messages
• Thinking subscribe() creates topics
• Assuming subscribe() closes connections

2. Which of the following is the correct way to subscribe a Kafka consumer to multiple topics named "topicA" and "topicB"?

easy

A. consumer.subscribe(["topicA", "topicB"])

B. consumer.subscribe("topicA", "topicB")

C. consumer.subscribe("topicA" + "," + "topicB")

D. consumer.subscribe({"topicA", "topicB"})

Solution

1. Step 1: Recall the subscribe() method signature

   In Kafka, subscribe() takes a list (or collection) of topic names as an argument.

2. Step 2: Identify correct syntax for multiple topics

   Passing a list like ["topicA", "topicB"] is the correct way to subscribe to multiple topics.

3. Final Answer:

   consumer.subscribe(["topicA", "topicB"]) -> Option A

4. Quick Check:

   subscribe() needs a list for multiple topics [OK]

Hint: Use a list to subscribe to multiple topics [OK]

Common Mistakes:

• Passing multiple string arguments instead of a list
• Concatenating topic names into one string
• Using curly braces instead of square brackets

3. Given the following Kafka consumer code snippet, what will be the output if the consumer is subscribed to topic "news" and there are two messages "Hello" and "World" in the topic?

consumer.subscribe(["news"])
records = consumer.poll(timeout_ms=1000)
for record in records:
    print(record.value)

medium

A. ['Hello', 'World']

B. Hello\nWorld

C. HelloWorld

D. No output because poll() is missing a loop

Solution

1. Step 1: Understand subscribe() and poll() behavior

   The consumer subscribes to "news" and calls poll() once, which fetches available messages.

2. Step 2: Analyze the for loop printing each record

   The loop prints each message value on its own line, so "Hello" and "World" appear separately.

3. Final Answer:

   Hello\nWorld -> Option B

4. Quick Check:

   poll() returns messages, loop prints each [OK]

Hint: poll() returns messages; loop prints each on new line [OK]

Common Mistakes:

• Thinking poll() returns a list printed directly
• Assuming messages print concatenated without newline
• Believing poll() needs a loop to work at all

4. What is wrong with the following Kafka consumer code snippet?

consumer.subscribe("topic1")
while True:
    records = consumer.poll(1000)
    for record in records:
        print(record.value)

medium

A. subscribe() expects a list, not a single string

B. poll() must be called outside the loop

C. print() cannot be used inside the loop

D. The loop should be a for loop, not while

Solution

1. Step 1: Check subscribe() argument type

   The subscribe() method requires a list of topics, even if subscribing to one topic.

2. Step 2: Identify the error in argument

   Passing a single string instead of a list causes a type error or unexpected behavior.

3. Final Answer:

   subscribe() expects a list, not a single string -> Option A

4. Quick Check:

   subscribe() needs list argument [OK]

Hint: Always pass a list to subscribe(), even for one topic [OK]

Common Mistakes:

• Passing a string instead of a list to subscribe()
• Thinking poll() can't be inside a loop
• Misunderstanding loop type requirements

5. You want to subscribe a Kafka consumer to all topics starting with "logs-" dynamically, without listing each topic name. Which approach correctly achieves this?

hard

A. Use consumer.assign() with topic partitions matching "logs-"

B. Manually list all topics starting with "logs-" in a list and pass to subscribe()

C. Call subscribe() multiple times, once per topic starting with "logs-"

D. Use consumer.subscribe(Pattern.compile("logs-.*")) to subscribe by regex pattern

Solution

1. Step 1: Understand dynamic topic subscription

   Kafka consumers can subscribe to topics matching a regex pattern using subscribe(Pattern).

2. Step 2: Identify correct method for pattern subscription

   Using consumer.subscribe(Pattern.compile("logs-.*")) subscribes to all topics starting with "logs-" automatically.

3. Final Answer:

   Use consumer.subscribe(Pattern.compile("logs-.*")) to subscribe by regex pattern -> Option D

4. Quick Check:

   subscribe() supports regex patterns [OK]

Hint: Use regex pattern with subscribe() for dynamic topic matching [OK]

Common Mistakes:

• Listing topics manually instead of using regex
• Calling subscribe() multiple times incorrectly
• Using assign() which requires partitions, not topics