Apache Kafka #

When a system needs to connect dozens of services exchanging data in real time, a point-to-point approach creates an unmanageable web of connections. Apache Kafka steps in as a distributed streaming platform that enables large-scale data exchange with low latency, without producers and consumers needing to know each other. Kafka isn’t just a message queue — it’s a distributed commit log that can store data for a configurable period, letting consumers re-read messages whenever needed. In the Java context, the Kafka ecosystem is very mature with a feature-rich official library, from simple producers to stream processing with Kafka Streams.

Kafka Basics #

Before writing a single line of code, it’s important to understand Kafka’s data model because it’s very different from traditional message brokers like RabbitMQ. Kafka uses an immutable log structure — messages that have been written can’t be modified or directly deleted.

Topics and Partitions #

A topic is a category or feed name where messages are published. Each topic is divided into one or more partitions. A partition is Kafka’s unit of parallelism — one partition can only be consumed by one consumer within the same consumer group at a time.

Topic: "order-events"
├── Partition 0: [msg-0] [msg-1] [msg-4] [msg-7]
├── Partition 1: [msg-2] [msg-5] [msg-8]
└── Partition 2: [msg-3] [msg-6] [msg-9]

Message order is only guaranteed within a single partition. If you need global ordering, use one partition — but that sacrifices parallelism. If ordering only needs to be guaranteed per entity (for example, all events for the same order ID must be in order), use a message key — Kafka will always send messages with the same key to the same partition.

Offsets #

Every message in a partition has an offset — an integer that increases monotonically. The offset is the consumer’s “bookmark” position. Consumers store the last processed offset, so they can resume from the correct position after a restart.

sequenceDiagram
    participant Producer
    participant Kafka as Kafka Broker
    participant Consumer

    Producer->>Kafka: publish("order-created", key="order-123")
    Kafka-->>Producer: offset=42 (partition=1)
    Consumer->>Kafka: poll() — starting from offset=40
    Kafka-->>Consumer: [msg offset=40, offset=41, offset=42]
    Consumer->>Kafka: commitOffset(partition=1, offset=43)

Brokers, Leaders, and Replicas #

Kafka runs as a cluster of one or more servers called brokers. Each partition has one leader and zero or more replicas. All reads and writes for a given partition go to its leader. Replicas follow the leader to provide fault tolerance.

flowchart TD
    P[Producer] --> B0[Broker 0\nLeader: P0, P2\nReplica: P1]
    P --> B1[Broker 1\nLeader: P1\nReplica: P0, P2]
    P --> B2[Broker 2\nLeader: —\nReplica: P0, P1, P2]

    B0 --> C1[Consumer Group A\nMember 1 — P0]
    B1 --> C2[Consumer Group A\nMember 2 — P1]
    B0 --> C3[Consumer Group A\nMember 3 — P2]

Setting Up Dependencies #

To use Kafka with Java, add the kafka-clients dependency to your project. With Maven:

<dependencies>
    <!-- Official Kafka client -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.7.0</version>
    </dependency>

    <!-- SLF4J for logging (required by the Kafka client) -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>2.0.12</version>
    </dependency>
</dependencies>

For Gradle:

dependencies {
    implementation 'org.apache.kafka:kafka-clients:3.7.0'
    implementation 'org.slf4j:slf4j-simple:2.0.12'
}

If you’re running Kafka locally for development, the easiest way is Docker:

# Run Kafka in KRaft mode (without Zookeeper, available since Kafka 3.3+)
docker run -d \
  --name kafka \
  -p 9092:9092 \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  apache/kafka:3.7.0

Producers #

The producer is responsible for sending messages to Kafka. Producer configuration affects throughput, latency, and delivery durability.

Producer Configuration #

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

public class OrderProducer {

    private final KafkaProducer<String, String> producer;
    private static final String TOPIC = "order-events";

    public OrderProducer() {
        Properties props = new Properties();

        // Bootstrap servers — the initial entry point to the cluster
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // Serializers for key and value
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // acks=all — wait for confirmation from all in-sync replicas
        // ANTI-PATTERN: acks=0 (fire-and-forget, messages can be lost)
        // ANTI-PATTERN: acks=1 (leader only, can be lost if the leader crashes before replication)
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        // Automatic retry on failure (transient errors)
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);

