Kafka — all questions in one page

Apache Kafka is a distributed, horizontally scalable event streaming platform that stores records in an append-only, partitioned log. Producers write records to topics, brokers persist them durably on disk, and consumers read them at their own pace by tracking offsets. Since version 4.0 (March 2025) Kafka runs only in KRaft mode, with no ZooKeeper dependency.

Point-to-point integrations create an N-to-N mesh of connections, so every new system must be wired to every other one. Kafka inserts a single durable log in the middle: producers publish once, any number of consumers subscribe independently, and new consumers can be added without touching producers. This decouples systems in time, in space and in throughput, and it lets you replay history instead of losing it.

Records are appended to segment files on disk and replicated to other brokers rather than being deleted as soon as they are read. Retention is time or size based, so a consumer can re-read old data by resetting its offset. This makes Kafka usable both as a messaging system and as a short-term source of truth for event replay and rebuilding derived state.

A classic queue deletes a message once it is acknowledged, and the broker usually pushes messages to consumers. Kafka keeps messages for a retention period regardless of consumption, and consumers pull at their own rate while the broker only tracks offsets. Kafka also scales throughput by partitioning a topic, which a traditional queue generally does not do in the same way.

Partition assignment decides the semantics. If all consumers share the same group.id, each partition is assigned to exactly one consumer, which behaves like a queue that load-balances work. If each consumer uses a different group.id, every group receives the full stream, which behaves like publish-subscribe. Both models can coexist on the same topic at the same time.

Kafka lets you publish and subscribe to streams of events, store those streams durably and for as long as needed, and process them either as they arrive or retrospectively. The ecosystem adds Kafka Connect for source and sink integration and Kafka Streams for stateful stream processing, so the same log underpins ingestion, transport and computation.

Kafka is a poor fit for request-response RPC, for very low-volume systems where a database table would do, and for workloads that need arbitrary per-message delays or complex routing rules. It is also heavy to operate for a single small application, and it does not replace a transactional database for querying current state. Choose it when you need high-throughput, ordered, replayable event distribution between multiple independent consumers.

A cluster consists of brokers that store data, plus controllers that manage cluster metadata; clients are producers, consumers and admin clients. Topics are split into partitions, and each partition is replicated across brokers. A bootstrap server list tells clients how to reach the cluster, after which metadata directs them to the correct leader for each partition.

A broker is one Kafka server process that owns a set of partitions, handles produce and fetch requests, and replicates data to peers. A cluster is a group of brokers that share metadata and identify themselves through a unique node.id and a common cluster.id. Adding brokers to a cluster increases storage capacity and parallelism, because partitions can be spread across them.

Producers write records to a topic and decide which partition each record goes to, either by key hashing or explicitly. Consumers read records from partitions in the order they were written and commit offsets to record their progress. Consumers are grouped with group.id, and group membership determines which partitions each member is responsible for.

A topic is a logical stream split into one or more partitions, and each partition is an ordered, immutable log. Every partition has one leader replica that serves reads and writes and zero or more follower replicas that copy it. The replication factor states how many copies exist, and those copies are placed on different brokers whenever possible.

The controller is the component that elects partition leaders, tracks broker liveness and propagates metadata changes. Up to 3.x the controller was a broker elected through ZooKeeper; in 4.x ZooKeeper is gone and a KRaft quorum of controllers owns the metadata log. Brokers then act as observers that replicate that metadata log, which removes the old scaling bottleneck.

process.roles=broker,controller
controller.quorum.voters=1@localhost:9093

Clients do not need every broker in configuration; they need at least one reachable address to fetch cluster metadata. Once connected, the client learns the full broker list, the partition layout and the current leader for each partition. It then connects directly to the relevant leaders, so the bootstrap list is only an entry point rather than a routing layer.

Kafka Connect is a separate cluster of worker processes that runs connectors to move data between Kafka and external systems without custom code. MirrorMaker 2 is built on Connect and replicates topics, consumer offsets and configuration between two Kafka clusters for disaster recovery or migration. Both are ordinary Kafka clients from the broker's point of view, so they do not change the core architecture.

In Kafka 4.x there is no ZooKeeper to start, so you generate a cluster id, format the storage directory with that id, and then start the broker. The shipped config/server.properties already defines a combined broker and controller node, so one command brings up a working single-node cluster.

bin/kafka-storage.sh random-uuid
bin/kafka-storage.sh format -t <uuid> -c config/server.properties
bin/kafka-server-start.sh config/server.properties

Formatting writes the cluster id and the initial metadata into the log directories, which is how a node knows which cluster it belongs to. Brokers and controllers refuse to start if their storage directory is unformatted or was formatted with a different cluster id. Reformatting an existing directory erases the metadata log, so it is only safe on fresh storage.

With process.roles=broker,controller a node does both jobs, which is convenient for development and small clusters. In production you normally run dedicated controller nodes and separate broker nodes so that metadata quorum latency and data traffic do not compete. The quorum size should be odd, typically three or five controllers, to tolerate failures.

# controller node
process.roles=controller
# broker node
process.roles=broker

Use the official apache/kafka image, set the node id, roles and quorum voters through environment variables, and map the listener ports. The container entrypoint formats the storage automatically when the KRaft variables are present. Do not expose only the internal listener, or clients outside the container network will receive unreachable advertised addresses.

services:
  kafka:
    image: apache/kafka:4.0.0
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

Ask the broker for a list of topics, which exercises a real metadata request, and inspect the log output for a successful startup. You can also list the cluster id and the registered brokers to confirm that the node joined the cluster. A successful describe of a test topic plus a round trip through the console producer and consumer is a complete smoke test.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092

The client listener, usually PLAINTEXT on 9092, serves producers, consumers and admin tools. The controller listener, usually 9093, is used only for the KRaft quorum, and in older setups 2181 belonged to ZooKeeper. If you enable SASL or SSL you often add separate listeners on other ports and must advertise each one correctly.

The client connects to the bootstrap address, then the broker returns its advertised listeners, and the client reconnects to those. If the broker advertises an internal container hostname such as kafka:9092, the host cannot resolve it and the connection fails. Fix it by advertising localhost for host clients and a separate internal listener for containers.

A broker receives produce requests, appends records to partition log segments, serves fetch requests to consumers, and replicates partitions where it is a follower. It also handles metadata requests, manages retention and log compaction, and participates in leader election through the controller. Health checks, throttling and quota enforcement are additional broker duties.

Each partition directory under log.dirs contains segment files plus an index and a time index. The active segment is appended to until it reaches segment.bytes or segment.ms, after which it is rolled and becomes read-only. Retention or compaction then removes or rewrites old segments, which is why the unit of deletion is a whole segment rather than a single record.

Only the leader accepts produce and fetch requests for a given partition, which keeps ordering unambiguous. Followers continuously fetch from the leader to stay in sync and never serve clients in normal operation. If the leader fails, the controller promotes an in-sync follower, and clients refresh metadata to find the new leader.

The controller detects the failure through the KRaft metadata log and missed heartbeats, then elects new leaders from the in-sync replicas. Clients receive a NOT_LEADER_OR_FOLLOWER error, refresh metadata and retry against the new leaders, so a short retriable failure is usually invisible. With replication factor three, the cluster keeps serving with two brokers remaining.

If no in-sync replica is available, unclean leader election allows an out-of-sync replica to become leader so the partition becomes writable again. The trade-off is silent data loss, because records that existed only on the failed leader are gone, and consumers may see the log jump backwards. It is disabled by default (unclean.leader.election.enable=false) and should only be enabled for availability-critical, loss-tolerant data.

Retention is configured per topic with retention.ms and retention.bytes, and the broker deletes closed segments that exceed the limit. The active segment is never deleted, so a low retention value can still leave one full segment on disk. Setting retention.ms=-1 together with a byte limit keeps data by size only.

retention.ms=604800000
retention.bytes=10737418240

Start the new broker with a unique node.id, and it registers automatically through the KRaft metadata log. New topics may then place replicas on it, but existing partitions stay where they are until you reassign them. Use kafka-reassign-partitions.sh to generate, execute and verify a reassignment plan, and throttle the movement to protect live traffic.

bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --reassignment-json-file plan.json --execute --throttle 50000000

A topic is a named, durable stream of records that producers write to and consumers read from. Physically it is a set of partitions, each an ordered append-only log, spread across brokers. Topics carry their own configuration such as retention, cleanup policy and replication factor.

Use kafka-topics.sh with --create, a name, the partition count and the replication factor. The replication factor cannot exceed the number of brokers, and the partition count is the main lever for consumer parallelism. You can pass topic-level overrides with --config at creation time.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic orders --partitions 6 --replication-factor 3 \
  --config retention.ms=604800000

cleanup.policy=delete removes segments once retention limits are exceeded, which is the default for event streams. cleanup.policy=compact keeps at least the last record for every key forever and garbage-collects older values, which suits changelog or state-snapshot topics. The two can be combined as compact,delete when you want keyed state with a time bound.

Compaction runs in the background and rewrites segments so that only the newest record per key survives, while offsets of removed records remain as gaps. A record with a null value is a tombstone, meaning the key was deleted, and it is retained for delete.retention.ms before being purged. Consumers must therefore tolerate missing offsets and null values.

__consumer_offsets stores committed consumer group offsets as a compacted topic, and __transaction_state stores transactional producer metadata. KRaft clusters also keep the metadata log in the controller quorum rather than in a user-visible topic. These topics are managed by Kafka and should not be written to or deleted manually.

If auto.create.topics.enable is true on the broker, the topic is created implicitly with the default partition and replication settings, which often produces a misconfigured topic by accident. When it is false, the produce request fails with UNKNOWN_TOPIC_OR_PARTITION and the client retries according to its retry settings. Production clusters normally disable auto-creation and provision topics explicitly.

Every topic inherits broker-level defaults such as log.retention.hours unless an override is set on the topic itself. Overrides are stored in the topic metadata and can be changed at runtime with kafka-configs.sh without restarting brokers. Dynamic broker defaults can also be set per broker, but a topic-level override always wins.

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter \
  --entity-type topics --entity-name orders \
  --add-config retention.ms=86400000

A partition is the unit of parallelism and ordering: an append-only log where each record gets a monotonically increasing offset. A topic is split into partitions so that a single topic can exceed the capacity of one broker. Records are ordered only within a partition, never across partitions of the same topic.

If the record has a key, the default partitioner hashes the key and maps it to a partition, so the same key always lands in the same partition while the topic layout is unchanged. Without a key, the sticky partitioner fills one partition for a batch and then moves on, which improves batching efficiency. You can also set the partition explicitly and bypass the partitioner entirely.

Start from target throughput divided by the throughput a single partition can handle, and keep the count at least equal to the maximum consumer parallelism you need. Partitions have a cost in metadata, open file handles and rebalance time, so thousands of partitions per broker are usually a bad sign. Because increasing partitions later breaks key-to-partition stability, it is better to over-provision slightly than to grow often.

The replication factor is the number of copies of each partition, one leader plus followers on other brokers. With replication factor three the cluster tolerates two broker failures without losing data as long as at least one in-sync copy remains. A replication factor of one has no redundancy and is only acceptable for local development.

The ISR is the set of replicas that are fully caught up with the leader, within replica.lag.time.max.ms. Only ISR members are eligible for leader election, so a shrunken ISR means reduced fault tolerance. Producers and consumers can inspect the ISR with kafka-topics.sh --describe and should alert when it is smaller than the replication factor.

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

If a follower does not fetch from the leader within replica.lag.time.max.ms, the leader removes it from the ISR and records the change in metadata. Writes continue as long as min.insync.replicas is still satisfied, otherwise the leader rejects produce requests. When the follower catches up again it rejoins the ISR automatically.

No, Kafka guarantees order only within a partition. Two records with different keys can be written to different partitions and become visible in a different order than they were produced. If ordering matters for a group of events, give them the same key, or use a single partition for that topic and accept the throughput limit.

The send call serializes the key and value, chooses a partition, and appends the record to an in-memory batch buffer per partition. A sender thread groups batches by broker, compresses them and sends one request per broker, then completes futures as responses arrive. Because batching is asynchronous, send() normally returns immediately with a Future.

acks=0 means the producer considers a record sent as soon as it hits the socket, so the broker never confirms it. acks=1 waits for the leader to write locally, which can still lose data if the leader dies before replication. acks=all waits for all in-sync replicas, which is the only setting that combines with min.insync.replicas for real durability.

acks=all
min.insync.replicas=2

The producer retries retriable errors such as timeouts and NOT_LEADER_OR_FOLLOWER up to retries times, and delivery.timeout.ms caps the total time a record may spend from send to final failure. That budget covers batching, retries and in-flight requests, so a large retry count with a small timeout still fails early. When the budget is exhausted the record fails with a TimeoutException.

An idempotent producer attaches a producer id and sequence number per partition, so the broker discards duplicate retries instead of writing the record twice. Since Kafka 3.0 idempotence is enabled by default and requires acks=all, retries greater than zero and max.in.flight.requests.per.connection of at most five. It prevents duplicates from retries but does not by itself make writes across multiple partitions atomic.

enable.idempotence=true
acks=all
max.in.flight.requests.per.connection=5

batch.size is the maximum bytes buffered per partition before a batch is sent, and linger.ms is how long the sender waits for a batch to fill. Raising them improves compression and reduces request count at the cost of added latency. Under low traffic, linger.ms is what actually decides how long a record waits.

With more than one in-flight request per connection, a retry of an earlier batch can arrive after a later batch succeeded, reordering records. Setting it to 1 guarantees ordering but serializes throughput, while idempotence solves the problem up to five in-flight requests. This is why the modern default combination is idempotence on with a limit of five.

If buffer.memory is exhausted, send() blocks for max.block.ms and then throws a TimeoutException, or fails immediately if the metadata is unavailable. Records larger than max.request.size are rejected before sending. The usual remedies are to raise the limits deliberately or, better, to reduce the record size and increase throughput instead.

group.id identifies a consumer group, and Kafka assigns each partition of the subscribed topics to exactly one member of that group. Members with the same group.id share the work and the committed offsets, while a different group.id gives an independent full copy of the stream. Offsets are stored per group in the __consumer_offsets topic.

It applies only when the group has no valid committed offset, for example on first start or after offsets expire. earliest starts from the beginning of the partition, latest starts from the end, and none throws NoOffsetForPartitionException. Setting it to earliest does not rewind an existing group, which is a frequent misunderstanding.

auto.offset.reset=earliest
enable.auto.commit=false

With auto-commit enabled, the consumer commits the offsets of records returned by poll every auto.commit.interval.ms. Those records are committed before your processing finishes, so a crash after the commit loses the unprocessed records. At-least-once processing therefore requires disabling auto-commit and committing after the work is done.

A rebalance redistributes partitions among group members so that every partition has exactly one owner. It is triggered by a member joining or leaving, by a heartbeat timeout, by a missed max.poll.interval.ms deadline, or by a subscription change. During the rebalance consumers stop fetching, so long or frequent rebalances show up as lag spikes.

The eager protocol revokes all partitions from every member and reassigns them from scratch, which pauses the whole group. The cooperative-sticky assignor revokes only the partitions that must move and keeps the rest, which greatly reduces the stop-the-world effect. Cooperative-sticky is the recommended assignor in current versions.

partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

max.poll.records caps how many records one poll returns, and max.poll.interval.ms is the maximum time allowed between two poll calls. If processing a batch takes longer than that interval, the member is considered dead, is removed from the group and triggers a rebalance. Tune both together: smaller batches or faster processing, not a larger interval, is usually the right fix.

Lag is the difference between the log end offset of a partition and the group's committed offset, so it measures how far behind the consumer is. It appears in kafka-consumer-groups.sh --describe as LAG per partition and as a total. Growing lag means consumers cannot keep up with production and is the primary consumer health signal.

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

Build a Properties object with at least bootstrap.servers and the key and value serializers, then construct a KafkaProducer. The producer is expensive to create and thread-safe, so create it once and share it across threads. Always close it in a shutdown hook or try-with-resources so buffered records are flushed.

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
try (Producer<String, String> producer = new KafkaProducer<>(props)) {
    producer.send(new ProducerRecord<>("orders", "o-1", "created"));
}

Calling get() on the Future returned by send() blocks the calling thread until the broker acknowledges the record. It makes error handling trivial and is fine for scripts or tests, but it destroys batching and caps throughput at one round trip per record. In services, prefer asynchronous sends with callbacks.

RecordMetadata md = producer.send(record).get();
System.out.println(md.topic() + "-" + md.partition() + "@" + md.offset());

Pass a Callback to send, and the producer invokes onCompletion with either the metadata or the exception once the record is acknowledged or has permanently failed. The callback runs on the producer's I/O thread, so it must be fast and must not block. Log or forward unrecoverable failures, because a dropped record is invisible otherwise.

producer.send(record, (metadata, ex) -> {
    if (ex != null) log.error("send failed", ex);
    else log.debug("offset {}", metadata.offset());
});

The key.serializer and value.serializer classes must be set explicitly, and they must match the key and value types of the ProducerRecord. Kafka ships String, Integer, ByteArray and ByteBuffer serializers; for JSON you either serialize to a String yourself or configure a JSON serializer from a library such as Spring Kafka or Confluent. The consumer must use a compatible deserializer or it will fail at read time.

Call close() so that buffered records are flushed and the producer deregisters from the cluster; close(Duration) bounds how long it may wait. Registering a JVM shutdown hook avoids losing the last batches on an orderly shutdown. Never call close() from inside a callback, because that runs on the producer thread and can deadlock.

Runtime.getRuntime().addShutdownHook(new Thread(producer::close));

The ProducerRecord constructors accept an explicit partition number, which bypasses the partitioner, and a list of Headers for metadata such as trace ids or schema versions. Explicit partitioning is useful for ordering-sensitive tooling but couples your code to the topic layout. Headers are not part of the key or value and are visible to consumers without deserializing the payload.

ProducerRecord<String, String> rec = new ProducerRecord<>("orders", 2, "o-1", "created");
rec.headers().add("trace-id", traceId.getBytes(StandardCharsets.UTF_8));

Transactions let a producer write to several partitions atomically and commit or abort the whole set, which is essential for exactly-once stream processing. You set a transactional.id, call initTransactions, beginTransaction, then commitTransaction or abortTransaction. Consumers must set isolation.level=read_committed to skip aborted records.

producer.initTransactions();
producer.beginTransaction();
producer.send(new ProducerRecord<>("a", "k", "v"));
producer.commitTransaction();

Provide bootstrap.servers, a group.id, a key and value deserializer, and auto.offset.reset, then construct a KafkaConsumer. A consumer is not thread-safe, so one instance per thread is required. It should also be closed in a finally block or shutdown hook because it holds sockets and group membership.

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

poll() is what fetches records, but it also sends heartbeats and participates in rebalances, so the loop must keep turning even when no data arrives. If the loop stops longer than max.poll.interval.ms the broker evicts the member and partitions are reassigned. Processing therefore belongs inside the loop, bounded by max.poll.records.

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    records.forEach(r -> handle(r));
}

commitSync blocks and retries automatically until it succeeds or the retriable time budget ends, which is simplest and safest but costs throughput. commitAsync returns immediately and takes a callback, so failures must be handled manually and it is normally used in the loop with a final commitSync on shutdown. Commit after processing, not before, if you need at-least-once guarantees.

records.forEach(this::handle);
consumer.commitSync();

subscribe registers the consumer with a group so that partitions are assigned by the group coordinator and rebalances are handled for you. assign gives the consumer explicit partitions and removes it from group management, so there is no rebalancing and no automatic failover. assign is used for manual partition routing or replay tools, not for normal scalable services.

seek(TopicPartition, long) moves the next read position for a partition, and seekToBeginning or seekToEnd are shorthands. It only takes effect on the next poll and is often combined with pause and resume to reprocess a bad batch. Remember that seeking does not change committed offsets, so the group position and the stored offset can diverge.

consumer.seek(new TopicPartition("orders", 0), 100L);
consumer.seekToBeginning(consumer.assignment());

Set a volatile flag from the shutdown hook and call consumer.wakeup(), which makes the blocking poll throw WakeupException. Catch that exception outside the loop, then call close() in a finally block so offsets and group membership are released. Closing the consumer directly from another thread is not safe.

try {
    while (running) consumer.poll(Duration.ofMillis(500));
} catch (WakeupException e) {
    // shutting down
} finally {
    consumer.close();
}

The deserializer runs inside poll, so a bad payload throws an exception from poll and the consumer cannot advance past that offset, effectively stalling the partition. Robust pipelines either use a schema with compatibility checks or an ErrorHandlingDeserializer that wraps failures and passes them to the application. Skipping the poison record requires seeking forward and committing deliberately.

An offset is a monotonically increasing 64-bit identifier that Kafka assigns to every record within a partition. It is unique only inside that partition, so a complete location is the triple topic, partition and offset. Offsets never change and are never reused, which makes them safe to store as a progress marker.

Offsets are committed to the internal compacted topic __consumer_offsets as records keyed by group, topic and partition. The group coordinator broker owns those partitions and caches the values in memory. Because the topic is compacted, only the latest offset per key survives rather than every commit.

The position is the offset of the next record the consumer will fetch in this session, and it advances in memory as records are returned. The committed offset is what was persisted to __consumer_offsets and is what a restarted consumer resumes from. They diverge between polls and whenever auto-commit has not fired yet.

Offsets expire after offsets.retention.minutes, which defaults to seven days, once the group is empty. When the group comes back with no valid offset, auto.offset.reset decides where to start, which is a common cause of unexpected reprocessing or data skipping. Long-lived batch jobs therefore need either a bounded downtime or an explicit reset before restart.

Use kafka-consumer-groups.sh with --reset-offsets, choosing a target such as --to-earliest, --to-latest, --to-offset or --to-datetime. The group must have no active members, so stop the consumers first, and add --execute to apply instead of just printing the plan. Resetting is the standard way to reprocess data after a bug fix.

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-service --topic orders \
  --reset-offsets --to-earliest --execute

Yes. Offsets are recorded per group, so each group has its own position in the same partition and reads independently at its own speed. This is the basis of publish-subscribe semantics in Kafka and why adding a new consumer never disturbs existing ones. Within one group, however, each partition has a single owner and a single shared offset.

The log start offset is the earliest offset still retained after deletion or compaction, and the log end offset is the next offset the leader will assign. The high watermark is the highest offset that has been replicated to all in-sync replicas, and consumers can only read below it. The gap between log end offset and high watermark is data that is written but not yet safely replicated.

A record carries a key, a value, a timestamp, optional headers and its offset once stored. The key and value are byte arrays, so any format works as long as producers and consumers agree on serialization. Records are grouped into batches that are compressed and stored together, which is why compression applies per batch rather than per record.

The key determines the partition, so all records with the same key are ordered together and processed by the same consumer. It also enables log compaction, where the latest value per key is retained, which turns a topic into a replayable table. Records with no key are distributed for throughput and cannot be compacted meaningfully.

Headers are a list of string-keyed byte values attached to a record, separate from the payload. They typically carry trace ids, schema identifiers, content types or retry counts without forcing consumers to deserialize the value. They are not indexed and not used for routing, so they are metadata rather than a query mechanism.

A schema registry stores versioned schemas and producers send only a small schema id in the payload, so consumers can look up and validate the contract. Compatibility rules such as backward or forward compatibility are enforced at publish time rather than discovered in production. Avro, Protobuf and JSON Schema are the common formats used this way.

A null key means the record was not keyed, so the partitioner uses the sticky or round-robin strategy and compaction cannot deduplicate it. A null value is a tombstone: on a compacted topic it signals deletion of that key. Consumers must handle null values defensively, because a tombstone deserializes to null rather than to an object.

The producer supports none, gzip, snappy, lz4 and zstd, and compression happens per batch, so it is most effective with larger batches. zstd generally gives the best ratio with good CPU cost, lz4 favours speed, and gzip is the middle ground with broad tool support. The broker does not decompress for standard consumers, so the consumer pays the CPU cost, which matters for high fan-out topics.

compression.type=zstd
linger.ms=20
batch.size=65536

Kafka stores raw bytes and does no validation, so the producer's serializer and the consumer's deserializer must agree on the same format and version. A change that removes a field or alters a type silently breaks consumers at read time, potentially after the data is already committed to the log. Versioned schemas with enforced compatibility rules, or at least a documented contract, are the safe way to evolve payloads.

KRaft is Kafka's built-in consensus layer, which uses a Raft quorum of controller nodes to store cluster metadata in an internal log. It replaces ZooKeeper entirely, so brokers, topics, ACLs and configuration are all managed through the metadata log. It became production ready in 3.3, was the default in 3.5, and is the only supported mode from 4.0 onward.

ZooKeeper was a second distributed system to secure, monitor and upgrade, and it was never a good fit for Kafka's metadata volume or for large partition counts. The old controller had to load all metadata from ZooKeeper, which limited cluster size and slowed failover. KRaft removes that duplication and makes metadata changes fast, atomic and easier to replicate to brokers.

Historically ZooKeeper stored the live broker list, topic and partition metadata, ACLs and controller election state, and it watched for broker session expiry. The Kafka controller was one broker that wrote changes through ZooKeeper while other brokers watched znodes. In Kafka 4.x none of this exists, so mentioning ZooKeeper is only relevant for legacy 2.x or 3.x clusters.

The quorum is the set of controller nodes that replicate the metadata log and elect an active controller; brokers then replicate that log as observers. An odd number of voters is required, typically three for production and five for large or multi-zone clusters. Raft needs a majority to commit, so three voters tolerate one failure and five tolerate two.

process.roles=controller
controller.quorum.voters=1@ctrl1:9093,2@ctrl2:9093,3@ctrl3:9093

