Consumer API basics in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka consumer poll code?

Consider the following Kafka consumer code snippet in Java. What will be printed to the console?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    System.out.println(record.key() + ":" + record.value());
}
consumer.close();

A Throws a SerializationException due to missing deserializer configuration

B Throws a NullPointerException because poll duration is too short

C Prints all key:value pairs from 'my-topic' available within 100ms poll duration

D Prints nothing if no messages are available within 100ms poll duration

💡 Hint

Think about what happens if the topic has no messages ready during the poll.

The poll method waits up to the specified duration for messages. If no messages are available, it returns an empty ConsumerRecords object, so the loop prints nothing.

🧠 Conceptual

intermediate

What does the 'group.id' property control in Kafka consumer?

In Kafka consumer configuration, what is the role of the group.id property?

A It sets the maximum number of messages to fetch per poll

B It identifies the consumer group for load balancing and offset management

C It specifies the Kafka broker address to connect to

D It defines the serialization format for message keys

💡 Hint

Think about how Kafka manages multiple consumers reading from the same topic.

The group.id identifies the consumer group. Kafka uses this to distribute partitions among consumers and track offsets for fault tolerance.

🔧 Debug

advanced

Why does this Kafka consumer code cause a runtime error?

Examine the following Kafka consumer code snippet. Why does it throw a SerializationException at runtime?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "group1");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// Missing value.deserializer configuration

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("topic1"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    System.out.println(record.key() + ":" + record.value());
}
consumer.close();

A Because the value.deserializer property is missing, Kafka cannot deserialize message values

B Because the topic name is incorrect, causing topic not found error

C Because the poll duration is too short, causing timeout exceptions

D Because the group.id is invalid, causing consumer group join failure

💡 Hint

Check the deserializer configurations carefully.

Kafka needs both key and value deserializers to convert bytes to objects. Missing value.deserializer causes a SerializationException when reading message values.

📝 Syntax

advanced

Which option correctly commits offsets synchronously in Kafka consumer?

Which of the following code snippets correctly commits offsets synchronously after processing records?

A consumer.commitAsync();

B consumer.commitSync(Duration.ofSeconds(1));

C consumer.commitSync();

D consumer.commitAsync(Duration.ofSeconds(1));

💡 Hint

Look for the method that commits offsets synchronously without parameters.

The commitSync() method commits offsets synchronously and throws exceptions on failure. The other methods are asynchronous or invalid.

🚀 Application

expert

How many partitions will be assigned to each consumer in this group?

Suppose a Kafka topic has 6 partitions and there are 3 consumers in the same consumer group subscribed to this topic. How many partitions will each consumer be assigned?

A Each consumer will be assigned 2 partitions

B Each consumer will be assigned 3 partitions

C One consumer will get all 6 partitions, others get none

D Partitions are assigned randomly with no guarantee of distribution

💡 Hint

Think about how Kafka balances partitions among consumers in a group.

Kafka evenly distributes partitions among consumers in a group. With 6 partitions and 3 consumers, each gets 2 partitions.

Practice

1. What is the primary purpose of the Kafka Consumer API?

easy

A. To send messages to Kafka topics

B. To delete Kafka topics

C. To create Kafka topics

D. To read messages from Kafka topics

Solution

1. Step 1: Understand Kafka Consumer API role

   The Consumer API is designed to read or consume messages from Kafka topics.

2. Step 2: Differentiate from Producer API

   The Producer API sends messages, while Consumer API reads them.

3. Final Answer:

   To read messages from Kafka topics -> Option D

4. Quick Check:

   Consumer API = Read messages [OK]

Hint: Consumer API reads messages, Producer API sends messages [OK]

Common Mistakes:

• Confusing Consumer API with Producer API
• Thinking Consumer API creates or deletes topics
• Mixing up reading and writing operations

2. Which of the following is the correct way to create a Kafka consumer in Java?

easy

A. new KafkaProducer<>(properties);

B. new KafkaConsumer<>(properties);

C. KafkaConsumer.create(properties);

D. KafkaConsumer.start(properties);

Solution

1. Step 1: Recall Kafka consumer instantiation syntax

   Kafka consumer is created by calling the constructor: new KafkaConsumer<>(properties);