        // Idempotent producer — prevents duplicates during retries
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // Batching for higher throughput
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);    // 16 KB per batch
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);         // wait 5ms for a fuller batch

        this.producer = new KafkaProducer<>(props);
    }

    public void close() {
        producer.close();
    }
}

Sending Messages #

There are three ways to send messages: fire-and-forget, synchronous, and asynchronous. Choose based on your needs:

public class OrderProducer {

    // ... (constructor above)

    // ✗ ANTI-PATTERN: Fire-and-forget without a callback
    // There's no way to know whether the message was sent successfully
    public void sendFireAndForget(String orderId, String payload) {
        producer.send(new ProducerRecord<>(TOPIC, orderId, payload));
    }

    // ✓ CORRECT: Asynchronous with a callback — high throughput, still know the result
    public void sendAsync(String orderId, String payload) {
        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, orderId, payload);

        producer.send(record, (RecordMetadata metadata, Exception exception) -> {
            if (exception != null) {
                System.err.println("Failed to send message for order " + orderId + ": " + exception.getMessage());
                // Here: send to a dead letter queue, alert monitoring, or retry logic
                return;
            }
            System.out.printf(
                "Message sent — topic=%s, partition=%d, offset=%d%n",
                metadata.topic(),
                metadata.partition(),
                metadata.offset()
            );
        });
    }

    // ✓ CORRECT: Synchronous — used when you need confirmation before continuing
    // Slower (blocks per message), use only when truly necessary
    public RecordMetadata sendSync(String orderId, String payload) throws Exception {
        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, orderId, payload);
        Future<RecordMetadata> future = producer.send(record);
        return future.get(); // blocks until confirmation
    }
}

Sending to a Specific Partition #

By default, Kafka determines the partition based on a hash of the message key. But sometimes you need to send to a specific partition explicitly:

// Kafka determines the partition from the key hash — enough for most cases
ProducerRecord<String, String> byKey = new ProducerRecord<>(
    "order-events",
    "order-123",   // key — all messages with this key go to the same partition
    "{\"status\": \"created\"}"
);

// Specify the partition explicitly
ProducerRecord<String, String> toPartition = new ProducerRecord<>(
    "order-events",
    0,             // partition index
    "order-456",
    "{\"status\": \"created\"}"
);

Don’t specify partitions explicitly unless you have a very strong reason. This complicates rebalancing and reduces load distribution flexibility. Use a message key and let Kafka decide the partition based on the hash.

Consumers #

Consumers read messages from Kafka. Unlike traditional message brokers, reading a message from Kafka doesn’t delete it — the message stays until the retention period expires.

Consumer Configuration #

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class OrderConsumer {

    private final KafkaConsumer<String, String> consumer;
    private static final String TOPIC = "order-events";
    private volatile boolean running = true;

    public OrderConsumer(String groupId) {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // Group ID — consumers with the same group ID share partitions
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        // Auto offset reset — what to do when no offset is stored?
        // "earliest" = start from the beginning, "latest" = only new messages
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // ANTI-PATTERN: enable.auto.commit=true — offsets are committed automatically
        // This can cause missed messages if the consumer crashes after commit but before processing finishes
        // CORRECT: disable auto commit, commit manually after successful processing
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        // Heartbeat interval — how often the consumer sends a liveness signal to the broker
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);

        // Session timeout — if there's no heartbeat for this long, the consumer is considered dead
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);

        // Max poll records — how many records per poll()
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);

        this.consumer = new KafkaConsumer<>(props);
    }
}

The Poll Loop #

Consumers work in a poll loop — they keep asking the broker whether new messages have arrived:

public void start() {
    // Subscribe to one or more topics
    consumer.subscribe(Collections.singletonList(TOPIC));

    try {
        while (running) {
            // Poll with a timeout — how long to wait if there are no new messages
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

            for (var record : records) {
                try {
                    processRecord(record.key(), record.value());

                    // ✓ Commit the offset after successful processing
                    // commitSync() — slower but safer (blocks until confirmation)
                    consumer.commitSync();

                } catch (Exception e) {
                    System.err.printf(
                        "Failed to process record at partition=%d offset=%d: %s%n",
                        record.partition(),
                        record.offset(),
                        e.getMessage()
                    );
                    // Don't commit the offset — the message will be reprocessed after restart
                }
            }
        }
    } finally {
        consumer.close();
    }
}