node.id must be unique and permanent for every broker and controller, because it is the identity used in the metadata log and in replica assignments. cluster.id identifies the whole cluster and is written at format time, so a node formatted with a different id cannot join. Changing node.id later effectively creates a new node and leaves stale assignments behind.

node.id=1
controller.quorum.voters=1@localhost:9093

The 3.x migration ran ZooKeeper and a KRaft quorum side by side, with a migration state machine moving from ZooKeeper to dual-write and finally to KRaft-only. It required specific broker configurations and a controlled sequence of restarts. In 4.x that path is gone, so upgrading from a ZooKeeper-based cluster means first going to a 3.x release and completing the migration there.

There is no ZooKeeper ensemble to deploy, but you now manage controller nodes, quorum health and metadata log snapshots. Controller failover is much faster because a standby is already up to date, and metadata changes are visible in the internal __cluster_metadata log. Topic and partition limits are far higher, but a healthy majority of controllers is a hard availability requirement.

Use kafka-topics.sh --create with the partition count and replication factor, then --describe to confirm the leader and ISR for each partition. --list quickly shows all topic names. Add --if-not-exists in scripts so that reruns are idempotent.

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic payments --partitions 3 --replication-factor 3 --if-not-exists
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments

kafka-console-producer.sh reads lines from stdin and sends them to the given topic. With parse.key=true and key.separator you can supply a key per line, which is how you test partitioning and compaction from a terminal. Without a key all lines are distributed by the sticky partitioner.

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic payments --property parse.key=true --property key.separator=:

kafka-console-consumer.sh needs --from-beginning to read existing records, otherwise it starts at the end. With --group it becomes a real group member and commits offsets; without a group it uses a random consumer and does not affect stored offsets. Add --property print.key=true to see keys alongside values.

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic payments --from-beginning --property print.key=true

--list shows all groups, --describe shows members, assigned partitions, current offset, log end offset and lag per partition, and --state shows whether the group is stable or rebalancing. --members --verbose adds the client host and assignment for each member. This is usually the fastest way to diagnose a stuck or lagging consumer.

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group payments-service

kafka-configs.sh --alter applies dynamic configuration without a restart, using --entity-type topics or brokers and --add-config. Use --describe to read the current overrides and --delete-config to remove one and fall back to the default. It is also the tool for creating SCRAM credentials with --entity-type users.

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter \
  --entity-type topics --entity-name payments \
  --add-config min.insync.replicas=2

kafka-get-offsets.sh prints the offset for a given position, where -2 means earliest and -1 means latest. The output is topic:partition:offset, which is ideal for scripting and for measuring the total record count between two points. It replaces the older kafka-run-class approach.

bin/kafka-get-offsets.sh --bootstrap-server localhost:9092 \
  --topic payments --time -1

kafka-dump-log.sh prints the contents of log segments, indexes and metadata, including batch headers, producer ids and compression. It is the tool of choice when you need to prove what is physically stored or debug a corrupt segment. Run it against a stopped or read-only copy where possible, because reading active files can be misleading.

bin/kafka-dump-log.sh --files /var/lib/kafka/data/payments-0/00000000000000000000.log --print-data-log

UnderReplicatedPartitions counts partitions whose ISR is smaller than the assigned replica set, meaning at least one copy is behind. A persistent non-zero value points to a slow or failing broker, disk saturation or network problems, and it shrinks your fault tolerance. It is the single most important cluster health metric and should normally be zero.

Lag grows when consumption is slower than production, which is the first visible symptom of an overloaded service, a slow downstream dependency or a stuck consumer. Alert on sustained growth and on time-based lag rather than on a fixed record count, because a few large batches can be harmless. Per-partition lag also exposes hot partitions that a total hides.

Track the 99th percentile of produce and fetch request latency, not the mean, because tail latency is what clients actually experience. Separate the total latency from request queue time and local processing time to see whether brokers are saturated or the disk is slow. Rising queue time with flat processing time means the broker needs more threads or capacity.

Watch disk usage and disk I/O latency together, since Kafka writes sequentially but reads depend on the page cache. Network in and out reveals throughput saturation and consumer fan-out, and CPU shows the cost of compression and TLS. Disk filling up is the classic outage cause, so keep retention and free space aggressively monitored.

Kafka relies heavily on the page cache but still allocates for request handling, and long GC pauses stall request processing and delay replication. Frequent full GCs or high pause times show up as latency spikes and shrinking ISRs. The usual fixes are a modest heap, G1 tuning and avoiding oversized caches inside the JVM.

KAFKA_HEAP_OPTS="-Xmx6g -Xms6g"

MessagesInPerSec and BytesInPerSec show ingest, while BytesOutPerSec shows egress, which is usually much larger because of consumer fan-out. Compare them per broker to detect imbalance, where one broker carries most of the traffic. Together with request rate they explain whether a slowdown is caused by volume or by latency.

OfflinePartitionsCount above zero means some partition has no leader and clients cannot read or write it, which is an immediate page. ISR shrink and expansion rates reveal instability even when the current ISR looks fine, and a leader count that is skewed across brokers signals poor balance. ActiveControllerCount should be exactly one in a ZooKeeper-era cluster and is replaced by quorum health in KRaft.

A topic is a named, durable, append-only log to which producers write records and from which consumers read. It is the logical stream name that decouples producers from consumers. Physically, a topic is split into partitions that are distributed across brokers. Topics are multi-subscriber: many independent consumer groups can read the same topic without interfering with each other.

Use the kafka-topics.sh tool shipped with the broker distribution, pointing the bootstrap server at a KRaft controller or broker listener. You specify the topic name, partition count and replication factor. In Kafka 4.x ZooKeeper is gone, so the cluster must run in KRaft mode, but the tool's usage is unchanged.

bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders \
  --partitions 6 --replication-factor 3

It depends on the broker setting auto.create.topics.enable. When it is true, the broker creates the topic on first use using num.partitions and default.replication.factor, which usually produces a badly sized topic. In production this setting is normally disabled so that topics are created deliberately with explicit partitions and replication. A producer then receives UNKNOWN_TOPIC_OR_PARTITION until the topic is created.

Partitions are the unit of parallelism: maximum consumer parallelism in a group equals the partition count, and throughput scales with partitions per broker. Size them from target throughput divided by per-partition throughput, leaving headroom for consumer growth. Avoid excessive counts because each partition costs file handles, memory and controller metadata, and more partitions lengthen leader election and rebalances. Partition count can be increased later but never decreased, and increasing it breaks key-to-partition ordering guarantees.

Replication factor is the number of brokers holding a copy of each partition, consisting of one leader and zero or more followers. It determines how many brokers can fail without data loss. A factor of 3 is the standard production choice because it tolerates two simultaneous broker failures while keeping min.insync.replicas=2 satisfiable. A factor of 1 offers no fault tolerance and is only acceptable for local development.

retention.ms sets the maximum age of a log segment before it becomes eligible for deletion, while retention.bytes caps the total size a partition log may occupy. Both are enforced on a per-partition basis by the log cleaner's deletion checks. Whichever limit is reached first triggers removal of the oldest segments. Setting retention.bytes to -1 disables the size limit so only time applies.

Use kafka-configs.sh with --alter against the topic entity. Dynamic topic-level configs take effect without restarting brokers. You can inspect current overrides with --describe and remove them by setting the value to null so the broker default applies again.

bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name orders \
  --alter --add-config retention.ms=604800000

acks controls how many replicas must acknowledge a write before the producer considers the send successful. acks=0 means fire and forget with no acknowledgement, acks=1 waits only for the leader, and acks=all waits for all in-sync replicas. Durability increases from 0 to all, at the cost of latency. In Kafka 4.x acks=all is the default and is strongly recommended together with idempotence.

The idempotent producer attaches a producer ID and per-partition sequence numbers so the broker can discard duplicate retries caused by network errors. This removes duplicates within a single producer session without any application logic. Since Kafka 3.0 enable.idempotence defaults to true, and in 4.x it stays on unless you explicitly disable it. Enabling it requires acks=all, retries greater than zero and max.in.flight.requests.per.connection of at most 5.

linger.ms is how long the producer waits before sending a batch that is not yet full, trading latency for larger batches. A small value such as 5 to 20 ms usually improves throughput dramatically with negligible added latency. Combined with batch.size, it controls how efficiently records are compressed and how few requests reach the broker. Setting linger.ms to 0 sends immediately and typically produces many tiny requests.

batch.size is the maximum size in bytes of a single batch per partition, which bounds how much data goes into one request. buffer.memory is the total memory the producer may use for all records waiting to be sent across all partitions. If records are produced faster than they can be sent and buffer.memory is exhausted, send() blocks for max.block.ms and then throws. Both must be raised together for high-throughput workloads.

If the record has an explicit partition, that partition is used. Otherwise, if a key is present, the partitioner hashes the key with murmur2 and takes the modulo of the partition count, so equal keys always map to the same partition. If there is no key, the producer uses a sticky partitioner that fills one partition's batch before moving on, which improves batching while distributing load over time.

Attach a callback to send() and treat a non-null exception as a real failure, then retry or route to a dead-letter store. Configure acks=all with enable.idempotence=true and leave retries at a high value so transient errors are absorbed automatically. Note that the callback runs on the producer I/O thread, so keep it fast and never block. For strict durability also make sure min.insync.replicas is at least 2 on the topic.

producer.send(record, (metadata, ex) -> {
    if (ex != null) {
        log.error("send failed for key {}", record.key(), ex);
    }
});

retries is the maximum number of retry attempts for a record, while delivery.timeout.ms is the total wall-clock budget from send() until the record is either acknowledged or failed, including all retries and time spent in the buffer. Once delivery.timeout.ms expires the record fails even if retries remain. The effective behaviour comes from whichever limit is hit first, so the two should be tuned together.

With cleanup.policy=delete the broker removes whole log segments once they exceed retention.ms or retention.bytes, so old records disappear entirely. With cleanup.policy=compact the broker keeps at least the most recent record for every key and removes older versions, so the log becomes a materialized view of latest values. Delete is right for event streams, compact for changelogs and state snapshots.

A background thread pool called the log cleaner scans the dirty portion of a partition and builds an offset map of the latest offset per key. It then rewrites segments, copying only records that are still the newest for their key, and skips records whose value is null because a null value is a tombstone marking deletion. Compaction is asynchronous, so duplicates can exist for a while and consumers must handle them.

A tombstone is a record with a key and a null value. It signals that the key should be deleted, and the cleaner retains it for delete.retention.ms so downstream consumers have time to observe the deletion before the tombstone itself is removed. Producers must send tombstones with the same key to actually delete a compacted entry. Consumers reading a compacted topic must treat null values as deletes.

min.cleanable.dirty.ratio is the fraction of the log that must be dirty, that is un-compacted, before the cleaner considers a partition. min.compaction.lag.ms prevents records from being compacted until they are at least that old, and max.compaction.lag.ms forces compaction within that window. These settings trade disk usage against cleaner CPU and I/O.

Yes, cleanup.policy accepts a comma-separated list such as compact,delete. Compaction keeps the latest record per key while deletion still removes segments older than the retention limits. This is common for changelog topics that should not grow forever. Be aware that delete can remove the last record for a key, so consumers can no longer reconstruct the full state from the log alone.

Because retention is only enforced at segment granularity, a segment is deleted only when the whole file is expired. A very large segment.ms or segment.bytes therefore delays deletion and can keep expired data on disk for a long time. Small segments give finer retention granularity but add file handles and index overhead. Time-based rolling with segment.ms around one day is a typical compromise.

Use the DeleteRecords Admin API, wrapped by kafka-delete-records.sh, which advances the log start offset of specified partitions up to a given offset. Consumers can no longer read anything below that offset, but the topic, its partitions and its configuration remain. The operation is irreversible and takes effect immediately, so it is mainly used for cleaning up bad data or resetting a stream. It requires Alter or Delete permission on the topic.

{"partitions": [{"topic": "orders", "partition": 0, "offset": 5000}],
 "version": 1}

Kafka treats keys and values as opaque byte arrays, so any serialization is possible. Common choices are plain JSON, Avro, Protobuf, or simple strings and primitives. The format is a contract between producers and consumers, not something the broker enforces. In practice schema-based formats are preferred in production because they make evolution explicit.

Avro is a compact binary format with a schema, and it stores data without field names, which makes payloads smaller than JSON. Its schema resolution rules support safe evolution such as adding fields with defaults. It integrates tightly with Schema Registry, and generic or specific records give type safety in clients. The main cost is that data is unreadable without the schema.

Protobuf is a good fit when the same contracts are already shared with gRPC services and polyglot clients, since the same .proto files can be reused. It has very efficient encoding and strong generated-code support in many languages. Avro is often preferred when deep integration with the Kafka ecosystem and Schema Registry's Avro-first tooling matters more.

JSON is verbose, has no enforced schema, and every consumer must defensively parse it. Without a schema contract, a producer can silently rename or retype a field and break downstream jobs. JSON Schema can be registered in Schema Registry to mitigate this, but payloads remain larger than binary formats. It is still a reasonable starting point for simple internal streams.

The consumer is configured with key.deserializer and value.deserializer classes that must match what the producer serialized. The broker does not record the format, so mismatch produces garbage or a SerializationException. With Schema Registry the deserializer reads the schema ID embedded in the record's header bytes. Teams therefore treat the serialization format as part of the topic contract.

Confluent serializers write a magic byte 0 followed by a 4-byte big-endian schema ID and then the serialized payload. Consumers read the ID and fetch the writer schema from the registry to deserialize correctly. This keeps the payload compact because the schema itself is not sent with every record.

byte 0     : magic byte (0x00)
bytes 1-4  : schema id (int, big endian)
bytes 5-n  : Avro/Protobuf/JSON payload

A serializer only converts an object into bytes, and a deserializer does the reverse; producers and consumers are configured with one each. A Serde is a single abstraction that bundles both directions, which is what Kafka Streams needs for its operators. Connect and Streams therefore speak in Serdes while plain producers and consumers speak in serializers. A Serde must be symmetric, otherwise records written by an application cannot be read back by itself.

Kafka supports gzip, snappy, lz4, and zstd at the producer's compression.type setting. lz4 is usually the best default balance of speed and ratio, while zstd gives the strongest ratio at higher CPU cost. Codec choice applies per batch, so bigger batches compress better. The broker stores batches as-is and decompresses only when needed.

The producer compresses each batch before sending, so compression saves both network bandwidth and broker disk space. The broker stores the compressed batch untouched and only decompresses for validation or when a consumer with a different codec requests it. This design means compression cost is paid once on the client, not repeatedly on the server. Consumers decompress after fetching.

Compression reduces bytes on the wire and on disk but adds CPU cost on producers and consumers. With good batching the savings usually far outweigh the CPU, often improving end-to-end throughput. With tiny batches the ratio is poor and latency may increase slightly. Measure with realistic payloads before choosing a codec.

Compression algorithms exploit redundancy across a window of bytes, so a batch of many similar records compresses far better than many single-record batches. Increasing linger.ms and batch.size therefore multiplies the benefit of any codec. This is why tuning batching first and codec second is the standard advice.

Nothing breaks: compression.type on the consumer side is only a preference, and consumers always decompress whatever codec the batch uses. The batch header records the codec, so the client knows how to decode it. Only the broker's topic-level compression.type setting affects re-compressed data during down-conversion.

none disables compression entirely, so batches travel and are stored uncompressed. The Kafka 4.x producer default is none, so compression must be enabled explicitly with a codec such as lz4 or zstd. Leaving it off is a common cause of unexpectedly high network and disk usage. Brokers can also recompress at the topic level via compression.type.

The limits are applied to the compressed batch: the producer's max.request.size and the broker's message.max.bytes are checked against the bytes that actually travel. This means compression can let a batch of logically large records pass, because only the compressed size counts. On the consumer side the decompressed data can be far larger, so heap and fetch settings must account for the expansion. Very high compression ratios can therefore cause unexpected consumer memory pressure.

Every partition has exactly one leader replica that accepts all produce requests and serves consumer fetches. Followers replicate the leader's log by fetching from it continuously. If the leader fails, one of the in-sync followers is promoted. Clients only need metadata about leaders, which brokers publish to them.

The ISR is the set of replicas that are fully caught up with the leader within replica.lag.time.max.ms. Followers that stop fetching or fall too far behind are removed from the ISR and rejoin once caught up. Only members of the ISR are eligible to become leader without data loss. The ISR is tracked by the KRaft controller, which persists it as cluster metadata.

It sets the minimum number of replicas that must acknowledge a write for a produce with acks=all to succeed. If the ISR shrinks below that number, the broker rejects writes with NOT_ENOUGH_REPLICAS instead of accepting data at lower durability. With replication.factor=3 and min.insync.replicas=2, one broker can fail and writes continue; two failures stop writes. This is the core durability-versus-availability trade-off.

Unclean leader election allows a non-ISR replica to become leader when no in-sync replica is available, restoring availability but losing records that only the old leader had. It is controlled by unclean.leader.election.enable and is disabled by default. Enabling it turns Kafka into an AP system for that topic, which is usually unacceptable for financial or event-sourcing data.

The high watermark is the highest offset that has been replicated to all in-sync replicas. Consumers can only read up to the high watermark, so they never see records that might be lost in a leader failover. The log end offset, by contrast, is the offset of the next record to be written on the leader. The gap between them shows replication lag.

When broker.rack is configured, Kafka assigns replicas of a partition to different racks where possible. This means a rack or availability-zone outage does not take down all replicas of the same partition. It also lets consumers fetch from the closest replica. Without rack awareness, three replicas may land in a single zone and provide no real resilience.

On restart the broker truncates its log to the leader's log start offset, then catches up by fetching from the leader, rejoining the ISR once fully caught up. If its local data is intact it catches up quickly; if the log directory was lost it must re-replicate entire partitions, which can be very expensive. That is why controlled shutdown and adequate replication bandwidth matter.

Consumer lag is the difference between the log end offset and the consumer group's committed offset, measured per partition. It directly tells you whether consumers keep up with production. Rising lag means the consumer is too slow, stuck, or partitioned incorrectly. It is best monitored per partition because an aggregate can hide a single hot partition.

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

UnderReplicatedPartitions should be zero, and OfflinePartitionsCount must be zero on a healthy cluster. ActiveControllerCount should be exactly one across the cluster. RequestHandlerAvgIdlePercent and NetworkProcessorAvgIdlePercent dropping near zero indicate saturation. Also watch log flush latency, disk usage per log dir, and ISR shrink and expand rates.

kafka-topics.sh with --describe prints partition count, replication factor, leader, replicas and ISR for each partition. You can filter with --topic or --exclude-internal. This is the first command to run when investigating imbalance or under-replication.

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

Generate a reassignment plan with kafka-reassign-partitions.sh --generate, review the JSON, execute it with --execute, then verify with --verify. Throttle the move with --throttle to avoid saturating network and disk. In KRaft clusters the same tool works against the broker bootstrap servers.

Cruise Control continuously monitors cluster load and produces rebalancing proposals that move partitions or replicas to even out disk, CPU and network usage. It also detects and fixes anomalies such as broker failures. It runs as a separate service and applies changes through the partition reassignment API.

Each broker has one or more log.dirs, and kafka-log-dirs.sh --describe reports segment size per partition per directory. Log dir failure takes partitions offline, so capacity planning must keep free space well above the largest segment. Retention and segment sizing are the main levers to control growth.

Metadata is now stored in an internal KRaft log replicated across controller nodes instead of in ZooKeeper. A quorum of controllers elects an active controller that manages metadata, and brokers register with it. Operational tooling moves to kafka-metadata-quorum.sh and kafka-storage.sh for formatting and describing the cluster. ZooKeeper-based clusters are no longer supported in 4.0.

Kafka Connect is a framework for streaming data between Kafka and external systems without writing custom producer or consumer code. It handles scaling, offset tracking, retries, serialization and error handling for you. Connectors are configured declaratively and run inside worker processes. It is the standard tool for database, file and object-store integration.

A source connector pulls data from an external system and writes it into Kafka topics. A sink connector reads from Kafka topics and writes into an external system. The same framework and worker run both, but the direction of data flow is opposite. Some connectors, such as mirroring tools, act in both directions.

Standalone mode runs a single worker process with local file offset and config storage, suitable for development or edge jobs. Distributed mode runs a cluster of workers that share connector configuration through Kafka internal topics, giving fault tolerance and horizontal scaling. Production deployments almost always use distributed mode.

It creates config, offset and status topics, by default named after the group ID with suffixes such as configs, offsets and status. These store connector configuration, source offsets and task state. They should be created ahead of time with an adequate replication factor, because losing them loses connector state. The cluster also relies on a consumer group to assign tasks to workers.

A converter translates between the connector's internal structured data and the bytes written to Kafka. Common converters are Avro, Protobuf and JSON Schema, usually paired with Schema Registry. The converter is configured at the worker level through key.converter and value.converter. Choosing a different converter than the rest of the platform produces unreadable topics.

Single message transforms modify each record in flight, for example to add a field, mask a value, route to another topic or drop fields. They are stateless and applied per record, one after another in the configured order. Converters, by contrast, handle serialization between the connector's data model and Kafka bytes. Transforms do not change the format on the wire.

The worker exposes an HTTP API on port 8083 by default for creating, updating, pausing, resuming and deleting connectors, plus inspecting status and tasks. Creating a connector with PUT is idempotent for a given name. The API is the standard way to automate deployments from CI.

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @s3-sink.json

Source connectors periodically commit their source offsets to the offset topic keyed by connector name and partition. On restart the connector seeks to those offsets instead of rereading everything. Connectors must therefore expose a stable offset representation, such as an incrementing ID or timestamp column. Incorrect offset handling causes duplicates or skipped rows.

It supports an incrementing mode based on a monotonically increasing column, and a timestamp mode based on a timestamp column, plus a bulk mode for full reloads. Incrementing mode is preferred because it is exact, while timestamp mode can miss rows written with the same timestamp or with clock skew. You can also use a custom query instead of a table name.

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "mode": "incrementing",
  "incrementing.column.name": "id",
  "table.whitelist": "orders",
  "topic.prefix": "db-"
}

The connector issues a query filtered by the last seen value, for example WHERE id > 1000, for every poll. Without an index on that column the database performs a full table scan each time, which becomes catastrophic on large tables. An indexed primary key or sequence column keeps polling cheap. Composite keys require care because only one incrementing column is tracked.

Use insert or upsert modes with the primary key fields configured, so replays overwrite rather than duplicate rows. Set pk.mode and pk.fields to the record key or a field in the value. Delete mode plus a tombstone handler covers removals. Without a primary key, at-least-once delivery inevitably creates duplicates.

"insert.mode": "upsert",
"pk.mode": "record_value",
"pk.fields": "order_id"

Check the connector and task status through the REST API, since a connector can be RUNNING while individual tasks are FAILED. Then read the worker log for the underlying SQL exception, such as a type mismatch or missing column. Retriable errors like connection loss are handled by the configured tolerance, while non-retriable ones fail the task and need a restart or a fix.

The connector attempts the insert or upsert and the database rejects it, failing the task. Schema compatibility between the topic and the target table is therefore enforced manually, with schema evolution settings such as auto-evolve only working for supported changes. A dead letter queue can capture the offending record so the pipeline continues. This is why schema contracts should be validated before reaching the sink.

Run the JDBC source connector in bulk mode, which selects all rows on every poll and ignores offsets, then stop the connector once it has finished. A common pattern is to combine a bulk load with a later incrementing run, or to use a custom query restricted to historical rows. Watch task throughput and set an appropriate batch size so the source database is not overloaded. Because bulk mode rereads everything, it must not be left running permanently.

"mode": "bulk",
"batch.max.rows": "10000",
"poll.interval.ms": "1000"

A KStream represents an append-only stream of independent facts, where every record matters. A KTable represents a changelog of updates keyed by record key, where only the latest value per key is retained. Reading the same topic as a KTable gives you a continuously updated view rather than a sequence of events. This distinction drives how joins and aggregations behave.

A topology is the directed acyclic graph of processors that describes how records flow from source nodes through operators to sink nodes. It is built with the Streams DSL or the low-level Processor API and then executed by the Streams runtime. The topology is the unit that gets distributed across stream tasks and threads. Printing it with describe() is the standard first debugging step.

Topology topology = builder.build();
System.out.println(topology.describe());

A stream-table join enriches each stream record with the latest value for its key from the table side, similar to a lookup against a database. The table is materialized in a local state store so the lookup is fast and does not require a network call. It is used for enrichment such as adding customer details to an order event. Only records arriving after the table has the key are enriched unless you use versioned or foreign-key variants.

State stores are local, embedded key-value stores that back stateful operations such as aggregations, joins and windowing. They are typically RocksDB, with an in-memory option for small state. Each store is backed by a changelog topic in Kafka so it can be restored after a failure. This combination gives fast local access with durable recovery.

A windowed aggregation groups records for the same key into time windows before aggregating. Kafka Streams offers tumbling windows of fixed, non-overlapping size, hopping windows that overlap, sliding windows that start with each record, and session windows based on activity gaps. The choice depends on whether you need periodic buckets or gap-based sessions. Window retention must be configured long enough for late records.

Streams uses at-least-once by default, so a crash between processing and committing offsets reprocesses the last batch. Enabling processing.guarantee=exactly_once_v2 makes writes and offset commits atomic through transactions. Even with exactly-once, a state store restore or a topology change can replay records from changelog topics. Applications should still be written to tolerate replayed input where possible.

The application runs as a consumer group, and partitions are assigned to instances, creating stream tasks. Each task owns a subset of partitions and its own state stores. Adding instances triggers a rebalance that redistributes tasks and migrates state through the changelog topics. Parallelism is therefore bounded by the number of input partitions.