2. Step 2: Eliminate incorrect options

   Producer is for sending messages, so new KafkaProducer<>(properties); is wrong. Options C and D are not valid KafkaConsumer methods.

3. Final Answer:

   new KafkaConsumer<>(properties); -> Option B

4. Quick Check:

   Constructor new KafkaConsumer<>(props) = Correct [OK]

Hint: Use new KafkaConsumer<>(props) to create consumer [OK]

Common Mistakes:

• Using KafkaProducer instead of KafkaConsumer
• Calling non-existent static methods
• Confusing method names for creation

3. Given the following Java code snippet, what will be the output if the topic has two messages "Hello" and "World"?

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
    System.out.println(record.value());
}

medium

A. Compilation error

B. No output

C. Hello\nWorld

D. Runtime exception

Solution

1. Step 1: Understand consumer subscription and poll

   The consumer subscribes to "my-topic" and polls messages for 1 second.

2. Step 2: Process messages in the topic

   Since the topic has two messages "Hello" and "World", the poll will fetch both, and the loop prints each value.

3. Final Answer:

   Hello\nWorld -> Option C

4. Quick Check:

   poll() fetches messages = Hello and World printed [OK]

Hint: poll() fetches messages; loop prints each value [OK]

Common Mistakes:

• Assuming poll returns no messages immediately
• Confusing subscribe with assign
• Expecting compile or runtime errors without cause

4. Identify the error in this Kafka consumer code snippet:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "group1");
props.put("key.deserializer", "StringDeserializer");
props.put("value.deserializer", "StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("topic1"));
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records) {
    System.out.println(record.value());
}

medium

A. Deserializer class names are incomplete and will cause ClassNotFoundException

B. Missing call to consumer.close()

C. Incorrect subscription method used

D. poll() duration is too short and causes timeout

Solution

1. Step 1: Check deserializer configuration

   The deserializer class names must be fully qualified, e.g., "org.apache.kafka.common.serialization.StringDeserializer".

2. Step 2: Identify impact of incorrect class names

   Using "StringDeserializer" alone will cause ClassNotFoundException at runtime.

3. Final Answer:

   Deserializer class names are incomplete and will cause ClassNotFoundException -> Option A

4. Quick Check:

   Deserializer full class name required = Deserializer class names are incomplete and will cause ClassNotFoundException [OK]

Hint: Use full deserializer class names to avoid errors [OK]

Common Mistakes:

• Using short class names for deserializers
• Ignoring missing consumer.close() (not error here)
• Assuming poll duration causes errors

5. You want to process messages from multiple topics and commit offsets manually after processing. Which code snippet correctly sets up the Kafka consumer for this use case?

hard

A. props.put("enable.auto.commit", "false");
consumer.subscribe(Arrays.asList("topicA", "topicB"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
process(record.value());
}
consumer.commitSync();
}

B. props.put("enable.auto.commit", "true");
consumer.subscribe(Collections.singletonList("topicA"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
process(record.value());
}
}

C. props.put("enable.auto.commit", "false");
consumer.assign(Arrays.asList(new TopicPartition("topicA", 0)));
consumer.subscribe(Arrays.asList("topicB"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
process(record.value());
}
consumer.commitAsync();
}

D. props.put("enable.auto.commit", "false");
consumer.subscribe(Arrays.asList("topicA", "topicB"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
process(record.value());
}
}

Solution

1. Step 1: Disable auto commit for manual offset control

   Setting "enable.auto.commit" to "false" allows manual offset commits.

2. Step 2: Subscribe to multiple topics correctly

   Use consumer.subscribe(Arrays.asList("topicA", "topicB")) to listen to multiple topics.

3. Step 3: Commit offsets after processing

   Calling consumer.commitSync() after processing ensures offsets are saved manually.

4. Final Answer:

   Code snippet with manual commit, multiple topic subscription, and commitSync() -> Option A

5. Quick Check:

   Manual commit + multiple topics = props.put("enable.auto.commit", "false"); consumer.subscribe(Arrays.asList("topicA", "topicB")); while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord record : records) { process(record.value()); } consumer.commitSync(); } [OK]

Hint: Disable auto commit and call commitSync() after processing [OK]

Common Mistakes:

• Using auto commit when manual commit is needed
• Mixing assign() and subscribe() incorrectly
• Forgetting to commit offsets after processing