private void processRecord(String key, String value) {
    System.out.printf("Processing order %s: %s%n", key, value);
    // business logic implementation
}

public void stop() {
    running = false;
}

Committing Offsets: Sync vs Async #

Your choice of offset commit method affects throughput and reliability:

// commitSync() — blocks until the broker confirms the commit
// WHEN: at the end of a batch, when the app is shutting down, when you need full guarantees
consumer.commitSync();

// commitAsync() — non-blocking, suitable for high throughput
// WHEN: when processing large batches and you can tolerate at-least-once
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        System.err.println("Commit failed: " + exception.getMessage());
        // commitAsync doesn't retry automatically to avoid out-of-order commits
        // retry manually here if needed
    }
});

// ✓ BEST PATTERN: combination — async during processing, sync at the end or on shutdown
while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (var record : records) {
        processRecord(record.key(), record.value());
    }
    consumer.commitAsync(); // async for throughput
}
consumer.commitSync(); // sync on shutdown for safety

Consumer Groups #

The consumer group is Kafka’s main mechanism for horizontal scalability. Several consumers in the same group collectively consume all partitions of a topic.

flowchart TD
    T[Topic: order-events\nPartition 0, 1, 2, 3]

    subgraph CGA[Consumer Group A — order-processor]
        C1[Consumer 1\nP0, P1]
        C2[Consumer 2\nP2, P3]
    end

    subgraph CGB[Consumer Group B — order-analytics]
        C3[Consumer 3\nP0, P1, P2, P3]
    end

    T --> C1
    T --> C2
    T --> C3

Key points about consumer groups:

One partition can only be consumed by one consumer within the same group — this guarantees per-partition ordering.
Multiple consumer groups can read the same topic independently — Group B above receives all the same messages as Group A, but each has its own offsets.
If the number of consumers > the number of partitions, the extra consumers sit idle — not an error, but wasted resources.

Rebalancing #

Rebalancing happens when:

A new consumer joins the group
A consumer leaves the group (crash or shutdown)
New partitions are added to a topic
The subscribe pattern changes

During rebalancing, all consumers in the group stop consuming. This pause is commonly called “stop the world”. To minimize its impact, implement a ConsumerRebalanceListener:

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;

import java.util.Collection;

public class OrderConsumerRebalanceListener implements ConsumerRebalanceListener {

    private final KafkaConsumer<String, String> consumer;

    public OrderConsumerRebalanceListener(KafkaConsumer<String, String> consumer) {
        this.consumer = consumer;
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Called BEFORE rebalancing — commit offsets for all partitions being taken away
        System.out.println("Partitions being revoked, committing offsets first...");
        consumer.commitSync(); // make sure no messages are missed
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Called AFTER rebalancing — new partitions have been assigned
        System.out.println("New partitions assigned: " + partitions);
    }
}

// Use the listener when subscribing
consumer.subscribe(
    Collections.singletonList("order-events"),
    new OrderConsumerRebalanceListener(consumer)
);

JSON Serialization #

In production, Kafka messages are almost always JSON or Avro. Here’s how to use Jackson for JSON serialization:

<!-- Add to pom.xml -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.17.1</version>
</dependency>

import com.fasterxml.jackson.databind.ObjectMapper;

// Model for the event
public record OrderEvent(
    String orderId,
    String status,
    double totalAmount,
    long timestamp
) {}

// Custom serializer for the producer
public class JsonSerializer<T> implements org.apache.kafka.common.serialization.Serializer<T> {

    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public byte[] serialize(String topic, T data) {
        if (data == null) return null;
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException("Failed to serialize to JSON: " + e.getMessage(), e);
        }
    }
}

// Custom deserializer for the consumer
public class JsonDeserializer<T> implements org.apache.kafka.common.serialization.Deserializer<T> {

    private final ObjectMapper objectMapper = new ObjectMapper();
    private final Class<T> targetClass;

    public JsonDeserializer(Class<T> targetClass) {
        this.targetClass = targetClass;
    }