You need Properties with at least application.id and bootstrap.servers, a StreamsBuilder that defines the topology, and a KafkaStreams instance started from that topology. The application id names the consumer group and prefixes the internal changelog and repartition topics. Once started, the app runs until close() is called.

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

StreamsBuilder builder = new StreamsBuilder();
builder.stream("words").flatMapValues(v -> Arrays.asList(v.split(" ")))
       .groupBy((k, v) -> v)
       .count()
       .toStream()
       .to("counts", Produced.with(Serdes.String(), Serdes.Long()));

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

It is the base for the consumer group id, the prefix of all internal topics such as changelogs and repartition topics, and the namespace for local state directories. Two deployments sharing an application id form one logical application and rebalance against each other. A new application id therefore starts from scratch and creates a new set of internal topics.

Use TopologyTestDriver, which runs the topology in-process with mock input and output topics. You pipe records into input topics and read results from output topics synchronously, advancing the driver's wall clock for windowed results. This makes unit tests fast and deterministic. Integration tests against a real or containerized broker are still useful for serialization and rebalance behaviour.

try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
    driver.pipeInput(inputTopic, new TestRecord<>("k", "a b a"));
}

to() writes the stream to a topic as a sink, ending that branch of the topology. through() writes to a topic and immediately reads it back, forcing a repartition so downstream key-based operations are correct. In modern Kafka Streams, groupBy and joins insert repartition topics automatically, so through() is rarely needed. Prefer to() for outputs and let the framework handle repartitioning.

Call close() on the KafkaStreams instance and register a JVM shutdown hook so SIGTERM triggers it. A clean close commits offsets and closes state stores, which avoids unnecessary reprocessing on restart. With exactly-once processing enabled, an unclean shutdown also means the application must wait for the transaction timeout before recovering. Never kill a Streams process abruptly in production.

Streams needs default.key.serde and default.value.serde, or explicit serdes on each operator. The default is a byte array serde, so forgetting to configure serdes produces ClassCastExceptions or unreadable data. When using Schema Registry, set the Avro, Protobuf or JSON serde explicitly. Serdes must match the actual wire format of the topic.

Start additional processes or containers with the identical application id and topology; they join the same consumer group and split partitions automatically. Each instance needs its own state directory, which the framework derives from the application id and task id. Do not run two different topology versions under the same application id, because state stores and changelogs will not be compatible.

The security.protocol setting accepts PLAINTEXT, SSL, SASL_PLAINTEXT and SASL_SSL. SASL_SSL is the recommended production choice because it combines TLS encryption with SASL authentication. PLAINTEXT offers neither and should be restricted to isolated development environments. The broker defines listeners for each protocol it supports.

The broker presents a certificate chain from its keystore and validates clients against its truststore when mutual TLS is enabled. Clients similarly need a truststore containing the CA that signed the broker certificates. Certificates must match the hostname clients use, so advertised listeners and DNS matter. TLS adds handshake and encryption CPU cost, but modern hardware handles it well.

Authentication establishes who the client is, through TLS client certificates, SASL credentials or a custom mechanism, producing a principal. Authorization decides what that principal may do, enforced by ACLs on topics, groups and the cluster. You can have authentication without any authorization checks if no authorizer is configured, which is insecure. Both must be configured for real security.

Kafka supports SASL/PLAIN, SASL/SCRAM-SHA-256 and SCRAM-SHA-512, SASL/GSSAPI for Kerberos, and SASL/OAUTHBEARER. SCRAM is the common choice for username-and-password setups because credentials are stored hashed and can be managed at runtime. PLAIN sends the password in clear text inside the TLS tunnel and is discouraged. OAUTHBEARER is used with external identity providers.

An ACL grants or denies an operation such as Read, Write, Describe or Alter on a resource such as a topic, consumer group or cluster to a principal. Matching is by resource type, name pattern and host. ACLs are managed with kafka-acls.sh and take effect immediately. The broker must have an authorizer configured, otherwise ACLs are ignored.

bin/kafka-acls.sh --bootstrap-server localhost:9092 \
  --add --allow-principal User:svc-orders \
  --operation Read --operation Write --topic orders

When true, any resource that has no matching ACL is accessible to all principals, which makes ACL rollout easier but leaves gaps open. Setting it to false means deny by default, so every access must be explicitly granted. Production clusters should use false and manage ACLs as code. The setting applies only when an authorizer is configured.

Replication traffic carries all topic data, so an unsecured inter-broker listener bypasses every client-side control. Configure security.inter.broker.protocol to match the secured listener and give brokers their own principal and ACLs for cluster operations. Many real incidents come from locking down client listeners while leaving replication open. Broker-to-controller traffic in KRaft must be protected as well.

Schema Registry is a separate service that stores versioned schemas and enforces compatibility rules for topics. Producers register a schema and get an ID, which is embedded in each record so consumers can deserialize without out-of-band coordination. This prevents silent breaking changes between teams. It supports Avro, Protobuf and JSON Schema.

The modes are BACKWARD, FORWARD, FULL and NONE, each with a transitive variant such as BACKWARD_TRANSITIVE. BACKWARD is the default and means new consumers can read data written with older schemas. FORWARD means old consumers can read data written with the new schema. FULL requires both, and NONE performs no checks.

You may delete fields and add fields that have default values, because a new consumer reading old data can fall back to the default. Renaming or retyping a field without a default breaks compatibility. This mode supports upgrading consumers before producers. It is the safest default for most event pipelines.

The new schema must be readable by consumers using any previously registered version, and must also be able to read data produced with any previous version. This is the strictest mode and permits only optional additions with defaults and careful removals. It suits long-lived shared event contracts where teams upgrade independently. The cost is that many convenient changes are rejected.

A subject is the scope under which schemas are versioned and compatibility is evaluated. By default the TopicNameStrategy uses the subject topic-value or topic-key. Alternative strategies include RecordNameStrategy, which versions by record type regardless of topic, and TopicRecordNameStrategy, which combines both. The strategy matters when the same event type appears in several topics.

Schema Registry rejects the registration with a 409 error describing the incompatibility, and the producer's send fails with a serialization error. The record is never written, which protects consumers from breaking data. The fix is to evolve the schema compatibly or to register a new subject with a deliberate mode change. Switching a subject to NONE to bypass the check should be a reviewed decision.

Run multiple registry instances behind a load balancer; they are stateless and share state through the internal _schemas compacted topic. Any instance can serve reads and writes, and the topic provides durability. Clients should be given a list of registry URLs for failover. Because the topic is compacted, registry state is bounded by schema count rather than history.

Retriable errors extend RetriableException, for example TimeoutException, NotLeaderOrFollowerException or NetworkException, and the producer retries them automatically within delivery.timeout.ms. Non-retriable errors such as RecordTooLargeException, SerializationException or InvalidRecordException fail the record immediately. Your callback should treat non-retriable failures as data or configuration bugs and route them aside rather than retrying forever. Idempotent retries also avoid duplicates.

The record waited in the accumulator longer than delivery.timeout.ms, usually because the broker was slow, the network was saturated, or batching was too aggressive relative to volume. Check broker latency and ISR health first, then tune linger.ms, batch.size and buffer.memory. Raising delivery.timeout.ms only masks an underlying throughput or availability problem.

When buffer.memory is full and metadata is available, send() blocks for up to max.block.ms, then throws TimeoutException; if metadata cannot be fetched it throws the same exception earlier. This indicates the producer is generating records faster than the brokers can accept them. Remedies are to increase buffer.memory, reduce record volume, or fix broker-side slowness. Blocking a shared application thread inside send() is the real danger, so use bounded queues or backpressure.

A poison pill is a record the consumer can never process, typically because its bytes do not match the expected format. Because the consumer cannot advance past it, the partition stalls. The standard remedy is ErrorHandlingDeserializer, which captures the deserialization failure and passes it to an error handler that can log and forward the record to a dead letter topic. The offset is then committed so consumption continues.

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    ErrorHandlingDeserializer.class);
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS,
    KafkaAvroDeserializer.class.getName());

It occurs when the consumer tries to commit but the group has already rebalanced, usually because processing took longer than max.poll.interval.ms. The consumer no longer owns the partitions, so its commit is rejected. Fixes are to reduce max.poll.records, process in parallel, or increase max.poll.interval.ms to match real processing time. Blindly retrying the commit will not help.

Retry a bounded number of times with backoff, then publish the record to a dead letter topic containing the payload, headers and failure reason, and commit the original offset. Blocking the poll loop for long retries risks exceeding max.poll.interval.ms and triggering a rebalance. Non-blocking retry topics avoid that but add complexity. Always alert on dead letter volume.

It means the broker received a sequence number it did not expect from an idempotent producer, which indicates broker-side state was lost, for example after log truncation, or that another instance reused the same transactional id. The producer's sequence state can no longer be trusted, so the instance must be closed and a new producer created. Retrying the same record on the same producer cannot recover. This is one reason transactional ids must be unique per instance.

A consumer group is a set of consumers sharing a group.id that collectively consume a topic, with each partition assigned to exactly one consumer in the group. Group membership is coordinated by a group coordinator broker, and assignments are computed by a partition assignment strategy. If there are more consumers than partitions, the extra consumers stay idle. Different groups read the topic independently.

Eager rebalancing revokes all partitions from all consumers before reassigning them, causing a stop-the-world pause in consumption. Cooperative rebalancing, used with the cooperative-sticky assignor, revokes only the partitions that need to move and can run in multiple phases. This greatly reduces pause time for large groups. It requires all members to support the protocol version.

Range assigns consecutive partition ranges per topic and can be unbalanced across multiple topics, while RoundRobin spreads partitions evenly across all subscribed topics. Sticky keeps existing assignments to minimize movement, and CooperativeSticky adds incremental rebalancing. CooperativeSticky is the usual production choice. Custom strategies can be plugged in via partition.assignment.strategy.

Setting group.instance.id gives a consumer a stable identity so that a restart within session.timeout.ms does not trigger a rebalance. Its partitions are retained while it is away, avoiding reassignment cost and state restoration. This is valuable for stateful Streams apps and rolling deployments. The id must be unique per instance and stable across restarts.

The coordinator is the broker responsible for a specific group: it manages membership through heartbeats, runs the rebalance protocol, and stores committed offsets. The group's partitions are hashed to a coordinator partition of __consumer_offsets. If the coordinator moves, clients rediscover it automatically. Unresponsive coordinators cause long rebalances and commit failures.

Committed offsets are written to the internal compacted topic __consumer_offsets, keyed by group, topic and partition. Because it is compacted, only the latest offset per key is retained, and offsets.retention.minutes controls how long offsets survive after a group becomes empty. Once expired, the group has no committed position and falls back to auto.offset.reset. Never delete this topic's data manually.

heartbeat.interval.ms is how often the background thread pings the coordinator, and session.timeout.ms is how long the coordinator waits without a heartbeat before declaring the consumer dead. max.poll.interval.ms is the maximum time between poll() calls before the consumer is considered stuck and forced out of the group. A short session timeout detects failures quickly, while a generous max.poll.interval.ms accommodates slow processing. Tuning one without the other is a common source of rebalance loops.

With cleanup.policy=delete Kafka removes whole segments once the retention threshold is crossed, so older records simply disappear. With cleanup.policy=compact Kafka keeps the log forever by time but retains only the last record for each key, discarding superseded values during compaction. Compaction is keyed: records with a null key are never compacted away by the cleaner, and the active segment plus segments newer than min.cleanable.dirty.ratio thresholds stay untouched. Use compaction for changelog-style topics where you need the latest state per key rather than an event history.

cleanup.policy=compact
min.cleanable.dirty.ratio=0.5
segment.ms=600000
delete.retention.ms=86400000

The topic is compacted first and then subject to time/size-based deletion, so the log keeps only the latest value per key but still expires segments older than retention.ms. This is useful for state topics where you want bounded storage but also want stale keys to eventually vanish. Note that compaction never guarantees a record survives until retention.ms; it is a best-effort background process and offsets are not renumbered, leaving gaps in the log.

A tombstone is a record with a key and a null value that instructs the log cleaner to delete all earlier records with the same key. It must be produced with a non-null key, otherwise there is nothing to delete. In the Java client you emit one by calling send(new ProducerRecord<>(topic, key, null)). Consumers must be prepared to receive a null value and treat it as a deletion signal rather than a deserialization error.

producer.send(new ProducerRecord<String, String>("accounts", "A-17", null));

Tombstones are ordinary records until the cleaner processes them, so they survive at least until the segment becomes cleanable. delete.retention.ms (default 24 hours) is the minimum time a tombstone is kept after compaction so that slow consumers can still observe the deletion. If a consumer is offline longer than that window it may never learn the key was deleted and will keep a stale value.

delete.retention.ms=86400000

The dirty portion is the tail of the log that has been written since the last cleaning pass, while the clean portion already holds at most one value per key. min.cleanable.dirty.ratio sets the ratio of dirty bytes to total bytes that triggers the cleaner, so a lower value compacts more aggressively at the cost of more I/O. Very low values cause near-continuous compaction, and very high values let duplicate keys accumulate. The active segment is never cleaned regardless of the ratio.

Use kafka-dump-log.sh with the --deep-iteration flag on a segment file to see the surviving records and offset gaps. There is no supported administrative command to force an immediate clean; you influence it through the dirty ratio, segment size and time settings, or by rolling segments with kafka-leader-election-free means such as lowering segment.ms. Compaction also requires the partition to be online on a broker that is the leader.

kafka-dump-log.sh --files 00000000000000000000.log --deep-iteration --print-data-log

Compacted topics are unbounded in key space, so an ever-growing number of distinct keys will keep the partition size large even though duplicates are removed. Consumers that need the full current state must read from the beginning, which means auto.offset.reset=earliest plus a full replay on startup. Also avoid depending on record order across keys after compaction and never rely on compaction for exactly-once semantics, since it only reclaims space.

Implement org.apache.kafka.clients.producer.Partitioner and set partitioner.class on the producer, or supply a Partitioner instance through the new ProducerConfig/builder API. It is justified when you need affinity such as routing a tenant to a fixed partition, weighting hot keys, or co-locating records that must be consumed in order together. Prefer keeping the default murmur2 key hashing unless there is a concrete ordering or load requirement.

public class TenantPartitioner implements Partitioner {
  public int partition(String topic, Object key, byte[] kb,
                       Object value, byte[] vb, Cluster cluster) {
    int n = cluster.partitionCountForTopic(topic);
    return Math.floorMod(key.hashCode(), n);
  }
  public void close() {}
  public void configure(Map<String, ?> cfg) {}
}

A new partitioning scheme means the same key can land on a different partition than before, which breaks per-key ordering for the transition period. Downstream consumers and state stores see the key in two partitions, so any per-partition aggregation or Streams state becomes split. Plan such a change with a dual-write or a full topic migration rather than a rolling producer upgrade.

With a null key the record has no ordering guarantee, so the default partitioner uses a sticky strategy: it fills one partition for a batch window then moves to the next, which improves batching efficiency while spreading load. In older clients this was round-robin. If you need ordering you must supply a key, because null-keyed records can land anywhere across batches.

A good key has high cardinality, is stable over the record's lifetime, and is naturally the unit of ordering you need, such as a device id or account id. Avoid keys with severe skew like country codes, since one partition then becomes the hot spot and limits parallelism. Also avoid very large keys because they add per-record overhead and hash cost.

Common approaches are salting the key with a bounded suffix to spread writes across partitions, splitting the hot entity into sub-keys, or isolating the hot key onto a dedicated topic with its own consumer. Salting breaks global ordering for that key, so downstream aggregation must re-merge the shards. Choose based on whether you need per-entity ordering or only eventual aggregation.

String sharded = accountId + "-" + (seq % 8);

Hashing spreads keys evenly only in expectation, so with few partitions the variance is large and some partitions can receive noticeably more traffic. Increasing partitions smooths the distribution but raises metadata, file handle and rebalance costs. A practical rule is to have at least a few times more partitions than the peak consumer parallelism you expect.

You can increase the count with kafka-topics.sh --alter --partitions but you cannot decrease it. Because the default hash is taken modulo the partition count, existing keys remap to different partitions after the change, breaking per-key ordering and any co-partitioning assumption in Streams joins. Consumers also need to rebalance, and compaction state is now split across new partitions.

kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic orders --partitions 24

replication.factor sets how many copies of a partition exist, min.insync.replicas sets how many of them must acknowledge a write for it to be considered durable, and acks on the producer decides how many acknowledgements it waits for. With acks=all and min.insync.replicas=2 on a replication factor of 3, a write succeeds only if two replicas have it. If fewer ISR members are available the broker rejects the write with NotEnoughReplicas rather than silently losing durability.

replication.factor=3
min.insync.replicas=2
acks=all

The leader tracks each follower's last fetched offset and time; a follower is in sync while it is caught up within replica.lag.time.max.ms. Only ISR members are eligible to become leader and count toward min.insync.replicas. A follower that falls behind is removed from the ISR, which can shrink the effective durability of the partition.

Rack awareness makes Kafka place replicas of a partition on different racks or availability zones so a single zone failure cannot take out all copies. Brokers advertise their location with broker.rack, and the default KafkaRaftReplicasAssignor uses it when assigning replicas. Producers and consumers can also set client.rack so fetches prefer a local replica.

broker.rack=us-east-1a
client.rack=us-east-1a

A consumer with client.rack set will fetch from a follower replica in the same rack when one is available, subject to replica.selector.class on the broker. This reduces cross-AZ data transfer cost and latency. Correctness is unchanged because only committed, in-sync data is served.

replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelector

With a replication factor of 2 spread over two racks, losing one rack loses one copy of every partition, so availability depends entirely on the surviving copy being in sync. Odd replication factors of 3 or more across 3 racks give a clear majority and are the recommended pattern. Document the failure domain assumption explicitly, because Kafka cannot protect against a failure domain larger than the replica spread.

Generate a reassignment plan with kafka-reassign-partitions.sh --generate using a broker list, edit the JSON to respect racks, then execute it with --execute and monitor with --verify. Use --throttle to limit replication traffic so reassignment does not starve normal produce and fetch requests. Reassignment temporarily increases load on the destination brokers.

kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file plan.json --execute --throttle 50000000

In Kafka 4.x there is no ZooKeeper; a KRaft controller quorum of 3 or 5 nodes commits metadata through Raft, requiring a majority to elect a leader and accept writes. Broker replication is independent of that quorum, so a partition can stay available for produce and fetch even while metadata changes are blocked. Size the controller quorum as an odd number across failure domains and keep it separate from broker load in production.

process.roles=controller
controller.quorum.voters=1@ctrl1:9093,2@ctrl2:9093,3@ctrl3:9093

batch.size caps the bytes accumulated per partition before a batch is sent, while linger.ms is how long the sender waits for more records to fill that batch. A batch is sent when either limit is reached, so a small batch size plus a long linger sends many small batches late. Tune them together: raise both for throughput, lower both for latency.

batch.size=65536
linger.ms=20
compression.type=lz4

It limits how many unacknowledged requests a producer may have in flight to a broker. With values above 1 and retries enabled, a failed batch can be retried while a later batch succeeds, reordering records. Setting enable.idempotence=true (the default in modern clients) preserves ordering even with up to 5 in-flight requests because the broker deduplicates by sequence number.

enable.idempotence=true
max.in.flight.requests.per.connection=5

delivery.timeout.ms is the upper bound on the total time from send() to success or failure, covering batching, retries and in-flight time. It must be at least linger.ms plus request.timeout.ms, otherwise the producer fails at startup validation. When it expires the producer stops retrying and surfaces a TimeoutException, even if the record might still have been written.

delivery.timeout.ms=120000
request.timeout.ms=30000
retries=2147483647

A retry after a timeout is ambiguous: the broker may already have appended the batch, so the retry writes it again. Idempotent producers add a producer id and per-partition sequence numbers so the broker discards duplicates, and transactional producers extend this to atomic multi-partition writes. Only the idempotent/transactional path gives real protection; retry tuning alone does not.

Compression is almost always worth it for throughput-bound workloads because it reduces network and disk usage on the broker and in transit. lz4 and zstd are the usual choices; zstd compresses better, lz4 has lower CPU cost. Compression happens per batch, so it only pays off when batching is effective, and brokers decompress for validation and recompress on replication in some paths.

The producer buffers records before sending, and when buffer.memory is full, send() blocks for up to max.block.ms and then throws. This indicates the producer is producing faster than the brokers can accept. Fix it by increasing buffer memory, improving compression, or adding backpressure rather than blindly raising the timeout.

buffer.memory=67108864
max.block.ms=60000

Use acks=all with idempotence enabled, a replication factor of at least 3 and min.insync.replicas=2, and keep delivery.timeout.ms generous enough to ride out broker rolling restarts. Enable transactions if you need atomic writes across partitions. This combination trades latency and availability during failures for no silent data loss.

fetch.min.bytes is the minimum data a broker should return for a fetch, and fetch.max.wait.ms is how long it may wait to reach that amount. Raising both improves throughput by returning bigger chunks but adds latency to low-volume partitions. On idle topics the wait dominates, so long values delay record delivery noticeably.

fetch.min.bytes=1048576
fetch.max.wait.ms=500

max.partition.fetch.bytes caps the response size for a single partition, while fetch.max.bytes caps the total response across partitions. A record larger than the per-partition limit can never be fetched, so the consumer stalls even though the broker accepted it. Keep these above your largest message size, and remember the consumer must also have heap for the returned batches.

The consumer is considered dead, leaves the group, and its partitions are reassigned, which can cause duplicate processing of the records it had already fetched but not committed. The classic cause is a slow per-record handler that blocks the poll loop. Fix it by processing in a worker pool with careful offset management or by reducing max.poll.records so each poll finishes faster.

max.poll.records=200
max.poll.interval.ms=300000

The consumer sends heartbeats in a background thread every heartbeat.interval.ms, and the group coordinator evicts a member that has not heartbeated within session.timeout.ms. The heartbeat interval should be roughly a third of the session timeout so a couple of missed heartbeats are tolerated. Since heartbeats run separately from polling, a long processing loop triggers max.poll.interval.ms rather than the session timeout.

With read_uncommitted the consumer sees all records including those from aborted transactions; with read_committed it only returns records from committed transactions and stops at the last stable offset. Any consumer reading output of a transactional or exactly-once pipeline must use read_committed, otherwise aborted writes leak into results. The cost is added latency while open transactions remain unresolved.

isolation.level=read_committed

partition.assignment.strategy selects the assignor: RangeAssignor, RoundRobinAssignor, StickyAssignor or CooperativeStickyAssignor. The cooperative sticky assignor performs incremental rebalances so only moved partitions are revoked, avoiding a stop-the-world pause. Also consider group.instance.id for static membership to avoid rebalances on rolling restarts.

partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
group.instance.id=consumer-a-1

A lagging consumer benefits from larger fetch.max.bytes and max.partition.fetch.bytes plus a bigger receive buffer so each round trip transfers more data. Also raise max.poll.records only if the handler is fast, since the real bottleneck is usually processing, not fetching. Monitor records-lag-max and fetch rate together to see which side limits throughput.

A state store is the local, per-task data structure that holds intermediate state such as aggregations, joins and lookups. Because it lives on local disk it is not durable by itself, so Streams writes every mutation to an internal changelog topic named <application.id>-<store>-changelog. On restart or task migration the store is rebuilt by replaying that changelog from the beginning. Disabling logging with withLoggingDisabled() makes the store purely local and non-recoverable.

StoreBuilder<KeyValueStore<String, Long>> builder =
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("counts"),
        Serdes.String(), Serdes.Long());

KeyValueStore is a simple key to value map; WindowStore keeps timestamped values per key for a retention period; SessionStore keeps session segments per key; and since 3.x there is also a versioned key-value store for offset-based lookups. Each type has a persistent RocksDB implementation and an in-memory implementation. Choose persistent stores for anything larger than memory, and in-memory stores for small, reconstructable state.

It is created automatically on startup for every logging-enabled store, with cleanup.policy=compact via the internal topic config. Partition count matches the source topic partitions so co-partitioning holds. You can override the config with StreamsConfig.topicPrefix(TopicConfig.CLEANUP_POLICY_CONFIG) or by pre-creating the topic.

props.put(StreamsConfig.topicPrefix(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG), "2");

Caching is an in-memory write buffer that coalesces updates to reduce downstream emission and disk writes, controlled by cache.max.bytes.buffering and per-store withCachingEnabled(). Logging controls whether the store is mirrored to a changelog for fault tolerance. They are independent: a store can be cached without being logged, which reduces changelog traffic but loses the store on failure.

A key-value store keeps one entry per distinct key forever, so an unbounded key space grows until disk fills. Options are applying a retention window store, periodically emitting and deleting entries via the Processor API, or using a compacted changelog plus tombstone deletion. For windowed stores the retention is set at store creation and old segments are dropped automatically.

TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30));

Use interactive queries: expose the store through KafkaStreams.store(StoreQueryParameters) and wrap it in an HTTP or RPC endpoint. Set application.server so instances can be discovered, and query the specific instance that hosts the key's partition. With StreamsMetadataState you can route the lookup, falling back across all instances when the local store does not contain the key.

ReadOnlyKeyValueStore<String, Long> store = streams.store(
    StoreQueryParameters.fromNameAndType("counts", QueryableStoreTypes.keyValueStore()));

A task is the unit of parallelism, one per input partition per sub-topology, and it owns the state stores for the keys in that partition. Tasks are assigned to instances during rebalancing, so state follows the task rather than staying on a node. This is why sticky assignment and standby replicas matter for minimizing restoration cost.

It is a windowed join: both streams are buffered in window stores and records are matched only if their timestamps fall within the join window. Because there is no table side, the join emits a result for each matching pair and cannot be a lookup. Both inputs must be co-partitioned so matching keys land in the same task.

