Apache Kafka — PE-Grade Deep Dive

5.1Pub/Sub: Producer and Consumer

Producer (idempotent, production defaults)

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Durability and ordering
props.put("acks", "all");                              // wait for ISR ack, default since 3.0
props.put("enable.idempotence", "true");                  // no duplicates on retry, default since 3.0
props.put("max.in.flight.requests.per.connection", 5); // safe with idempotence
props.put("retries", Integer.MAX_VALUE);                  // rely on delivery.timeout.ms instead
props.put("delivery.timeout.ms", 120000);                // 2 min total deadline per record

// Throughput
props.put("compression.type", "lz4");                    // zstd is often better for archival
props.put("linger.ms", 20);                              // wait up to 20ms to batch
props.put("batch.size", 65536);                          // 64 KB per partition batch
props.put("buffer.memory", 67108864);                    // 64 MB total buffer

try (Producer<String, String> producer = new KafkaProducer<>(props)) {
    // Key drives partition assignment. Same key → same partition → ordered.
    ProducerRecord<String, String> record =
        new ProducerRecord<>("orders", orderId, payload);

    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            log.error("Send failed for {}", orderId, exception);
        } else {
            log.debug("Sent {} → partition {} offset {}",
                orderId, metadata.partition(), metadata.offset());
        }
    });
}

Consumer (KIP-848 new protocol, manual commit)

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092");
props.put("group.id", "order-processor");
props.put("group.protocol", "consumer");  // KIP-848 protocol (Kafka 4.0+)
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// Manual commit, read only committed records (skip aborted transactions)
props.put("enable.auto.commit", "false");
props.put("isolation.level", "read_committed");

// Tune throughput vs latency
props.put("max.poll.records", 500);            // records per poll()
props.put("max.poll.interval.ms", 300000);    // processing deadline, exceed → kicked from group
props.put("fetch.min.bytes", 1048576);          // wait for 1 MB of data
props.put("fetch.max.wait.ms", 500);           // or 500 ms, whichever first

// Fetch from same-AZ follower to cut cross-AZ traffic (KIP-392)
props.put("client.rack", "us-east-1a");

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

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

        for (ConsumerRecord<String, String> record : records) {
            // Process. If processOrder throws, the loop bails and we DO NOT commit.
            // Next poll will re-deliver from the last committed offset. At-least-once.
            processOrder(record.value());
        }

        if (!records.isEmpty()) {
            consumer.commitSync();  // commit after successful processing
        }
    }
}

5.2Kafka Streams: Stateful Processing

Mental model

KStream is a stream of records (insert log). KTable is a changelog interpreted as a materialized view (latest value per key). GlobalKTable is a KTable fully replicated to every Streams instance (no co-partitioning needed). Joins between these have different semantics and different cost.

Example: enrich orders with customer data, aggregate by region in 5-min windows

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-enrichment");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/lib/streams");

// KIP-1071: new server-side Streams rebalance protocol (GA in Kafka 4.2)
props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, "streams");

StreamsBuilder builder = new StreamsBuilder();

// Source: stream of orders, keyed by customerId
KStream<String, Order> orders = builder.stream("orders",
    Consumed.with(Serdes.String(), orderSerde));

// Materialized table of customers, keyed by customerId
KTable<String, Customer> customers = builder.table("customers",
    Consumed.with(Serdes.String(), customerSerde));

// Stream-table join: enrich each order with customer attributes.
// Requires orders and customers co-partitioned on the same key.
KStream<String, EnrichedOrder> enriched = orders
    .leftJoin(customers, (order, customer) -> new EnrichedOrder(order, customer));

// Aggregate revenue per region in 5-minute tumbling windows
KTable<Windowed<String>, RevenueAgg> revenue = enriched
    .selectKey((k, v) -> v.region())
    .groupByKey(Grouped.with(Serdes.String(), enrichedOrderSerde))
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5), Duration.ofSeconds(30)))
    .aggregate(
        RevenueAgg::new,
        (region, order, agg) -> agg.add(order.amount()),
        Materialized.<String, RevenueAgg, WindowStore<Bytes, byte[]>>
            as("revenue-store")
            .withValueSerde(revenueAggSerde));

