Consumer configuration in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka consumer configuration code?

Consider the following Kafka consumer configuration snippet in Java. What will be the value of props.getProperty("enable.auto.commit") after execution?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "false");
props.put("auto.commit.interval.ms", "1000");
String autoCommit = props.getProperty("enable.auto.commit");

A "true"

B "false"

C null

D Throws NullPointerException

💡 Hint

Check the value assigned to the key "enable.auto.commit" in the properties.

The property "enable.auto.commit" is explicitly set to "false" in the properties. So, calling getProperty returns "false".

❓ Predict Output

intermediate

What will be the output of this Kafka consumer poll timeout code?

Given the following Kafka consumer poll call, what is the type and meaning of the returned value?

Kafka

Consumer<String, String> consumer = new KafkaConsumer<>(props);
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
System.out.println(records.count());

A Throws TimeoutException

B Always zero because poll timeout is too short

C Number of partitions assigned to the consumer

D Number of records fetched in 500 milliseconds

💡 Hint

The poll method returns records fetched within the timeout duration.

The poll method waits up to the specified timeout to fetch records. The count() method returns how many records were fetched.

❓ Predict Output

advanced

What error does this Kafka consumer configuration cause?

Analyze the following Kafka consumer configuration snippet. What error will occur when creating the consumer?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "maybe");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

A IllegalArgumentException due to invalid enable.auto.commit value

B TimeoutException on consumer creation

C NullPointerException because group.id is missing

D No error, consumer created successfully

💡 Hint

The enable.auto.commit property expects "true" or "false" only.

The property "enable.auto.commit" must be "true" or "false". Setting it to "maybe" causes an IllegalArgumentException.

🧠 Conceptual

advanced

How many partitions will the consumer subscribe to after this code?

Given a Kafka topic with 4 partitions, what is the number of partitions assigned to the consumer after this subscription code?

Kafka

consumer.subscribe(Collections.singletonList("my-topic"));
Set<TopicPartition> assigned = consumer.assignment();
int count = assigned.size();
System.out.println(count);

A 0

B 4

C 1

D Throws IllegalStateException

💡 Hint

Partitions are assigned only after poll() is called.

The assignment set is empty until consumer.poll() is called to trigger partition assignment.

❓ Predict Output

expert

What is the output of this Kafka consumer manual commit code?