left.join(right,
    JoinWindows.ofTimeDifferenceAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30)),
    (l, r) -> l.getId() + ":" + r.getStatus());

A stream-table join enriches each stream record by looking up the current table value, producing one output per stream record and emitting nothing if the key is absent. A table-table join is a changelog-style join whose result updates whenever either side changes. The stream-table form requires the table to be materialized so lookups are local.

Two infinite streams have no notion of "current" value, so a boundary is required to decide which records are candidates to match, which the window provides. A table has a well-defined current state per key, so a stream record can look it up at processing time without a window. This is also why stream-table joins are the cheaper and more common enrichment pattern.

A GlobalKTable is fully replicated to every Streams instance rather than being partitioned, so any instance can look up any key locally. It is populated by consuming the whole topic with a custom mapping from record to key. Use it for small reference data such as currency rates or configuration, not for large high-cardinality topics because every instance stores all of it and restores it all.

GlobalKTable<String, Config> conf = builder.globalTable("config-topic",
    Materialized.<String, Config>as("config-store"));

A GlobalKTable join does not require co-partitioning because the lookups are local on every instance, which makes it convenient for enriching a stream with static data. KTable joins do require both sides to have the same partition count and partitioning scheme. The trade-off is that GlobalKTable state is duplicated across all instances and cannot be scaled by adding partitions.

Stream records arriving before the table has seen the corresponding key find no match and are silently dropped, which is the classic cause of missing enrichments after a restart. Mitigations are to materialize the table from a compacted topic replayed on startup, use a GlobalKTable seeded before processing, or buffer and retry. Monitoring unmatched lookups is essential in production.

An inner join emits only when both sides have a record for the key, while a left join emits the stream record even when no table value exists, using a null right-hand side. Left joins are common for enrichment where a missing reference should not drop the event. Be careful downstream: null values then flow into serializers and aggregations.

stream.leftJoin(table, (v, cfg) -> cfg == null ? v : enrich(v, cfg));

A tumbling window has a fixed size and no overlap, so every record belongs to exactly one window. A hopping window has a size larger than its advance interval, so records belong to several overlapping windows and are counted multiple times. Use tumbling for periodic reporting and hopping for smoothed moving aggregates.

TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1));
TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(10))
           .advanceBy(Duration.ofMinutes(1));

A sliding window is defined by a fixed time difference rather than a fixed boundary: for each record, the window covers records whose timestamps are within that difference of it, based on the stream's observed event times. It produces the most precise "events in the last N minutes" semantics but creates the most windows and therefore more state and output. It is available for join and aggregation via SlidingWindows.

A session window groups records per key into activity bursts separated by a defined inactivity gap, so windows are data-driven and variable-length rather than fixed. Sessions merge when a late record bridges two previously separate sessions. This makes them ideal for user-session analytics but harder to reason about for exact-once output timing.

SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30));

Grace extends how long a window stays open for late-arriving records after its nominal end, and only after the grace expires does Streams emit the final result and purge the window state. Without grace, a window closes as soon as stream time passes its end. A larger grace improves correctness for out-of-order data at the cost of delayed results and more retained state.

Stream time per task is the maximum timestamp seen so far, so it only advances when records with newer timestamps arrive on that task. If a partition goes idle, its stream time stalls and windows remain open, holding state indefinitely. max.task.idle.ms and emiting dummy records or using WallclockTimestampExtractor are the usual mitigations.

max.task.idle.ms=5000

They are considered late and dropped for that window; the aggregate is not updated. Streams exposes this through the late-record-drop sensor and the dropped-records-total metric. If drops are frequent, either increase grace or fix the upstream producer timestamp handling.

suppress(Suppressed.untilWindowCloses(...)) buffers window results and emits only the final value when the window closes after its grace period. This converts an updating stream of intermediate aggregates into one final record per window. It increases state usage and output latency, and requires the grace period to be set explicitly.

.suppress(Suppressed.untilWindowCloses(
    Suppressed.BufferConfig.unbounded()))

Watch process-rate and process-latency at the thread level, commit-latency and poll-latency for health, and records-lag-max per task for backlog. Store metrics such as put-rate and flush-latency reveal RocksDB pressure. All are exposed through the kafka.streams metric group and can be reported via JMX or the Micrometer integration.

restore-rate and restore-total show changelog replay progress, while restoration-time and the state metric (REBALANCING to RUNNING) show how long instances stay unavailable. A high records-lag on changelog partitions during startup confirms slow recovery. Rebalance rate and task migration counts explain repeated restorations.

streams.metrics().forEach((name, metric) ->
    System.out.println(name.name() + "=" + metric.metricValue()));

When a task is assigned, Streams first replays the changelog topic from the checkpointed offset stored in the local RocksDB directory, then resumes processing. If the local state directory is intact the replay is short; if the instance is new or the directory was wiped, the whole changelog is replayed. The task remains in RESTORING and does not process records until restoration completes.

num.standby.replicas makes other instances continuously replay a task's changelog into a local copy of the store. On failover the promoted instance has warm state and restores in seconds instead of minutes. The cost is extra network and disk usage proportional to the number of standbys.

num.standby.replicas=1

Options include increasing num.standby.replicas, tuning restore.consumer.<client config> settings such as fetch sizes, keeping the state directory on fast local disks, and increasing num.stream.threads so tasks restore in parallel. Also consider compacting the changelog aggressively and aligning commit.interval.ms so fewer records need replay from the last checkpoint.

restore.consumer.max.partition.fetch.bytes=52428800

Streams metrics describe local processing behaviour while broker metrics describe partition throughput and request handling. Correlating consumer records-lag-max with broker BytesInPerSec and request queue time tells you whether lag comes from slow processing or a saturated cluster. Emit both with a shared application tag so dashboards can align them.

Streams writes RocksDB SST files and the changelog restore data under state.dir; if the volume fills, the task fails and the application may enter an error state. Set state.dir to a dedicated volume with monitoring on free space and size the store retention carefully. Increasing disk is the only real fix once the store legitimately needs the space.

Use the Processor API when you need explicit control over forwarding, punctuation (scheduled callbacks), custom state stores, or access to record headers and metadata that the DSL hides. It is also the right choice for complex routing that would require awkward workarounds in the DSL. For standard map, filter, join and aggregate pipelines the DSL is shorter and less error prone.

Call process() or transform() on a KStream to insert a processor into a DSL topology, and use builder.addStateStore() to attach a store to that processor. In modern versions process replaces the deprecated transform and transformValues. The result stays a KStream so DSL operators can continue after it.

stream.process(() -> new DedupProcessor(), Named.as("dedup"), "dedup-store");

Punctuation schedules a callback either on stream time (WALL_CLOCK_TIME is the other option) so a processor can act without new input. Typical uses are flushing buffered records, expiring stale state, or emitting heartbeat records. Stream-time punctuation only fires as data advances, so an idle partition never punctuates.

context.schedule(Duration.ofSeconds(30), PunctuationType.WALL_CLOCK_TIME,
    ts -> context.forward(KeyValue.pair("flush", buffer.size())));

In the Processor API a processor emits records by calling context.forward(key, value, To.child("name")), which can send to a specific downstream node or all of them. There is no return value, so a single process() call may emit zero, one, or many records. This is what enables flatMap-style and conditional routing logic.

The DSL builds a logical graph and splits it into sub-topologies at repartition points; each sub-topology has its own tasks and its own state. When you insert a processor with a state store, that store is scoped to the sub-topology containing it, so a repartition downstream creates a new task boundary. Understanding this avoids surprises where state you expected to be shared is actually per-sub-topology.

A custom StateStore implementation gives you a specialized index or in-memory structure the DSL cannot express, but you must implement init, flush, close and restoration behaviour correctly. Fault tolerance still comes from the changelog only if you write through the record collector the framework provides. Most teams should wrap an existing store type rather than implement one from scratch.

public class DedupProcessor implements Processor<String, String, String, String> {
  private KeyValueStore<String, Long> store;
  public void init(ProcessorContext ctx) {
    this.store = ctx.getStateStore("dedup-store");
  }
  public void process(Record<String, String> r) {
    if (store.putIfAbsent(r.key(), r.timestamp()) == null) {
      context.forward(r);
    }
  }
}

Use TopologyTestDriver with the same properties as production, then push input through TestInputTopic and assert on TestOutputTopic or query the state store directly. Advancing wall-clock or stream time with the driver lets you test punctuators deterministically. This runs in milliseconds without a broker, making it suitable for CI.

try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
  TestInputTopic<String, String> in = driver.createInputTopic("in", new StringSerializer(), new StringSerializer());
  TestOutputTopic<String, String> out = driver.createOutputTopic("out", new StringDeserializer(), new StringDeserializer());
  in.pipeInput("k", "v");
}

tasks.max is the maximum number of tasks the connector may create, and each task runs on some worker in the cluster. It is an upper bound, not a guarantee: a source connector may only create one task if its partitioning scheme cannot be split, and a sink connector is limited by the number of partitions it consumes. Raising it beyond the partition count adds no throughput and only adds overhead.

{"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
 "tasks.max": "4"}

Converters translate between Connect's internal Schema/Struct representation and the bytes on the wire, while serializers only handle raw byte conversion for the clients. This separation lets one connector work with any format by swapping key.converter and value.converter. If the converter does not match the actual data, records arrive as opaque bytes and SMTs can no longer inspect fields.

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081

Source connectors report offsets through SourceRecord.sourceOffset() and the framework persists them to the offset.storage.topic, a compacted internal topic keyed by connector and partition. This is why a source connector can resume from the correct position after a restart or rebalance. Losing that topic means the connector restarts from its configured initial offset and can duplicate data.

offset.storage.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status

config.storage.topic holds connector and task configurations and must be compacted with a single partition. offset.storage.topic holds source offsets and should be compacted with many partitions. status.storage.topic holds connector and task status plus partition-level sink offsets and should also be compacted. All three must have replication factors of at least 3 in production.

Set errors.tolerance=all to skip failing records and optionally configure a dead letter queue with errors.deadletterqueue.topic.name plus errors.deadletterqueue.context.headers.enable=true. Without the DLQ the failed record is silently dropped, so always pair tolerance with a DLQ and alarms on its size. Keep errors.tolerance=none when data loss is unacceptable.

errors.tolerance=all
errors.deadletterqueue.topic.name=dlq-connect
errors.log.enable=true

Standalone mode runs a single worker process, stores offsets and configs in local files, and has no fault tolerance or rebalancing, which suits development and one-off jobs. Distributed mode runs multiple workers sharing the internal topics, so connectors survive worker failure and tasks rebalance automatically. Production deployments should always use distributed mode.

The framework tracks which offsets the sink task has flushed and, when offset.flush.interval.ms elapses, commits the lowest unflushed offset per partition. This gives at-least-once delivery by default, so a crash between writing to the sink and committing causes reprocessing. Sinks that need idempotency must implement upserts keyed by a stable record key.

offset.flush.interval.ms=10000

An SMT is a lightweight transform applied to each record as it passes through a connector, in the worker process, after conversion from bytes on source and before conversion to bytes on sink. It can rename, mask, filter or route individual records. It has no state, no windowing and no access to other records, and it never writes to Kafka on its own.

transforms=addSource
transforms.addSource.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.addSource.static.field=origin
transforms.addSource.static.value=orders-db

SMTs are per-record, stateless and run inside the Connect worker without creating another topic or application. Streams processors can aggregate, join, window and keep state across records. Use an SMT for cheap field-level fixes at the connector boundary and Streams when you need stateful logic or cross-record correlation.

List the SMT aliases in the transforms property and they are applied in that exact order, so a rename before a mask sees different field names than the reverse. Each transform receives the output of the previous one, and a failed or dropped record stops the chain. Order matters most when combining routing with field changes.

transforms=rename,mask
transforms.rename.type=org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.rename.renames=ssn:taxId
transforms.mask.type=org.apache.kafka.connect.transforms.MaskField$Value
transforms.mask.fields=taxId

Use RegexRouter to rewrite the topic name with a regular expression, combined with ExtractField or a header-based transform to make the field part of the name. For arbitrary per-record routing logic you need a custom SMT implementing Transformation that returns a record with a new topic. Routing to a topic that does not exist fails unless auto topic creation is enabled.

transforms.route.type=org.apache.kafka.connect.transforms.RegexRouter
transforms.route.regex=(.*)orders(.*)
transforms.route.replacement=$1orders-v2$2

Many SMTs are typed for Struct and expect a schema; a tombstone or schemaless record causes a DataException or a no-op depending on the transform. Apply the transform with the correct $Key or $Value variant and guard for nulls. Concretely, a MaskField$Value on tombstones will throw unless errors.tolerance is set.

Returning null from Transformation.apply() drops the record. The built-in Filter transform does this for a simple predicate on a field, and DropHeaders or a custom predicate covers more complex cases. Dropping is permanent and invisible downstream unless you log it, so add metrics when filtering at volume.

transforms.drop.type=org.apache.kafka.connect.transforms.Filter
transforms.drop.predicate=isTest
predicates=isTest
predicates.isTest.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.isTest.pattern=.*\\.test

Use Flatten$Value to collapse nested structs into a single level using a configurable delimiter, or ExtractField$Value to pick one nested field as the new value. Flattening simplifies sinks that only understand flat columns but produces long column names and can collide when delimiters repeat. Do it in the connector only when the sink cannot model nesting, otherwise you lose information.

transforms.flat.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flat.delimiter=_

Log-based CDC reads the database's write-ahead log, binlog or redo log through a native replication protocol, so every insert, update and delete is captured in commit order without polling tables. Query-based polling needs a timestamp or sequence column, misses hard deletes, and adds read load on the primary. Log-based capture has far lower impact and captures the full change stream.

Each change event carries before and after field images, an op code (c create, u update, d delete, r snapshot read), a source block with connector and position metadata, and ts_ms timestamps. The Kafka key is the primary key so all changes for a row land in one partition in order. Downstream consumers read the envelope or flatten it with ExtractNewRecordState.

{"before":null,"after":{"id":1,"status":"NEW"},
 "source":{"db":"shop","table":"orders","lsn":12345},
 "op":"c","ts_ms":1712345678901}

A delete arrives with op=d and a populated before image but a null after. With ExtractNewRecordState and delete.handling.mode=rewrite you get a tombstone with the key and a null value, which is exactly what a sink needs to delete a row. If you use drop mode the delete disappears and the sink keeps stale data.

transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.delete.handling.mode=rewrite
transforms.unwrap.drop.tombstones=false

On first start Debezium takes a consistent snapshot of existing rows and emits them as op=r events before streaming live changes. A blocking snapshot locks or holds a transaction and can be disruptive on large tables, so incremental snapshots use watermarks and signalling tables to chunk the read while streaming continues. Incremental snapshots can also be triggered later for a single table.

Log-based capture connects as a replication client, so the database user needs REPLICATION plus read access to the captured tables and, on Postgres, a logical replication slot. The slot tracks the confirmed position so the server retains WAL until Debezium acknowledges it. An abandoned slot prevents WAL truncation and can fill the disk, which is a common production incident.

ALTER ROLE debezium WITH REPLICATION LOGIN;
SELECT * FROM pg_replication_slots;

Because the pipeline is at-least-once, duplicates are normal; the fix is idempotent sinks. Use the primary key as the Kafka key and configure the sink for upsert with primary-key semantics instead of plain insert, so replaying an update rewrites the same row. Where the sink cannot upsert, deduplicate downstream on key plus log position.

"insert.mode":"upsert",
"pk.mode":"record_key"

Keep schemas enabled and register them in a Schema Registry so each change has a versioned schema, then let the sink connector apply compatible alterations or manage them explicitly. Set the registry compatibility mode to BACKWARD or FULL to reject breaking changes early. Flattening with ExtractNewRecordState hides the envelope but still requires the underlying schema to evolve compatibly.

Implement Connector with start, stop, taskConfigs and config, and a SourceTask with start, poll and stop. taskConfigs returns one config map per task, which is how you decide the parallelism. Version information comes from a Version class, and config() should return a ConfigDef so settings are validated and documented.

public class MySourceConnector extends SourceConnector {
  public List<Map<String, String>> taskConfigs(int maxTasks) {
    return IntStream.range(0, maxTasks)
        .mapToObj(i -> Map.of("task.id", String.valueOf(i)))
        .collect(Collectors.toList());
  }
}

SourceTask.poll() is called repeatedly and returns a list of SourceRecord objects, or an empty list when there is nothing to read; the worker then produces them to Kafka. SinkTask.put() is called with a collection of SinkRecords that the worker already consumed, and the method must block until the writes are flushed. A slow put() blocks the task's consumer and increases rebalance risk.

public List<SourceRecord> poll() {
  return List.of(new SourceRecord(
      Map.of("pos", next), null, "out-topic", null,
      Schema.STRING_SCHEMA, "payload"));
}

Attach the source position to each record via SourceRecord.sourceOffset() and the framework commits it to the offset topic after the record is written. On restart, the offsets passed to start() let the task resume exactly where it stopped. Offsets must be per source partition and monotonically advancing, otherwise the connector may replay or skip data.

Retry with bounded exponential backoff inside put(), and throw RetriableException so the framework retries the batch instead of failing the task. Throw ConnectException only for non-recoverable errors. Since offsets are committed after a successful flush, an in-place retry keeps at-least-once semantics without duplicating work across batches.

catch (SQLTransientException e) {
  throw new RetriableException("db unavailable", e);
}

Return a ConfigDef from Connector.config() with types, defaults, importance and validators, so Connect rejects bad configs at submission time and renders a UI form. Use ConfigProvider for secrets so passwords are not stored in plaintext in the config topic. Validate cross-field constraints in validate() for rules a single field validator cannot express.

CONFIG_DEF.define("bootstrap.servers", ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "Kafka brokers")
          .define("batch.size", ConfigDef.Type.INT, 500, ConfigDef.Range.atLeast(1), ConfigDef.Importance.MEDIUM, "rows per batch");

Build a jar with all non-provided dependencies and place the directory under a path listed in plugin.path, then restart the worker so it scans for plugins. In distributed mode the plugin must exist on every worker, since tasks can be scheduled anywhere. Use plugin.discovery=service_load with the Connector service file, and isolate versions per plugin directory to avoid classloader conflicts.

plugin.path=/opt/connectors
plugin.discovery=hybrid_warn

Implement Version.getVersion() returning the plugin version string; it surfaces in the Connect REST API and logs so operators can confirm what is deployed. Without it, troubleshooting mixed plugin versions across workers becomes guesswork. It has no effect on behaviour but is required for a well-behaved plugin.

Start from the required consumer parallelism and peak throughput per partition, then add headroom because you cannot easily reduce partitions later. A common approach is to size for peak produce rate divided by a measured per-partition throughput, rounded up and multiplied by a safety factor. Also keep the count a multiple of your consumer instance count to avoid idle consumers.

Both limits apply, and a segment is deleted when either one is exceeded, so whichever is tighter wins. Setting retention.bytes protects disk from a burst while retention.ms expresses how long consumers may be offline. If you need to keep data for seven days regardless of size, leave retention.bytes=-1.

retention.ms=604800000
retention.bytes=-1
segment.bytes=268435456

Use lowercase, dot or dash separated names that encode domain, entity and version, such as orders.payments.v1, and keep a registry of owners. Encode the environment in the cluster or a prefix rather than scattering it inconsistently, and never put consumer group names into topic names. Consistent naming makes ACLs, quotas and monitoring rules far easier to express as patterns.

One topic per event type gives clean schemas, independent retention and simpler consumer subscriptions, but multiplies topic count. A shared topic with a type field reduces count and preserves cross-type ordering per key, but forces every consumer to filter and couples schema evolution. Most designs use per-type topics for high-volume streams and a shared topic only for tightly related events of one aggregate.

Each partition costs file handles, memory for indexes, controller metadata and replication threads, and rebalances and leader elections slow down as counts grow. Cluster-wide recovery after a broker restart also takes much longer with a high partition count. Keep per-broker partition counts within tested limits and split into more clusters rather than growing one indefinitely.

Retention deletes whole segments, so a segment size larger than the retention target means data outlives its configured retention until the segment rolls. Very small segments increase file count and index overhead and slow down compaction. Default is 1 GiB; lower it only when you need tighter retention granularity on low-throughput topics.

Cluster defaults should encode the safe baseline: replication factor, min.insync.replicas, compression and reasonable retention. Per-topic overrides then express genuine exceptions like compaction or a short retention for transient data. Never let a per-topic override weaken durability below the baseline without an explicit review, because it is invisible to consumers.

log.message.format.version=4.0
num.partitions=12
default.replication.factor=3

Retriable exceptions are transient conditions such as NotLeaderForPartition, NetworkException, NotEnoughReplicas, or coordinator disconnects, where the broker state may self-heal after metadata refresh and retrying can succeed. Non-retriable exceptions are deterministic failures such as RecordTooLargeException, SerializationException, an invalid topic name, or ClusterAuthorizationException, where repeating the same send will always fail. The client classifies them internally via RetriableException, so custom interceptors should rely on that flag instead of string matching.

retries caps total attempts per record, retry.backoff.ms sets the base wait between attempts, and retry.backoff.max.ms bounds exponential backoff under sustained failure. delivery.timeout.ms is the hard upper bound covering batching, retries, and in-flight time, so it must be greater than linger.ms + request.timeout.ms. enable.idempotence=true is the default in 4.x and makes retries safe by deduplicating at the broker.

With idempotence the producer attaches a producer ID and a monotonically increasing sequence number per partition, and the broker rejects or acknowledges duplicates based on that sequence. A retried batch that was actually committed before a network timeout is therefore not written twice, which removes the classic at-least-once duplication at the producer level. This is why idempotence is enabled by default in Kafka 4.x and why acks=all is now implied by it.

The Callback fires once per record after final acknowledgement or terminal failure, exposing the record metadata (topic, partition, offset, timestamp) or the exception. It executes on the producer I/O thread, so blocking work in it stalls all other sends. Use it to route failed events to an error store, not to perform synchronous retries.

producer.send(record, (metadata, ex) -> {
    if (ex != null) {
        log.error("send failed for key {}", record.key(), ex);
        errorStore.put(record);
    }
});

If max.in.flight.requests.per.connection is greater than one and retries are allowed, a failed batch may be retried while later batches already succeeded, so records land out of order. Idempotence solves this for up to five in-flight requests by preserving per-partition sequence ordering. Beyond that limit, you must lower in-flight to one or accept reordering.

TimeoutException from the future only means the client stopped waiting; the record may still be in flight and may yet be committed, so treating it as a definitive failure risks duplicates. Either use the callback as the single source of truth or deduplicate downstream with a business key. Never retry synchronously on the same thread while the original batch is still pending.

A single bad record should be diverted to a dead letter path and the transaction continued, because aborting discards all previously written records in that transaction. Abort is correct when an invariant of the whole batch is violated, for example a missing reference entity, since a partial commit would leave consumers with inconsistent state. Aborts are cheap but they rewind consumer offsets to the last committed position, so they should be rare.

In the default poll loop the Deserializer.deserialize throws inside poll(), which surfaces as a SerializationException and prevents the batch from being processed; no offset is committed for that record, so the next poll hits the same record again. Without handling, this becomes a permanent stall of the partition. The usual fix is a deserializer that returns a wrapper carrying raw bytes and the error instead of throwing.

Catch the failure per record inside the processing loop, publish the original key, value, headers, and error metadata to a DLT, and only then commit the offset of the failed record. Recording the source topic, partition, offset, and exception class in headers makes replay possible. The DLT producer must use acks=all so the offset is never committed without a durable copy of the failed record.

try {
    handle(record);
} catch (Exception e) {
    dlt.send(new ProducerRecord<>("orders.DLT", record.key(), record.value(),
        headersWithOrigin(record, e)));
}
consumer.commitSync();

A poison pill is a record that deterministically fails processing, for example malformed JSON or a schema mismatch, so every redelivery fails identically. Retrying only blocks the partition and inflates lag while other records behind it starve. The correct response is bounded retries with backoff for transient causes, then quarantine to a DLT for deterministic ones.

Transient errors come from external dependencies, such as a database timeout or an HTTP 503, and are usually expressed as retriable exceptions with a clear recovery signal. Permanent errors come from the record itself, such as validation failures, unknown enum values, or schema incompatibility, and will never succeed on redelivery. Classify by exception type explicitly rather than by message text, and default unknown errors to bounded retry plus DLT.

Committing first converts the pipeline to at-most-once: a crash between commit and completion loses the record silently. It is acceptable only for idempotent, low-value workloads where loss is tolerable. For anything with side effects, process then commit, and make the side effect idempotent so at-least-once is safe.

A retry topic is a separate topic consumed with a pause on the partition, or a scheduled re-poll, so the failed record is retried after a delay without blocking the main topic for unrelated keys. Ordering per key is preserved only if all records for that key follow the same retry path and the main topic stops advancing that key. This trade-off is why many teams accept per-key ordering loss on retry topics and document it.

Replay with a dedicated consumer that reads the DLT, strips or rewrites the error headers, and republishes to the original topic, preserving the original key so partition affinity is kept. Do it only after the root cause is fixed, otherwise the records return to the DLT immediately and amplification hides the real problem. Record a replay marker header so that repeated replays are detectable and deduplication stays possible.

The event carries only a small notification, typically an ID and a type, and consumers call back into a service or read a store to fetch the full state. Messages stay small and the producer stays decoupled from consumer data needs. The cost is a synchronous dependency at consumption time and a risk of reading state newer than the event.

Event-carried state transfer embeds the full entity snapshot in the event so consumers can build local materialized views without calling back to the source. This removes runtime coupling and enables offline or high-throughput consumers, at the cost of larger messages and duplicated storage. A common hybrid sends a delta plus the changed fields only.