revenue.toStream()
       .map((windowedKey, agg) -> KeyValue.pair(windowedKey.key(),
            new RevenueResult(windowedKey.window().startTime(), agg.total())))
       .to("revenue-by-region-5min", Produced.with(Serdes.String(), revenueResultSerde));

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

What's actually happening under the hood

• Local state in RocksDB. The revenue-store is an embedded RocksDB instance per partition assigned to this Streams instance. Reads and writes are local; the disk is yours.
• Changelog topic for durability. Every update to local state is also written to a changelog topic (Streams creates this for you with the name {application-id}-{store-name}-changelog). On restart or rebalance, state is restored by replaying the changelog. This is why "Streams state" survives a host failure.
• Co-partitioning required for joins. If orders has 12 partitions and customers has 6, the join is rejected at topology build time. Both must have the same partition count and same partitioner.
• EOS v2 uses transactions. Read input → update state + write changelog + write output, all in one Kafka transaction. Crashes leave no half-state.

5.3Kafka Connect: Source and Sink Pipelines

Architecture in one paragraph

Connect runs as a separate JVM cluster of "worker" nodes. Connectors (plugins) define source and sink logic. Tasks are the parallel work units within a connector. State (offsets, configs, status) is persisted to Kafka topics (connect-offsets, connect-configs, connect-status). The REST API on each worker exposes connector lifecycle. Workers coordinate via the same KIP-848 protocol used by consumers.

Source: Postgres CDC via Debezium with SMTs

POST http://connect-worker:8083/connectors
Content-Type: application/json

{
  "name": "orders-pg-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",  // CDC is single-task per database

    "database.hostname": "pg-primary.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/secrets/db-pw:debezium}",
    "database.dbname": "orders_db",
    "topic.prefix": "pg.orders_db",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_orders",

    "table.include.list": "public.orders,public.order_items",
    "snapshot.mode": "initial",

    // Single Message Transforms (SMTs)
    "transforms": "unwrap,route,filter",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "pg.orders_db.public.(.*)",
    "transforms.route.replacement": "cdc.orders.$1",
    "transforms.filter.type": "org.apache.kafka.connect.transforms.Filter",
    "transforms.filter.predicate": "is-not-system-table",

    // Exactly-once support (Kafka Connect 3.3+)
    "exactly.once.support": "required",

    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

Sink: write enriched orders to S3 in Parquet, partitioned by date

POST http://connect-worker:8083/connectors

{
  "name": "orders-s3-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "8",    // parallel tasks, capped at partition count
    "topics": "enriched-orders",

    "s3.bucket.name": "data-lake-prod",
    "s3.region": "us-east-1",
    "topics.dir": "raw/kafka",

    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "parquet.codec": "snappy",

    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "partition.duration.ms": "3600000",         // 1 hour partitions
    "path.format": "'dt'=YYYY-MM-dd/'hr'=HH",
    "locale": "en-US",
    "timezone": "UTC",
    "timestamp.extractor": "Record",

    "flush.size": "100000",                     // 100k records per S3 file
    "rotate.interval.ms": "600000",              // or 10 min, whichever first

    "behavior.on.null.values": "ignore",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "orders-s3-sink-dlq",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

5.4Schema Registry: Schemas as a Service

What it actually does

Schema Registry stores Avro/Protobuf/JSON Schemas, assigns each one an integer schema ID, and serves them via REST. The serializer in the producer registers the schema (or looks it up) and prepends the 5-byte ID to every record (1 magic byte + 4-byte schema ID + payload). The consumer reads the ID, fetches the schema, deserializes. Without Schema Registry, every consumer needs the schema baked in. With it, schema becomes a runtime concern.

Compatibility modes

Producer with Avro and Schema Registry

import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.generic.GenericRecord;

Properties props = new Properties();
props.put("bootstrap.servers", "broker:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", KafkaAvroSerializer.class.getName());

props.put("schema.registry.url", "http://schema-registry:8081");

// PE production defaults:
props.put("auto.register.schemas", "false");   // register via CI, not at runtime
props.put("use.latest.version", "true");       // always serialize with the registered latest
props.put("latest.compatibility.strict", "true"); // fail at startup if schemas drift

try (Producer<String, GenericRecord> producer = new KafkaProducer<>(props)) {
    GenericRecord order = new GenericData.Record(orderSchema);
    order.put("orderId", "ord-7821");
    order.put("customerId", "cust-42");
    order.put("amount", 29.99);

    producer.send(new ProducerRecord<>("orders", "ord-7821", order));
}

Schema evolution: add an optional field

// V1 schema (already registered)
{
  "type": "record", "name": "Order",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "amount", "type": "double"}
  ]
}