Consider the following code snippet for manual offset commit. What will be printed?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "false");
props.put("auto.offset.reset", "earliest");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props, new StringDeserializer(), new StringDeserializer());
consumer.subscribe(List.of("my-topic"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
    System.out.println(record.offset() + ":" + record.value());
}
consumer.commitSync();
System.out.println("Committed offsets");

A Prints nothing and then 'Committed offsets'

B Prints records but never prints 'Committed offsets'

C Prints all records with offsets and values, then 'Committed offsets'

D Throws CommitFailedException

💡 Hint

Manual commit commits offsets after processing records.

The code polls records, prints each offset and value, then commits offsets synchronously and prints confirmation.

Practice

1.

What is the purpose of setting group.id in Kafka consumer configuration?

easy

A. To specify the Kafka broker address

B. To define the message deserializer class

C. To identify the consumer group for coordinating message consumption

D. To set the maximum number of messages to fetch

Solution

1. Step 1: Understand the role of group.id

   The group.id identifies the consumer group that the consumer belongs to, allowing Kafka to coordinate message consumption among multiple consumers.

2. Step 2: Differentiate from other configurations

   Other options like broker address or deserializer do not relate to consumer group coordination.

3. Final Answer:

   To identify the consumer group for coordinating message consumption -> Option C

4. Quick Check:

   group.id = consumer group ID [OK]

Hint: group.id always sets the consumer group name [OK]

Common Mistakes:

• Confusing group.id with broker address
• Thinking group.id sets deserializer
• Assuming group.id controls message fetch size

2.

Which of the following is the correct way to set the Kafka consumer's auto.offset.reset property to start reading from the earliest message?

easy

A. "auto.offset.reset" = "latest"

B. "auto.offset.reset" = "earliest"

C. "auto.offset.reset" = "none"

D. "auto.offset.reset" = "first"

Solution

1. Step 1: Recall valid values for auto.offset.reset

   The valid values are "earliest", "latest", and "none". "earliest" means start from the oldest message.

2. Step 2: Identify the value for reading from earliest

   Setting it to "earliest" makes the consumer read from the beginning if no offset is found.

3. Final Answer:

   "auto.offset.reset" = "earliest" -> Option B

4. Quick Check:

   earliest = start from oldest message [OK]

Hint: earliest means start from oldest message [OK]

Common Mistakes:

• Using "first" which is invalid
• Confusing "latest" with earliest
• Setting "none" expecting to read messages

3.

Given the following Kafka consumer configuration snippet, what will be the output behavior when consuming messages?

props.put("enable.auto.commit", "false");
props.put("auto.offset.reset", "earliest");
props.put("group.id", "my-group");

Assuming this consumer starts fresh with no committed offsets, what happens?

medium

A. Consumer reads messages from the earliest offset but does not auto-commit offsets

B. Consumer reads messages from the latest offset and auto-commits offsets

C. Consumer throws an error due to missing deserializer configuration

D. Consumer reads messages from the earliest offset and auto-commits offsets

Solution

1. Step 1: Analyze enable.auto.commit setting

   It is set to "false", so the consumer will not automatically commit offsets after reading messages.

2. Step 2: Analyze auto.offset.reset and group.id

   With "earliest" and a new group "my-group", the consumer starts reading from the earliest messages available.

3. Final Answer:

   Consumer reads messages from the earliest offset but does not auto-commit offsets -> Option A

4. Quick Check:

   enable.auto.commit=false + earliest = read earliest, no auto commit [OK]

Hint: enable.auto.commit=false means manual offset commit [OK]

Common Mistakes:

• Assuming auto-commit is enabled by default
• Confusing earliest with latest offset reset
• Expecting error without deserializer (not shown here)

4.

Identify the error in this Kafka consumer configuration snippet:

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

medium

A. "enable.auto.commit" should be a String "true", not boolean true

B. "group.id" is missing and must be added

C. "bootstrap.servers" value is incorrect format

D. Deserializers should be set to IntegerDeserializer

Solution

1. Step 1: Check data types for properties

   Kafka consumer properties expect String values, so "enable.auto.commit" should be set as "true" (a String), not boolean true.

2. Step 2: Verify other properties

   "group.id" is present, "bootstrap.servers" format is correct, and deserializers are valid for String keys and values.

3. Final Answer:

   "enable.auto.commit" should be a String "true", not boolean true -> Option A

4. Quick Check:

   Properties values must be Strings [OK]

Hint: Property values must be strings, not booleans [OK]

Common Mistakes:

• Using boolean true instead of "true" string
• Omitting group.id property
• Incorrect bootstrap.servers format
• Wrong deserializer class for String data

5.

You want a Kafka consumer to read messages from multiple topics and manually commit offsets only after processing batches successfully. Which configuration settings should you use?

hard

A. "auto.offset.reset" = "latest" and use consumer.assign() for topics

B. "enable.auto.commit" = "true" and assign partitions manually

C. "enable.auto.commit" = "false" and set auto.offset.reset to "none"

D. "enable.auto.commit" = "false" and subscribe to multiple topics with consumer.subscribe(List.of(...))

Solution

1. Step 1: Understand manual offset commit requirement

   Setting "enable.auto.commit" to "false" disables automatic commits, allowing manual control after processing.

2. Step 2: Subscribe to multiple topics

   Using consumer.subscribe(List.of(...)) allows subscribing to multiple topics easily.

3. Step 3: Evaluate other options

   "enable.auto.commit" = "true" and assign partitions manually enables auto commit which conflicts with manual commit need; "auto.offset.reset" = "latest" and use consumer.assign() for topics uses latest offset reset which may skip old messages; "enable.auto.commit" = "false" and set auto.offset.reset to "none" sets offset reset to "none" which causes errors if no offset found.

4. Final Answer:

   "enable.auto.commit" = "false" and subscribe to multiple topics with consumer.subscribe(List.of(...)) -> Option D

5. Quick Check:

   manual commit + subscribe multiple topics = "enable.auto.commit" = "false" and subscribe to multiple topics with consumer.subscribe(List.of(...)) [OK]

Hint: Disable auto commit and subscribe to topics for manual control [OK]

Common Mistakes:

• Enabling auto commit when manual commit is needed
• Using assign() instead of subscribe() for multiple topics
• Setting auto.offset.reset to "none" causing errors