Commands go to a write side that validates and persists, then publishes domain events to compacted topics keyed by aggregate ID. Read-side services consume those events into purpose-built projections such as Elasticsearch or a denormalized table. Kafka acts as the durable, replayable log that lets a projection be rebuilt from offset zero at any time.

A saga is a sequence of local transactions linked by events, with compensating actions for rollback, coordinated by either an orchestrator topic or choreographed events. Kafka provides durable ordering per key and replay, but not distributed transactionality across services. Failures usually come from missing idempotency in compensations, which double-refund on redelivery, and from unbounded saga timeouts.

Compaction suits state transfer where only the latest value per key matters, such as entity snapshots or configuration, because it bounds storage by key cardinality rather than time. Time-based retention suits event streams where every record is a fact that must be replayed in order. Note that compaction is lazy and does not guarantee immediate removal of older versions.

Make every sink operation idempotent using a natural key or an idempotency table keyed by topic-partition-offset, or use Kafka transactions to write output and offsets atomically. For external systems without transactions, an upsert with a deterministic key is the practical answer. Deduplication windows on the consumer side are only a partial mitigation because they lose state on restart.

The service writes the business change and the corresponding event into the same local database transaction, into an outbox table, so the event exists if and only if the change committed. A relay or CDC connector then publishes outbox rows to Kafka and deletes or marks them, giving exactly-once handoff from database to log. This avoids the dual-write problem where a database commit succeeds but the Kafka send fails.

BEGIN;
UPDATE orders SET status = 'PAID' WHERE id = 42;
INSERT INTO outbox(topic, k, v) VALUES ('orders', '42', '{"status":"PAID"}');
COMMIT;

It sets how often Streams commits processed offsets and flushes state stores, defaulting to 30 seconds in recent versions. A smaller interval reduces the replay window after a crash and lowers end-to-end latency, but increases commit overhead and broker load. Very small values on high-throughput topologies can dominate CPU with commit traffic.

The cache buffers updates per key and forwards only the latest value on flush or eviction, so downstream operators and repartition topics see fewer records. cache.max.bytes.buffering controls the total budget, and setting it to zero disables caching, which raises write amplification substantially. Caching trades latency for throughput and should be tuned with commit interval in mind.

Add instances or threads up to the number of input partitions, because a task is bound to one partition and extra capacity stays idle beyond that point. To go further you must increase topic partitions and ensure the partitioner keeps related keys co-located. Scaling down requires care because state stores migrate through the changelog and cause rebalance-time restoration.

Large windows, high key cardinality, and infrequent compaction of the changelog all inflate local store size, and windowed stores retain data until the retention period passes. statestore.cache.max.bytes and rocksdb.config.setter tuning affect write amplification more than raw size. Monitor state-store-size metrics and set retention.ms on windowed stores deliberately.

A KTable is partitioned and co-partitioned with the stream it joins, so each instance holds only its share of keys. A GlobalKTable replicates the entire topic to every instance, enabling joins without co-partitioning but consuming memory and restoration time proportional to the full dataset. Use GlobalKTable only for small reference data.

num.standby.replicas keeps warm copies of state stores on other instances, so a rebalance can reassign a task without replaying the whole changelog. This shortens restoration from minutes to seconds for large stores. The trade-off is extra disk, extra network replication, and more restoration chatter during rolling deploys.

Streams processing is single-threaded per task, so a blocking call stalls the whole task and eventually triggers max.poll.interval.ms violations. Wrap remote calls with strict timeouts and circuit breakers, or move enrichment to an async stage with a bounded queue. Unbounded blocking is the most common cause of Streams rebalance loops.

The position is where the consumer will fetch next in memory, while the committed offset is what the group coordinator has durably stored for restart and rebalance recovery. They diverge between polls and commits, which is why a crash replays everything after the last commit. position() and committed() expose both values for diagnostics.

commitSync blocks and retries until success or a fatal error, which is correct at shutdown and after a batch where correctness matters. commitAsync is non-blocking and faster but does not retry, so a failed commit can be silently superseded by a later one. The common pattern is async during the loop and sync in the shutdown hook.

seek() only moves the local position; it does not change the committed offset until the next commit. This makes it a precise tool for replaying a range or skipping a corrupt record without altering group state. Call it between polls, ideally after pause(), to avoid fetching records you intend to discard.

consumer.pause(partitions);
consumer.seek(partition, offsetToReplay);
consumer.resume(partitions);

The position is clamped to the log end offset on fetch, and with auto.offset.reset the behaviour applies only when no committed offset exists. A seek past the end is legal but yields no records until new data arrives. Seeking before the log start offset triggers an OffsetOutOfRangeException handled by the reset policy.

The reset policy only applies when the group has no valid committed offset, for example a brand-new group or an offset aged out by retention. Once a commit exists, the consumer resumes there regardless of the policy. This is why a new group name unexpectedly reprocesses the entire retained log.

Use offsetsForTimes to translate a timestamp into per-partition offsets, then seek to those offsets and commit. This is far more accurate than subtracting a fixed record count because ingestion rates vary by partition. The operation is safe only while the target timestamp is still within retention.

Map<TopicPartition, OffsetAndTimestamp> ts =
    consumer.offsetsForTimes(targets);
ts.forEach((tp, o) -> { if (o != null) consumer.seek(tp, o.offset()); });

Setting a static member ID makes the consumer a static group member, so a restart within session.timeout.ms rejoins without triggering a rebalance and keeps its assigned partitions. This preserves local state and avoids offset churn during rolling restarts. The catch is that a stuck or duplicate instance with the same ID is fenced out of the group.

MM2 periodically emits offset-sync records mapping source partition offsets to target offsets into an internal topic, and the checkpoint connector writes consumer group offsets into __consumer_offsets on the target. Consumers can then resume at the translated position after failover. Translation is only as fresh as the last sync interval, so some reprocessing is expected.

It creates the replicated data topics, plus internal topics such as mm2-offset-syncs.<source>.internal, <source>.checkpoints.internal, and heartbeat topics per source cluster. Topic names are prefixed by default, which keeps multiple source clusters separable. Renaming policies can strip the prefix if the target must mirror names exactly.

Each source has a heartbeat topic that MM2 writes to and monitors; missed heartbeats increment the ReplicationCheckpoint lag and trigger alerts via the heartbeat connector metrics. There is no automatic failover, detection only signals operators or external automation. This is deliberate because automatic promotion risks split-brain writes.

Active-active runs producers in both clusters, so each cluster replicates the other's topics with prefixes and consumers must merge streams carefully to avoid duplicate processing. Active-passive keeps all writes in one cluster and uses MM2 only for standby readiness, which is simpler but leaves the standby idle. Active-active needs conflict resolution because the same key can be written in both directions.

MM2 copies a defined subset of topic configs and can sync topic ACLs when enabled, but it does not replicate broker-level or dynamic cluster-wide settings, and some configs are intentionally reset. Replication of configs is also subject to sync.topic.configs.enabled. Anything outside that subset must be managed by your provisioning tooling.

Checkpointing overwrites target group offsets only when sync.group.offsets.enabled is set, and it can regress a group that already progressed on the target. In failover scenarios, disable automatic offset sync and translate once at promotion time. Always validate the translated offset against target retention before switching traffic.

Throughput is bounded by the number of source partitions per task and by tasks.max, with each task owning a set of partitions for a source cluster. Raising tasks.max up to the partition count increases parallelism, and dedicating a connect cluster per source cluster avoids cross-source contention. Network bandwidth and target disk are usually the real ceiling.

The standard is replication factor three with min.insync.replicas=2 and producer acks=all, which survives one broker loss without losing acknowledged writes. RF two with min ISR two gives no fault tolerance for writes, since any single failure blocks producers. RF three with min ISR three is stricter but stops writes on any single broker loss.

Kafka's rack awareness assigns replicas to distinct racks so a zone outage cannot take out a whole replica set. Without it, a rack failure can drop a partition below min ISR and halt producers. Set broker.rack consistently, since a missing value silently disables the placement logic.

A stretched cluster keeps one KRaft metadata quorum across sites, giving low recovery time but making the metadata quorum latency-sensitive and vulnerable to cross-site partitions. Two independent clusters with MM2 tolerate full-site loss and allow independent upgrades, at the cost of asynchronous replication and manual failover. Most regulated environments choose the replicated pair.

The RPO equals the replication lag at the moment of failure, which is driven by MM2 throughput, network latency, and checkpoint interval. You can measure it as the offset delta between source and target using MM2 lag metrics. Reducing it means more replication capacity, not a configuration flag.

Size the standby for the full production write throughput plus consumer catch-up, because failover moves all traffic at once. Disk must cover retention for the replicated volume, and broker count should match the source to preserve partition distribution. A standby that is sized for replication only will fail under production load.

One source cluster replicates to several downstream clusters, typically one per region for locality or one per environment for analytics. MM2 supports this by running one replication flow per target, each with its own Connect cluster. Fan-out increases source egress linearly and must be accounted for in network capacity.

Run game days against a staging mirror of the topology, promote the standby, and validate that consumer offsets translate and producers reconnect with the new bootstrap servers. Include DNS or configuration switchover in the drill, since that is where most failures hide. Measure actual RTO rather than assuming it from architecture diagrams.

SCRAM authenticates with a username and password over a TLS channel, with credentials stored as salted hashes in KRaft metadata, which makes rotation easy but keeps a shared secret. mTLS authenticates with client certificates signed by a trusted CA, giving stronger non-repudiation and per-client identity but requiring PKI lifecycle management. Many clusters enable both, SCRAM for applications and mTLS for service mesh traffic.

Credentials are managed with kafka-configs.sh --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=...]' against a user entity, and the salted digest is stored in the metadata log. ZooKeeper-based kafka-configs is gone in 4.x, so all credential operations go through the KRaft controller. Rotation must be done with a dual-credential window to avoid client downtime.

The StandardAuthorizer stores ACLs in KRaft metadata and is the only authorizer available in 4.x, replacing kafka.security.authorizer.AclAuthorizer. ACL changes propagate through the metadata log, so they are consistent across brokers without ZooKeeper watches. It supports the same resource and operation model, so existing ACL expressions mostly carry over.

Kafka evaluates ACLs by resource pattern specificity, then by principal and operation, and a deny always wins over an allow at the same or higher specificity. There are no explicit deny entries; absence of an allow is a deny. This means broad allows on a wildcard resource can silently grant access unless allow.everyone.if.no.acl.found is false.

Quotas limit produce and consume byte rates and request rates per user, client ID, or IP, and they are enforced by throttling rather than rejection. They can be set cluster-wide in broker config or per entity via kafka-configs.sh, with the most specific match winning. Throttling shows up as increased latency, so quotas must be paired with alerting on throttle-time metrics.

The listener name declares both transport encryption and authentication mechanism, so SASL_SSL means TLS plus SASL, while SASL_PLAINTEXT authenticates but does not encrypt. SSL relies on mTLS for identity with no SASL exchange. Each listener must have matching listener.security.protocol.map and advertised listener entries.

Set security.inter.broker.protocol and the matching listener configs, and verify that the controller quorum uses its own listener with the same credentials. Rolling changes should be staged so brokers can still speak the old protocol during the restart window. A mismatch here typically appears as replicas failing to fetch and ISR shrinking immediately.

request.timeout.ms bounds a single request, delivery.timeout.ms bounds the whole send including retries, and max.block.ms bounds how long send() can wait for metadata or buffer space. Spurious failures usually come from max.block.ms being hit because the buffer is full, which looks like a send error but is really backpressure. Keep delivery.timeout.ms larger than linger.ms plus request.timeout.ms.

They set the socket send and receive buffer sizes, and on high bandwidth-delay product links the default of 128 KB can cap throughput well below link capacity. Raising them to a few megabytes per connection helps on cross-region links but multiplies memory across many connections. Buffer size is not the fix for a slow broker; it only removes a client-side ceiling.

Raising fetch.min.bytes makes the broker wait until enough data accumulates, which trades latency for fewer, larger responses and lower CPU per byte. It is the right tuning for high-throughput batch consumers and wrong for low-latency interactive ones. fetch.max.wait.ms bounds that wait so an idle partition does not stall forever.

It caps how many records a single poll() returns, which bounds the work per iteration and therefore the risk of exceeding max.poll.interval.ms. Lowering it improves heartbeat stability for slow processing at the cost of more round trips. It does not change fetch size, only how much of the fetched data is handed to the application per call.

Idle connections are closed by the client after this interval, and if a firewall or load balancer closes them silently first, the next send on that connection fails and triggers a metadata refresh. The symptom is periodic latency spikes on otherwise idle producers. Setting the client value below the network device idle timeout avoids the race.

linger.ms lets the sender wait to accumulate a larger batch, which raises throughput and compression efficiency while adding up to that delay per record. At low load the delay is the dominant latency term, so interactive producers should keep it small. It only helps when there is concurrent traffic to batch with.

Set batch.size large enough that batches actually fill, typically 64 KB to 256 KB, and use compression.type=lz4 or zstd to cut network and disk usage. Larger batches amortize per-request overhead but increase memory per partition and tail latency. Measure with batch-size-avg and compression-rate-avg before changing defaults.

With remote.log.storage.system.enable=true, closed log segments are uploaded to remote object storage, so local disk only holds the active segment and a configurable cache. Retention can then be measured in weeks or months at object-storage cost instead of local disk cost. Fetch latency for historical reads increases, so consumers replaying old data need larger timeouts.

local.retention.ms and local.retention.bytes bound the local copy independently of the topic's overall retention.ms, and retention.ms governs the remote copy. This lets you keep hours locally and months remotely. If local retention is unset, the local copy follows the topic retention and the disk savings disappear.

Compaction is right for changelog-style topics where only the last value per key matters, because it bounds size by key count and enables state restoration. Deletion is right for event streams where every record must be preserved in order for the retention window. Compaction never removes the latest record for a key, so a tombstone is required to actually delete a key.

Retention deletes whole segments, so a large segment.bytes or long segment.ms means data lingers well past its retention deadline. Smaller segments delete more precisely but increase file count and index overhead. A common tuning is segment sizes of a few hundred megabytes to one gigabyte with hourly rolls.

Enable compression at the topic level and let producers batch, since broker recompression is costly and should be avoided by matching producer and topic compression. Reduce retention only after confirming consumer lag and replay requirements, and use tiered storage to move cold segments off local disk. Increasing partitions does not reduce size and can increase it through replica overhead.

retention.bytes applies per partition, so total disk usage equals the limit multiplied by the partition count and replication factor. A limit that looks modest per partition can be enormous cluster-wide on a topic with hundreds of partitions. Size it from cluster capacity, not from a single partition's expected volume.

The hybrid applies both rules: segments are compacted to keep the latest value per key and also deleted once they exceed the retention bounds. This is useful for state topics where you want bounded history but also a hard size ceiling. Ordering guarantees still hold per key, but older values may disappear before the retention window ends.

New partitions become available immediately and a rebalance assigns them, but existing keys are not redistributed because the partitioner hashes against the new count. Records for a key written before and after the change can therefore land on different partitions, breaking per-key ordering across the boundary. Plan partition increases for topics where cross-boundary key ordering matters.

No, Kafka does not support reducing partitions in place because it would break key-to-partition mapping and offset semantics. The workaround is to create a new topic with the desired count, migrate producers and consumers, and delete the old one after draining. Deletion also requires delete.topic.enable=true and is asynchronous.

Use kafka-reassign-partitions.sh to generate a reassignment plan, execute it with --throttle, then verify completion with --verify before cancelling the throttle. Throttling prevents the movement from saturating replication bandwidth and pushing partitions below min ISR. Always confirm that UnderReplicatedPartitions returns to zero between batches.

kafka-reassign-partitions.sh --bootstrap-server b:9092 \
  --reassignment-json-file plan.json --execute --throttle 50000000

Deleting a topic removes metadata, all partitions, and the underlying log directories asynchronously, and it cannot be undone. Deleting data means letting retention or a delete-records call expire specific offsets while the topic and its configuration survive. Recreating a topic with the same name does not restore the old data, but it does reset consumer offsets for groups whose commits were deleted with __consumer_offsets retention.

Recreation loses all data and resets the topic's identity, so consumers with the same group may fall back to auto.offset.reset and either reprocess or skip. It also breaks anything holding a partition reference, such as Streams state or Connect offsets keyed by topic name. Prefer kafka-configs.sh --alter for dynamic configs and a blue-green topic for structural changes.

Adding a broker alone changes nothing until partitions are reassigned, but a poorly throttled reassignment can flood the new broker and trigger ISR shrink. Throttle the movement, move a bounded number of partitions at a time, and watch network and disk on both source and target. Use a plan that balances leadership as well as replica placement.

Kafka has no rename operation, so the practical path is to create the new topic with the target configuration, copy retained data with a mirroring job, and have consumers read both topics during a transition window. Offset continuity must be translated explicitly, since consumer group commits are keyed by topic name and will not carry over. Only after consumers have switched should producers be repointed and the old topic deleted.

Latency is the sum of batching wait (linger.ms), serialization and compression, time in the send buffer, network transit, broker append and replication acknowledgement, and the response path. With acks=all the replication leg dominates because the leader waits for in-sync followers. Each component is visible in producer metrics such as record-queue-time-avg and request-latency-avg.

It is usually not slower in isolation, but acks=1 can acknowledge a write that is later lost when the leader fails, forcing consumers to reprocess and inflating perceived latency. With idempotence enabled, acks=all is enforced anyway, so the comparison is often not real. Measure with the same idempotence setting before drawing conclusions.

The chain is fetch request wait (fetch.max.wait.ms), network transfer, decompression and deserialization, then application processing per record. Deserialization and per-record processing usually dominate, not the broker. fetch.min.bytes and larger fetch sizes reduce round trips but add wait time on idle partitions.

max.partition.fetch.bytes caps the response per partition, and fetch.max.bytes caps the total response, so one large partition can consume the whole budget and starve others in the same fetch. If a single record exceeds the per-partition cap, the consumer cannot make progress and loops. Keep both above the largest expected record size.

Commits flush state stores and send offset commits, and with a large cache the flush can be expensive, producing periodic pauses aligned to commit.interval.ms. Reducing cache size or staggering commits across instances smooths the curve. Persistent spikes usually indicate RocksDB flush or compaction work rather than the commit itself.

Compare producer request-latency-avg with broker RequestQueueTimeMs and ResponseQueueTimeMs; if the broker queue times are low but client latency is high, the delay is network or client-side batching. Broker TotalTimeMs percentiles show whether local append or replication is the bottleneck. Correlating the two sides avoids tuning the wrong component.

Extra instances trigger rebalances that pause consumption, and if the instance count exceeds partitions the surplus members idle while still participating in group coordination. Frequent scaling also resets fetch sessions and can cause repeated partition revocation. Scale to partition count and use cooperative sticky assignment to reduce the cost.

Network bandwidth and page cache typically saturate before CPU, because Kafka relies on the OS page cache for reads and zero-copy for sends. Disk throughput matters for replication and for consumers that fall behind the cache. As broker count grows, controller and metadata load also become visible, which is why KRaft separates the controller quorum.

Watch file descriptor limits, num.io.threads and num.network.threads, and heap pressure from per-partition request state; thousands of partitions per broker mostly stress memory and open files rather than CPU. Raise num.replica.fetchers to parallelize replication across leaders. Avoid very small partitions, since overhead is per partition, not per byte.

Kafka benefits from leaving most RAM to the page cache, so a heap beyond roughly six to eight gigabytes adds GC pause risk without improving throughput. Large heaps increase worst-case pause time, which delays requests and can push consumers past session timeouts. Use G1 with a bounded pause target and monitor GC time as a first-class metric.

It controls how many threads each follower uses to fetch from leaders, so raising it increases replication parallelism on brokers that follow many busy partitions. Too high a value competes with request handler threads for disk and network. It is a common fix for replication lag on large clusters with uneven leader distribution.

Each partition consumes memory, file handles, and metadata, and request batching becomes inefficient when each partition carries little traffic. Thousands of tiny topics increase controller work and make reassignment slow. Consolidating into fewer topics with a key or header discriminator often improves throughput substantially.

Apply client quotas per user or client ID, separate critical workloads onto dedicated topics with their own retention and replication settings, and consider dedicated broker pools for the heaviest tenants. Quotas throttle rather than reject, so they need alerting to be effective. Isolation by cluster remains the strongest boundary when SLAs differ sharply.

TLS adds handshakes per connection, per-record encryption on the zero-copy path, and additional CPU that competes with request handler threads. Throughput loss is modest on modern hardware for large batches but becomes significant when records are small and requests are numerous. Compensate with larger batches, connection reuse, more network threads, and TLS session resumption rather than by disabling encryption.

records-lag-max and records-lag per partition show the gap, while records-consumed-rate against records-produced-rate shows whether the gap is closing. Lag alone is misleading when traffic is bursty, so track lag over time and its trend. Export these through JMX Exporter or the Confluent metrics reporter rather than scraping JMX directly.

It counts partitions whose in-sync replica count is below the expected replication factor, which signals a broker or replication problem. It does not tell you which broker, why, or whether writes are still possible, since that depends on min.insync.replicas. Always alert on it together with IsrShrinksPerSec and IsrExpandsPerSec.

Track the KRaft quorum metrics for commit latency and the number of active controllers, plus ActiveControllerCount on brokers, which should be zero everywhere except the active controller. Metadata log lag between controllers indicates a follower falling behind. A stuck quorum blocks topic creation and leader election even while data-plane traffic continues.

RequestHandlerAvgIdlePercent falling and TotalTimeMs p99 rising together usually indicate disk contention, and LogFlushStats latency confirms it. Host-level metrics such as iowait and queue depth complete the picture. Disk full is a hard failure, so free-space alerting must fire well before the last segment fills.

The reassignment tool exposes throttle-time metrics, and a persistent positive throttle indicates movement is rate-limited, which is expected; the real signal is progress stalling. Track bytes moved per unit time and the count of partitions still not in the plan's final state. Alert if no progress occurs for a defined window rather than on throttle presence.

RequestQueueTimeMs measures how long a request waits before an I/O thread picks it up, so sustained growth means the broker cannot keep up with arrival rate. Rising queue time with flat CPU usually points to disk or network saturation rather than a code problem. Track p99 rather than averages, because averages hide the queueing that clients actually experience.

The log start offset moves forward as retention or compaction removes old segments, so a fast-rising value means data is expiring sooner than consumers expect. Comparing a group's committed offset against the log start offset shows how much safety margin exists before an OffsetOutOfRangeException. A sudden jump also signals an accidental retention change or a disk-pressure intervention.

It is repeated group rebalances in quick succession, each pausing consumption and revoking partitions, which drives lag up and can starve progress entirely. Common triggers are processing exceeding max.poll.interval.ms, session timeouts from long GC pauses, and rolling deploys that restart members rapidly. Cooperative sticky assignment plus tuning poll bounds usually resolves it.

A follower falls out of the ISR when it cannot fetch from the leader within replica.lag.time.max.ms, usually because of network saturation, slow disk, or an overloaded broker. Recovery is automatic once the follower catches up, but repeated shrink and expand cycles point to sustained resource pressure. Check per-broker network and disk before assuming a Kafka bug.

Log writes fail, the broker goes offline or becomes unable to serve partitions, and producers see errors on the affected partitions while leadership moves elsewhere. Kafka does not automatically delete data to free space. Prevention means free-space alerting, bounded retention, and moving log directories to larger volumes rather than deleting segments under pressure.

Compare per-partition byte-in rates and request rates across a topic; a partition carrying a disproportionate share means a skewed key or a null-key round-robin pattern. Fix it by salting the key, adding a random suffix for high-cardinality fan-out, or redesigning the partitioner. Increasing partitions does not help when a single key is the cause.

The group's committed offset fell outside the retained range because retention deleted the segments while the consumer was stopped. With auto.offset.reset=latest the consumer silently jumps to the end and skips data, which is often worse than an error. Extend retention for topics with intermittent consumers and monitor the gap between committed offset and log start offset.

This means fewer than min.insync.replicas are in sync at append time, so check UnderReplicatedPartitions, ISR history, and the health of the follower brokers for that partition. It is usually a symptom of replication lag rather than a producer issue. Fixing broker disk or network pressure resolves it; lowering min ISR hides it at the cost of durability.

First check whether the group still has active members, because a group with zero members keeps its last committed offsets and therefore shows frozen lag. If members exist, compare records-consumed-rate with the log end offset trend: a stalled log end offset means the producer side stopped, not the consumer. A consumer that is alive but blocked shows no commits while heartbeats continue, which points to a stuck processing call or an exhausted thread pool.

The producer registers its transactional.id with the transaction coordinator, which hands back a producer id and epoch, then writes the records to every target partition before sending a commit marker to each of them and recording the outcome in the coordinator's state. Consumers using read_committed never advance past the Last Stable Offset, so none of the records become visible until the commit marker exists. Atomicity therefore holds across all partitions of all topics touched by the single transaction, not per publish call.

Each transactional.id is hashed onto one partition of __transaction_state, and the leader of that partition acts as the coordinator that persists the transaction's lifecycle states. Because that state is itself a replicated Kafka log, a coordinator failover can resume by reading it and then either complete or time out any in-doubt transaction. The replication factor and min.insync.replicas of __transaction_state are what actually decide whether transaction state survives a broker loss.

Idempotence, on by default since Kafka 3.0, deduplicates retries within a single producer session using the producer id and per-partition sequence numbers, so a network retry cannot duplicate a record. A transactional.id is a stable user-supplied identity that lets a restarted producer recover the same transactional state and fence its predecessor. Idempotence fixes duplicate writes; transactional.id adds atomic multi-partition writes plus continuity across process restarts.