// V2 schema: add couponCode (optional, default null) → BACKWARD compatible
{
  "type": "record", "name": "Order",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "couponCode", "type": ["null", "string"], "default": null}
  ]
}

# Register via CLI (use in CI, not at runtime)
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data @order-v2.json \
  http://schema-registry:8081/subjects/orders-value/versions

5.5Transactions: Exactly-Once Semantics

What Kafka transactions actually guarantee

Transactional producers write all-or-nothing across multiple partitions and topics, and (in consume-process-produce) atomically commit consumer offsets and produced records together. Consumers with isolation.level=read_committed skip records from aborted transactions. The guarantee scope is Kafka-to-Kafka only.

Consume-process-produce with EOS

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;

// Producer config
Properties prodProps = new Properties();
prodProps.put("bootstrap.servers", "broker:9092");
prodProps.put("transactional.id", "order-enricher-instance-1");  // unique per process
prodProps.put("enable.idempotence", "true");                       // required for txn
prodProps.put("transaction.timeout.ms", 60000);

// Consumer config
Properties consProps = new Properties();
consProps.put("bootstrap.servers", "broker:9092");
consProps.put("group.id", "order-enricher");
consProps.put("isolation.level", "read_committed");
consProps.put("enable.auto.commit", "false");  // MUST be false for EOS

Producer<String, EnrichedOrder> producer = new KafkaProducer<>(prodProps);
Consumer<String, Order> consumer = new KafkaConsumer<>(consProps);

producer.initTransactions();
consumer.subscribe(List.of("orders"));

while (true) {
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofSeconds(1));
    if (records.isEmpty()) continue;

    try {
        producer.beginTransaction();

        for (ConsumerRecord<String, Order> record : records) {
            EnrichedOrder enriched = enrich(record.value());
            producer.send(new ProducerRecord<>("enriched-orders",
                record.key(), enriched));
        }

        // Critical: commit offsets WITHIN the transaction, not via consumer.commitSync()
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        for (TopicPartition tp : records.partitions()) {
            List<ConsumerRecord<String, Order>> partRecs = records.records(tp);
            long lastOffset = partRecs.get(partRecs.size() - 1).offset();
            offsets.put(tp, new OffsetAndMetadata(lastOffset + 1));
        }
        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());

        producer.commitTransaction();

    } catch (ProducerFencedException | OutOfOrderSequenceException e) {
        // Producer is fenced (another instance took over) or fatal error. Bail.
        producer.close();
        throw e;
    } catch (KafkaException e) {
        // Transient error. Abort and reprocess.
        producer.abortTransaction();
    }
}

5.6Tiered Storage: Hot Local, Cold S3

Broker-side config

# server.properties
remote.log.storage.system.enable=true
remote.log.storage.manager.class.name=org.apache.kafka.server.log.remote.storage.RemoteLogManager
remote.log.metadata.manager.class.name=org.apache.kafka.server.log.remote.storage.TopicBasedRemoteLogMetadataManager