    @Override
    public T deserialize(String topic, byte[] data) {
        if (data == null) return null;
        try {
            return objectMapper.readValue(data, targetClass);
        } catch (Exception e) {
            throw new RuntimeException("Failed to deserialize from JSON: " + e.getMessage(), e);
        }
    }
}

Use this custom serializer in the producer and consumer configuration:

// In the producer
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());

// In the consumer
// For generic classes, a different approach is needed due to Java type erasure
KafkaConsumer<String, OrderEvent> consumer = new KafkaConsumer<>(props,
    new StringDeserializer(),
    new JsonDeserializer<>(OrderEvent.class)
);

Error Handling and Retries #

Kafka doesn’t have a built-in dead letter queue (DLQ), but you can build your own. A common pattern is sending failed messages to a separate topic:

public class ResilientConsumer {

    private final KafkaConsumer<String, String> consumer;
    private final KafkaProducer<String, String> dlqProducer;
    private static final String TOPIC = "order-events";
    private static final String DLQ_TOPIC = "order-events.DLQ";
    private static final int MAX_RETRY = 3;

    // ... constructor

    public void startWithDLQ() {
        consumer.subscribe(Collections.singletonList(TOPIC));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

            for (var record : records) {
                boolean processed = false;

                for (int attempt = 1; attempt <= MAX_RETRY; attempt++) {
                    try {
                        processRecord(record.key(), record.value());
                        processed = true;
                        break; // exit the retry loop on success

                    } catch (Exception e) {
                        System.err.printf(
                            "Attempt %d/%d failed for key=%s: %s%n",
                            attempt, MAX_RETRY, record.key(), e.getMessage()
                        );

                        if (attempt < MAX_RETRY) {
                            try {
                                Thread.sleep(100L * attempt); // exponential-like backoff
                            } catch (InterruptedException ie) {
                                Thread.currentThread().interrupt();
                            }
                        }
                    }
                }

                if (!processed) {
                    // Send to the DLQ for manual investigation
                    sendToDLQ(record.key(), record.value(), "Failed after " + MAX_RETRY + " attempts");
                }
            }

            consumer.commitSync();
        }
    }

    private void sendToDLQ(String key, String value, String reason) {
        // Add error context as a header or value prefix
        String dlqPayload = String.format("{\"original\": %s, \"dlqReason\": \"%s\", \"timestamp\": %d}",
            value, reason, System.currentTimeMillis());

        dlqProducer.send(
            new ProducerRecord<>(DLQ_TOPIC, key, dlqPayload),
            (metadata, ex) -> {
                if (ex != null) {
                    System.err.println("CRITICAL: Failed to send to DLQ! Key=" + key);
                }
            }
        );
    }

    private void processRecord(String key, String value) {
        // business logic
    }
}

flowchart TD
    A[Consumer poll\nnew message] --> B{Processed successfully?}
    B -- Yes --> C[Commit offset]
    B -- No --> D{Max retries\nreached?}
    D -- Not yet --> E[Wait backoff] --> B
    D -- Reached --> F[Send to DLQ topic]
    F --> G[Commit offset\ndespite failure]
    G --> H[Alert ops team]

Don’t let a consumer crash endlessly without a DLQ. A consumer that keeps failing and never commits offsets will read the same messages repeatedly, blocking the entire partition from processing new messages.

Managing Topics Programmatically #

