Apache Kafka

Apache Kafka is a distributed append-log system for durable event streams. Producers append records to topic partitions, brokers replicate and retain those records, and consumers read them by offset. Kafka is therefore useful when several independent applications need to observe the same ordered history without coupling their availability or processing rate.

Cluster Roles and Metadata #

A Kafka cluster contains two server roles:

• Brokers store partition replicas and serve produce and fetch requests.
• KRaft controllers maintain cluster metadata, broker membership, topic configuration, replica placement, and partition leadership.

A process can perform one or both roles in a development cluster. Production clusters separate controller and broker roles because data-plane load should not interfere with metadata quorum decisions. ZooKeeper belongs to Kafka’s legacy metadata architecture and is absent from Kafka 4.x.

Clients receive a list of bootstrap addresses rather than a complete routing table. A metadata request identifies the broker that currently leads each partition, after which produce and fetch requests go directly to the relevant leaders. Every advertised broker address must therefore be resolvable and reachable from the client’s network; publishing container-only hostnames to an external client is a common deployment error.

Kafka uses a versioned binary protocol over TCP. The org.apache.kafka:kafka-clients artifact provides Java producer, consumer, and administration APIs. Other language clients implement the same protocol.

Topics, Partitions, and Records #

A topic is divided into ordered partitions. Each record can contain:

1. a key;
2. a value;
3. a timestamp; and
4. zero or more headers.

Keys and values are byte arrays at the protocol boundary. Serializers convert application objects into those bytes, and deserializers perform the inverse operation.

A broker assigns monotonically increasing offsets to records appended to one partition. An offset identifies a position within that partition; it is not a globally ordered event ID. Kafka guarantees ordering within one partition, not across a topic.

When a producer does not specify a partition, the partitioner uses the record key and current topic metadata. Equal non-null keys are consequently routed to the same partition while the partition count remains unchanged. This property supports per-entity ordering for keys such as an account ID or device ID. Increasing the partition count can change the key-to-partition mapping, so partition expansion requires an explicit ordering and migration analysis.

Each partition is stored as segment files under a broker log directory. Segment files contain binary record batches and have offset and timestamp indexes; they are not line-oriented text logs.

Retention and Compaction #

Consumption never deletes a record. Brokers remove records according to the topic’s cleanup policy:

• cleanup.policy=delete removes old segments after the configured time or size boundary.
• cleanup.policy=compact retains the latest value for each key.
• cleanup.policy=compact,delete applies both policies.

Compaction preserves offsets and does not produce a dense sequence. A consumer can therefore encounter gaps between offsets. A keyed record with a null value is a tombstone; after delete.retention.ms, compaction can remove both the tombstone and the older value for that key.

Configure a compacted state topic with the administration tool:

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --create --topic customer-state \
  --partitions 12 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.5

Compaction is asynchronous. Applications must not assume that an obsolete value disappears immediately after a replacement or tombstone is appended.

Replication and Durability #

A partition has one leader and one or more replicas. Producers and consumers ordinarily communicate with the leader. Followers fetch the leader’s log, and the controller elects a replacement when leadership changes.

Several independent settings determine acknowledged-write durability:

• replication.factor determines how many brokers store the partition.
• acks=all makes the producer wait for the leader to satisfy the in-sync replica requirement.
• min.insync.replicas rejects an acks=all write when too few replicas remain in sync.
• unclean leader election permits a stale replica to become leader and can therefore lose acknowledged records.

For example, replication factor 3 with min.insync.replicas=2 tolerates one replica being unavailable while preserving the two-copy acknowledgement rule. It does not tolerate two unavailable replicas for writes. Availability and durability are coupled by this choice.

Replica placement must cross host, rack, or availability-zone failure domains. Three replicas on one physical host do not provide host-failure tolerance.

Creating and Inspecting Topics #