# S3-backed plugin (Aiven's TS plugin is the de facto open-source choice)
remote.log.storage.manager.class.path=/opt/kafka/plugins/tiered-storage-s3/*
rsm.config.storage.backend.class=io.aiven.kafka.tieredstorage.storage.s3.S3Storage
rsm.config.storage.s3.bucket.name=kafka-tiered-storage-prod
rsm.config.storage.s3.region=us-east-1
rsm.config.chunk.size=4194304    # 4 MB chunks
rsm.config.compression.enabled=true

# Quota (KIP-956): cap remote fetch/upload throughput
remote.log.manager.copy.max.bytes.per.second=104857600   # 100 MB/s upload
remote.log.manager.fetch.max.bytes.per.second=209715200  # 200 MB/s fetch

Per-topic enablement

# Enable tiered storage on a topic, keep 1 day local, 90 days total
kafka-configs.sh --bootstrap-server broker:9092 \
  --entity-type topics --entity-name events \
  --alter \
  --add-config remote.storage.enable=true,\
local.retention.ms=86400000,\
retention.ms=7776000000

# Dynamically disable and re-enable (KIP-950, KRaft mode only)
kafka-configs.sh --bootstrap-server broker:9092 \
  --entity-type topics --entity-name events \
  --alter --add-config remote.storage.enable=false

5.7KRaft: Operating Kafka Without ZooKeeper

What changed

ZooKeeper was Kafka's external metadata service for 14 years. KRaft (Kafka Raft) replaces it with an internal Raft quorum running on dedicated "controller" nodes (or co-located with brokers in small clusters). Metadata lives in a single internal topic, __cluster_metadata, replicated via Raft. Controller failover is now sub-second instead of tens of seconds. Partition scale ceiling moved from ~200k to ~1.9M per cluster.

Bootstrap a fresh KRaft cluster

# Step 1: generate a cluster UUID once, share across all nodes
KAFKA_CLUSTER_ID="$(./bin/kafka-storage.sh random-uuid)"

# Step 2: format storage on each node with that UUID
./bin/kafka-storage.sh format -t "$KAFKA_CLUSTER_ID" -c config/kraft/server.properties

# Step 3: start brokers
./bin/kafka-server-start.sh config/kraft/server.properties

Combined mode (broker + controller in one process)

# config/kraft/server.properties
process.roles=broker,controller
node.id=1

# All controller nodes listed here, on every node
controller.quorum.voters=1@host1:9093,2@host2:9093,3@host3:9093
controller.listener.names=CONTROLLER

listeners=PLAINTEXT://host1:9092,CONTROLLER://host1:9093
advertised.listeners=PLAINTEXT://host1:9092

log.dirs=/var/lib/kafka/data
num.partitions=3
default.replication.factor=3
min.insync.replicas=2

Operational verbs that changed

5.8Share Groups (Queues for Kafka): KIP-932

Why this exists

For 14 years, the rule was "partition = unit of parallelism, max consumers per group = partition count." Want 1000 concurrent workers? Make 1000 partitions. That model fights real workloads where you want to scale consumers freely without committing to a partition count upfront.

Share Groups break the rule. A share group lets many consumers cooperatively consume from the same partition with per-record acknowledgment. The whole topic becomes a queue. You lose strict per-key ordering. You gain elastic scaling and SQS-like semantics, inside Kafka.

Share group consumer

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.consumer.AcknowledgeType;

Properties props = new Properties();
props.put("bootstrap.servers", "broker:9092");
props.put("group.id", "image-processors");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");

// Per-record ack mode
props.put("share.acknowledgement.mode", "explicit");

try (ShareConsumer<String, byte[]> consumer = new KafkaShareConsumer<>(props)) {
    consumer.subscribe(List.of("image-jobs"));

    while (true) {
        ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofSeconds(2));

        for (ConsumerRecord<String, byte[]> record : records) {
            try {
                processImage(record.value());
                // Acknowledge as ACCEPT: remove from delivery set
                consumer.acknowledge(record, AcknowledgeType.ACCEPT);
            } catch (TransientException e) {
                // Release: redeliver to another consumer, increment delivery count
                consumer.acknowledge(record, AcknowledgeType.RELEASE);
            } catch (PoisonException e) {
                // Reject: send to dead-letter handling; will not be redelivered
                consumer.acknowledge(record, AcknowledgeType.REJECT);
            }
        }

        consumer.commitSync();  // flush ack state to share coordinator
    }
}

Topic-side: enable share group on a topic

# Cluster-wide feature flag (one-time, KRaft)
kafka-features.sh --bootstrap-server broker:9092 \
  upgrade --feature share.version=1

# Configure share-group settings (delivery count threshold, lock timeout)
kafka-configs.sh --bootstrap-server broker:9092 \
  --entity-type groups --entity-name image-processors \
  --alter --add-config \
group.share.delivery.count.limit=5,\
group.share.record.lock.duration.ms=30000

# Inspect
kafka-share-groups.sh --bootstrap-server broker:9092 --describe \
  --group image-processors