The Last Stable Offset is the offset of the first record belonging to a transaction that is still open, and a read_committed consumer is not allowed to read beyond it. Below that boundary the broker filters out records from aborted transactions using the aborted transaction index it maintains per segment. The cost is added read latency and a stalled consumer whenever a long-running transaction stays open.

The coordinator increments the producer epoch for that transactional.id, which immediately fences the older instance: its next transactional request fails with ProducerFencedException and it must shut down. This is the mechanism that gives single-writer semantics per transactional.id across rolling restarts and zombied pods. The new instance's initTransactions() call also aborts any transaction the previous instance left open.

props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "orders-etl-0");
producer.initTransactions(); // fences any older instance

Configure a transactional.id, enable idempotence, and set isolation.level to read_committed on the input consumer, then loop per batch: beginTransaction, produce outputs, send the input offsets via sendOffsetsToTransaction together with the group metadata, and finally commitTransaction. Because the offsets are written into __consumer_offsets inside the same transaction, output records and consumption progress commit atomically or not at all. On any failure call abortTransaction and resume from the last committed offsets.

while (true) {
  var records = consumer.poll(Duration.ofMillis(500));
  if (records.isEmpty()) continue;
  producer.beginTransaction();
  records.forEach(r -> producer.send(transform(r)));
  producer.sendOffsetsToTransaction(offsetsOf(records), consumer.groupMetadata());
  producer.commitTransaction();
}

Transaction duration is bounded by transaction.timeout.ms, which the broker caps at transaction.max.timeout.ms, so long batch jobs must be split into multiple transactions. A producer crash or coordinator failover leaves in-doubt transactions that block the Last Stable Offset for every read_committed consumer on those partitions until they are expired. Transactions also add per-record overhead, cannot span clusters, and do not protect against duplicates introduced downstream of the commit.

Call createTopics with explicit NewTopic definitions and inspect the per-topic results, treating TopicExistsException as success, or diff against listTopics before creating. Always pass partition count, replication factor and configs explicitly so repeated runs converge instead of inheriting whatever broker defaults are current. Because a batch returns a result per topic, partial failures must be handled individually rather than assuming all-or-nothing behaviour.

CreateTopicsResult res = admin.createTopics(List.of(
    new NewTopic("orders.v1", 24, (short) 3)
        .configs(Map.of("cleanup.policy", "delete",
                        "retention.ms", "604800000"))));
res.values().forEach((name, f) -> {
  try { f.get(); }
  catch (ExecutionException e) {
    if (!(e.getCause() instanceof TopicExistsException)) throw new RuntimeException(e);
  }
});

Partitions can only be added with createPartitions, never removed, and adding them permanently breaks the key-to-partition mapping for existing keys. Replication changes go through alterPartitionReassignments with explicit replica lists, which is the incremental API that replaced the older JSON-plan workflow. Apply leader.replication.throttled.rate and follower.replication.throttled.rate before starting so a large reassignment does not saturate inter-broker traffic.

Drive ACLs declaratively with describeAcls to read current bindings, diff against the desired set, then call createAcls and deleteAcls only for the delta; use describeUserScramCredentials and alterUserScramCredentials for SCRAM-SHA-256 and SCRAM-SHA-512 users. The automation principal needs cluster-level ALTER or super-user rights, so isolate it and audit its usage. Credentials must come from a secret manager and be rotated on a schedule, never embedded in the reconciler's own config.

admin.createAcls(List.of(new AclBinding(
  new ResourcePattern(ResourceType.TOPIC, "orders.v1", PatternType.LITERAL),
  new AccessControlEntry("User:etl-svc", "*", AclOperation.READ, AclPermissionType.ALLOW)))).all().get();

AdminClient exposes listConsumerGroupOffsets, alterConsumerGroupOffsets, deleteConsumerGroups and describeConsumerGroups, the latter returning member assignments and lag context under the new consumer group protocol. Resetting offsets of a group that still has active members is unsafe or rejected, so stop the consumers first or remove the members explicitly. Always persist the offsets you read before the change to a backup topic or artifact, because the operation is not reversible from Kafka itself.

Poll listPartitionReassignments until it returns an empty map, and while it is non-empty inspect addingReplicas and removingReplicas per partition to detect stalled movement. Progress is also visible in per-broker replica fetcher metrics and in controller log events, which help distinguish a slow reassignment from a stuck one. On completion the automation should clear the throttles it applied, otherwise steady-state replication stays capped.

Set default.api.timeout.ms and request.timeout.ms explicitly instead of accepting defaults, and let the client handle retriable exceptions for genuinely transient errors. Mutating calls such as createTopics or alterPartitionReassignments may have succeeded before the timeout, so on TimeoutException re-describe the resource rather than blindly resubmitting. Bound the retries and close the client in a finally block so long-running scheduled jobs do not leak connections.

Topic and broker-level dynamic configs can be changed at runtime through incrementalAlterConfigs, covering retention, throttles, log cleaner settings and most listener-independent tuning. Kafka 4.x keeps this metadata in the KRaft metadata log, so changes propagate through the controller without touching ZooKeeper or restarting a broker. Settings marked non-dynamic, such as broker.id or log.dirs, still require a restart and must be treated as a rolling operation.

Without a contract, producers and consumers drift silently: a renamed field or widened type breaks consumers at runtime with no compile-time or deploy-time signal. Move to Avro, Protobuf or JSON Schema with a registry so compatibility is checked when the schema is registered rather than when a consumer deserializes. If raw JSON is unavoidable, wrap it in an explicit envelope carrying a schema version and validate before producing.

The subject naming strategy decides whether schemas are registered per topic, per topic-and-record-type, or per record type across all topics. TopicNameStrategy is the default and forces exactly one schema per topic, which conflicts with topics carrying several event types. RecordNameStrategy lets a shared type evolve once across topics but couples unrelated pipelines, so the choice determines which compatibility checks actually protect you.

Implement org.apache.kafka.common.serialization.Serializer for the outbound side with a deterministic byte representation, and the mirrored Deserializer for the inbound side, then wire them through value.serializer and value.deserializer or a Serde. The client may instantiate them per partition, so they must be cheap to construct and must release resources in close(). A serializer that is not the exact inverse of its deserializer will corrupt state stores and changelogs long before it fails a test.

public class OrderSerializer implements Serializer<Order> {
  private final ObjectMapper mapper = new ObjectMapper();
  @Override public byte[] serialize(String topic, Order data) {
    if (data == null) return null;
    try { return mapper.writeValueAsBytes(data); }
    catch (JsonProcessingException e) { throw new SerializationException(e); }
  }
}

Put the event type in a record header or in an explicit envelope field, then dispatch on it in a thin routing layer or a delegating deserializer. With Schema Registry, use TopicRecordNameStrategy or RecordNameStrategy so each type gets its own subject and its own compatibility rules. Inferring the type from payload shape is a trap that breaks as soon as two events share a field name.

A Serializer or Deserializer handles one direction of a single record, while a Serde bundles both directions plus configuration so the Streams runtime can build state stores, repartition topics and changelogs with consistent encoding. Streams also needs the Serde for its internal topics, so an asymmetric implementation silently corrupts state stores rather than failing loudly. Build them with Serdes.serdeFrom(serializer, deserializer) or a typed wrapper that owns the configuration.

Wrap the real deserializer in ErrorHandlingDeserializer so poll returns a DeserializationException record instead of throwing and stalling the whole consumer. Route those records to a dead-letter topic with headers capturing the source topic, partition, offset and stack trace, then commit past them deliberately. Skipping them without a DLQ is silent data loss, so that decision needs to be explicit, monitored and alertable.

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
          ErrorHandlingDeserializer.class);
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS,
          SpecificAvroDeserializer.class);

Avoid reflection-based mappers per record and reuse a thread-safe mapper or parsed schema, caching resolved schema ids so you do not hit the registry on every message. Prefer compact binary formats at high throughput and confirm compression is applied after serialization, otherwise the benefit disappears. Deserialization usually dominates consumer CPU, so decode in batches and avoid materializing fields the pipeline never reads.

Key the topic by aggregate id so every event for one aggregate lands in the same partition and is therefore strictly ordered, and rebuild state by replaying from offset zero into a projection. Compaction keeps only the latest record per key, which makes a compacted topic suitable for a snapshot or changelog rather than for the full event history. The usual design keeps the event stream on delete retention and a compacted snapshot topic alongside it, because mixing both on one topic destroys history.

Periodically write the aggregate's full state together with the source offset or version it reflects into a compacted snapshot topic keyed by aggregate id. Loading then reads the newest snapshot and replays only the events that follow the recorded position, which bounds recovery time. Snapshot frequency trades recovery latency against extra write amplification and storage, so it is tuned per aggregate size and event rate.

Start a new consumer group at the earliest offset and write into a separate or versioned target so the serving projection stays online during the rebuild. Throttle the replay consumer and track progress through its committed offsets and lag rather than guessing. Cut traffic over to the rebuilt projection only after it has caught up to the live head, then decommission the old one.

kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group rebuild-orders-v2 --topic orders.events \
  --reset-offsets --to-earliest --execute

Compaction never guarantees a single record per key at any instant, because the active segment tail is left untouched and the latest value may legitimately appear several times. It also does not preserve anything beyond per-partition offset order, and tombstones survive until delete.retention.ms elapses. Segments can disappear through time or size based retention before the cleaner ever runs, so compaction is not a substitute for retention configuration.

Enforce backward compatibility in the registry and upcast old events inside the consumer rather than rewriting history. Keep payload changes additive: add fields with defaults, deprecate rather than repurpose, and never redefine the meaning of an existing field name. When a breaking change is unavoidable, introduce a new event type name so old and new representations can coexist during replay.

Compaction only rewrites closed segments, so segment.ms and segment.bytes must be small enough at your event rate for segments to roll and be cleaned regularly. Keep min.cleanable.dirty.ratio low enough that cleaning actually triggers, and set delete.retention.ms longer than the maximum consumer replay window so tombstones are not purged too early. Size retention by the replay-time SLA the business needs, not by available disk.

The cheapest guarantee is a single writer per key: key by aggregate id so exactly one consumer instance owns each aggregate at a time. Where multiple writers are unavoidable, carry an expected version in every command and reject or re-drive any command whose version does not match current state. Optimistic concurrency on the projection is far cheaper and more observable than introducing distributed locks into the pipeline.

Emit lineage metadata as record headers at produce time, carrying source system, producing job, schema id and correlation id, so the edge is recorded by the pipeline itself rather than by documentation. Derive the topic graph in a catalogue either by consuming those headers or by parsing job definitions such as Streams topologies, Flink SQL and Connect configs. Lineage that is maintained manually rots within a quarter, so generation must be automatic and continuously reconciled.

Field-level encryption or tokenization at the producer keeps sensitive values out of the log while preserving join keys, whereas segregating PII into a dedicated topic with tighter ACLs and shorter retention limits the blast radius. Whichever you choose, the record key must not itself be raw PII unless it is keyed-hashed, since keys are visible in tooling and offsets. Headers and dead-letter copies leak data just as easily as payloads and need the same treatment.

Extract topic metadata from AdminClient and schema metadata from the registry into a catalogue, tagging each entry with domain, owner, sensitivity classification and retention class. Sync automatically and on a schedule so the catalogue reflects drift instead of the state at last manual edit. Cataloguing only holds together if topic naming, ownership and classification are enforced at creation time by automation.

Translate each data classification into a concrete topic configuration of retention.ms, retention.bytes and cleanup.policy, then enforce maximums in CI or through a policy controller so no team can set infinite retention on a sensitive topic. Legal-hold exceptions must be explicit, owned and time-bounded, otherwise they become permanent. Verify actual deletion through log segment metrics rather than trusting the config value you set.

Kafka cannot delete a single record by key on a delete-retention topic, so the practical options are a tombstone on a compacted topic, crypto-shredding by encrypting each subject's data under a deletable key, or rewriting the partition and swapping the topic. Compaction only removes older values and only after cleaning, so it leaves a residual window governed by delete.retention.ms. Crypto-shredding is the only approach that scales to per-subject erasure while keeping the log immutable.

Express allowed topic configs, naming patterns, partition and replication bounds and required classifications as code, check declared topics in CI before deploy, and reconcile the live cluster against the same rules continuously via AdminClient. Broker-side guardrails such as the authorizer, quotas and restricted dynamic configs prevent teams from bypassing the checks. Every violation should raise an alert addressed to a named owner, not a page in a wiki.

Enable authorizer logging and ship broker audit logs to a tamper-evident store, then correlate principal, operation and resource for Read and Describe on classified topics. Consumer group metadata combined with ACLs reconstructs who consumed what, and alerting on new groups or unexpected principals catches drift early. Kafka cannot log per-record access, so audit granularity is capped at topic and consumer-group level and must be stated as such to auditors.

Start from the number of distinct keys per task multiplied by the serialized key and value size, then add RocksDB's own overhead of memtables, block cache, SST files and compaction headroom, which typically means planning for roughly 1.5 to 2 times the raw data on disk. Divide by the task count to get the per-instance footprint and multiply by the number of instances that will share the partitions. Leave additional headroom for restoration and for compaction peaks, which can transiently double the on-disk size.

Changelog topics retain history unless compacted, tombstones are only purged after delete.retention.ms, and windowed stores keep every window until its retention elapses. Cardinality is usually the real culprit: a composite key that embeds a timestamp or a request id grows with time rather than with the number of entities. Aggregating on a high-cardinality dimension silently turns a small store into an unbounded one.

Each input partition belongs to exactly one task, and each task owns its own state directory, so the task count equals the largest partition count among the input topics. When a rebalance moves a task, the new owner restores its stores from the changelog before it resumes processing. Standby replicas maintained through num.standby.replicas keep warm copies so the restore window is short, and cooperative rebalancing limits how many tasks move at once.

Design against a target restore rate derived from changelog size, not against steady-state throughput: increase num.standby.replicas so a warm copy already exists, and tune num.restore.threads plus the restore consumer's fetch sizes. Keep changelogs compacted so restoration replays the latest value per key instead of the full history. Finally cap per-instance state so a single restore fits inside the deployment's readiness budget.

props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
props.put(StreamsConfig.NUM_RESTORE_THREADS_CONFIG, 4);
props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 64L * 1024 * 1024);

In-memory stores fit when the working set comfortably fits in heap and you accept that a restart means a full replay from the changelog, since there is no local persistence. RocksDB is the default for large state because it spills to disk and supports persistent volumes, standby replicas and incremental restoration. A custom store is justified only when you control the indexing and segmenting strategy and can implement the Store interface's range and batching semantics correctly.

Ensure the changelog is compacted, which Streams does by default for key-value stores, and emit a tombstone whenever an entity is genuinely deleted so the key can eventually disappear. Set window retention to the business window actually needed and let the store purge expired segments instead of keeping them forever. Where the domain requires TTLs on aggregates, implement them with a punctuator that emits tombstones rather than leaving orphaned entries.

Track restore-consumer lag and the restore rate during startup, commit latency and commit rate in steady state, and the RocksDB block-cache hit ratio, SST file count and store size. Compare per-task process rate against the input topic's produce rate to spot backpressure before lag explodes. Alert on rebalance rate and on commit failures, since both precede duplicate processing or stalled tasks.

Streams suits JVM teams doing per-record transformations, joins and aggregations that can ship as ordinary services with no separate cluster to operate. Flink is the better fit when you need event-time watermarks with sophisticated late-data handling, savepoint-based upgrades, or SQL over streams with a managed job lifecycle. Spark Structured Streaming is appropriate for micro-batch ETL and lakehouse writes where throughput matters more than per-record latency.

Use a GlobalKTable for small, slowly changing reference data that every task needs, remembering that it is fully replicated on each instance and must therefore stay small. For large dimensions prefer a co-partitioned KTable join, and reserve interactive queries for serving lookups to external callers rather than for in-topology enrichment. A synchronous call to an external database per record is the pattern most likely to collapse under load.

Streams derives stream time from the maximum timestamp observed and applies a per-window grace period, dropping records that arrive after window end plus grace, so grace must be set explicitly for the business tolerance. Flink's watermark model offers finer control over out-of-orderness, allowed lateness and side outputs for late records. In either engine, emit window results as revisions keyed by window id so downstream consumers can apply corrections idempotently.

Deploy a second instance with a fresh application.id, reset its input offsets to the desired start, and write to a new or versioned sink so the live pipeline keeps serving. Keep the topology free of external side effects so replay is deterministic, and throttle consumption to avoid saturating the cluster during the backfill. Cut over by swapping the sink or the downstream consumer once the backfill catches up to the live head.

Streams co-deploys with your service, so scaling is just adding pods and there is no cluster to operate, but you own rebalancing, restoration and upgrade mechanics. Flink and Spark externalize job management, checkpointing and savepoints, which pays off for long-running jobs with large state and for stateful upgrades. Streams has no savepoint equivalent, so reprocessing is normally expressed as a replay from specific offsets into a new sink.

Set processing.guarantee to exactly_once_v2 so the runtime wraps consumed offsets, state store changelogs and produced records in a single transaction. The sink still decides the real guarantee: a Kafka sink inherits EOS, while a JDBC or HTTP sink needs an idempotent upsert keyed by a deterministic record id. Confirm downstream consumers use read_committed, otherwise they will read records that may later be aborted.

props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.ISOLATION_LEVEL_CONFIG),
          "read_committed");

TopologyTestDriver gives fast broker-free unit tests where you pipe records with explicit timestamps and assert on output topics and state store contents. Cover windowing, grace and out-of-order behaviour explicitly, then add a Testcontainers or embedded-broker integration test for serialization, EOS and restoration concerns the driver cannot model. A golden-file check on the topology description catches accidental repartition topics or changelog changes during refactors.

You declare Kafka, KafkaNodePool, KafkaTopic and KafkaUser custom resources, and the operator continuously computes the desired StatefulSets, ConfigMaps, Services and certificate secrets and applies them in an ordered, rolling fashion. Reconciliation is level-triggered, so manual drift is corrected automatically rather than persisting. In KRaft mode node pools carry broker and controller roles instead of the older separate zookeeper stanza.

Use volumeClaimTemplates in the node pool so every broker gets its own PersistentVolumeClaim, and pick a StorageClass with a Retain reclaim policy so deleting a pod never deletes the log. Size each claim for retention plus compaction headroom plus at least thirty percent free space, and insist on dedicated volumes because network storage with low IOPS caps throughput. JBOD across several volumes raises throughput but loses the whole broker when a single volume fails.

A PDB with maxUnavailable of one per broker set stops Kubernetes from evicting more than one broker at a time during voluntary disruptions such as drains and cluster upgrades. That alone is not enough: min.insync.replicas must sit below the replication factor and producers must use acks=all, or a single eviction still takes a partition offline. Always confirm under-replicated partitions return to zero before allowing the next eviction.

Populate broker.rack from the node's topology.kubernetes.io/zone label through the downward API, then spread brokers across zones with topologySpreadConstraints or node affinity so replicas of a partition land in different zones. With a replication factor of three across three zones the cluster survives losing a zone without data loss. Account for cross-zone traffic cost and for the added latency when a leader serves consumers in another zone.

The operator restarts one pod at a time and waits for the broker to rejoin and its partitions to be fully replicated before moving on, which only works if min.insync.replicas and the replication factor leave headroom. Before starting, verify no partition is already under-replicated and that Cruise Control is not simultaneously moving data. Clients should rely on retriable producer errors and correct rebalance handling rather than expecting zero disruption.

Controllers are modest in CPU but latency and durability sensitive, so give them their own node pool with dedicated PVCs and no broker role. Run three or five controllers so the quorum tolerates one or two failures, and never place two controllers on the same node. Co-locating controllers with broker load is dangerous because metadata commit latency gates every cluster operation, including partition leadership changes.

A CPU limit throttles the broker exactly during GC pauses and compaction bursts, so prefer requests with generous or absent limits for brokers. Memory limits must cover the JVM heap plus page cache plus off-heap buffers, because an OOM kill discards the page cache and throughput collapses after restart. Bursty shared storage classes are equally harmful, showing up as consumer lag rather than as an obvious error.

Cluster per tenant gives the strongest isolation of failure, upgrades and configuration, but multiplies operational cost and makes cross-tenant data flow awkward. A shared cluster with quotas, ACLs and naming conventions is much cheaper and is adequate for internal teams with predictable load. Most organisations land on a shared cluster for standard tenants plus dedicated clusters for regulated or extremely high-throughput ones.

User and client-id quotas bound producer_byte_rate, consumer_byte_rate, request_percentage and controller_mutation_rate, and the broker enforces them by throttling responses rather than rejecting traffic. Set explicit per-user quotas for tenant principals and keep a conservative cluster-wide default so an unconfigured tenant cannot monopolise the cluster. Because throttling is silent, alert on throttle time per user or tenants will degrade without anyone noticing.

kafka-configs.sh --bootstrap-server broker:9092 --alter \
  --add-config 'producer_byte_rate=10485760,consumer_byte_rate=20971520,request_percentage=200' \
  --entity-type users --entity-name tenant-a

Adopt a deterministic convention such as tenant.domain.entity.version and enforce it with authorization so a tenant can only create or read topics under its own prefix. Treat that prefix as the namespace for quotas, ACLs, retention policy, lineage and chargeback, which keeps every governance concern addressable by one key. Without prefix enforcement, one typo in a consumer config silently crosses a tenant boundary.

A shared topic couples every tenant's schema, retention, throughput and error handling, so one tenant's malformed payload can poison the stream for all of them. A hot tenant also dominates partition leadership and compaction, and per-tenant deletion or isolation becomes impossible without rewriting the log. Shared topics work for identical high-volume machine telemetry, but not for tenant-specific business events with independent lifecycles.

Bound partition count and replication factor per tenant through policy and quota, because every partition costs controller metadata, file handles and replication traffic. Cap controller_mutation_rate so a tenant creating thousands of topics cannot stall cluster-wide operations. Watch request handler idle percentage per user as well, since metadata-heavy workloads degrade everyone even when byte-rate quotas are respected.

Give each tenant a distinct principal, either a SCRAM user or a certificate subject with mTLS, and grant ACLs only on its prefixed topics plus its own consumer groups. Avoid wildcard grants and review any need for DescribeCluster, because that permission leaks topic names and metadata across tenants. Manage ACLs from a declarative per-tenant source and reconcile them, so a manual grant cannot quietly persist.

Aggregate broker metrics by principal and client-id: bytes in and out, request counts, throttle time, partition count and disk consumed by prefixed topics. Disk is the hardest to attribute because retention is time based, so derive it from partition size metrics rather than from produce volume. Publish those numbers back to tenants so quota negotiations rest on observed usage instead of anecdotes.

Consume the feature topic into an in-process cache or a low-latency key-value store keyed by entity id, and treat consumer lag as the freshness SLO. Kafka is a log, not a point-lookup store, so the serving path must be the materialized view rather than a per-request read from the topic. Block readiness until the cache is primed, otherwise early requests score against missing features.

Generate both from the same transformation code over the same event stream so the offline table is a materialization of the serving logic rather than a SQL reimplementation of it. Record both event time and ingestion time so point-in-time joins during training reconstruct exactly what was known at prediction time. Without those two timestamps, offline metrics look excellent while online quality degrades silently.

Yes for the latest-value-per-entity case: key by entity id, compact the topic, and any consumer can rebuild the current feature vector by replaying it. It does not provide historical feature values, so training still needs an append-only history or a lakehouse table for point-in-time correctness. Compaction lag and tombstone retention also mean the view is eventually consistent, not immediately consistent.

Emit corrections as new records for the same key carrying a monotonically increasing version or event timestamp, and have the serving layer reject any update older than what it already holds. Compaction keeps the record with the highest offset, not the highest event time, so an out-of-order correction can otherwise overwrite a newer value. Log both applied and rejected corrections so drift investigations have evidence to work from.

Publish predictions with the model version, feature snapshot reference and request id to a dedicated topic, then later join them with observed outcomes to compute online metrics. Keying by request id makes that join with delayed labels straightforward in a stream processor with a long window or in the lakehouse. The same loop feeds retraining triggers and exposes training-serving skew early.

Keep records far below max.request.size and the broker's message.max.bytes, and use the claim-check pattern: place a reference in the Kafka record and store the embedding in object storage or a key-value store. Multi-megabyte payloads inflate replication traffic, page cache pressure and fetch latency for every consumer on the partition. If inline payloads are unavoidable, compress them and give that topic its own tuned segment and fetch sizes.

Compute per-window statistics such as null rate, cardinality, distribution moments and range violations in a Streams or Flink job, and publish them to a metrics topic that drives alerting. Compare live distributions against the training baseline using PSI or a KS test rather than eyeballing means. Include schema and volume checks too, because a silent upstream producer change is the most common cause of a feature pipeline outage.

Compare per-partition byte-in rate and record count against the topic average and against the partition's own history, then confirm the skew is persistent rather than a spike. Per-broker metrics will show one leader saturated while its followers on other brokers are idle, which distinguishes skew from a uniformly loaded cluster. Sampling keys on the producer side or in a side aggregation identifies the specific offending key.

If per-key ordering is not required, salt the key with a bounded suffix to spread writes across partitions and aggregate downstream. If ordering matters, keep the single key but move it to a dedicated topic with fewer, larger consumers so it stops competing with normal traffic. Two-stage aggregation, where salted partial aggregates are combined by the real key, is the standard pattern when both throughput and correctness are needed.

String salted = key + "-" + ThreadLocalRandom.current().nextInt(16);
// downstream: re-aggregate by the unsalted key

With a key the default partitioner hashes the key bytes with murmur2 and takes the modulo of the partition count, so distribution is even only when keys are well distributed. Null keys use the sticky partitioner, which fills one partition's batch before moving on to improve batching efficiency. Sticky behaviour concerns keyless records only and does not fix skew from a dominant key.