Topic count and partition count are operational capacity decisions, so create important topics explicitly instead of relying on broker-side automatic creation:

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --create --topic orders \
  --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --describe --topic orders

bin/kafka-configs.sh --bootstrap-server broker:9092 \
  --entity-type topics --entity-name orders --describe

The Java Admin API provides the same control plane:

import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;

Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");

try (Admin admin = Admin.create(config)) {
    NewTopic orders = new NewTopic("orders", 12, (short) 3)
        .configs(Map.of(
            "min.insync.replicas", "2",
            "retention.ms", "604800000"
        ));
    admin.createTopics(List.of(orders)).all().get();
}

createTopics is asynchronous; waiting on the returned future surfaces authorization, validation, and timeout failures.

Java Client Dependency #

Producer, consumer, and Admin clients are in one artifact:

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>${kafka.version}</version>
</dependency>

Select a client version from Kafka’s compatibility documentation. Protocol negotiation permits many mixed client and broker versions, but new APIs and record formats still have minimum broker requirements.

Producing Records #

KafkaProducer is thread-safe. It accumulates records into per-partition batches, compresses each batch, and sends requests asynchronously. A long-lived producer therefore performs better than constructing one producer per record.

import java.nio.charset.StandardCharsets;
import java.util.Properties;
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.common.serialization.StringSerializer;

Properties config = new Properties();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");
config.put(ProducerConfig.LINGER_MS_CONFIG, 5);

try (KafkaProducer<String, String> producer = new KafkaProducer<>(config)) {
    ProducerRecord<String, String> record =
        new ProducerRecord<>("orders", "customer-42", "{\"orderId\":123}");
    record.headers().add("schema-version",
        "1".getBytes(StandardCharsets.UTF_8));

    producer.send(record, (metadata, error) -> {
        if (error != null) {
            System.err.println("Produce failed: " + error);
            return;
        }
        System.out.printf("%s-%d@%d%n",
            metadata.topic(), metadata.partition(), metadata.offset());
    });
}

Closing the producer flushes outstanding sends. Applications should propagate or record callback failures; accepting a record into the client’s buffer does not mean the broker acknowledged it.

Important throughput and latency controls include:

• batch.size: maximum bytes accumulated for one partition batch;
• linger.ms: time available for a partially filled batch;
• compression.type: compression applied to whole batches;
• buffer.memory: producer memory available to unsent records;
• delivery.timeout.ms: upper bound for completing a send; and
• max.request.size: maximum request size accepted by the client.

Larger batches improve compression and request efficiency but consume memory and can add latency. Broker and topic message-size limits must also accept the resulting batch.

Idempotence and Retries #

Network failure can leave a producer uncertain whether a broker appended a request. Blind retries can then create duplicates. Idempotent production attaches a producer identity and sequence numbers so the broker can reject duplicate retries within the producer session.

Idempotence does not deduplicate two independent business requests, records sent by different producer instances, or records resent after application-level replay. Stable business IDs and downstream idempotency remain necessary.

Consuming Records #

KafkaConsumer is not thread-safe. One application thread should own each consumer, call poll, process assignment callbacks, and commit offsets. Parallel processing requires careful coordination between completed work and offset commits.

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;

Properties config = new Properties();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-indexer");
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config)) {
    consumer.subscribe(List.of("orders"));

    while (!Thread.currentThread().isInterrupted()) {
        ConsumerRecords<String, String> records =
            consumer.poll(Duration.ofSeconds(1));

        for (ConsumerRecord<String, String> record : records) {
            indexOrder(record.key(), record.value());
        }

        // Commits the next offsets after the complete batch succeeds.
        consumer.commitSync();
    }
}

The example provides at-least-once processing: a failure after indexOrder but before commitSync causes replay. indexOrder must consequently be idempotent, or the destination must deduplicate by a stable event ID.

auto.offset.reset applies only when the group has no valid committed offset. It does not force an established group to reread a topic. Use seek, a new group ID, or the consumer-groups reset operation for deliberate replay.

