Subscribing to topics in Kafka - Commands & Configuration

Introduction

When you want to receive messages from a Kafka topic, you subscribe to it. This lets your application listen and react to new data as it arrives.

When you want to process user activity logs in real-time.

When you need to update a dashboard with live data from sensors.

When you want to trigger actions based on events like orders placed.

When you want to build a chat app that shows messages instantly.

When you want to collect and analyze streaming data continuously.

Commands

This command starts a Kafka consumer that connects to the Kafka server at localhost on port 9092. It subscribes to the topic named 'example-topic' and reads all messages from the start of the topic.

Terminal

kafka-console-consumer --bootstrap-server localhost:9092 --topic example-topic --from-beginning

Expected Output

This will print all messages from 'example-topic' to the terminal as they arrive, starting from the oldest message.

For example:

Hello Kafka

New event received

User logged in

→

--bootstrap-server - Specifies the Kafka server address to connect to.

→

--topic - Specifies the topic to subscribe to.

→

--from-beginning - Reads all messages from the start of the topic instead of only new messages.

This command subscribes to 'example-topic' but only shows new messages arriving after the command starts, ignoring old messages.

Terminal

kafka-console-consumer --bootstrap-server localhost:9092 --topic example-topic

Expected Output

Only new messages sent to 'example-topic' after this command runs will appear here.

For example:

New order placed

Payment processed

→

--bootstrap-server - Specifies the Kafka server address.

→

--topic - Specifies the topic to listen to.

Key Concept

If you remember nothing else from this pattern, remember: subscribing to a Kafka topic lets your app receive messages as they come in, either from the start or only new ones.

Common Mistakes

Not specifying the correct Kafka server address with --bootstrap-server.

The consumer cannot connect to Kafka and will fail to receive messages.

Always provide the correct Kafka server address and port using --bootstrap-server.

Forgetting to specify the topic name with --topic.

Kafka consumer will not know which topic to listen to and will error out.

Always include the --topic flag followed by the exact topic name.

Expecting old messages without using --from-beginning.

By default, the consumer only shows new messages after it starts.

Add --from-beginning if you want to read all existing messages from the topic.

Summary

Use kafka-console-consumer with --bootstrap-server and --topic to subscribe to a Kafka topic.

Add --from-beginning to read all messages from the start, otherwise only new messages appear.

Check your Kafka server address and topic name carefully to avoid connection errors.

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