Size for the target throughput per partition in bytes per second, which is bounded by a single partition's consumer parallelism and replication cost, then add future growth headroom. Remember that increasing partitions later breaks key-to-partition assignment, so plan for the next two or three years rather than current peak. Over-provisioning has real costs too, since each partition consumes controller metadata, file handles and replication bandwidth.

Enable automatic leader rebalancing so leadership spreads evenly across brokers after failovers and reassignments. Where manual correction is needed, trigger preferred leader election, which only moves leadership when the preferred replica is already in sync. For data-volume skew rather than leadership skew, Cruise Control's rebalance proposals redistribute partitions by load and must be run with throttles and during low traffic.

A single in-flight batch is limited by max.in.flight.requests.per.connection, so a slow or throttled partition can hold a request slot and block batches destined for other partitions under the same producer. Retries for that partition also consume the delivery timeout budget for records that have nothing to do with the hot key. Raising delivery.timeout.ms and separating hot-key traffic into its own producer or topic are the usual remedies.

Because a task owns one partition, a hot key makes one task and one instance do most of the work while the rest idle, so scaling out does not help. Co-partitioned joins additionally require identical partition counts and the same partitioner on both sides, or records silently fail to match. If the hot key is unavoidable, pre-aggregate or salt before the join and re-key afterward into a lower-volume stream.

Kappa keeps a single stream processing path and rebuilds history by replaying the log, which is viable when retention or tiered storage covers the replay window and the batch logic can be expressed streaming. Lambda adds a separate batch layer for corrections and historical accuracy, at the cost of two codebases that inevitably diverge. Retaining long history in Kafka through tiered storage is what makes kappa practical for more than a few days of data.

Run a continuous aggregation that materializes results into a queryable store, using ksqlDB or a Flink job with an external sink, and push deltas to the UI over websockets rather than polling an API that scans Kafka. Pre-aggregate at the granularity the dashboard needs so a refresh is a key lookup instead of a range scan. Keep the aggregation window small enough that freshness is bounded by event time plus processing latency, and monitor that end-to-end lag as the actual SLO.

Aggregate on event time with an explicit watermark or grace period, and make the choice of lateness tolerance a product decision rather than a default. Records beyond the tolerance should go to a side output or a retraction path instead of being silently dropped. Never aggregate on processing time for business metrics, because replay and consumer restarts will produce different numbers from the same data.

Publish aggregates as upserts keyed by window and dimension, and let the downstream store apply the latest version, so a correction is just a newer record for the same key. Where the store cannot upsert, emit a retraction and a new value so a summing consumer nets out correctly. Dashboards must read the versioned key rather than append-and-sum a stream that contains both old and corrected values.

Do not let every dashboard query Kafka directly: materialize into a serving store sized for the query pattern and let the UI hit that. Where consumers must read topics, scale by adding consumer instances within the partition limit and consider a dedicated read-only cluster fed by replication for analytical workloads. Caching aggregates at the API layer is usually far cheaper than adding broker capacity.

Interactive queries read the local state store of the task, so under exactly_once_v2 the store reflects committed transactional data only when the commit succeeds. Reading a standby replica returns data that is not yet committed on the active task, so routes must be aware of which copy they hit. For cross-instance queries, resolve the owning instance through metadata rather than querying every host and merging results.

Tiered storage moves closed log segments to remote object storage while keeping the local tail hot, so retention can extend from days to months without local disk growth. That makes replay-based backfills, long-window stream processing and audit queries practical on the same cluster. Reads of historical data hit remote storage with higher latency and cost, so consumers doing long scans should be isolated from latency-sensitive ones.

Workers in the same group form a cluster, and the leader assigns connector and task instances to workers and stores the configuration and offsets in the internal Connect topics. When a worker joins or leaves, the cluster rebalances and tasks are reassigned, which is why task count must exceed worker count to allow parallelism across nodes. A connector's tasks.max caps how many tasks the framework will create regardless of cluster size.

Build a shaded JAR with all connector dependencies and no conflicting Kafka client classes, place it under plugin.path on every worker, and restart or roll workers so the plugin scanner picks it up. Keep one plugin directory per connector version so upgrades are a directory swap rather than an in-place overwrite. Version the plugin in its connector config so you can tell which build is running during an incident.

plugin.path=/opt/connectors
# /opt/connectors/my-sink-1.4.2/  contains the shaded jar and its libs

The framework tracks source offsets per partition and commits them after your put() call returns, giving at-least-once delivery by default when offsets.flush.interval or the periodic commit runs. Returning a sink record with null offsets for a failed record signals the framework not to advance past it, which is how you express partial progress. Any stronger guarantee must come from the sink itself, using idempotent writes keyed by a deterministic identifier.

Set tasks.max to at least the source partition count, and raise batch sizes such as max.batch.size and max.poll.records together with fetch.max.bytes so each task moves more data per round trip. For sink connectors, tune the embedded producer through producer.override settings and check that the bottleneck is not the external system. Measure per-task throughput rather than aggregate, because one slow task after a rebalance distorts the picture.

Set errors.tolerance to all only when you also configure a dead letter queue topic, so failed records and their headers land somewhere inspectable instead of being skipped silently. Use Converters with a schema registry so schema changes are validated at conversion time rather than at the sink. Alert on the DLQ produce rate, since a steady trickle means a systematic incompatibility rather than isolated bad data.

errors.tolerance=all
errors.deadletterqueue.topic.name=dlq.connect.orders
errors.deadletterqueue.context.headers.enable=true
errors.log.enable=true

Use incremental cooperative rebalancing so tasks migrate without revoking everything, and avoid frequent worker restarts or config updates that each trigger a rebalance. Set a sensible rebalance timeout and keep the number of connectors per worker moderate so assignment computation stays fast. A storm usually shows as connectors cycling through RUNNING and UNASSIGNED, so alert on task state churn rather than only on failures.

Track connector and task status through the REST API and export those states as metrics, alerting on FAILED or repeated restart of the same task. Watch the worker JVM heap and GC alongside source-record-poll and sink-record-send rates, since a leaking plugin typically shows as rising heap with flat throughput. Offset commit failures and DLQ growth are the two signals that most often precede visible data loss.

At the source when a producer retries without idempotence, in processing when a consumer commits offsets after processing but crashes before committing, and at the sink when a non-transactional write succeeds but its offset commit fails. Each boundary needs its own mechanism, which is why enabling transactions only on the middle hop rarely delivers end-to-end exactly-once. Map the guarantee hop by hop before claiming EOS.

Idempotence removes duplicates from producer retries within a session, transactions make multi-partition writes and offset commits atomic, and read_committed hides records from aborted transactions from consumers. Together they give exactly-once for the Kafka-to-Kafka segment, provided every consumer of the output uses read_committed. Any consumer on read_uncommitted silently reintroduces the aborted records the design was meant to hide.

Write idempotently using a deterministic key derived from topic, partition and offset, or from a business identifier plus version, and make the write an upsert or an insert that is deduplicated by a unique constraint. Alternatively write to a staging area and swap atomically, or use a sink with native transactions. Then store the consumer offset in the same transaction as the data whenever the target supports it, so progress and data cannot diverge.

MERGE INTO orders o USING (SELECT ? AS id, ? AS v) s
ON o.id = s.id
WHEN MATCHED AND o.version < s.v THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...;

Read a batch, write it idempotently, and only then commit the offsets, so a crash between write and commit causes a replay that the idempotent write absorbs. Committing before the write inverts the failure mode into permanent data loss. Batch the writes and commits together so the replay window stays small and the deduplication work stays bounded.

MirrorMaker 2 replicates records but does not preserve transactional boundaries, so downstream consumers see a stream that is at-least-once at the mirror level and must deduplicate or be idempotent. Preserving the source topic, partition and offset in headers lets the target deduplicate deterministically. For strict EOS across sites, run the transform on the target cluster after replication rather than expecting the mirror to carry the guarantee.

Transactions add per-record overhead, force extra round trips for begin and commit, hold the Last Stable Offset forward which raises read latency, and consume coordinator resources proportional to partition count. They are a poor fit for very high fan-out writes, very low latency requirements, or pipelines whose sink is already idempotent and where at-least-once plus deduplication is cheaper. Choose EOS where duplicates are expensive and measurable, not by default everywhere.

Reconcile counts and sums between source and sink per time window using a deterministic business key, and monitor for keys appearing twice with different versions. Rehearse failure modes deliberately: kill a consumer mid-batch, restart a producer with the same transactional.id to confirm fencing, and verify aborted transactions never surface downstream. A reconciliation job that runs continuously catches the regressions that one-off tests miss.

Consume entity and relationship events into a projection that issues MERGE statements keyed by stable business ids, with per-key ordering guaranteed by the topic key. Batch writes per transaction and keep the consumer single-writer per key space where the graph engine does not handle concurrent merges well. Store the applied offset with the write so replays after a crash are idempotent rather than duplicating nodes.

MERGE (p:Person {id: $id})
SET p.name = $name, p.version = $version
WITH p WHERE p.version <= $version

Key the topic by the entity id so all events for one node or edge arrive in one partition in order, and attach a version or event time that the writer checks before applying. Graph engines' MERGE is not idempotent by default for relationships unless the match pattern is precise, so a retried event can create a duplicate edge. Verify with a uniqueness constraint on the business id and a check on the relationship's identity properties.

Model deletion as an explicit tombstone event rather than absence of updates, then delete the node or detach its edges in the writer with a defined policy for dangling references. Compacted topics make the tombstone the retained state for that key, so replayed rebuilds converge to the deleted state. Cascading deletes need to be emitted as their own events by the source domain, since the graph writer cannot infer them reliably.

Start CDC capture before or during the snapshot, record the snapshot's consistent point, then replay events from that point so anything changed during the load is applied afterward. Because writes are idempotent upserts with version checks, overlapping the snapshot and the stream is safe and preferable to a gap. Cut over only after the consumer's lag reaches zero at the snapshot boundary.

Use a sink connector writing with external versioning so the index rejects writes older than the document already holds, which makes replay safe without deduplication tables. Set the document id from a stable business key rather than letting the engine generate one, otherwise retries create duplicate documents. Manage retention with index lifecycle policies driven by the same timestamps in the payload.

"write.method": "UPSERT",
"behavior.on.null.values": "DELETE",
"flush.timeout.ms": "60000"

Pause and resume the consumer or connector based on lag and on the sink's own latency, instead of buffering unboundedly in memory until the process dies. Configure bounded internal queues and let the framework throttle rather than fail, but alert when pause duration exceeds the freshness SLO. If the sink is persistently slower than the source, the answer is to scale the sink or to decouple through an intermediate compacted topic, not to grow consumer heap.

Periodically rebuild a sample or a full partition from the compacted topic into a shadow index and diff document counts and field hashes against the live one. Run full rebuilds on a schedule proportional to the acceptable staleness of the worst-case document, since compaction means the topic itself is the authoritative snapshot. Track a per-document last-applied version so reconciliation reports the specific keys that diverged rather than only an aggregate mismatch.

Kafka record timestamps are producer-supplied (CreateTime) or broker-assigned (LogAppendTime), so they are only as trustworthy as the producing client. If upstream devices emit skewed clocks, you must stamp an explicit event-time field in the payload and have the stream processor extract it, keeping the Kafka timestamp for retention and lag metrics only. Switching the topic to LogAppendTime destroys event-time semantics entirely and should be treated as a deliberate ingestion-time decision. Document which clock owns which field, because watermarks downstream inherit every upstream timestamp bug.

record.timestamp()        // broker-visible, retention clock
payload.get("event_ts")   // application clock, used for watermarks

A watermark is a monotonic assertion that no further records with an event timestamp below the watermark are expected, allowing window operators to fire and state to be garbage collected. It is a heuristic derived from observed timestamps, not a proof: records older than the watermark can still arrive and are then classified as late. Watermark generation policy therefore trades completeness against latency, and the choice must be aligned with the downstream sink's tolerance for revisions.

With an upsert-capable sink you can keep windows open past the watermark and emit retractions or corrected aggregates keyed by window plus dimension, so a late record simply rewrites the affected row. With an append-only sink, late data must be routed to a side output or a dedicated late-events topic and reconciled by a batch recomputation. The architectural decision is really about whether the serving layer accepts mutation, which should be settled before window sizes are chosen.

if (ts < currentWatermark) {
  sideOutput.collect(new LateEvent(ts, record));
}

Prefer sliding or hopping windows with explicit allowed lateness plus a TTL on state, rather than unbounded session windows, and enable incremental or RocksDB-backed state with compaction. Emit partial aggregates early and finalize on watermark so the operator can clear per-key accumulators. Also partition by the same key used for windowing, otherwise a single task holds keys for the whole topic and checkpoint size grows with cardinality.

Compare the producer event timestamp with the broker LogAppendTime and with ingestion-side wall clock, then publish the delta as a metric per producer instance or device fleet. A sustained negative delta beyond a few seconds indicates a device clock running fast, which will push watermarks forward and cause healthy records to look late. Remediation is NTP discipline plus an optional per-source offset correction applied before watermark assignment.

Each partition is consumed independently and may be idle or lagging, so taking a maximum would let a fast partition advance the global watermark and mark records from slow partitions as late. The correct alignment is the minimum of the per-subtask watermarks, with idle partitions explicitly excluded so a silent partition does not stall progress forever. This is why an idle partition with no traffic must be marked idle rather than contributing an old watermark.

Build a deterministic replay harness that feeds a fixed trace with deliberately permuted timestamps and duplicated records, then assert exact window outputs against expected results. Keep the harness in CI so watermark policy, allowed lateness and state TTL changes are regression-tested, since these bugs appear only under specific arrival orders. Complement it with a chaos test that pauses one partition consumer to simulate skew during a live run.

The source uses the Kafka connector with partition discovery and offset commit controlled by Flink checkpoints, so offsets are part of the checkpoint rather than committed eagerly. Sinks use at-least-once by default, and exactly-once requires either the transactional sink with a checkpoint interval aligned to transaction timeouts or an idempotent downstream. The critical coupling is that checkpoint interval, transaction timeout and transaction.max.timeout.ms must be mutually consistent.

execution.checkpointing.interval = 60s
transaction.timeout.ms = 900000
transaction.max.timeout.ms = 900000

The transactional sink writes records inside a Kafka transaction and commits them only when the checkpoint completes, so readers must use isolation.level=read_committed or they will see uncommitted data. Aborted transactions from failed checkpoints are then filtered by the broker, which means downstream latency includes the checkpoint interval. Any consumer left at read_uncommitted silently breaks the end-to-end guarantee.

Spark is attractive when the same logic must run in batch and streaming with one codebase, when the team already operates a Spark platform, and when micro-batch latency of hundreds of milliseconds to seconds is acceptable. Flink wins on true per-record processing, richer event-time and state semantics, and lower tail latency. The choice is usually organizational as much as technical, but the latency SLO should drive it.

Set maxOffsetsPerTrigger or the newer maxBytesPerTrigger to bound each micro-batch, and let trigger(availableNow=True) handle catch-up backfills without unbounded batches. Without these limits a large backlog produces ever-growing batches, unstable shuffle sizes and checkpoint timeouts. Also monitor numInputRows and batch duration to confirm the limiter is actually binding.

.option("maxOffsetsPerTrigger", "500000")
.option("maxBytesPerTrigger", "1g")

Two offset owners diverge: the checkpoint records what Spark processed, while the consumer group commits whatever the Kafka source tracked, and after a restart the group position can be ahead of the checkpoint. The safe pattern is to disable group commits for the streaming query and treat the checkpoint as the single source of truth, or to accept at-least-once and design idempotent sinks. Mixing both without a clear owner causes silent data loss on redeploy.

Model the dimension as a compacted Kafka topic keyed by entity id and consume it as a broadcast or lookup state, so each stream record is enriched from local state rather than a per-record remote call. For Flink use a broadcast state or a temporal table join with versioned rows; for Spark use a stream-static or stream-stream join with watermark bounds. Versioning the dimension rows with valid-from and valid-to timestamps makes the join reproducible during replay.

Run the backfill into a versioned or shadow output location keyed by a deterministic run id, then swap or reconcile at the serving layer instead of writing into the live table. Alternatively reset the consumer group to the historical offset only if the sink is idempotent and keyed, and aggregate state is rebuilt from scratch. Blindly replaying into an append-only aggregate sink double-counts, which is the most common backfill incident.

Normal CPU with high latency points to I/O waits, page cache misses, network saturation or lock contention rather than compute, so start with per-request-type latency percentiles in the broker metrics and correlate with disk await and network retransmits. Thread dumps or async-profiler wall-clock sampling reveal whether request handler threads are blocked on flush, on the purgatory, or on a contended lock. Splitting produce versus fetch versus metadata latency usually isolates the layer within minutes.

kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce
kafka.log:type=LogFlushStats,name=LogFlushRateAndTimeMs

Use a modern generational collector such as G1 with a modest max heap, typically six to eight gigabytes, and leave the rest of the node to the page cache because Kafka relies on it heavily. Explicitly set -Xms equal to -Xmx, enable heap dumps on OOM, and avoid gigantic heaps that turn young collections into long pauses. Aggressive manual GC tuning and huge survivor spaces usually hurt more than they help with modern collectors.

-Xms6g -Xmx6g -XX:+UseG1GC -XX:MaxGCPauseMillis=20
-XX:InitiatingHeapOccupancyPercent=35

A large heap displaces page cache, and Kafka serves most reads from the OS page cache, so heap growth directly reduces the cache available for log segments. It also lengthens GC pauses and increases the cost of object graphs held by request state. The correct sizing keeps the heap just large enough for in-flight requests, replication buffers and metadata, and lets the kernel cache the log.

Size from target throughput per partition, measured rather than assumed, then add headroom for consumer parallelism and future growth while keeping the count a small multiple of broker and consumer counts. Each partition costs file handles, memory for indexes, replication traffic and controller metadata, and it lengthens leader election and rebalance times. Over-partitioning also increases end-to-end latency because producers batch less per partition.

The purgatory holds requests waiting on a condition, such as acks=all replication or fetch min-bytes accumulation, and expires them on timeout. When it grows, produce and fetch latency rise while CPU stays flat, and PurgatorySize plus request queue time climb together. Persistent growth usually means replication is slow or fetch.min.bytes is set too high for the arrival rate.

Check producer metrics for record queue time, batch size, compression ratio and in-flight requests before blaming the broker, because a saturated buffer or tiny batches produce the same timeout symptom. On the broker side, look at request queue time and throttling metrics to see whether the request even reached a handler. If queue time is near zero but total time is high, the delay is in the client's accumulator or network path.

Raise file descriptor limits, use a low swappiness value to protect the page cache, and ensure the dirty page ratio does not let the kernel stall on writeback. Disable transparent huge pages for latency-sensitive brokers and tune the network stack for larger socket buffers. Never let the JVM or Kafka run on swap, since a single swap event turns into multi-second request stalls.

vm.swappiness=1
vm.dirty_ratio=80
vm.dirty_background_ratio=5
net.core.somaxconn=32768

acks=all requires a round trip to the in-sync replicas, adding at least one network hop and the follower fetch latency, while acks=1 waits only for the leader's local append. acks=1 is defensible only for loss-tolerant telemetry where a few dropped records are acceptable and the topic has replication for durability of what was acknowledged. For anything financial or state-changing, acks=all with min.insync.replicas of at least two is the baseline.

linger.ms sets how long the sender waits to fill a batch, and batch.size caps how large it can grow, so a large batch with zero linger rarely fills and gains nothing. For low latency keep linger in the single-digit milliseconds and size batches to match your record rate times that interval. Measure the resulting batch size and compression ratio, because a small linger that never fills batches wastes the compression win.

linger.ms=5
batch.size=65536
compression.type=lz4

lz4 gives the best latency-to-ratio balance for most event streams, zstd gives a better ratio at higher CPU cost and is a good fit for large batches or cross-region replication, and uncompressed is only reasonable on a fast local network with tiny records. Compression happens on the producer's sender thread and decompression on the broker and consumer, so a slow codec shows up as client CPU and as broker request time. Always benchmark with production record shapes rather than synthetic strings.

fsync forces the page cache to durable media, protecting against power loss, but it costs milliseconds per call and would cap throughput severely. Kafka instead relies on replication to other brokers' page caches for durability, flushing segments periodically and on close or shutdown. This means acks=all protects against broker process and disk failure, but a simultaneous power loss across replicas can still lose the unsynced tail.

Lowering flush thresholds tightens the durability window at a direct throughput cost, and on a replicated topic it is usually redundant with acks=all. If the requirement is truly zero loss on power failure, the honest answer is that Kafka alone cannot guarantee it without synchronous flushes, so you either accept the window or add an upstream durable store. Quantify the window in seconds and get sign-off rather than silently setting flush.messages=1.

A request passes the network processor, the request queue, a handler thread, the log append with possible index updates, and the purgatory waiting for replication, then back through the response queue. Queue time, purgatory time and disk append dominate, and each maps to a distinct metric. Optimizing without knowing which stage dominates leads to tuning the wrong layer.

Roll segments on size rather than time when possible, and keep segment.bytes large enough that rolling is infrequent but small enough that recovery and retention deletion stay fast. Index files are rebuilt from the log on unclean shutdown, so segment size directly bounds recovery time. Avoid very small segments, which multiply file handles and make every fetch span more files.

In KRaft a set of controller nodes form a Raft quorum that stores cluster metadata in an internal replicated log, and one of them is the active controller. Metadata writes are committed by a majority of that quorum rather than by an external ensemble, and brokers fetch metadata from the controller instead of watching ZooKeeper. ZooKeeper was removed in Kafka 4.0, so all new clusters must run KRaft.

Use three controllers for tolerance of one failure, or five when you need to survive two simultaneous failures during maintenance, always an odd number for clean majority semantics. A two-node quorum tolerates no failure and is strictly worse than three. Controllers should be dedicated nodes with low-latency disks and should not share broker roles in large clusters.

Every topic, partition and replica is a metadata record, so partition count drives snapshot size, log replay time on controller restart and the cost of propagating changes to brokers. Very large partition counts slow controller failover because the new active controller must load the full metadata log. Keeping partition counts proportional to real throughput need, and avoiding per-request topic creation, is the main defense.

The broker continues serving existing partitions for a grace period but cannot process metadata changes, and it eventually shuts down if contact is not restored. Clients see stale metadata and may route to partitions whose leadership changed. Monitoring controller-to-broker channel health and the metadata log lag is essential because this failure mode degrades quietly.

Provision the controller quorum, migrate metadata with the provided tooling while brokers run in migration mode, verify that all metadata is mirrored, then roll brokers into KRaft mode and finally decommission ZooKeeper. Each phase needs a validation gate, and the rollback path disappears once ZooKeeper is removed. In Kafka 4.x this is historical, but the same staged discipline applies to any metadata migration.

Look for rising event queue processing time on the active controller, growing metadata log lag on followers, and slow responses to admin operations such as topic creation or partition reassignment. Brokers may be healthy on produce and fetch while admin calls time out. Correlating controller metrics with partition churn usually confirms the diagnosis.

Each broker registers with the controller quorum using a lease and an incarnation id, and a restarted process obtains a new incarnation that fences the previous one. If an old process is still alive, its heartbeats are rejected and it shuts down rather than serving stale leadership. This fencing is what prevents split-brain behavior that ZooKeeper sessions handled less explicitly, and it is why broker session timeouts and controller failover time must be tuned together.

broker.session.timeout.ms=9000
controller.quorum.fetch.timeout.ms=2000

Connect is a good fit for landing raw topics into object storage as Parquet or Avro files on a fixed interval, using single message transforms only for light shaping. It stops being appropriate when you need schema evolution handling, deduplication, exactly-once semantics or upserts, because sink connectors provide at-least-once file commits. Beyond that point a stream processor or a table format writer is the correct layer.

Small intervals reduce data visibility latency but create many small files, which degrade query planning and increase object storage request costs. Large intervals produce healthier file sizes but delay downstream freshness and risk data loss on unclean shutdown since the connector commits on interval boundaries. Most teams target one to five minutes and run a separate compaction job to merge small files.

Kafka transactions cover Kafka writes, not object storage commits, so the sink must make file publication atomic and idempotent through a commit protocol. Table formats achieve this with a metadata commit that lists the new files, so partial uploads are invisible until the commit succeeds. Without such a protocol you get duplicate rows after every retry, which downstream deduplication can mask but not fix.

Write both a compact columnar projection for analytics and the raw bytes, or a raw JSON and Avro pair, in separate columns or sibling tables partitioned by ingestion date. Keeping the raw payload makes reprocessing possible when the projected schema proves wrong. Storage cost is modest relative to the value of replayability.

Use a schema registry with backward-compatible evolution, and make the lake writer apply the registered schema at write time rather than trusting producer serialization. Additive columns are safe; type changes and column drops require a new table version plus a backfill. Partitioning by ingestion date, not event date, keeps late-arriving schema versions from rewriting history.

Frequent commits plus high partition counts multiply output files, and each file carries object storage request overhead and query planning cost. Solve it with a compaction or clustering job that rewrites partitions into target file sizes, scheduled after the ingestion window closes. Tuning the sink's commit interval and partition strategy first reduces how much compaction is needed.

Land late records into the correct event-date partition using partition pruning on the event timestamp, and make downstream reports read through a layer that tolerates partition rewrites. For already-published aggregates, either recompute the affected window or publish corrections with an explicit version. The design question is whether the lake is the source of truth or a derived view, and that determines the reconciliation strategy.

RPO is the maximum acceptable data loss measured in records or seconds, driven by replication lag and whether acks=all is enforced; RTO is the time to restore service, driven by failover automation, DNS or client reconfiguration, and metadata recovery. Both must be stated per failure scenario, because a broker loss and a region loss have completely different answers. Committing to a single RPO number without scenario qualification is a red flag in a design review.