Consumer Groups and Rebalancing #

Consumers with the same group.id form a group. Each partition is assigned to at most one active consumer in that group:

• fewer consumers than partitions means some consumers own several partitions;
• equal counts permit one partition per consumer; and
• more consumers than partitions leaves excess consumers idle.

Separate groups receive independent copies of the topic history. A fraud detector and a search indexer should therefore use different group IDs.

Membership or subscription changes update assignments. Cooperative assignors reduce partition movement, while static membership can avoid unnecessary reassignment during short restarts. Applications must still stop work for revoked partitions and commit only completed offsets.

Inspect assignments and lag with:

bin/kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group order-indexer

Lag is log-end-offset - committed-offset. It measures uncommitted distance, not processing latency. A low-volume partition can have small lag while its oldest unprocessed record is old, so monitor both record lag and time-based service objectives.

max.poll.interval.ms limits the time between calls to poll. Processing a large batch synchronously can exceed this interval and remove the consumer from the group. Bound the batch with max.poll.records, increase the interval for legitimate slow work, or move work to executors while pausing partitions and committing only contiguous completed offsets.

Delivery Semantics #

Kafka applications can implement three outcomes:

• At most once: advance or commit before processing, accepting loss after a failure.
• At least once: process before committing, accepting replay and duplicates.
• Exactly-once Kafka processing: atomically commit produced records and consumed offsets with Kafka transactions.

Kafka transactions do not make database writes, emails, or HTTP calls atomic. External effects require idempotency keys, an outbox/inbox design, deduplication, or another coordination protocol.

Transactional Consume-Transform-Produce #

Each live producer instance needs a stable, unique transactional.id. Reusing one ID concurrently fences the older producer, which prevents two instances from committing as the same transaction owner.

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
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.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

Properties consumerConfig = new Properties();
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "order-normalizer");
consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());

Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
    "order-normalizer-instance-1");
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());

try (KafkaConsumer<String, String> consumer =
         new KafkaConsumer<>(consumerConfig);
     KafkaProducer<String, String> producer =
         new KafkaProducer<>(producerConfig)) {
    consumer.subscribe(List.of("orders"));
    producer.initTransactions();

    while (true) {
        var records = consumer.poll(Duration.ofSeconds(1));
        if (records.isEmpty()) {
            continue;
        }

        try {
            producer.beginTransaction();
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

            for (var record : records) {
                producer.send(new ProducerRecord<>(
                    "normalized-orders", record.key(),
                    normalize(record.value())));
                offsets.put(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
                );
            }

            producer.sendOffsetsToTransaction(
                offsets, consumer.groupMetadata());
            producer.commitTransaction();
        } catch (ProducerFencedException fatal) {
            throw fatal;
        } catch (KafkaException retryableBatchFailure) {
            producer.abortTransaction();
        }
    }
}

Consumers that must hide aborted records use isolation.level=read_committed. Transactions increase coordination and latency, so use them when atomic Kafka read-process-write behavior is required rather than as a default substitute for idempotent processing.

Kafka Streams #

Kafka Streams is an embeddable Java library, not a separate cluster service. Each application instance joins a consumer group, partitions work into tasks, and stores local state backed by changelog topics.

import java.util.Properties;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;

Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "paid-order-counts");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
    Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
    Serdes.String().getClass());
config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");

StreamsBuilder builder = new StreamsBuilder();
builder.<String, String>stream("orders")
    .filter((customerId, json) -> json.contains("\"status\":\"PAID\""))
    .groupByKey()
    .count()
    .toStream()
    .to("paid-orders-per-customer");

KafkaStreams streams = new KafkaStreams(builder.build(), config);
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
streams.start();

Changing application.id creates a new processing identity with different consumer groups, internal topics, and state directories. Treat it as persistent application state rather than a display name.

Kafka Connect and Replication #

