How can Kafka avoid duplicate messages?
A Kafka consumer is a component in Apache Kafka that reads (consumes) messages from Kafka topics. Simply put, if a producer sends data to Kafka, the consumer is the one that receives and processes that data.
This guide explains how consumers work, how consumer groups share partitions, how offsets and commits affect delivery guarantees, and how to handle rebalancing, shutdown, errors and threads, with formatted Java and Spring Boot examples. It pairs with Kafka Producer and builds on Kafka Partition.
1. What Is a Kafka Consumer?
Think of Kafka as a newspaper delivery system. The producer publishes newspapers, Kafka stores them, and the consumer reads the newspapers whenever it wants. Unlike a real newspaper, reading does not remove the paper: other readers can read the same copy, and the same reader can go back to an older edition.
Order Service Kafka topic "orders" Consumers
(producer) -----> [P0] [P1] [P2] -----> Billing Service
-----> Email Service2. How Does a Kafka Consumer Work?
A Kafka consumer connects to a Kafka cluster and subscribes to one or more topics. It then runs a poll loop, continuously asking Kafka for new messages and processing them:
+--------------------------------------------------+
| |
v |
poll() --> get a batch of records --> process each --> commit offsets
(fetch from the partition leaders) (save progress)- The consumer asks Kafka for new messages with
poll(). - Kafka returns the next batch of messages, in order within each partition.
- The consumer processes the messages.
- The consumer keeps track of what it has already read by committing offsets.
Calling poll() regularly also tells Kafka that the consumer is still alive and working.
3. Key Features of a Kafka Consumer
| Feature | Explanation |
|---|---|
| Pull-based consumption | Consumers pull messages instead of Kafka pushing them, so each consumer reads at the speed it can handle |
| Consumer groups | Consumers can work together in a group; each message is read by only one consumer in the group, which balances the load |
| Message ordering | Messages are consumed in the same order they were written, within a single partition |
| Offset management | Each group tracks its position with offsets, so consumers resume where they left off, even after a failure |
| Fault tolerance | If a consumer crashes, another consumer in the same group takes over its partitions and continues from the last committed offset |
| Scalability | Add more consumers to a group and Kafka automatically spreads the partitions among them |
| Delivery guarantees | At-most-once, at-least-once or exactly-once processing, depending on configuration |
| Replay | Consumers can move back to earlier offsets and read messages again while they are retained |
4. Writing a Kafka Consumer in Java
Add the Kafka client library to your project:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>4.0.0</version>
</dependency>This consumer reads order messages from the orders topic as part of the billing-service group:
package com.example.kafka;
import java.time.Duration;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
public class OrderConsumer {
public static void main(String[] args) {
// 1. Configure the consumer
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "billing-service");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
// 2. Create the consumer and subscribe to the topic
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(List.of("orders"));
// 3. Poll loop
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Order %s -> %s (partition=%d, offset=%d)%n",
record.key(), record.value(), record.partition(), record.offset());
createInvoice(record.value());
}
// 4. Save progress after the batch is processed
if (!records.isEmpty()) {
consumer.commitSync();
}
}
}
}
private static void createInvoice(String order) {
// business logic
}
}| Part | What it does |
|---|---|
group.id | Name of the consumer group; consumers with the same ID share the work |
| Deserializers | Turn the stored bytes back into Java objects (here, Strings) |
auto.offset.reset=earliest | A brand-new group starts from the oldest available message |
subscribe() | Joins the group; Kafka assigns partitions to this consumer |
poll() | Returns the next batch of records, waiting up to 500 ms if none are ready |
commitSync() | Saves the group’s position only after processing succeeds |
5. Important Consumer Settings
| Setting | Default | Meaning |
|---|---|---|
bootstrap.servers | – | Brokers to connect to first (required) |
group.id | – | Consumer group name; needed for subscribe() and commits |
key.deserializer / value.deserializer | – | How to convert bytes to objects (required) |
auto.offset.reset | latest | Where to start when the group has no committed offset |
enable.auto.commit | true | Commit offsets automatically every auto.commit.interval.ms (5 s) |
max.poll.records | 500 | Maximum records returned by one poll() |
max.poll.interval.ms | 300000 (5 min) | Maximum time between polls before the consumer is removed from the group |
session.timeout.ms | 45000 | How long without heartbeats before the consumer is considered dead |
fetch.min.bytes / fetch.max.wait.ms | 1 / 500 | How much data to wait for, and for how long, before a fetch returns |
isolation.level | read_uncommitted | Set to read_committed to skip messages from aborted transactions |
6. Consumer Groups and Partition Assignment
Consumers with the same group.id form a consumer group. Kafka assigns each partition of the subscribed topics to exactly one member of the group:
| Partitions | Consumers in group | Result |
|---|---|---|
| 6 | 1 | One consumer reads all 6 partitions |
| 6 | 3 | Each consumer reads 2 partitions |
| 6 | 6 | Each consumer reads 1 partition |
| 6 | 8 | 6 consumers read 1 partition each; 2 consumers stay idle as standbys |
Different groups are completely independent: a billing-service group and an email-service group both receive every message of the topic.
The partition.assignment.strategy setting chooses how partitions are divided. The Java client’s default list includes the cooperative sticky assignor, which keeps existing assignments where possible and moves only the partitions that need to move. Kafka 4.0 also offers a newer group protocol (group.protocol=consumer) in which the broker calculates the assignment.
7. Offsets and Commits
Each record has an offset, its position in the partition. The group saves its progress by committing the offset of the next record to read. After a restart or rebalance, reading continues from the committed offset.
| Commit style | How | Trade-off |
|---|---|---|
| Automatic | enable.auto.commit=true; the client commits every 5 seconds during poll() | Simple; a crash can cause some repeated messages |
| Manual, synchronous | commitSync() after processing | Reliable; waits for the broker |
| Manual, asynchronous | commitAsync() after processing | Faster; failed commits are not retried |
See Kafka Offset for committing specific offsets, resetting offsets and replaying data.
8. Delivery Guarantees
| Guarantee | How to get it | Risk |
|---|---|---|
| At-most-once | Commit before processing | Messages can be lost if the consumer crashes during processing |
| At-least-once | Commit after processing (the usual choice) | Messages can be processed twice after a crash |
| Exactly-once | Read-process-write with a transactional producer and isolation.level=read_committed | Only for Kafka-to-Kafka pipelines; more setup |
With at-least-once, make processing idempotent: for example, store the order ID with each invoice and skip orders that already have one. Then a repeated message causes no harm.
9. Rebalancing and How to Avoid Surprises
A rebalance moves partitions between the members of a group. It happens when a consumer joins, leaves, crashes or is considered stuck. During a rebalance, the affected partitions pause briefly.
| Cause | Setting involved | How to avoid surprises |
|---|---|---|
| Processing a batch takes too long | max.poll.interval.ms (5 min) | Lower max.poll.records or speed up processing |
| Consumer stops sending heartbeats | session.timeout.ms (45 s) | Check for long garbage-collection pauses or network problems |
| Instances restart during deployments | group.instance.id | Static membership: a restarted instance gets its partitions back without a full rebalance |
| Consumer leaves without closing | – | Always call close() so the group reacts immediately |
To run code when partitions are assigned or taken away, for example to commit progress, pass a ConsumerRebalanceListener to subscribe().
10. Shutting Down a Consumer Cleanly
A consumer should leave its group cleanly when the application stops, so its partitions are reassigned immediately and its last offsets are committed. wakeup() is the only consumer method that is safe to call from another thread; it makes a blocked poll() throw a WakeupException:
public class GracefulConsumer {
public static void main(String[] args) {
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(ConsumerProps.create("billing-service"));
Thread mainThread = Thread.currentThread();
// Runs when the JVM is asked to stop (for example Ctrl+C or a container stop)
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
consumer.wakeup();
try {
mainThread.join(); // wait for the poll loop to finish
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}));
try {
consumer.subscribe(List.of("orders"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
records.forEach(record -> System.out.println("Processing " + record.value()));
consumer.commitAsync();
}
} catch (WakeupException e) {
// expected during shutdown
} finally {
try {
consumer.commitSync(); // final, reliable commit
} finally {
consumer.close(); // leave the group immediately
}
}
}
}Here ConsumerProps.create() is a small helper that returns the same properties as in section 4.
11. Reading JSON Messages
Most messages carry JSON. A small custom deserializer turns the bytes into a Java object, using the Jackson library:
package com.example.kafka;
import java.math.BigDecimal;
public record OrderEvent(String orderId, String customerId, BigDecimal amount) {
}package com.example.kafka;
import java.io.IOException;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import com.fasterxml.jackson.databind.ObjectMapper;
public class OrderEventDeserializer implements Deserializer<OrderEvent> {
private final ObjectMapper mapper = new ObjectMapper();
@Override
public OrderEvent deserialize(String topic, byte[] data) {
if (data == null) {
return null;
}
try {
return mapper.readValue(data, OrderEvent.class);
} catch (IOException e) {
throw new SerializationException("Invalid order JSON on topic " + topic, e);
}
}
}Register it with value.deserializer=com.example.kafka.OrderEventDeserializer and declare the consumer as KafkaConsumer<String, OrderEvent>.
12. Handling Errors and Bad Messages
Two kinds of problems are common:
- A message that cannot be deserialized (a “poison pill”), such as broken JSON.
poll()throws aRecordDeserializationException, which includes the partition and offset of the bad record. - Processing fails, for example because a database is temporarily down, or because the data is invalid.
try {
ConsumerRecords<String, OrderEvent> records = consumer.poll(Duration.ofMillis(500));
process(records);
} catch (RecordDeserializationException e) {
log.error("Bad record at {} offset {}", e.topicPartition(), e.offset(), e);
consumer.seek(e.topicPartition(), e.offset() + 1); // move past the bad record
}| Failure type | Recommended handling |
|---|---|
| Temporary (database or service down) | Retry a few times with a pause between attempts |
| Permanent (invalid data) | Send the message to a dead-letter topic with error details, then continue |
| Poison pill (cannot deserialize) | Log it, store the raw bytes if possible, and skip it |
Never let one bad message stop the whole consumer, and never skip messages silently: always log them or send them somewhere for investigation.
13. Consumers and Multithreading
KafkaConsumer is not thread-safe. Calling it from several threads at once causes an error. There are two simple ways to process in parallel:
| Approach | How | Notes |
|---|---|---|
| One consumer per thread | Start several consumers with the same group.id, each in its own thread | Simple; parallelism is limited by the partition count |
| One consumer, worker threads | One thread polls and hands records to a thread pool | More parallelism, but commits and ordering need careful handling |
public class ConsumerPool {
public static void main(String[] args) {
int threads = 3; // at most the number of partitions
ExecutorService executor = Executors.newFixedThreadPool(threads);
for (int i = 0; i < threads; i++) {
executor.submit(() -> {
try (KafkaConsumer<String, String> consumer =
new KafkaConsumer<>(ConsumerProps.create("billing-service"))) {
consumer.subscribe(List.of("orders"));
while (!Thread.currentThread().isInterrupted()) {
for (ConsumerRecord<String, String> record : consumer.poll(Duration.ofMillis(500))) {
System.out.printf("%s handled partition %d offset %d%n",
Thread.currentThread().getName(), record.partition(), record.offset());
}
consumer.commitSync();
}
}
});
}
}
}Each thread has its own consumer, so Kafka assigns different partitions to each one. Order is kept within every partition.
14. Kafka Consumer in Spring Boot
Spring Boot manages consumers for you. Add spring-kafka, configure it, and annotate a method with @KafkaListener:
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
group-id: billing-service
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializerpackage com.example.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class OrderListener {
private final InvoiceService invoiceService;
public OrderListener(InvoiceService invoiceService) {
this.invoiceService = invoiceService;
}
@KafkaListener(topics = "orders", concurrency = "3")
public void onOrder(ConsumerRecord<String, String> record) {
invoiceService.createInvoice(record.key(), record.value());
}
}concurrency = "3" starts three consumers in the group, each on its own thread. Spring commits offsets after the listener returns successfully. To retry failed records and then send them to a dead-letter topic, define an error handler bean:
package com.example.kafka;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;
@Configuration
public class KafkaErrorConfig {
@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
// retry 3 more times, 2 seconds apart, then publish to the dead-letter topic
return new DefaultErrorHandler(recoverer, new FixedBackOff(2000L, 3));
}
}Spring Boot attaches this error handler to the listener containers automatically. Failed records go to a topic with a .DLT suffix, such as orders.DLT.
15. Simple Example of Kafka Consumers
Imagine an online shopping application:
- The producer sends order details to a Kafka topic.
- A Kafka consumer reads the order messages, processes the order and updates the database.
- Another consumer sends a confirmation email.
| Consumer group | Instances | Responsibility |
|---|---|---|
order-processing | 3 | Saves orders and updates stock |
email-service | 1 | Sends confirmation emails |
analytics | 2 | Updates sales dashboards |
Each group receives every order and handles its own responsibility without affecting the others. If the email service is down for an hour, orders are still processed; when it comes back, it continues from its last committed offset and sends the delayed emails.
16. Monitoring Consumer Lag
Consumer lag is how many messages a group still has to read. Check it with the command-line tool:
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group billing-serviceGROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
billing-service orders 0 1520 1525 5
billing-service orders 1 1498 1498 0
billing-service orders 2 1310 1902 592Low, steady lag is healthy. Growing lag means consumers cannot keep up: add consumers (up to the number of partitions), speed up processing, or look for a stuck consumer on the lagging partition.
17. When Should You Use a Kafka Consumer?
- When you need to process large volumes of data.
- When systems must work independently of each other.
- When you need reliable and fault-tolerant message processing.
- When several applications need the same data.
- When scalability is important.
- When you may need to reprocess past events later.
18. Best Practices and Common Mistakes
| Do | Avoid |
|---|---|
| Commit offsets after processing | Committing before processing important data |
| Make processing idempotent | Assuming every message arrives exactly once |
| Use one consumer per thread | Sharing one KafkaConsumer across threads |
Keep each batch’s processing well under max.poll.interval.ms | Slow work that causes repeated rebalances |
| Close consumers on shutdown | Killing the process and waiting for the session timeout |
| Send bad messages to a dead-letter topic | Letting one bad message block a partition forever |
Set auto.offset.reset deliberately | Silently skipping old data with the default latest |
| Monitor consumer lag | Discovering backlogs only when users complain |
| Run no more consumers than partitions (plus standbys) | Expecting extra consumers to add speed |
19. Summary
A Kafka consumer reads messages from Kafka topics and processes them. It pulls records in a poll loop, tracks its progress with committed offsets, and works with other consumers in a group so each partition is handled by one member. Committing after processing gives at-least-once delivery; idempotent processing makes repeats harmless. Clean shutdown, sensible poll settings, dead-letter topics and lag monitoring keep consumers reliable, and Spring Boot’s @KafkaListener makes all of this simple to use.
In simple words, Kafka consumers help applications listen to events, process data and react in real time.
20. Frequently Asked Questions (FAQ)
Q1. What is a Kafka consumer in simple words?
A Kafka consumer is an application, or part of one, that reads messages from Kafka topics and processes them. It pulls messages from the brokers, remembers how far it has read using offsets, and can share the work with other consumers in a group.
Q2. Does Kafka push messages to consumers?
No. Kafka consumers pull messages by calling poll(). This lets each consumer read at its own speed, read in batches, and go back to earlier messages when needed.
Q3. What is a consumer group?
A set of consumers that share the same group.id and divide a topic’s partitions among themselves. Each partition is read by one member of the group, so every message is processed once per group. Different groups read the same data independently.
Q4. How many consumers should a group have?
At most as many as the topic has partitions; extra consumers stay idle. Start with enough consumers to keep lag low, and add more, up to the partition count, when load grows. One or two idle standbys can be useful for faster failover.
Q5. Can one consumer read from several topics?
Yes. Pass several topic names to subscribe(), or a regular expression such as Pattern.compile("orders-.*"). Each record tells you which topic it came from with record.topic().
Q6. What is the difference between subscribe() and assign()?
subscribe() joins a consumer group, and Kafka assigns partitions automatically and rebalances them when members change. assign() lets you pick exact partitions yourself, with no group management; it is useful for tools and special cases.
Q7. What happens when a consumer crashes?
It stops sending heartbeats. After session.timeout.ms (45 seconds by default), the group rebalances and another member takes over its partitions, starting from the last committed offset. Messages processed after that commit are processed again.
Q8. What is a rebalance?
A rebalance redistributes partitions among the members of a consumer group. It happens when consumers join, leave, crash, or take longer than max.poll.interval.ms between polls. The cooperative sticky assignor and static membership keep rebalances short.
Q9. Why does my consumer keep rebalancing?
Usually because processing one batch takes longer than max.poll.interval.ms (5 minutes by default), so Kafka thinks the consumer is stuck. Reduce max.poll.records, make processing faster, or increase the interval. Frequent restarts without close() also cause extra rebalances.
Q10. Should I use automatic or manual offset commits?
Manual commits after processing give the clearest at-least-once behaviour and are common in production. Automatic commits are fine for simple consumers that finish processing each batch inside the poll loop. Spring Kafka commits for you after the listener succeeds.
Q11. Why does my new consumer not read old messages?
Because auto.offset.reset defaults to latest, so a new group starts with new messages only. Set it to earliest to start from the oldest retained message. If the group already has committed offsets, this setting is ignored.
Q12. Is KafkaConsumer thread-safe?
No. Use each KafkaConsumer from a single thread. The only exception is wakeup(), which can be called from another thread to stop a blocked poll(). To process in parallel, run one consumer per thread.
Q13. How do I stop a consumer cleanly?
Call consumer.wakeup() from a shutdown hook. The poll loop then gets a WakeupException; catch it, commit the final offsets, and call close() so the consumer leaves the group immediately.
Q14. How can I avoid processing a message twice?
Kafka’s normal guarantee is at-least-once, so repeats are possible after crashes or rebalances. Make processing idempotent, for example by saving the message key or a unique event ID and skipping IDs already processed. For Kafka-to-Kafka flows, transactions give exactly-once results.
Q15. What is a poison pill message?
A message the consumer cannot deserialize, such as broken JSON. Without handling, it makes poll() fail again and again. Catch RecordDeserializationException, log the record, and seek() past it, or use Spring’s ErrorHandlingDeserializer.
Q16. What is a dead-letter topic?
A separate topic, such as orders.DLT, where messages that keep failing are sent with error details. The consumer can continue with the next messages, and the failed ones can be inspected, fixed and reprocessed later.
Q17. What is consumer lag and how do I check it?
Lag is the number of messages a group has not yet read: the latest offset minus the committed offset, per partition. Check it with:
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group billing-serviceQ18. What is static membership?
When each consumer instance sets a unique group.instance.id, Kafka remembers it. If the instance restarts within the session timeout, it gets the same partitions back without a full rebalance. This is helpful for rolling deployments.
Q19. How do I create a consumer in Spring Boot?
Add spring-kafka, set spring.kafka.consumer.group-id and deserializers, and annotate a method:
@KafkaListener(topics = "orders")
public void onOrder(String order) {
invoiceService.createInvoice(order);
}Use concurrency on the annotation to run several consumers in parallel.
Q20. Can a consumer read only committed transactional messages?
Yes. Set isolation.level=read_committed. The consumer then skips messages from aborted transactions and waits for open transactions to finish before reading past them. The default, read_uncommitted, returns all messages.