MM2 runs as a Connect source and sink pair per direction, replicating records with offset translation and heartbeats, checkpoints and topic configuration sync. It does not replicate consumer group offsets unless offset sync is enabled and consumers are translated correctly, and it never replicates ACLs for you. Client-side failover also requires deliberate consumer logic, since MM2 does not move clients.

Offset translation relies on periodic checkpoint records mapping source to target offsets, so any replication lag or checkpoint gap leaves the mapping imprecise and consumers may skip or re-read records. For strict correctness, fail over with an explicit application-level decision, such as a control record or a start timestamp, rather than trusting the translated offset. Always validate translated positions in a game day before relying on them.

Active-active means both clusters accept writes for the same logical topics, so consumers must tolerate duplicates and ordering is per-cluster only, with replication loops prevented by topic naming conventions. Active-passive keeps one writer and uses replication as a warm standby, which is far simpler to reason about for aggregates. Most teams that claim active-active actually run partitioned ownership of keys.

Cross-region replication adds round-trip latency to every batch and consumes egress bandwidth, so throughput is bounded by the link and by the consumer-side fetch settings of the replicator. Keep replication compression enabled with zstd, raise fetch sizes, and monitor replication lag as a first-class SLO. If lag exceeds your RPO, the replication link is effectively the RPO.

Topic configuration, partition counts, ACLs, schema registry subjects, client quotas and consumer group offsets all need a replication and validation path, and none of them are covered by data replication alone. A cluster with all the data but missing schemas or ACLs is not a recovery target. Rehearse failover end to end, including client reconfiguration, at least quarterly.

Run scheduled game days that cut traffic to the primary and measure time to first successful produce and consume on the secondary, including DNS, credential and client-config steps. Measure the human decision time separately from the mechanical failover, because approvals often dominate. Record the result and feed regressions back into automation.

Hub-and-spoke centralizes aggregation and keeps the number of replication links linear in regions, which is far easier to operate and monitor, but it adds a hop of latency for region-to-region traffic. A mesh gives direct paths and lower latency but multiplies links, bandwidth cost and loop-prevention complexity. Most organizations start hub-and-spoke and add directed edges only where a latency SLO demands it.

Rack awareness spreads replicas across failure domains inside one cluster, so a single rack or availability zone loss does not take a partition offline. It does not place replicas in another region, and a cluster stretched across regions will pay cross-region latency on every acks=all write. Geo-durability therefore requires separate clusters plus replication, not a stretched cluster.

broker.rack=az-a

Follower fetching lets a consumer read from the replica in its own zone or region instead of the leader, avoiding egress charges and long round trips. It requires rack-aware consumers so the client can select a local replica, and it accepts slightly higher staleness because the follower lags the leader. This is one of the cheapest large wins in a multi-zone deployment.

client.rack=az-b

Apply quotas on the replicator's client id or on the inter-broker path, and run the replicator as a distinct principal so its usage is measurable and separable. Without quotas a backlog catch-up saturates the link and inflates produce latency for real applications. Also cap consumer fetch sizes so a single catch-up burst cannot monopolize the network.

Kafka guarantees order within a partition, and replication preserves that per-partition order in the target cluster, but there is no global ordering across partitions and no atomicity across topics. Any cross-cluster workflow that assumes causal ordering between two topics must encode causality in the payload or use a single partition. Consumers must also tolerate that the target has no record of transactions from the source unless the replicator preserves them.

Cluster linking keeps a direct mirror of partitions with offset preservation on the destination, so consumers can fail over without offset translation and can even read the mirrored topic as if it were local. MM2 instead runs as a Connect job that copies records and stores offset mappings as checkpoints, which is more flexible but less transparent to clients. The trade-off is operational simplicity and offset fidelity versus connector-level extensibility and multi-hop flexibility.

Front the clusters with a discovery layer such as DNS with health checks, a service mesh, or a bootstrap endpoint that returns region-local brokers, and have clients resolve bootstrap servers at startup from that layer. The client must also be able to refresh metadata when leadership moves, so avoid long-lived static broker lists. Failover policy belongs in this layer, not in each application's configuration file.

Prefer local state populated from a compacted reference topic, so enrichment is a local lookup with no per-record network hop. Where the reference set is too large, use a partitioned lookup service keyed by the same key as the stream, or a cached remote call with strict timeouts and a fallback. The pattern must define behavior on lookup miss, because that is the normal case during startup.

Consume the reference topic as a changelog into a local store, apply updates in order, and treat the store as a materialized view that is rebuilt on restart from the compacted topic. Compacted topics guarantee the latest value per key survives, so rebuild time is proportional to key count rather than history. Add a periodic version check to detect silent divergence.

KTable<String, Ref> refs = builder.table("ref.customers");

A join across differently partitioned streams requires a repartition, which adds a shuffle, state and latency, and doing it implicitly can silently drop matches. Either repartition the enrichment side by the join key or perform the lookup through a keyed service that does not depend on co-partitioning. Never assume that matching key names imply matching partitioning.

Serve from the last known good snapshot with an explicit staleness flag on the enriched record, and never block the pipeline on a remote lookup without a timeout. Emit a dead-letter or retry topic for records that could not be enriched if the business requires completeness. Silent nulls in enriched fields are the worst outcome because they corrupt downstream aggregates invisibly.

Broadcast state is replicated to every parallel subtask, so memory and checkpoint size scale with dataset size times parallelism, which caps it at tens to low hundreds of megabytes in practice. Beyond that, use a partitioned keyed lookup or an external store with a local cache. The constraint is checkpoint duration and recovery time, not raw heap.

Version reference rows with valid-from and valid-to timestamps and join on event time rather than on processing time, so a replay reproduces the value that was correct when the event occurred. If the reference topic only stores current state, historical replay will enrich with today's values and produce different results. This is a design decision made at the start, not something you can retrofit cheaply.

When the reference dataset exceeds practical state size, changes very frequently, or is owned by a system that cannot publish a changelog topic, a keyed lookup service with a bounded cache is the pragmatic choice. The cost is per-record latency and a new availability dependency, both of which must be covered by timeouts and fallbacks. Local state remains preferable whenever the dataset fits.

Boundaries should follow transactional consistency and ownership of a business fact, not team structure or database tables. A service should own the events it publishes and be the only writer of its topics, with consumers reacting rather than reaching into its state. If two services must commit atomically, they are probably one service or need a saga.

A notification carries only an id and forces consumers to call back, creating temporal coupling and read amplification; state transfer carries enough data for consumers to act without a call. State transfer decouples services but couples them to a schema and leaks internal detail. Most mature systems use thin notifications for rare, high-value events and state transfer for high-volume reference updates.

Enforce single-writer ownership per topic, publish a versioned contract, and treat consumers as independent so a producer cannot require coordinated deploys. The smell is a topic that many teams write to with different semantics, or a schema change that requires every consumer to ship simultaneously. Shared topics with shared ownership are the most common form of accidental coupling.

Consuming directly is fine for reactions that need the raw fact and can tolerate ordering per key; a materialized view is better when the service needs joins, point lookups, or a stable read model independent of topic retention. Views add storage and rebuild time but remove runtime dependency on the producer's availability. Choose based on whether the consumer needs the stream or the current state.

Model the workflow as a saga with explicit compensating events and an orchestrator or choreography that tracks state, rather than distributed transactions. Each step must be idempotent and each compensation must be safe to run more than once. The orchestrator needs its own durable state, typically a compacted topic, so recovery after a crash is deterministic.

The contract should include stable business identifiers, the event type and version, the event timestamp, and the facts consumers genuinely need, with internal database ids and implementation details excluded. Internal fields become de facto public API the moment they ship, so removing them later is a breaking change. Version the envelope and evolve payloads additively.

Give each service its own consumer group and never share a group across services, since a group rebalance affects all members and forces coordinated pauses. Use static membership and cooperative sticky assignment to reduce churn during rolling deploys. Shared groups are only appropriate for horizontally scaled instances of one service.

A topic-level dynamic config wins over a per-broker dynamic override, which wins over a cluster-wide dynamic default, which wins over the static value in server.properties. The static file is only the fallback for keys that support dynamic update, so an operator who edits server.properties and restarts will often see no change because a dynamic override is still in place. Always resolve effective config with the admin tool rather than assuming the file is authoritative.

kafka-configs.sh --bootstrap-server b:9092 --entity-type topics \
  --entity-name orders --describe --all

Listener and security-protocol settings, log directory layout, broker id and role, and most KRaft quorum settings require a restart, so they belong to a versioned rollout rather than an ad hoc change. Plan them as a rolling restart one broker at a time with under-replicated partitions back to zero between steps. Verify that min.insync.replicas and producer acks settings still allow writes while a replica is offline.

Use a cluster default for settings that should apply to every new topic, such as a baseline replication factor, minimum in-sync replicas or retention, then override only the topics that genuinely differ. Topic-level overrides multiply configuration surface and drift, and they silently survive cluster-wide policy changes. Keeping overrides few and justified makes audits tractable.

Declare topics and their configs in version control and reconcile with an operator or a GitOps controller that computes a diff against the live cluster, rather than letting teams run ad hoc alter commands. The reconciler must treat desired state as authoritative and report drift before overwriting it, so genuine emergencies are not clobbered. Kafka itself keeps no config change history, so your Git history becomes the audit trail.

Reducing retention deletes whole segments immediately once they exceed the limit, so check consumer lag and downstream batch jobs before shrinking it, and prefer reducing in steps. Changing segment.bytes only affects newly rolled segments, so the topic keeps mixed segment sizes until old ones age out, which matters for recovery time estimates. Neither change requires a restart, but both should be announced because they alter data availability.

Quotas are enforced per broker for the clients connected to it, so a change reaches a client only after the broker refreshes its quota configuration, and a client spread across brokers sees uneven throttling in the interim. Verify by reading the throttle-time metrics for the specific user or client id, not by watching aggregate throughput. Also confirm the entity name matches exactly, since user, client-id and ip quotas are separate namespaces.

kafka-configs.sh --bootstrap-server b:9092 --entity-type clients \
  --entity-name svc-ingest --alter --add-config producer_byte_rate=10485760

Capture the effective configuration before every change, either through your declarative repo or an exported snapshot, so rollback is a restore of known values rather than a guess. Automated reconcilers should compute and store the pre-change diff as part of the apply step. Without that, rollback depends on memory and is the usual cause of prolonged incidents.

Start from peak ingress bytes per second, multiply by replication factor plus a replication overhead, add consumer egress, then divide by a per-broker network budget at no more than about seventy percent utilization to leave headroom for recovery and rebalances. Convert bytes per second into disk per day with retention to size storage, and check that partition count supports the required consumer parallelism. Validate the model with a load test rather than trusting vendor throughput numbers.

disk_per_day = ingress_Bps * 86400 * RF / compression_ratio

A partition is served by one leader and appended serially, so its ceiling is set by sequential disk write rate, replication round trips and the batching the producer can build for that partition. With many small partitions the producer cannot fill batches, and with one huge partition you lose parallelism. The practical fix is enough partitions to allow batching and parallelism without so many that per-partition batches shrink.

Compare per-partition byte-in rates and lag, not just topic-level totals, because a hot key hides inside healthy aggregates. Mitigate by salting the key with a bounded suffix for aggregation-free workloads, splitting the hot entity into its own topic, or moving aggregation upstream. Salting trades ordering for parallelism and must be paired with a downstream merge.

String key = entityId + "-" + (hash % 8);

Size for the peak sustained for the longest expected burst rather than the annual peak instant, and let Kafka's retention buffer absorb short spikes that consumers drain afterwards. Quantify the maximum tolerable consumer lag in time, then size retention and consumer throughput so the backlog clears before it breaches that bound. This is cheaper than provisioning for instantaneous peak in most telemetry workloads.

Break the loop into fetch time, deserialization, processing and commit, using consumer metrics plus application tracing, because lag alone hides which stage dominates. A common finding is that per-record synchronous processing or downstream calls dominate, not the poll itself. Increasing consumer count only helps if partition count and downstream capacity allow it.

Consumer parallelism is capped by partition count, so extra members beyond that stay idle. Beyond that cap, the bottleneck is usually downstream throughput, per-record processing cost, or rebalance churn from long processing exceeding max.poll.interval.ms. Check assigned partition counts, rebalance rate and downstream latency before scaling the group.

Embed a produce timestamp and trace id in the payload, then compare it with the consumption or sink-write timestamp at each hop to obtain per-stage latency distributions. Broker metrics only cover request handling inside the cluster and say nothing about consumer processing or downstream commits. Publishing these deltas as percentiles, not averages, is the only way to see tail latency that breaches SLOs.

The main risks are an unauthenticated or over-privileged producer injecting forged records into another team's topic, a consumer reading data outside its tenant boundary, and an operator or compromised host exfiltrating data at rest. Secondary risks include metadata disclosure through describe operations and denial of service by an unbounded producer. Each risk maps to a specific control: authentication, least-privilege ACLs, encryption in transit and at rest, and quotas.

Missing ACLs on consumer group and transactional id resources, wildcard or prefixed ACLs granted for convenience, and super users that bypass authorization entirely are the usual findings. A principal with write access to a topic but no describe permission produces confusing client failures, which teams then fix by over-granting. Review ACLs by resource type, including cluster-level actions such as alter and idempotent write.

--allow-principal User:svc.a --operation Write --topic orders
--allow-principal User:svc.a --operation Read --group orders-cg

Enable authorizer logging with a structured logger and ship it to a SIEM, then accept that it records decisions rather than complete read events unless you log successes at a level that is operationally affordable. For stronger data-access audit, use per-team principals, per-topic ACLs and a broker-level access log pipeline, or front sensitive topics with a proxy that logs every fetch. Design the audit requirement explicitly, because Kafka has no built-in per-record read audit.

TLS protects the wire, not the log segments on disk, broker memory dumps, or a compromised broker with a valid principal. An attacker with filesystem access reads plaintext segments, and a super user can still read every topic. Encryption at rest via volume or filesystem encryption is therefore a separate control, as is restricting who can run a broker.

Run listeners that accept both old and new credentials during the overlap window, distribute the new material, roll clients, then remove the old configuration in a second phase. With SASL/OAUTHBEARER and a token issuer this becomes a short-lived token refresh rather than a coordinated restart. Certificate rotation requires the same two-phase discipline, including truststore updates on both brokers and clients.

ACLs separate who may read or write, and quotas limit throughput per client, but neither isolates latency, disk consumption or metadata visibility completely. A noisy tenant can still consume shared page cache and network, and cluster-wide metadata is visible to any principal allowed to describe it. Strong isolation for regulated tenants usually means separate clusters with their own principals and quotas.

Scope each credential to a narrow set of topics, groups and operations, avoid shared principals across services, and keep transactional id and group prefixes under the same ownership. Pair that with quotas so a hijacked client cannot saturate the cluster, and with alerting on anomalous produce volume per principal. Blast radius is decided at design time by how much any single credential can touch.

Key by a stable device or tenant identifier so per-device ordering is preserved, and never use a random or null key because it destroys ordering and prevents any future compaction by entity. If cardinality is extreme and ordering is not required, key by a coarser grouping to keep partition count manageable. Record the keying decision in the topic contract, since changing it later requires a new topic and a migration.

Compaction keeps only the latest record per key, which for a device emitting measurements means you retain one sample and discard the time series. Telemetry needs time-based retention with tiered storage for long lookback, and compaction only for the latest-state view of a device registry. Mixing the two in one topic silently destroys history.

Stamp both a device timestamp and an ingestion timestamp, then use ingestion time for retention and watermarks when device clocks are untrusted, while keeping device time as a payload field for analysis. Allow generous lateness windows and route very old records to a quarantine topic instead of dropping them. Buffering on the device with sequence numbers lets you detect gaps after reconnect.

Apply per-tenant or per-client quotas, split high-volume device classes into their own topics, and downsample at the edge so the cluster receives aggregated values rather than raw samples. Because partitioning is by key, one hot device can only saturate its own partition, but enough hot devices still saturate brokers. Edge aggregation is usually the cheapest fix.

Use a common envelope with a typed payload union or a map of optional fields rather than one schema per device model, and register schemas per model family with backward-compatible evolution. Thousands of near-duplicate schemas make registry operations and consumer deserialization fragile. Store raw payloads in the lake so a schema mistake remains recoverable.

Partition count should follow throughput and consumer parallelism, not device count, because devices can share partitions while ordering is preserved per key. Each partition costs broker memory, file handles and controller metadata, so tens of thousands of partitions per cluster is already a large operational footprint. Group devices into topics by volume class rather than creating a topic per fleet.

Aggregate at the edge: terminate device protocols at a gateway or MQTT proxy and have a small pool of gateway instances produce to Kafka with large batches, instead of every device holding a broker connection. This also centralizes authentication, payload validation and rate limiting. Direct device-to-broker connections multiply TLS handshakes, metadata requests and per-connection broker memory, which becomes the limiting factor long before throughput does.

Keep a short local retention on broker disks for the hot window and offload older segments to tiered storage, so long lookback reads come from object storage at a fraction of the cost per gigabyte. Set local retention by time or size to bound broker disk, and monitor remote read latency since historical fetches are slower. This keeps the cluster small while satisfying compliance retention.

local.retention.ms=86400000
retention.ms=7776000000
remote.storage.enable=true

Broker disk sizing becomes a function of the hot window plus upload headroom rather than total retention, which can shrink disks by an order of magnitude for long-retention topics. You still need enough local space to absorb a remote-storage outage, otherwise the broker cannot accept writes while segments cannot be offloaded. Size for the largest expected offload stall, not the average.

Reads of segments that are no longer local require a fetch from object storage, with first-byte latency in the tens to hundreds of milliseconds, so a consumer replaying history or a new consumer group starting from the earliest offset will be slower. Keep the local retention window larger than any normal consumer lag or replay distance, and prefer consumers that start near the tail. Catch-up consumers should be treated as batch workloads with their own throughput budget.

Compare cost per gigabyte month including replication factor and headroom on broker disks against object storage pricing plus retrieval and request charges for the expected read pattern. Broker disks cost several times more per gigabyte and are replicated, so long cold retention is almost always cheaper remotely. The decision flips only when the data is read frequently, because retrieval and egress charges dominate.

Separate them: keep an event topic with delete and time-based retention for the history, and a compacted topic keyed by entity for current state. Compaction with a delete retention as well is valid but subtle, because the delete window then bounds how long tombstones and history survive. Mixing both policies on a high-volume event stream usually means unpredictable storage and confusing replay behavior.

cleanup.policy=compact,delete
delete.retention.ms=86400000

Compression is applied per batch and per segment, so larger segments and larger producer batches compress better and reduce both local and remote bytes. Very large segments slow down retention deletion and recovery, so there is a practical middle ground. Measure the achieved ratio on real payloads, because JSON with repeated keys compresses very differently from binary protobuf.

Leadership moves to another replica, and if that replica has already uploaded the segment it serves the read from remote storage; if it has not, it must fetch from the remaining local replica or wait for upload. This is why remote log metadata must be replicated and why local retention should exceed the worst-case upload lag. Test broker loss explicitly, since tiered storage changes the failure characteristics of historical reads.

Run the official Apache Kafka image with the combined controller and broker role; KRaft formats its own metadata log from CLUSTER_ID, so no ZooKeeper container is involved.

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_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  -e CLUSTER_ID=4L6g3nShT-eMCtK--X86sw \
  apache/kafka:4.1.0

Use kafka-topics.sh with --partitions and --replication-factor; the replication factor cannot exceed the number of brokers, so a single-node cluster must use 1.

kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders \
  --partitions 6 --replication-factor 1 \
  --config min.insync.replicas=1 \
  --config retention.ms=604800000

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders

Enable parse.key and set a key.separator on the producer, then ask the consumer to print keys so ordering per key stays visible in the output.

printf 'order-1:created\norder-2:paid\norder-1:shipped\n' | \
kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic orders --property parse.key=true --property key.separator=:

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic orders --from-beginning \
  --property print.key=true --property key.separator=' => '

Idempotence is on by default in Kafka 4.x, but you should state it explicitly together with acks=all and enough in-flight requests; the broker then deduplicates retried batches using producer IDs and sequence numbers.

# producer.properties
enable.idempotence=true
acks=all
retries=2147483647
max.in.flight.requests.per.connection=5
delivery.timeout.ms=120000

# CLI usage
kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic orders --producer.config producer.properties

Set enable.auto.commit=false and call commitSync after the batch is fully processed, so a crash replays uncommitted records rather than losing them.

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "orders-worker");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(List.of("orders"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<String, String> r : records) {
            process(r);
        }
        if (!records.isEmpty()) consumer.commitSync();
    }
}

Use partition.assignment.strategy=CooperativeStickyAssignor to avoid stop-the-world rebalances and a stable group.instance.id so a restarting pod reclaims its partitions without triggering one at all.

# consumer.properties
bootstrap.servers=localhost:9092
group.id=orders-cg
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
group.instance.id=worker-1
session.timeout.ms=45000
heartbeat.interval.ms=3000

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic orders --group orders-cg --consumer.config consumer.properties

Split each line into words, group by word, then count into a KTable and materialize it to an output topic; the counting is backed by a state store that Streams manages for you.

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> lines = builder.stream("text-input");

KTable<String, Long> counts = lines
    .flatMapValues(v -> Arrays.asList(v.toLowerCase().split("\\W+")))
    .groupBy((k, word) -> word)
    .count(Materialized.as("word-counts"));

counts.toStream().to("word-count-output",
    Produced.with(Serdes.String(), Serdes.Long()));

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

Enable logging on the store so it gets a changelog topic, and raise the restore threads and restore batch size to rebuild local state quicker after a restart.

# streams.properties
application.id=wordcount-app
bootstrap.servers=localhost:9092
num.stream.threads=4
num.standby.replicas=1
state.dir=/var/lib/kafka-streams

# store-level settings in code
StoreBuilder<KeyValueStore<String, Long>> store =
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("word-counts"),
        Serdes.String(), Serdes.Long())
      .withLoggingEnabled(Map.of("cleanup.policy", "compact"))
      .withCachingEnabled();

Run a Connect worker and POST a JSON connector config that points at the database, choosing timestamp or incrementing mode so new rows are picked up continuously.

{
  "name": "orders-jdbc-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "2",
    "connection.url": "jdbc:postgresql://db:5432/shop",
    "connection.user": "connect",
    "connection.password": "secret",
    "table.whitelist": "orders",
    "mode": "timestamp+incrementing",
    "timestamp.column.name": "updated_at",
    "incrementing.column.name": "id",
    "topic.prefix": "db-"
  }
}

Wrap the consume-transform-produce cycle in producer transactions and send the consumer offsets inside that transaction, using read_committed downstream so aborted data is invisible.

props.put("transactional.id", "orders-tx-1");
producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    if (records.isEmpty()) continue;
    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> r : records) {
            producer.send(new ProducerRecord<>("orders-enriched", r.key(), enrich(r.value())));
        }
        producer.sendOffsetsToTransaction(currentOffsets(records),
            consumer.groupMetadata());
        producer.commitTransaction();
    } catch (Exception e) {
        producer.abortTransaction();
    }
}

Republish the failed record with error headers to a DLT and commit past it, then bound retries in the handler itself so a poison message is never consumed forever.

private static final int MAX_ATTEMPTS = 3;

void handle(ConsumerRecord<String, String> r, KafkaProducer<String, String> dlq) {
    try {
        process(r);
    } catch (Exception e) {
        int attempts = attemptsOf(r);
        if (attempts + 1 < MAX_ATTEMPTS) {
            dlq.send(new ProducerRecord<>("orders-retry", r.key(), r.value()));
            return;
        }
        ProducerRecord<String, String> dead = new ProducerRecord<>("orders-dlt", r.key(), r.value());
        dead.headers().add("x-error", e.getMessage().getBytes(StandardCharsets.UTF_8));
        dead.headers().add("x-original-topic", r.topic().getBytes(StandardCharsets.UTF_8));
        dlq.send(dead);
    }
}

Enable a SASL_PLAINTEXT or SASL_SSL listener, create SCRAM credentials with kafka-configs.sh, and give clients the matching mechanism plus JAAS username and password.

# broker: server.properties
listeners=SASL_PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=SASL_PLAINTEXT://localhost:9092
sasl.enabled.mechanisms=SCRAM-SHA-512
listener.name.sasl_plaintext.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required;

# create a credential
kafka-configs.sh --bootstrap-server localhost:9092 --alter \
  --add-config 'SCRAM-SHA-512=[password=secret]' --entity-type users --entity-name app

# client: client.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="app" password="secret";

Use Admin.create with a NewTopic list and read the resulting futures, so a per-topic error such as TOPIC_ALREADY_EXISTS is handled instead of silently ignored.

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");

try (Admin admin = Admin.create(props)) {
    NewTopic topic = new NewTopic("orders", 6, (short) 1)
        .configs(Map.of("min.insync.replicas", "1",
                        "cleanup.policy", "delete"));
    CreateTopicsResult result = admin.createTopics(List.of(topic));
    result.values().forEach((name, future) -> {
        try {
            future.get();
        } catch (ExecutionException e) {
            System.out.println(name + " failed: " + e.getCause().getMessage());
        }
    });
}

Describe the group to see per-partition lag, then poll it on a schedule and alert on a rising total rather than a single snapshot, since lag spikes are normal during rebalances.

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group orders-cg

# machine-readable output for a monitoring script
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group orders-cg --members --verbose

# reset offsets for a stuck partition (group must be inactive)
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group orders-cg --topic orders --reset-offsets --to-latest --execute