Kafka Connect runs reusable source and sink connectors. Source connectors import records from external systems; sink connectors export Kafka records. Distributed workers store connector configuration, offsets, and status in Kafka topics and rebalance connector tasks across workers.

Connect simplifies standard integration but does not remove the need to define:

• record keys and partitioning;
• schema compatibility;
• delete and tombstone semantics;
• retry and dead-letter behavior;
• destination idempotency; and
• credentials and network boundaries.

MirrorMaker 2 uses Connect to replicate topics between Kafka clusters. Replication is asynchronous and does not turn two clusters into one synchronous consensus system. Disaster-recovery procedures must account for replication lag, consumer offset translation, topic configuration, and failback.

Schemas and Evolution #

Kafka stores bytes and does not enforce an application schema. Avro, Protobuf, JSON Schema, or another encoding can define record structure. A schema registry can associate schema IDs with payloads and reject incompatible changes.

Compatibility rules should reflect how data is replayed:

• new consumers may encounter old records retained for months;
• old consumers may remain deployed while producers change;
• compacted topics can preserve old values indefinitely; and
• restoration from an archive can reintroduce older schemas.

Adding an optional field with a defined default is easier to evolve than changing a field’s meaning or reusing its name. Event names and field semantics form a long-lived contract even when the serialization format permits a mechanical change.

Security #

Kafka separates transport security, authentication, and authorization:

• TLS encrypts traffic and authenticates brokers to clients.
• mTLS or SASL mechanisms authenticate clients.
• ACLs authorize operations on topics, groups, transactional IDs, and cluster resources.

A TLS client configuration can be supplied through properties:

security.protocol=SSL
ssl.truststore.location=/run/secrets/kafka.truststore.p12
ssl.truststore.password=${TRUSTSTORE_PASSWORD}
ssl.keystore.location=/run/secrets/client.keystore.p12
ssl.keystore.password=${KEYSTORE_PASSWORD}
ssl.key.password=${KEY_PASSWORD}

Do not place real passwords in committed property files. Inject them through the deployment platform’s secret mechanism and restrict filesystem access.

Partition and Capacity Planning #

Partition count bounds consumer-group parallelism, but more partitions are not free. Each partition adds metadata, open files, replica fetch work, elections, and recovery traffic. Estimate a starting count from:

1. peak ingress and egress throughput;
2. maximum useful consumer concurrency;
3. per-broker network and disk throughput;
4. key skew;
5. replication and recovery objectives; and
6. expected growth before repartitioning is acceptable.

Retention capacity is approximately:

ingress bytes/second × retention seconds × replication factor

Add headroom for segment granularity, indexes, compaction, temporary imbalance, reassignment, and broker recovery. A cluster that serves steady-state traffic but cannot rebuild a failed broker within its recovery objective is underprovisioned.

Operations and Failure Diagnosis #

Monitor at least:

• offline and under-replicated partitions;
• controller and broker availability;
• produce and fetch latency and error rates;
• request-handler and network-thread saturation;
• disk use and disk latency;
• replica fetch lag;
• consumer offset lag and record age;
• transaction and authentication failures; and
• partition imbalance and reassignment progress.

Useful diagnosis commands include:

# Observe records without joining an application consumer group.
bin/kafka-console-consumer.sh --bootstrap-server broker:9092 \
  --topic orders --from-beginning \
  --property print.key=true --property print.offset=true

# Inspect one group's committed offsets and lag.
bin/kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group order-indexer --describe

# Read broker API compatibility information.
bin/kafka-broker-api-versions.sh --bootstrap-server broker:9092

An unavailable leader, insufficient in-sync replicas, unreachable advertised listener, authorization failure, and overloaded broker can all appear to an application as a send timeout. Correlate client exceptions with broker metrics, partition state, and controller events before changing retry limits.

References #

• Apache Kafka Documentation
• Kafka Protocol
• Kafka Java API