Besides using the command line, you can create and manage topics from Java code using AdminClient:

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.CreateTopicsResult;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaTopicManager {

    private final AdminClient adminClient;

    public KafkaTopicManager() {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        this.adminClient = AdminClient.create(props);
    }

    public void createTopic(String topicName, int partitions, short replicationFactor) {
        NewTopic newTopic = new NewTopic(topicName, partitions, replicationFactor);

        // Retention configuration: 7 days (in milliseconds)
        newTopic.configs(Map.of(
            "retention.ms", String.valueOf(7 * 24 * 60 * 60 * 1000L),
            "compression.type", "snappy",
            "cleanup.policy", "delete"
        ));

        CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));

        try {
            result.all().get(); // wait until finished
            System.out.println("Topic created successfully: " + topicName);
        } catch (ExecutionException e) {
            if (e.getCause() instanceof org.apache.kafka.common.errors.TopicExistsException) {
                System.out.println("Topic already exists: " + topicName);
            } else {
                throw new RuntimeException("Failed to create topic: " + e.getMessage(), e);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public void listTopics() throws Exception {
        var topics = adminClient.listTopics().names().get();
        topics.forEach(System.out::println);
    }

    public void close() {
        adminClient.close();
    }
}

Running Together — Complete Example #

Here’s a complete example of a producer and consumer working together in one application:

public class KafkaDemo {

    public static void main(String[] args) throws Exception {
        String topic = "demo-orders";
        String groupId = "order-processor";

        // Create the topic first
        KafkaTopicManager topicManager = new KafkaTopicManager();
        topicManager.createTopic(topic, 3, (short) 1);
        topicManager.close();

        // Run the consumer on a separate thread
        Thread consumerThread = new Thread(() -> {
            OrderConsumer consumer = new OrderConsumer(groupId);
            consumer.start();
        });
        consumerThread.setDaemon(true);
        consumerThread.start();

        // The producer sends a few messages
        OrderProducer producer = new OrderProducer();
        for (int i = 1; i <= 10; i++) {
            String orderId = "order-" + i;
            String payload = String.format("{\"id\": \"%s\", \"status\": \"created\", \"amount\": %.2f}",
                orderId, Math.random() * 1000);
            producer.sendAsync(orderId, payload);
        }

        // Wait a moment so the consumer has time to process
        Thread.sleep(5000);
        producer.close();
    }
}

When to Use Kafka and When Not To #

Kafka isn’t the solution for every messaging need. Choosing Kafka in the wrong situation adds operational complexity without proportionate benefits.

USE KAFKA WHEN:
  ✓ Message volume is very high (hundreds of thousands per second)
  ✓ You need replay — consumers need to re-read old messages
  ✓ Many independent consumers need to read the same data
  ✓ Event sourcing — events as the system's source of truth
  ✓ Stream processing — real-time analytics, aggregation, transformation
  ✓ Audit logs — you need a history of all events in the system
  ✓ Decoupling between many services in a microservices architecture

CONSIDER ALTERNATIVES WHEN:
  ✗ Low volume and high routing complexity → RabbitMQ fits better
  ✗ You need request-reply (RPC pattern) → gRPC or REST is more natural
  ✗ A simple task queue (jobs that must execute once) → RabbitMQ or Amazon SQS
  ✗ A small team without ops capacity to maintain a Kafka cluster → manage operational costs first
  ✗ Sub-millisecond latency is critical → Kafka isn't the best choice

flowchart TD
    A{Message volume\nvery high?} -- Yes --> B{Need\nreplay?}
    A -- No --> C{Many independent\nconsumers?}

    B -- Yes --> KAFKA[Kafka]
    B -- No --> D{Need durability\nand ordering?}

    C -- Yes --> KAFKA
    C -- No --> E{Simple task\nqueue?}

    D -- Yes --> KAFKA
    D -- No --> RABBIT[RabbitMQ or SQS]

    E -- Yes --> RABBIT
    E -- No --> REST[REST or gRPC]

Summary #

Topics and partitions are Kafka’s foundation — a partition is the unit of parallelism, and ordering is only guaranteed within one partition. Use a message key for deterministic routing to the same partition.
Async producers with a callback are the best choice for high throughput — use acks=all and enable.idempotence=true for durability without duplicates.
Disable automatic offset commits (enable.auto.commit=false) and commit manually after successful processing to avoid missed messages.
Consumer groups enable horizontal scalability — add consumers to increase throughput, but the effective consumer count is capped by the number of partitions.
Implement a Dead Letter Queue (DLQ) as a separate topic to handle failed messages — don’t let a consumer loop forever on a broken message.
ConsumerRebalanceListener is important for committing offsets before rebalancing starts, preventing messages from being processed twice after partitions move between consumers.
AdminClient enables topic management from Java code — ideal for ensuring a topic exists before a producer starts sending.
Kafka fits best for high volume, event sourcing, stream processing, and many independent consumers — not for simple task queues or request-reply patterns.

← Previous: Elasticsearch Next: RabbitMQ →