Java consumer client in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka consumer poll loop?

Consider the following Java Kafka consumer code snippet. What will be printed to the console when the consumer polls messages from the topic?

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(1000));
for (ConsumerRecord<String, String> record : records) {
    System.out.println("Received message: " + record.value());
}
consumer.close();

A Compilation error due to missing imports

B

Received message: Hello
Received message: World

C No output because poll returns empty records

D Runtime exception due to missing group.id property

💡 Hint

Assume the topic 'my-topic' has two messages: "Hello" and "World".

The consumer subscribes to 'my-topic' and polls messages. Since the topic contains two messages, the loop prints each message's value. The group.id is set, so no runtime error occurs. Imports are assumed present.

🧠 Conceptual

intermediate

What does the 'auto.offset.reset' property control in a Kafka consumer?

In Kafka consumer configuration, what is the role of the 'auto.offset.reset' property?

A It controls what happens when there is no initial offset or the current offset is invalid

B It sets the maximum number of messages to fetch in one poll

C It defines the timeout for consumer session heartbeat

D It specifies the serializer class for message keys

💡 Hint

Think about what happens if the consumer group is new or the offset is lost.

The 'auto.offset.reset' property tells the consumer where to start reading if no committed offset exists or the offset is out of range. Common values are 'earliest' or 'latest'.

🔧 Debug

advanced

Why does this Kafka consumer code throw a SerializationException?

Examine the code below. 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 property

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("topic1"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

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

B Because the group.id is invalid and causes consumer failure

C Because the topic name is incorrect and causes deserialization failure

D Because the poll timeout is too short and causes serialization error

💡 Hint

Check the deserializer properties carefully.

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

📝 Syntax

advanced

Which option correctly creates a Kafka consumer with generic types for String keys and Integer values?

Select the correct Java code snippet that creates a KafkaConsumer instance with String keys and Integer values.

A KafkaConsumer<String, Integer> consumer = new KafkaConsumer<String, Integer>(props);

B KafkaConsumer<String, Integer> consumer = new KafkaConsumer(props);

C KafkaConsumer<String, Integer> consumer = new KafkaConsumer<>(props);

D KafkaConsumer<String, Integer> consumer = new KafkaConsumer<Integer, String>(props);

💡 Hint

Consider Java's diamond operator and generic type matching.

Option C uses the diamond operator <> to infer types correctly. Option C is valid Java but verbose. Option C lacks generics on constructor, causing raw type warning. Option C swaps key and value types incorrectly.

🚀 Application

expert

How many partitions will the consumer group consume concurrently with this configuration?

Given a Kafka topic with 4 partitions and a consumer group with 3 consumers, how many partitions will each consumer consume concurrently?

A Each consumer will consume 4 partitions concurrently

B Each consumer will consume exactly 1 partition

C One consumer will consume all 4 partitions, others none

D Two consumers will consume 1 partition each, and one consumer will consume 2 partitions

💡 Hint

Partitions are divided among consumers in a group, but not always evenly.

Kafka assigns partitions to consumers in a group as evenly as possible. With 4 partitions and 3 consumers, two consumers get 1 partition each, and one consumer gets 2 partitions.

Practice

1.

What is the main purpose of a Java consumer client in Kafka?

easy

A. To create Kafka topics

B. To send messages to Kafka topics

C. To read messages from Kafka topics

D. To delete Kafka topics

Solution

1. Step 1: Understand Kafka client roles

   Kafka clients include producers (send messages) and consumers (read messages).

2. Step 2: Identify the consumer's role

   The Java consumer client reads messages from Kafka topics to process or display them.

3. Final Answer:

   To read messages from Kafka topics -> Option C

4. Quick Check:

   Consumer = reads messages [OK]

Hint: Consumer reads, producer writes messages in Kafka [OK]

Common Mistakes:

• Confusing consumer with producer
• Thinking consumer creates topics
• Assuming consumer deletes topics

2.

Which of the following is the correct way to subscribe a Kafka consumer to a topic named my-topic?

consumer._____ (Collections.singletonList("my-topic"));

easy

A. subscribe

B. assign

C. poll

D. commitSync

Solution

1. Step 1: Understand consumer methods

   The subscribe method registers the consumer to listen to topics.

2. Step 2: Match method to code

   Using consumer.subscribe(Collections.singletonList("my-topic")) correctly subscribes to the topic.

3. Final Answer:

   subscribe -> Option A

4. Quick Check:

   Subscribe = listen to topics [OK]

Hint: Use subscribe() to listen to topics, not assign() [OK]

Common Mistakes:

• Using assign() instead of subscribe()
• Calling poll() to subscribe
• Using commitSync() incorrectly

3.

What will the following Java Kafka consumer code print if the topic has one message with value "Hello Kafka"?

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

medium

A. Hello Kafka

B. null

C. ConsumerRecords object

D. No output

Solution

1. Step 1: Poll messages from topic

   The poll method fetches available messages; here it gets one message.

2. Step 2: Print each message value

   The loop prints the value() of each record, which is "Hello Kafka".

3. Final Answer:

   Hello Kafka -> Option A

4. Quick Check:

   Print record.value() = Hello Kafka [OK]

Hint: poll() fetches messages; print record.value() to see content [OK]

Common Mistakes:

• Printing the record object instead of value
• Expecting poll() to return null if no messages
• Confusing key and value fields

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", "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(100);
for (ConsumerRecord<String, String> record : records) {
    System.out.println(record.value());
}

medium

A. group.id property is missing

B. subscribe() should use assign()

C. Deserializers are incorrectly set

D. poll() argument should be Duration, not int

Solution

1. Step 1: Check poll() method signature

   In modern Kafka clients, poll requires a Duration argument, not an int.

2. Step 2: Identify incorrect argument type

   The code uses poll(100) (int), should be poll(Duration.ofMillis(100)).

3. Final Answer:

   poll() argument should be Duration, not int -> Option D

4. Quick Check:

   poll() needs Duration argument [OK]

Hint: Use Duration.ofMillis() with poll(), not plain int [OK]

Common Mistakes:

• Passing int to poll() instead of Duration
• Missing group.id property
• Using assign() instead of subscribe()

5.

You want to consume messages from two topics topicA and topicB and print their keys and values. Which code snippet correctly subscribes and processes messages?

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");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer._____ (Arrays.asList("topicA", "topicB"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("Key: " + record.key() + ", Value: " + record.value());
    }
}

hard

A. assign

B. subscribe

C. poll

D. commitSync

Solution

1. Step 1: Choose method to listen to multiple topics

   subscribe accepts a list of topics to listen to, while assign assigns specific partitions.

2. Step 2: Confirm correct usage in code

   Using consumer.subscribe(Arrays.asList("topicA", "topicB")) subscribes to both topics properly.

3. Final Answer:

   subscribe -> Option B

4. Quick Check:

   subscribe() listens to multiple topics [OK]

Hint: Use subscribe() with list to consume multiple topics [OK]

Common Mistakes:

• Using assign() instead of subscribe() for topics
• Calling poll() to subscribe
• Forgetting to print keys