Apache Kafka — Cheat Sheet

Kafka is a distributed, durable, high-throughput event streaming platform. It acts as a fault-tolerant, ordered message log. Problems it solves: • Decoupling: Producers and consumers are independent — producers don't know who consumes their events • Durability: Messages are persisted to disk and replicated across brokers • High throughput: Millions of messages/second via sequential disk writes and zero-copy reads • Replay: Consumers can re-read historical events by resetting their offset • Fan-out: One event can be consumed by multiple independent consumer groups Use cases: Event sourcing, change data capture, real-time analytics, audit logs, inter-service communication, stream processing.

Topic: A named category of messages (like a table in a DB). Producers write to topics; consumers read from topics. Partition: A topic is split into ordered, immutable partitions. Partitions are the unit of parallelism. Each partition is a sequential log. Within a partition, messages are ordered. Across partitions, there is no global order. Offset: A unique, sequential integer ID assigned to each message within a partition. Consumers track their position by committing offsets. Resetting an offset allows message replay. Key insight: Messages with the same partition key always go to the same partition (hash-based routing), guaranteeing ordering for a key (e.g., all events for user_id=123 are in order).

A consumer group is a set of consumers that jointly consume a topic. Kafka assigns each partition to exactly one consumer within the group. Parallelism: With 12 partitions and 4 consumers in a group, each consumer gets 3 partitions. Maximum parallelism = number of partitions. If consumers > partitions: Extra consumers are idle. If consumers < partitions: Some consumers handle multiple partitions. Multiple groups: Multiple independent consumer groups can all read the same topic at their own pace (one for real-time processing, one for analytics). Each group maintains its own committed offsets. Rebalancing: When a consumer joins or leaves the group, Kafka rebalances partition assignments. During rebalance, consumption is paused — design for idempotency.

At-most-once: Consumer commits offset before processing. If the consumer crashes after commit but before processing, the message is lost. Use when losing messages is acceptable (metrics). At-least-once: Consumer commits offset after successful processing. If it crashes after processing but before commit, the message is reprocessed. Most common. Design consumers to be idempotent (same message processed twice = same outcome). Exactly-once: The holy grail. Achieved in Kafka via: • Idempotent producer: Kafka deduplicates retried sends using a sequence number (enable.idempotence=true) • Transactions: Producer uses beginTransaction/commitTransaction to atomically write to multiple partitions • Kafka Streams: Built-in exactly-once processing with transactional producers and atomic offset commits Exactly-once outside Kafka (e.g., Kafka → database) requires application-level idempotency.

Kafka guarantees ordering within a partition. There is no global ordering across partitions. To order messages for a specific entity (e.g., all order events for order_id=123), use the entity ID as the partition key. Kafka routes all messages with the same key to the same partition using hash(key) % num_partitions. Caveats: • Adding partitions changes the key-to-partition mapping — messages for the same key may go to different partitions • Consumer group rebalancing can cause temporary out-of-order processing between reassigned consumers • With multiple partitions, ordering across keys is not guaranteed For strict global ordering: Use a single-partition topic (limits throughput to one consumer).

The acks setting controls how many broker acknowledgments the producer waits for before considering a send successful. • acks=0: No acknowledgment. Fire and forget. Highest throughput, risk of message loss. • acks=1: Wait for the leader to write to its local log. Risk: if leader crashes before replicating, message is lost. • acks=all (or -1): Wait for all in-sync replicas (ISR) to acknowledge. Strongest durability guarantee. Combined with min.insync.replicas=2 ensures at least 2 replicas have the message before acknowledging. Recommendation for critical data: acks=all + min.insync.replicas=2 + enable.idempotence=true. For high-throughput analytics where some loss is acceptable: acks=1.

Log compaction is a retention policy that keeps only the latest value for each message key, discarding older values with the same key. Use case: Maintain the latest state for each entity. Example: user profile updates — only the most recent profile per user_id is needed. A consumer reading from the beginning gets the complete current state of all entities without reading every historical update. Compares to time-based retention: Time-based retention deletes old messages regardless of key. Compaction retains at least one message per key. Configure: log.cleanup.policy=compact. Tombstone records: Produce a message with the key and null value to delete a key from the compacted log. Used for: Kafka as a database (Kafka Streams stateful applications, materialized views).

Kafka Streams is a Java library for building stateful stream processing applications that read from and write to Kafka topics. No separate cluster needed — runs inside your application. Key abstractions: • KStream: An unbounded, continuous stream of records (like a SQL INSERT) • KTable: A changelog stream representing the latest value per key (like a SQL table) • GlobalKTable: A KTable replicated to all application instances Capabilities: Filtering, mapping, grouping, joining streams and tables, windowed aggregations, stateful processing with RocksDB-backed state stores. Use when: You need stateful processing close to Kafka, low-latency transformations, or joining Kafka streams. For heavy-duty batch processing, use Flink or Spark. For simple stateless transformations, a simple consumer loop is sufficient.

Consumer lag is the difference between the latest offset produced to a partition and the last offset committed by a consumer group. High lag means the consumer is falling behind the producer. Formula: lag = latest_offset - committed_offset (per partition) Monitoring: • kafka-consumer-groups.sh --describe: CLI tool to view lag per consumer group • Kafka JMX metrics: kafka.consumer.group.lag exposed via JMX • Burrow (LinkedIn): Dedicated Kafka consumer lag monitoring • Prometheus + Grafana: kafka_consumer_group_lag metric via Kafka exporter Alerting: Set alerts on lag spikes, not just absolute lag (some lag at low throughput is fine). Alert on rate of lag increase. Causes: Consumer too slow (add more instances or partitions), message processing too expensive (optimize or use async), GC pauses, rebalancing.

Architecture: • Kafka: Distributed log. Messages are retained by broker for a configurable time and pulled by consumers. Consumers track their own offsets. Supports replay. • RabbitMQ: Traditional message broker. Messages are pushed to consumers and deleted after acknowledgment (by default). Consumer controls flow. Ordering: Kafka guarantees order within a partition. RabbitMQ queue is FIFO but doesn't guarantee order across consumers. Throughput: Kafka is orders of magnitude higher (millions/sec per node vs ~50k/sec for RabbitMQ) due to sequential disk writes. Use Kafka for: Event streaming, audit logs, high-throughput pipelines, event sourcing, replay. Use RabbitMQ for: Task queues, complex routing (exchanges, bindings), request-reply patterns, lower-latency message delivery.

Historically, Kafka relied on Apache ZooKeeper for cluster metadata management: broker registration, topic configuration, leader election for partition leadership, and consumer group coordination. ZooKeeper limitations: External dependency, separate cluster to manage, limits Kafka scalability (ZooKeeper bottleneck at high partition counts). KRaft (Kafka Raft): Introduced in Kafka 2.8, GA in 3.3, ZooKeeper-free mode. Kafka uses a built-in Raft consensus algorithm for metadata management. A subset of brokers (controllers) manage metadata natively. Benefits of KRaft: Simpler deployment (one less system), supports millions of partitions, faster controller failover, and easier Kafka-as-a-service offerings. Kafka 4.0 removes ZooKeeper support entirely. New deployments should use KRaft mode.

Kafka default max message size is 1MB. Sending large messages (images, blobs) exceeds this and harms throughput. Options: 1. Increase limits: message.max.bytes (broker) + max.request.size (producer) + max.partition.fetch.bytes (consumer). Works but large messages are serialized/deserialized per-consumer, increasing GC pressure. 2. Claim-check pattern (recommended for very large payloads): Store the large payload in object storage (S3, GCS). Publish a Kafka message with a reference pointer to the object. Consumer fetches the large data from storage. 3. Compression: Enable compression (snappy, lz4, gzip, zstd) to reduce message size. Transparently handled by producer/consumer. 4. Message chunking: Split large messages into chunks at the producer; reassemble at the consumer. Complex to implement correctly.

Kafka Connect is a framework for reliably streaming data between Kafka and external systems (databases, Elasticsearch, S3, JDBC, LDAP) without writing custom producer/consumer code. Two directions: • Source connectors: Pull data from an external system and publish to Kafka (e.g., Debezium reads DB changelog → Kafka) • Sink connectors: Consume from Kafka and write to an external system (e.g., Kafka → Elasticsearch indexer) Benefits: Declarative configuration (JSON), horizontal scaling (tasks distribute work), built-in retry/error handling, large connector ecosystem (Confluent Hub, Apache). Debezium (popular source connector): Reads binlog from MySQL/PostgreSQL and publishes row-level changes as Kafka events — the standard for CDC (Change Data Capture) in microservices.

Each Kafka partition has one leader and N-1 followers (replicas). replication.factor controls N (typically 3 for production). Leader handles all reads and writes. Followers continuously fetch from the leader and replicate in order. ISR (In-Sync Replicas): Replicas that are fully caught up with the leader (within replica.lag.time.max.ms). Only ISR members are eligible for leader election on failover. Acknowledgment flow: With acks=all, the producer waits for all ISR members to acknowledge. min.insync.replicas=2 ensures at least 2 replicas confirm before success. Leader election: If the leader dies, the controller elects a new leader from ISR members. Unclean leader election (electing an out-of-sync replica) risks data loss but improves availability — disabled by default.

Schema Registry (Confluent) is a service that stores and enforces Avro/Protobuf/JSON Schema definitions for Kafka topics. Problem without it: Producers and consumers must agree on message format. If a producer changes the schema (adds/removes/renames a field), consumers break silently or crash. How it works: Producer serializes messages with a schema ID (registered in Registry). Consumer deserializes using the schema ID fetched from Registry. Schema evolution rules (backward, forward, full compatibility) are enforced at registration time. Compatibility modes: • Backward: New schema can read data written by old schema • Forward: Old schema can read data written by new schema • Full: Both backward and forward compatible Essential in production microservices to prevent API breaking changes from cascading across Kafka consumers.

A Kafka topic is a named, ordered log of records. Producers write to topics; consumers read from them. Structure: • Topic: Logical category (e.g., "orders", "user-events") • Partition: Each topic is split into N partitions. Partitions are the unit of parallelism and ordering. Records within a partition are strictly ordered; across partitions, ordering is not guaranteed. • Segment: Each partition is stored as a sequence of segment files on disk. Kafka appends to the active segment, seals it when full, and starts a new one. • Offset: Each record in a partition has a monotonically increasing offset number (0, 1, 2...). Consumers track their position by offset. Partition count: More partitions = more parallelism (more consumers in a group can read simultaneously). But more partitions = more file handles, more memory per broker, longer leader election time on failure. Retention: Records are retained for a configurable period (default 7 days) regardless of whether they are consumed. Consumers can read at any offset within the retention window.

Core philosophy difference: RabbitMQ: Traditional message broker. Messages are routed to queues, consumed, then deleted. Designed for task distribution — each message processed by exactly one consumer. Complex routing (exchanges: fanout, direct, topic, headers). Good for short-lived tasks. Kafka: Distributed commit log. Messages persisted to disk for a retention period, regardless of consumption. Multiple consumer groups each read all messages independently. No message routing complexity — consumers pull from partitions. Designed for high-throughput event streaming and replay. Key differences: • Message retention: RabbitMQ deletes after consume. Kafka retains for configured period. • Replay: Kafka consumers can seek back and reprocess. RabbitMQ cannot. • Fan-out: Kafka — unlimited consumer groups each get all messages. RabbitMQ — requires fanout exchanges and separate queues per consumer. • Throughput: Kafka handles millions/sec. RabbitMQ handles tens of thousands/sec. • Ordering: Kafka — ordered within partition. RabbitMQ — no ordering guarantees across consumers. • Consumer scaling: Kafka — consumers in group, each gets subset of partitions. RabbitMQ — competing consumers on same queue. Choose RabbitMQ: Complex routing rules, task queues, short-lived messages, native AMQP interoperability. Choose Kafka: High throughput, event sourcing, replay, multiple independent consumers, audit log.

Rebalance: When consumers join or leave a consumer group, Kafka redistributes partitions among the remaining consumers. During rebalance, all consumption pauses — a "stop the world" event. Triggers: New consumer joins, consumer leaves (crash or graceful shutdown), consumer fails to heartbeat within session.timeout.ms, topic partition count changes. Rebalance protocols: • Eager (default): All consumers stop, revoke all partitions, rejoin, get new assignment. Full pause. • Cooperative/Incremental (Kafka 2.4+): Only affected partitions are reassigned. Other consumers continue processing unaffected partitions. Much lower impact — preferred for production. Minimizing rebalance impact: • Set session.timeout.ms appropriately — too low causes spurious rebalances from slow consumers. Too high delays detection of crashed consumers. • Use cooperative rebalancing: partition.assignment.strategy=CooperativeStickyAssignor • Static membership: assign.group.instance.id to each consumer. Kafka considers it the same consumer even after restart — no rebalance on graceful restart. Timeout still triggers rebalance if truly dead. • Process records quickly — slow processing may trigger heartbeat timeout → rebalance. Rebalance listener: onPartitionsRevoked() — commit current offsets before rebalance. onPartitionsAssigned() — reset state for newly assigned partitions.

Kafka Streams: A client-side Java library for stream processing on top of Kafka. No separate cluster needed — runs inside your application. Scales by running multiple instances of the app. Key concepts: • KStream: Unbounded stream of records (each record is an independent event) • KTable: Changelog stream interpreted as a table (latest value per key — like a DB table) • GlobalKTable: Full copy of a topic on every instance — good for broadcast lookups • Topology: The processing graph (source → operations → sink) Operations: filter, map, flatMap, join (stream-stream, stream-table), groupByKey, aggregate, windowing. Windowing: Time-based grouping of records. • Tumbling windows: Fixed non-overlapping (events per minute) • Hopping windows: Fixed, overlapping (events in last 5 min, updated every 1 min) • Session windows: Activity-based, gap-triggered State stores: Local RocksDB stores per instance for aggregations. State backed by Kafka changelog topics — durable and recoverable on restart. Example: ```java KStream<String, Order> orders = builder.stream("orders"); orders .filter((k, v) -> v.getAmount() > 1000) .groupByKey() .windowedBy(TimeWindows.of(Duration.ofMinutes(1))) .count() .toStream() .to("large-order-counts"); ``` Vs Flink: Kafka Streams is simpler (no separate cluster), Flink is more powerful (exactly-once, complex CEP, better latency guarantees).

Kafka transactions: Enable atomic writes across multiple topics/partitions and exactly-once semantics between producer and consumer. Exactly-once semantics (EOS): Read from topic A, process, write to topic B — all atomically. If processing fails, both the output write and the offset commit are rolled back. No duplicate processing, no data loss. Producer transactions: ```java producer.initTransactions(); try { producer.beginTransaction(); producer.send(new ProducerRecord<>("output-topic", key, value)); producer.sendOffsetsToTransaction(offsets, consumerGroupId); producer.commitTransaction(); } catch (Exception e) { producer.abortTransaction(); } ``` How it works: • Transactional producer gets a transactional.id — persistent across restarts • Kafka assigns a PID (producer ID) and epoch — fences zombie producers • Producer writes records + offset commits to a transaction coordinator (a special Kafka broker) • On commit, coordinator marks transaction complete — records become visible • On abort, records are there but marked aborted — consumers skip them Consumer side: isolation.level=read_committed — only sees committed records (default is read_uncommitted). Overhead: Transactions add latency (coordinator involved, two-phase commit). Batch transactions to amortize cost — don't use one transaction per message. Use case: Kafka Streams exactly-once processing, financial event pipelines where double processing is unacceptable.

Message delivery semantics describe what guarantees a system makes about message delivery in the presence of failures. At-most-once: Messages may be lost, never duplicated. Producer fires and forgets (acks=0). Consumer commits offset before processing — if crash during processing, message is lost. Use when: metrics/telemetry where occasional loss is acceptable. Fastest — no retries, no overhead. At-least-once (default): No message loss, but duplicates possible. Producer retries on failure (acks=1 or all, retries > 0). Consumer commits offset after processing — if crash after processing but before commit, message reprocessed on restart. Use when: business can tolerate (and handle) duplicate processing with idempotent consumers. Exactly-once: No loss, no duplicates. Hardest to achieve. Kafka mechanisms: • Idempotent producer (enable.idempotence=true): Broker deduplicates retries using producer ID + sequence number. Exactly-once within a single producer session to a single partition. • Transactions: Atomic write across partitions + atomic offset commit. Exactly-once end-to-end in Kafka Streams. Exactly-once end-to-end (producer → Kafka → consumer → DB): Nearly impossible to guarantee truly end-to-end. Approximate with: idempotent producer + transactional consumer + idempotent consumer (deduplication on event_id in DB). Practical approach: At-least-once + idempotent consumers is the industry standard for most use cases.

Partition count determines maximum parallelism — you can have at most as many consumers in a group as partitions. Factors to consider: Target throughput: Estimate throughput per partition on both producer and consumer side. If one partition handles 10 MB/s, and you need 100 MB/s: 100/10 = 10 partitions minimum. Consumer parallelism: How many consumers will run in the group? You need at least that many partitions. More partitions than consumers = fine (consumers share). More consumers than partitions = wasted consumers sitting idle. Ordering requirements: If you need strict ordering for a key (all orders for user-123 in order), they must land on the same partition. Fine with any count — hash(key) % N routes to consistent partition. Retention and disk: More partitions = more files on disk. Each partition stores data. With 1000 topics × 100 partitions × 3 replicas = 300,000 files — broker memory and file descriptor limits. Replication overhead: Each additional partition adds replication work per broker. Rule of thumb: keep partitions per broker under 4000 (Confluent recommendation). Latency: Fewer partitions = lower replication lag, faster leader election on failure. Recommendation: Start with max(target_parallelism, throughput_based) × 2 as a buffer. You can increase partitions later (add more), but cannot decrease. Increasing disrupts ordering (rehashing of keys).

KRaft (Kafka Raft Metadata mode): Kafka's replacement for ZooKeeper dependency, introduced as production-ready in Kafka 3.3. Problem with ZooKeeper: • External dependency to manage, deploy, and monitor separately • Metadata stored in ZooKeeper limits scalability (millions of partitions problematic) • Leader election and metadata propagation slower (ZooKeeper round trip) • Two separate systems to understand and operate KRaft solution: Kafka brokers themselves manage metadata using the Raft consensus protocol. A subset of brokers act as controllers (quorum of 3 or 5). Controller quorum stores metadata in a replicated Kafka log. Benefits: • No ZooKeeper: Simpler deployment and operations. One system instead of two. • Faster recovery: Controller failover in milliseconds vs seconds with ZooKeeper. • Scalability: Millions of partitions supported (vs hundreds of thousands with ZooKeeper). • Predictable: Raft is a well-understood consensus algorithm. Modes: • Combined mode: Same nodes act as both controller and broker (small clusters, dev) • Isolated mode: Separate controller-only nodes (production recommendation) Migration: Kafka 2.8-3.2 offered KRaft as preview. 3.3+ production-ready. ZooKeeper support removed in Kafka 4.0. Conflict: Confluent Cloud has been KRaft-only for new clusters since 2023.

Log compaction: An alternative retention policy where Kafka keeps only the latest record for each key, instead of time-based retention. Old records with the same key are cleaned up, but the latest value is always retained. Use case: When Kafka topic represents a "current state" table rather than an event stream. After compaction, reading the entire topic gives you the latest state for every key — a full snapshot. Examples: • User profile updates: Keep only the latest profile per user_id • Database change capture (Debezium): Latest row state per primary key • Configuration changes: Latest config per config_key • KTable materialization: Kafka Streams uses compacted changelog topics to restore state stores Tombstone: A record with key=X and value=null. Signals that key X should be deleted. After compaction, this tombstone itself is eventually removed after a grace period. How it works: Log cleaner threads periodically scan "dirty" (uncompacted) segments. For each key in the dirty section, if a newer record with the same key exists (in clean section), the old record is removed. Active segment is never cleaned. Configuration: cleanup.policy=compact (vs delete for time-based). min.compaction.lag.ms controls how old a record must be before it's eligible for compaction. Mixed policy: cleanup.policy=compact,delete — both compact by key AND delete old segments. Useful for bounded-size compacted topics.

Kafka's pull-based consumer model inherently handles backpressure — consumers control the flow rate by polling at their own pace. Producer backpressure: • Producer has an internal record accumulator (buffer). If the buffer fills (buffer.memory exhausted), producer.send() blocks or throws an exception (max.block.ms). • The buffer fills when the broker can't keep up with the send rate — natural backpressure signal. • Reduce throughput: Increase batch.size and linger.ms to batch more records per request. Consumer backpressure: • Consumer calls poll() to fetch records. It processes them before calling poll() again. If processing is slow, next poll() is delayed — consumer naturally reads at the processing rate. • If processing is very slow, the broker doesn't push more data. Consumer falls behind — consumer lag grows. • Alert on consumer lag (via Kafka JMX, Burrow, or Grafana Kafka dashboard). Kafka consumer lag: The offset difference between the latest produced message and the latest consumed message. High lag = consumer can't keep up. Resolution: add more consumer instances (up to partition count), optimize consumer processing, or scale the consumer group. Reactive Streams: If using Project Reactor Kafka or Alpakka Kafka, backpressure is propagated through the reactive pipeline — upstream slows down when downstream is full. KEDA: Scale consumer pods based on Kafka consumer lag. Automatically adds pods when lag grows, reduces when caught up.

Kafka controller: One broker in the cluster is elected as the controller. It is responsible for cluster-wide coordination and metadata management. Controller responsibilities: • Leader election: When a partition leader fails (broker down), the controller selects a new leader from the ISR (In-Sync Replicas) and notifies all brokers of the change. • Partition reassignment: When brokers are added or removed, the controller orchestrates partition rebalancing. • Broker registration: Tracks which brokers are alive (via ZooKeeper ephemeral nodes in classic mode, or Raft heartbeat in KRaft). • Topic/partition lifecycle: Handles topic creation, deletion, and partition count changes. Controller election: In ZooKeeper mode: first broker to create /controller ephemeral node becomes controller. When it dies, ZooKeeper notifies others → new election. In KRaft: Raft quorum elects a leader from controller nodes. Controller failure: When the controller broker crashes, a new controller is elected within seconds. During this window, leader elections for failed partitions are delayed — unavailability window. Controller bottleneck: In large clusters (thousands of partitions), the single controller can become a bottleneck for metadata propagation. KRaft improves this significantly — controller operations are faster and more scalable. Active controller: SHOW controller with kafka-broker-api-versions.sh or via JMX (kafka.controller:type=KafkaController,name=ActiveControllerCount should be 1 on exactly one broker).

acks setting controls how many broker acknowledgments the producer waits for before considering a send successful. acks=0 (fire and forget): Producer sends and immediately considers it done. No acknowledgment from broker. Highest throughput, lowest latency. Risk: if broker is down or message is lost before writing, producer doesn't know. Use for: high-volume telemetry/metrics where occasional loss is acceptable. acks=1 (leader only): Producer waits for the leader broker to write the message to its local log before acknowledging. If leader crashes after ack but before followers replicate → data loss. Moderate throughput. Use for: non-critical events where some loss is tolerable. acks=all (or acks=-1): Producer waits for ALL in-sync replicas (ISR) to acknowledge. No data loss as long as min.insync.replicas requirement is met. Lowest throughput, highest latency. Use for: financial transactions, order events, any data where loss is unacceptable. min.insync.replicas (min.isr): Works with acks=all. Minimum number of ISR members that must acknowledge. If fewer replicas are in sync (e.g., broker down), producer receives NotEnoughReplicasException. Recommended: min.isr=2, replication.factor=3. Combination for strong durability: acks=all + min.insync.replicas=2 + replication.factor=3. Tolerates one broker failure without data loss.

Key Kafka metrics to monitor: Broker metrics: • UnderReplicatedPartitions: Partitions with fewer replicas than expected. Should be 0. High value = broker failure or slow follower. • ActiveControllerCount: Should be exactly 1 across the cluster. 0 = no controller (problem). >1 = split-brain. • OfflinePartitionsCount: Partitions with no leader. Should be 0. • RequestHandlerAvgIdlePercent: How busy the request handler threads are. Below 30% = broker under heavy load. • NetworkProcessorAvgIdlePercent: Network thread utilization. Producer metrics: • record-error-rate: Failed sends. • record-send-rate: Throughput. • request-latency-avg: How long broker takes to acknowledge. Consumer metrics: • records-lag (consumer lag): Difference between latest offset and consumer's committed offset. Most important consumer metric. Should be near 0 for real-time consumers. • fetch-latency-avg: Time to fetch records from broker. Tool stack: • JMX exporter + Prometheus: Expose Kafka JMX metrics as Prometheus format. • Grafana: Official Kafka dashboards (from Confluent or community). • Kafka UI, AKHQ, Redpanda Console: Web UIs for topic/consumer group management. • Burrow (LinkedIn): Dedicated consumer lag monitoring with SLA alerting. • Confluent Control Center: Commercial, fully managed monitoring. Alerting thresholds: UnderReplicatedPartitions > 0, consumer lag > 10K and growing, broker disk > 80%.

Poll loop: The fundamental pattern for consuming messages from Kafka. The consumer repeatedly calls poll() to receive batches of records. ```java KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(List.of("orders")); try { while (running) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500)); for (ConsumerRecord<String, String> record : records) { process(record); } consumer.commitSync(); // commit after processing batch } } finally { consumer.close(); } ``` poll(duration): The duration is the max time to block waiting if no records are available. Returns immediately if records are ready. Why poll matters for group management: Poll loop also sends heartbeats and triggers partition assignment callbacks. If you don't call poll() within max.poll.interval.ms (default 5 minutes), Kafka considers the consumer dead and triggers rebalance. max.poll.records: Max number of records returned per poll() call. Lower this if processing each record takes a long time — prevents max.poll.interval.ms from being exceeded. Commit strategies: • commitSync() after processing: At-least-once — safe but blocks • commitAsync() after processing: At-least-once, non-blocking, failures silently lost • Auto commit (enable.auto.commit=true): Commits periodically — may lose messages on crash (committed but not processed) Best practice: Manual commitSync() after processing batch = at-least-once with full control.

Kafka headers: Key-value metadata attached to a ProducerRecord. Headers are separate from the message key and value — they carry cross-cutting context without polluting the business payload. Common uses: • Distributed tracing: Propagate trace-id and span-id headers across Kafka messages. Kafka consumers extract and continue the trace. • Schema version: Carry schema version identifier separately from the payload. • Message type: "event-type: OrderPlaced" — consumer can route without deserializing the full payload. • Source service: "source-service: order-service" — auditing and debugging. • Idempotency key: Pass deduplication key without embedding it in the domain payload. • Content type: "content-type: application/json" for format negotiation. • Retry metadata: "retry-count: 2", "original-topic: orders" for DLQ processing. Producing headers: ```java ProducerRecord<String, String> record = new ProducerRecord<>("orders", key, value); record.headers().add("trace-id", traceId.getBytes(StandardCharsets.UTF_8)); record.headers().add("source-service", "order-service".getBytes()); producer.send(record); ``` Consuming headers: ```java for (Header header : record.headers()) { String key = header.key(); String value = new String(header.value(), StandardCharsets.UTF_8); } ``` Spring Kafka: MessageListenerAdapter automatically extracts headers into Spring Message headers for use with @Header annotation in @KafkaListener methods.

Consumer group coordinator: A specific Kafka broker responsible for managing the lifecycle of one or more consumer groups. Handles member join/leave, rebalancing, and offset commits. How it's selected: Each consumer group is assigned to a coordinator broker based on: hash(group.id) % number_of_partitions_in___consumer_offsets_topic → maps to a partition of __consumer_offsets → the leader of that partition is the coordinator. __consumer_offsets topic: An internal Kafka topic with 50 partitions by default. Stores committed offsets for all consumer groups. The coordinator manages this. Group join protocol: 1. Consumer sends JoinGroup request to coordinator 2. First consumer to join becomes the group leader (not the coordinator) 3. Coordinator sends JoinGroupResponse with member list to all members 4. Group leader performs partition assignment (using configured assignor) 5. Group leader sends SyncGroup request with assignments 6. Coordinator distributes assignments to all members via SyncGroupResponse Heartbeat: Each consumer sends periodic Heartbeat to coordinator within heartbeat.interval.ms. If coordinator doesn't receive a heartbeat within session.timeout.ms → declares consumer dead → triggers rebalance. Offset commit: Consumer sends OffsetCommit request to coordinator. Coordinator writes to __consumer_offsets topic. Durable, replicated like any Kafka topic.

Idempotent consumer: Processes each event exactly once by detecting and ignoring duplicate events, which naturally occur with at-least-once delivery. Why duplicates happen: Consumer crashes after processing but before committing offset → on restart, the same messages are redelivered. Also: Kafka producer retries, transient network issues, rebalances. Deduplication strategies: 1. Database unique constraint: Include event_id in the INSERT/UPDATE. If the event was already processed, the unique constraint violation is caught and ignored. ```sql INSERT INTO order_events (event_id, order_id, type, ...) VALUES (?, ?, ?, ...) ON CONFLICT (event_id) DO NOTHING; ``` 2. Redis set membership: Before processing, check SET.ismember(event_id). If present: skip. If not: add to set (with TTL) and process. TTL should match event retention period. 3. Idempotent state machine: Design operations so repeating them is a no-op. UPDATE orders SET status='paid' WHERE id=? AND status='pending' — if status is already 'paid', UPDATE affects 0 rows, no harm done. 4. Event ID tracking table: Dedicated table (event_id, processed_at) with unique constraint on event_id. Scope of deduplication: Must cover the at-least-once window (how long before a message is considered too old to be a duplicate). TTL in Redis or cleanup job for old event IDs. Performance: Redis check is sub-millisecond. DB unique constraint adds one write per event. Design for throughput.

Kafka storage is designed for high-throughput sequential writes and reads, optimized for disk I/O patterns. Storage layout: • Each partition is stored as a directory: /data/kafka/topic-name-partition-number/ • Inside: sequence of segment files (.log), index files (.index, .timeindex) • Active segment: The current segment being written to. All others are sealed (immutable). Segment files: • .log: The actual record data. Binary format — Kafka's internal record batch format. • .index: Sparse offset index mapping logical offsets to physical byte positions in the .log file. Enables O(log N) offset lookup. • .timeindex: Sparse timestamp index for time-based offset lookup (seek to timestamp). Write path: Records are appended to the active segment. Sequential writes = maximum disk throughput (100+ MB/s on spinning disk, much more on SSD). No random writes. Read path: Consumer requests records at offset X. Kafka uses binary search on the .index file to find the byte position of offset X. Seeks to that position in the .log file. Reads sequentially from there. Zero-copy: Kafka uses sendfile() system call (Java FileChannel.transferTo) to transfer data from page cache directly to network socket — bypasses user space. Dramatically reduces CPU overhead for consumers that can read from OS page cache. Page cache: Kafka relies on OS page cache heavily. Hot data (recently written) stays in page cache — reads are from memory, not disk. This is why Kafka needs significant RAM for the broker OS, not just the JVM.

Consumer lag: The number of messages a consumer group is behind the latest produced offset. Lag = latest_offset - consumer_committed_offset. Lag = 0: Consumer is caught up with producers. Real-time processing. Lag growing: Consumer can't keep up with production rate. Need to scale or optimize. Lag stable (non-zero): Consumer maintains a consistent buffer behind — may be intentional (batch processing) or concerning. Measuring lag: • kafka-consumer-groups.sh --describe --group my-group: Shows lag per partition. • Kafka JMX metrics: kafka.consumer:type=consumer-fetch-manager-metrics,partition=X,topic=Y records-lag • Burrow: LinkedIn's dedicated consumer lag monitor with SLA alerting. • Prometheus + JMX exporter: Scrape consumer lag metrics, alert in Grafana. Resolving high lag: 1. Add more consumers to the group (up to partition count limit) 2. Optimize consumer processing (faster DB writes, batch inserts, async processing) 3. Increase max.poll.records (fetch larger batches per poll) 4. Check if consumer is stuck (GC pause, slow external call, deadlock) 5. If lag is due to one "poison pill" message: skip it (seek past offset) or send to DLQ Lag alerting: Alert when lag > threshold AND growing (lag rate > 0). If lag is stable even at 10K, it may be fine. A growing lag is urgent — it will eventually exhaust retention and messages will be lost.

Kafka retention: How long Kafka keeps messages before deleting them. Unlike traditional message queues, Kafka retains messages regardless of consumption. Time-based retention (default): log.retention.hours (default 168 = 7 days). Segments older than the retention period are deleted when the log cleaner runs. Active segment is never deleted. Size-based retention: log.retention.bytes — total size per partition. When partition exceeds this size, oldest segments are deleted. -1 = unlimited. Both together: log.retention.bytes AND log.retention.hours — whichever limit is hit first triggers deletion. Segment rolling: log.segment.bytes (default 1 GB) — size at which a new segment is started. log.segment.ms — time-based rolling. Segments are sealed and eligible for deletion as a unit. Compaction as alternative: cleanup.policy=compact — keep only latest value per key instead of time-based deletion. Retention design considerations: • Enough retention for: consumer groups to catch up after incidents (days of lag possible), replay scenarios (re-processing from beginning), forensics. • Not too much: storage cost grows linearly with retention × throughput × replication.factor. • Different topics need different retention: raw click events (2 days), financial transactions (7 years for compliance), user sessions (24 hours). Offset reset: If a consumer's committed offset is older than retention, Kafka throws OffsetOutOfRange. Configure auto.offset.reset=earliest (start from oldest available) or =latest (start from newest).

Poison pill: A message that causes the consumer to fail every time it tries to process it. Without handling, it blocks the entire consumer and partition — no other messages can be processed. Example: Malformed JSON, message with an unexpected null field, a message that triggers a bug in processing logic. Detection: Consumer throws exception processing the message. Retry logic keeps retrying the same message. Consumer offset never advances — lag grows for that partition. Strategies: 1. Dead Letter Queue (DLQ): After N failed retries, publish the poison pill to a separate DLQ topic. Log the error with full context. Continue processing subsequent messages. ```java try { process(record); } catch (Exception e) { if (attempts > maxRetries) { producer.send(new ProducerRecord<>("orders-dlt", record.key(), record.value())); // commit offset to move past this record } } ``` 2. Skip and log: Log the bad record with full context, increment a counter metric, and commit the offset to move past it. Acceptable for non-critical data streams. 3. Fix and replay: Identify the root cause (bug in consumer code). Fix. Seek to the offset of the poison pill and replay it. Only works if the fix makes the message processable. 4. Spring Kafka SeekToCurrentErrorHandler / DefaultErrorHandler: Configures retry with backoff, then sends to DLQ automatically. Simplifies error handling. DLQ monitoring: Alert on any message appearing in DLQ topics — it always means something failed and needs investigation.

MirrorMaker 2 (MM2): Kafka's built-in tool for replicating data between Kafka clusters. Built on Kafka Connect framework. Use cases: • Geo-replication: Replicate topics from a US cluster to an EU cluster for local consumers. • Disaster recovery: Maintain a standby cluster that is continuously synced with the primary. On primary failure, failover to standby. • Data migration: Move data between clusters during a Kafka upgrade or infrastructure change. • Aggregation: Collect events from multiple regional clusters into a central analytics cluster. How it works: MM2 runs as a Kafka Connect cluster. Source connector reads from source cluster, sink connector writes to target cluster. Preserves message key, value, headers, timestamps. Tracks consumer group offsets — enables consumer groups to resume from correct position after failover. Offset translation: MM2 stores offset mapping between source and target offsets. On failover, consumers can look up their source offset → find equivalent target offset → resume without reprocessing or missing messages. Topic naming: By default, target topics are prefixed with source cluster alias: source.topic-name. Configurable. Alternatives: • Confluent Replicator: Commercial, part of Confluent Platform. More features, better support. • Uber uReplicator: High-scale replication with better dynamic topic management. • Brooklin (LinkedIn): Multi-source, multi-destination replication platform. MM2 is production-ready for most replication use cases as of Kafka 2.7+.

Kafka achieves millions of messages per second through multiple complementary optimizations: 1. Sequential disk writes: All writes are append-only to the end of log segments. Sequential I/O is orders of magnitude faster than random I/O on both HDD and SSD. 2. OS page cache: Kafka writes to the OS page cache (via memory-mapped files) and lets the OS flush to disk asynchronously. Reads also serve from page cache — effectively memory-speed access for hot data. 3. Zero-copy: sendfile() system call transfers data from page cache directly to network socket. Eliminates copy from kernel space → user space → kernel space. Reduces CPU usage by 60-70% for consumers. 4. Batching: Producers batch records (controlled by batch.size and linger.ms) into a single network request. Consumers fetch batches (max.partition.fetch.bytes). Fewer, larger I/O operations = better throughput. 5. Compression: GZIP, Snappy, LZ4, ZSTD compression at batch level. Reduces network bandwidth and disk usage. CPU overhead is offset by reduced I/O. LZ4/Snappy for latency, ZSTD for best ratio. 6. Horizontal partitioning: Work distributed across partitions → multiple consumers read in parallel → aggregate throughput scales linearly with partitions. 7. Efficient binary protocol: Custom binary protocol over TCP — minimal overhead vs HTTP. No JSON parsing at the broker level. 8. Replication optimization: Followers fetch in batches, not record by record. Leader uses zero-copy to send to followers too.

ISR (In-Sync Replicas): The subset of a partition's replicas that are fully caught up with the leader — within replica.lag.time.max.ms (default 30 seconds) of the leader's latest offset. Leader maintains the ISR list. Followers fetch from the leader continuously. If a follower falls behind (network issue, slow disk, broker overloaded), it is removed from ISR. When it catches up, it rejoins ISR. Durability connection: • acks=all: Producer waits for all current ISR members to acknowledge. • min.insync.replicas: Minimum ISR size required for produce to succeed. If ISR shrinks below this, broker returns NotEnoughReplicasException to producers. Example (replication.factor=3, min.insync.replicas=2): • ISR = [broker1 (leader), broker2, broker3] → produces succeed • broker3 falls behind → ISR = [broker1, broker2] → still succeeds (ISR size = 2 = min.isr) • broker2 also fails → ISR = [broker1] → produces FAIL (ISR size = 1 < min.isr=2) → topic becomes unavailable until a broker rejoins Trade-off: Higher min.insync.replicas = better durability but lower availability (more scenarios cause produce failures). Lower = more available but more data loss risk on multiple simultaneous failures. Unclean leader election (unclean.leader.election.enable): If ISR is empty and no ISR member is available, allow an out-of-sync replica to become leader. Risk: data loss. Default: disabled in Kafka 0.11+.

Stateful operations: Stream processing that requires maintaining state across multiple records (aggregations, joins). Kafka Streams manages this state in local RocksDB stores backed by Kafka changelog topics. Aggregations: ```java KGroupedStream<String, Order> grouped = orders.groupByKey(); KTable<String, Long> orderCounts = grouped.count(Materialized.as("order-counts-store")); ``` Reducing: ```java KTable<String, Double> totalRevenue = grouped.reduce((v1, v2) -> v1 + v2); ``` Windowed aggregation: ```java TimeWindowedKStream<String, Order> windowed = grouped.windowedBy( TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))); KTable<Windowed<String>, Long> windowedCounts = windowed.count(); ``` Joins: • Stream-Stream join: Join two streams within a time window. Both sides must arrive within the window. • Stream-Table join: Enrich stream records with table lookups (table = latest value per key). • Table-Table join: Materialized view of joined tables, updated when either changes. State store management: • Local RocksDB: Fast local reads/writes. Persisted to disk. • Changelog topic: Every state update is written to a Kafka topic. On restart or rebalance, restore state by replaying the changelog. • Standby replicas: num.standby.replicas > 0 — pre-warm state on standby instances to reduce restore time after rebalance. Interactive queries: Query state stores directly from the Kafka Streams app via REST API pattern — expose current aggregate counts without publishing to a separate topic.

Both are stream processing frameworks for real-time computation on Kafka data — but with different design philosophies and capabilities. Kafka Streams: • Deployment: Library — runs inside your Java application. No separate cluster. • Scaling: Scale by running more instances of your app. Kafka manages partition assignment. • State: Local RocksDB backed by Kafka changelog topics. Simple but limited. • Exactly-once: Supported with Kafka transactions (EOS). • Language: Java/Kotlin only. • Learning curve: Lower — it's a Java library with familiar Kafka concepts. • Best for: Simple to medium stream processing embedded in a microservice. No ops overhead. Apache Flink: • Deployment: Dedicated cluster (JobManager + TaskManagers) or managed (Confluent Cloud, AWS KDA, Google Dataflow). • Scaling: Flink manages parallelism internally across TaskManagers. More flexible. • State: RocksDB or heap-based. Flink savepoints (snapshots) for migration and recovery. • Exactly-once: Flink's checkpointing + Kafka sources/sinks gives true end-to-end exactly-once. • Language: Java, Scala, Python (PyFlink), SQL (Flink SQL). • Features: Complex event processing (CEP), more flexible window types, better late event handling, native SQL. • Best for: Complex stream processing, high-scale analytics, teams that can operate a Flink cluster. Choose Kafka Streams: Embedded in a microservice, familiar Java devs, simple pipelines, no new infrastructure. Choose Flink: Complex processing logic, SQL-based pipelines, language flexibility, enterprise analytics.

Kafka ordering guarantees: • Within a partition: Messages are strictly ordered (offset 0, 1, 2, 3...). Producers write sequentially. Consumers read in order. • Across partitions: No ordering guarantee. Messages in partition 0 and partition 1 may be interleaved arbitrarily. Partition key determines ordering: Messages with the same key always go to the same partition (via hash(key) % num_partitions). This ensures all events for a specific entity (user-123, order-456) are ordered relative to each other. Ordering vs parallelism trade-off: More partitions = more parallelism but within a partition, only one consumer in a group reads (serialized). To maintain ordering for a key while scaling, increase parallelism by having more distinct keys — don't increase consumers for the same key's partition. Producer ordering gotcha: With retries enabled, if a batch fails and retries while a later batch succeeds, messages can arrive out of order at the broker. Fix: enable.idempotence=true (automatically sets max.in.flight.requests.per.connection=5 with deduplication guarantees ordering). Global ordering (across partitions): Cannot be done efficiently in Kafka. If truly required: use single partition (limits throughput to one partition's capacity). Or: use an external sequencer/watermarking system. Rarely worth the trade-off — redesign the problem to not need global ordering. Consumer ordering: Consumer reads one partition at a time per thread (per-partition ordered). To maintain order, process one record completely before polling next within a partition.

__consumer_offsets: An internal Kafka topic that stores committed offsets for all consumer groups. This is how Kafka persists consumer progress durably. Structure: • 50 partitions by default (offsets.topic.num.partitions) • Replication factor 3 (offsets.topic.replication.factor) • The group coordinator broker owns the partition: hash(group.id) % 50 → coordinator partition → leader of that partition is the coordinator Format: • Key: (group_id, topic, partition) • Value: committed_offset, metadata, commit_timestamp Compaction: __consumer_offsets uses log compaction — only the latest committed offset per (group, topic, partition) is retained. Offset commit flow: Consumer calls commitSync() or commitAsync() → request goes to group coordinator → coordinator writes to __consumer_offsets → acknowledgment returned. Reading offsets: Use kafka-consumer-groups.sh to inspect committed offsets. Internally, the tool reads __consumer_offsets. Failure impact: If the coordinator broker fails, Kafka elects a new coordinator → new coordinator rebuilds state by reading its __consumer_offsets partition. Consumers reconnect to new coordinator and resume. Retention: Default 7 days (offsets.retention.minutes). Offsets for inactive groups are eventually cleaned up. If a consumer group is inactive for 7 days and restarts, it triggers auto.offset.reset. Never write to __consumer_offsets directly — only through the Kafka consumer API or kafka-consumer-groups.sh --reset-offsets.

CDC (Change Data Capture) pipeline: Capture every row-level change (INSERT/UPDATE/DELETE) from a database and stream them to Kafka in real-time. Debezium: Open-source CDC platform that reads from database transaction logs (MySQL binlog, PostgreSQL WAL, MongoDB oplog) and publishes changes as Kafka events. Setup (PostgreSQL): 1. Enable logical replication: wal_level=logical in postgresql.conf 2. Create replication slot: Debezium uses a replication slot to track WAL position 3. Configure Debezium connector: ```json { "name": "inventory-connector", "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "database.hostname": "postgres", "database.port": "5432", "database.user": "debezium", "database.password": "...", "database.dbname": "inventory", "table.include.list": "public.orders,public.products", "topic.prefix": "dbserver1" } ``` 4. Debezium publishes to: dbserver1.public.orders topic Event format (Debezium envelope): ```json { "before": {"id": 1, "status": "pending"}, "after": {"id": 1, "status": "shipped"}, "op": "u", // c=create, u=update, d=delete, r=read/snapshot "ts_ms": 1234567890, "source": {"db": "inventory", "table": "orders"} } ``` Consumers: Elasticsearch sync, cache invalidation, audit log, event-driven business logic. Snapshot mode: On first start, Debezium performs a full snapshot (reads all existing rows). Then switches to streaming mode from the WAL.

Kafka uses a pull-based consumer model — consumers pull messages from brokers on their own schedule. Pull (Kafka's approach): Consumer calls poll() when ready. Consumer controls the rate. If processing is slow, consumer polls less frequently — natural backpressure. Consumer can batch large fetches. If broker has no new messages, consumer waits (long poll) up to fetch.max.wait.ms (default 500ms) before returning empty. Advantages of pull: • Consumer controls pace — no overwhelming a slow consumer • Efficient batching — consumer can request as many messages as it can handle • Rewind/replay — consumer can seek to any offset and re-read • Simpler broker — broker doesn't track per-consumer push state Push (traditional message queues like RabbitMQ): Broker pushes messages to registered consumers. Low latency when messages arrive. Risk: fast broker can overwhelm slow consumer. Requires per-consumer flow control. Producer to Kafka: Also push — producer pushes to broker. Broker buffers. Consumer pulls from buffer. Long polling: Kafka implements efficient long polling. Consumer sends fetch request to broker. If no data, broker waits up to fetch.max.wait.ms before responding. This prevents consumer from burning CPU with empty poll() calls. Hybrid in practice: Some Kafka integrations use a push model on top of Kafka pull — Kafka Connect sink connectors push to external systems. Kafka consumer processes pull, then pushes to a webhook or SSE endpoint. The core Kafka consumption remains pull-based.

Rolling restart: Restart each broker in the cluster one at a time, allowing the cluster to remain operational throughout. Why needed: JVM heap tuning, broker config changes, OS patches, Kafka version upgrades. Pre-restart checklist: 1. Verify cluster health: No under-replicated partitions (UnderReplicatedPartitions = 0) 2. Confirm partition leadership distribution is balanced 3. Check all ISR lists are full (no degraded replicas) 4. Increase min.insync.replicas tolerance if needed Steps per broker: 1. Trigger a controlled shutdown: kafka-server-stop.sh or graceful shutdown via system service. Kafka sends a LeaderEpoch to hand off leadership before shutting down — faster than timeout-based failover. 2. Wait for partition leadership to transfer to other brokers. 3. Wait for the broker to fully stop. 4. Apply changes (config update, JVM flags, package install). 5. Start the broker. 6. Wait for it to rejoin ISR (UnderReplicatedPartitions returns to 0). 7. Move to next broker. Safety: Never restart more than one broker at a time with replication.factor=3 and min.isr=2. Two brokers down simultaneously can make partitions unavailable. Preferred replica election: After rolling restart, trigger a preferred replica election (kafka-leader-election.sh) to redistribute leadership evenly — restarts often leave all leaders on the last restarted brokers. Tools: Confluent Control Center automates rolling restarts. Cruise Control rebalances leadership after restarts.

Offset management: Tracking and committing how far each consumer group has read in each partition. The foundation of Kafka's "at-least-once" delivery guarantee. Offset commit methods: Auto commit (enable.auto.commit=true): Kafka commits the latest polled offset every auto.commit.interval.ms (default 5000ms). Risk: records polled but not yet processed may be committed → on crash, those records are lost (at-most-once behavior in failure scenarios). Manual commit sync: ```java consumer.commitSync(); // blocks until committed // Use after processing the full batch ``` Manual commit async: ```java consumer.commitAsync((offsets, exception) -> { if (exception != null) log.error("Commit failed: {}", offsets, exception); }); // Non-blocking; use in the poll loop ``` Per-partition commit: ```java Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(); offsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1)); consumer.commitSync(offsets); // Commit specific offset, not entire batch ``` Offset reset (auto.offset.reset): LATEST — start from end of topic (miss historical). EARLIEST — start from beginning. NONE — throw exception if no committed offset found. Manual reset: kafka-consumer-groups.sh --reset-offsets --to-earliest / --to-datetime / --to-offset — seek the consumer group to a specific position. Useful for replaying events.

Kafka supports efficient batch processing through producer-side batching and consumer-side batch fetching. Producer batching: • batch.size (default 16KB): Max bytes to accumulate before sending. Records destined for the same partition accumulate in a batch. • linger.ms (default 0): Producer waits this many milliseconds for the batch to fill before sending. 0 = send immediately. Setting linger.ms=5 allows more records to join the batch → better compression, fewer requests. • Compression (compression.type=snappy/lz4/zstd): Applied per batch. More records per batch = better compression ratio. Consumer batch fetching: • max.poll.records (default 500): Maximum records returned per poll() call. • fetch.min.bytes (default 1): Broker waits until at least this many bytes are available before responding. Higher value = larger batches but more latency. • fetch.max.wait.ms (default 500): Max time broker waits for fetch.min.bytes threshold. • max.partition.fetch.bytes (default 1MB): Max bytes per partition per fetch. Batch processing pattern: Accumulate a batch of ConsumerRecords, process as a batch (bulk insert to DB, batch API call), commit offsets once for the entire batch. ```java List<ConsumerRecord<String, String>> batch = new ArrayList<>(); for (ConsumerRecord<String, String> record : records) { batch.add(record); if (batch.size() >= 100) { dbService.batchInsert(batch); batch.clear(); } } if (!batch.isEmpty()) dbService.batchInsert(batch); consumer.commitSync(); ```

Core rule: Within a consumer group, each partition is consumed by exactly ONE consumer at a time. A consumer can read from multiple partitions, but a partition cannot be shared between consumers in the same group. Implications: • Maximum parallelism = number of partitions. If you have 10 partitions and 10 consumers in a group, each consumer reads from 1 partition. • Excess consumers sit idle: 15 consumers + 10 partitions = 5 idle consumers. They're on standby for failover. • Fewer consumers than partitions: Consumers compensate. 3 consumers + 10 partitions = each consumer reads from ~3 partitions. Partition assignment strategies: • RangeAssignor: Assigns contiguous partition ranges per consumer. Simple but can be unbalanced. • RoundRobinAssignor: Distributes partitions evenly across consumers in round-robin order. • StickyAssignor: Minimizes partition movement during rebalance. Assigns partitions that consumers previously had when possible. • CooperativeStickyAssignor: Same as Sticky but incremental — doesn't pause all partitions during rebalance. Scaling: Add partitions AND consumers together for more throughput. Adding consumers beyond partition count gives no throughput benefit but does improve failover speed (idle consumers immediately take over on failure). Independent groups: Multiple consumer groups each get ALL partitions independently. Group A and Group B both read all 10 partitions. The partition-per-consumer rule is within a group.

Producer record accumulator: An in-memory buffer where records wait to be batched and sent to the broker. The buffer is organized by TopicPartition — each TopicPartition has its own batch queue. buffer.memory (default 32MB): Total memory for the producer buffer across all partitions. When the buffer is full, producer.send() blocks (up to max.block.ms) or throws BufferExhaustedException. Batch formation: • batch.size (default 16KB): Max size of a single batch for one TopicPartition. When batch fills, it's sent immediately. • linger.ms (default 0): How long to wait for the batch to fill before sending. 0 = send immediately (low latency, small batches). 5-20ms = wait for more records to join (higher throughput, slightly more latency). Sender thread: A background thread (I/O thread) drains the accumulator — picks ready batches and sends them to the broker. Multiple in-flight requests per broker: max.in.flight.requests.per.connection (default 5). Back-pressure signal: If buffer.memory exhausted → send() blocks → application slows down → natural flow control. Tuning for throughput: • Increase buffer.memory to handle more in-flight data • Increase batch.size to 64KB-256KB for fewer, larger requests • Set linger.ms=5-20ms to allow batches to fill • Use compression (lz4/zstd) to reduce payload size Tuning for low latency: • linger.ms=0 (send immediately) • Smaller batch.size • acks=1 (don't wait for all ISR)

True exactly-once end-to-end (producer → Kafka → consumer → external system) requires combining multiple mechanisms. Step 1 — Idempotent producer: enable.idempotence=true. Each message gets a producer ID + sequence number. Broker deduplicates retries. Exactly-once producer-to-broker for a single session. Automatically enables acks=all and max.in.flight=5. Step 2 — Transactional producer: Atomic writes across multiple partitions. Commit offsets + output records atomically. If processing fails, both are rolled back — no partial state. Step 3 — Transactional consumer: isolation.level=read_committed. Consumer only sees committed transactions. Uncommitted or aborted records are invisible. Step 4 — Idempotent external writes: For writes to external systems (DB, Elasticsearch), use idempotent write patterns (INSERT ON CONFLICT DO NOTHING with event_id). The transactional Kafka guarantees get you exactly-once within Kafka, but the external write must also be idempotent. Kafka Streams EOS: Kafka Streams wraps all of this automatically with processing.guarantee=exactly_once_v2. Read → process → write output + commit offsets atomically. Most complete EOS in the ecosystem. Cost: Transactions add latency (2-phase commit with coordinator). Transactional IDs must be stable across restarts. Not suitable for every use case — use for financial pipelines, inventory, any domain where double-processing has business consequences.

Kafka stores data in a binary format optimized for sequential writes and network efficiency. Record batch (MessageSet v2, introduced in Kafka 0.11): The unit of storage and transfer. A batch contains multiple records, compressed together. Record batch header: • baseOffset: Offset of the first record in the batch • batchLength: Total batch size in bytes • partitionLeaderEpoch: Leader epoch for fencing • magic: Format version (currently 2) • attributes: Compression codec (0=none, 1=gzip, 2=snappy, 3=lz4, 4=zstd), transactional flag, control batch flag • lastOffsetDelta: Relative offset of last record • firstTimestamp and maxTimestamp • producerId, producerEpoch, baseSequence: Idempotent/transactional producer info • records: Array of record entries Record entry (within a batch): • attributes: Per-record attributes (currently unused) • timestampDelta: Relative to batch firstTimestamp • offsetDelta: Relative to baseOffset • keyLength + key • valueLength + value • headers: Array of (key, value) byte pairs Why this format matters: • Batch-level compression: Compress all records together → better ratio than per-record compression • Relative offsets/timestamps: Smaller deltas take less space • Headers: Available without deserializing the value — useful for routing/filtering • Zero-copy compatible: Fixed-length prefix allows sendfile() without parsing

Broker failure sequence: 1. Broker B goes down: ZooKeeper (or KRaft) session expires. Broker B's ephemeral ZooKeeper node is deleted. The controller is notified. 2. Controller reacts: The controller identifies all partitions where B was the leader. For each such partition, controller selects a new leader from the ISR list of that partition. Controller writes new leader info to ZooKeeper. Controller sends LeaderAndIsr request to all brokers informing them of the new leaders. 3. Producers and consumers update: Metadata is cached by clients. Clients refresh metadata when they receive a Not Leader error. They discover the new leader and resume. 4. Follower recovery: When B comes back online, it reconnects and fetches the log end offset for its partitions from the new leaders. It fetches all missing records and catches up. Once within replica.lag.time.max.ms, it rejoins the ISR. Recovery time: Leader election is fast — seconds in healthy clusters. min.insync.replicas ensures no data loss (producers fail if ISR < min.isr). Unclean leader election (disabled by default) would be faster but risks data loss. Optimization: kafka-preferred-replica-election.sh redistributes leadership after recovery so that leadership doesn't concentrate on surviving brokers. Auto leader rebalancing: auto.leader.rebalance.enable=true (enabled by default). Broker data: Data on the failed broker's disk is still there when it comes back. Only the records that were in-flight (produced after the last sync to followers) may need to be re-produced.

Kafka wire protocol: A custom binary TCP protocol used for all communication between Kafka clients (producers, consumers, admin) and brokers, and between brokers themselves. Protocol structure: • Request: [length (4 bytes)] [api_key (2 bytes)] [api_version (2 bytes)] [correlation_id (4 bytes)] [client_id] [request body] • Response: [length (4 bytes)] [correlation_id (4 bytes)] [response body] • Correlation ID: Matches responses to requests — allows multiple in-flight requests on one connection. API keys: Each operation has a numeric key: Produce (0), Fetch (1), ListOffsets (2), Metadata (3), OffsetCommit (8), OffsetFetch (9), JoinGroup (11), SyncGroup (14), Heartbeat (12)... API versioning: Each API key has versions. Brokers advertise supported versions. Clients negotiate the highest mutually supported version. This enables protocol evolution without breaking backward compatibility. TCP connection: Clients maintain persistent TCP connections to brokers. One connection per broker per client (or pool of connections for high-throughput producers). No HTTP overhead — binary framing is compact and fast. Admin API: All admin operations (create topics, describe consumer groups, list offsets) use the same wire protocol. kafka-topics.sh, kafka-consumer-groups.sh are just thin wrappers around admin client API calls. Security: TLS for encryption (PLAINTEXT, SSL). SASL for authentication (SASL_PLAINTEXT, SASL_SSL). SCRAM-SHA-256/512, GSSAPI (Kerberos), OAUTHBEARER mechanisms.

Kafka interceptors: Plugin hooks that intercept messages at the producer or consumer layer without changing application code. Useful for cross-cutting concerns: monitoring, audit, transformation, correlation ID injection. Consumer interceptor interface: ```java public interface ConsumerInterceptor<K, V> { ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records); void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets); void close(); void configure(Map<String, ?> configs); } ``` onConsume(): Called after records are fetched but before they are returned to the poll() caller. Can modify, filter, or enrich records. Can extract tracing headers and set MDC (logging context). onCommit(): Called when offsets are committed. Useful for audit logging of consumer progress. Producer interceptor: ```java public interface ProducerInterceptor<K, V> { ProducerRecord<K, V> onSend(ProducerRecord<K, V> record); // called before send void onAcknowledgement(RecordMetadata metadata, Exception exception); // called on ack } ``` Configuration: ```properties consumer.interceptor.classes=com.example.TracingInterceptor,com.example.MetricsInterceptor ``` Common uses: • Inject trace-id from Kafka header into MDC for structured logging • Record consumer metrics (lag, processing rate) • Audit log: log every message received with topic/partition/offset • DLQ routing: intercept deserialization errors and route to DLQ • Schema validation: validate message format before processing

Kafka security covers three dimensions: encryption, authentication, and authorization. Encryption (TLS/SSL): • Configure SSL listener on brokers: listeners=SSL://0.0.0.0:9093 • Clients configure: security.protocol=SSL, ssl.truststore.location, ssl.truststore.password • mTLS: Require client certificates too: ssl.client.auth=required. Clients present certificate — mutual authentication. Authentication (SASL): • SASL_PLAINTEXT: SASL authentication without TLS (dev only — credentials visible in transit) • SASL_SSL: SASL + TLS encryption (production) • Mechanisms: PLAIN (username/password), SCRAM-SHA-256/512 (salted challenge-response, credentials stored in ZooKeeper), GSSAPI (Kerberos — enterprise), OAUTHBEARER (JWT from OAuth2 provider) Authorization (ACLs): • Kafka ACLs control which principals can perform which operations on which resources • kafka-acls.sh --add --allow-principal User:order-service --operation Write --topic orders • Operations: Read, Write, Create, Delete, Alter, Describe, ClusterAction, All • Resources: Topic, Group (consumer group), Cluster, TransactionalId Network isolation: • Separate listeners for internal (broker-to-broker) and external (client) traffic • Firewall rules to restrict broker ports to authorized networks • VPC private subnets for broker nodes Secrets rotation: Kafka supports SCRAM credential updates without cluster restart. Certificate rotation with rolling update.

Kafka stores each partition as a sequence of log segments on disk. Understanding segment management is essential for capacity planning and performance. Segment lifecycle: 1. Active segment: Current segment being written to. Only one per partition. 2. Sealed segments: Full segments waiting for retention cleanup. 3. Deleted segments: Removed when beyond retention policy. Segment configuration: • log.segment.bytes (default 1GB): Create a new segment when active segment reaches this size. • log.segment.ms (default 7 days): Create a new segment after this time even if not full. Ensures time-based retention works for low-volume topics. Retention check: Log cleaner runs periodically (log.retention.check.interval.ms, default 5 minutes). Deletes segments where the largest timestamp in the segment is older than retention period. Files per segment: For each segment, Kafka creates three files: • 00000000000000000000.log — record data • 00000000000000000000.index — offset index • 00000000000000000000.timeindex — timestamp index The filename prefix is the base offset of the segment. Disk capacity planning: retention_hours × throughput_bytes_per_hour × replication_factor + 20% headroom. Compaction segments: Compacted topics also use segments. The "dirty ratio" (dirty log size / clean log size) triggers compaction when it exceeds min.cleanable.dirty.ratio.

Producer throughput tuning — multiple levers: Batching (biggest impact): • batch.size: Increase from 16KB to 64KB-256KB. Larger batches = fewer network requests. • linger.ms: Set 5-20ms. Allows more records to join a batch before sending. Small latency cost, significant throughput gain. Compression: • compression.type=lz4 or zstd. Reduces payload size → less network bandwidth → broker writes less data. LZ4 for low CPU, ZSTD for best ratio. Must be configured on producer; broker stores compressed batches. Acknowledgments: • acks=1 instead of acks=all: Eliminates ISR replication wait. Faster but risks data loss on leader failure. Only use for non-critical data. Buffer memory: • buffer.memory=64MB or more: Larger buffer allows more in-flight batches. • max.block.ms: Increase if producers block waiting for buffer space. Parallelism: • More producer instances or threads: Each instance sends to its own partitions. • max.in.flight.requests.per.connection=5: Multiple requests in-flight per connection (default). Reduce to 1 only for strict ordering without enable.idempotence. Network: • send.buffer.bytes and receive.buffer.bytes: Increase TCP socket buffers for high-throughput scenarios. • Ensure producer and broker are in same region (low latency). Monitoring: Watch record-send-rate, batch-size-avg, compression-rate, request-latency-avg to validate improvements.

AdminClient: A programmatic Java API for performing administrative operations on a Kafka cluster without using command-line tools. Common operations: ```java Properties props = new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); try (AdminClient admin = AdminClient.create(props)) { // Create topic NewTopic newTopic = new NewTopic("orders", 10, (short) 3) .configs(Map.of("retention.ms", "604800000")); admin.createTopics(List.of(newTopic)).all().get(); // List topics Set<String> topics = admin.listTopics().names().get(); // Describe topic Map<String, TopicDescription> desc = admin.describeTopics(List.of("orders")).all().get(); // Alter topic config admin.alterConfigs(Map.of( new ConfigResource(ConfigResource.Type.TOPIC, "orders"), new Config(List.of(new ConfigEntry("retention.ms", "86400000"))) )).all().get(); // List consumer groups Collection<ConsumerGroupListing> groups = admin.listConsumerGroups().all().get(); // Describe consumer group offsets Map<TopicPartition, OffsetAndMetadata> offsets = admin.listConsumerGroupOffsets("my-group").partitionsToOffsetAndMetadata().get(); } ``` Use cases: Dynamic topic creation at application startup, monitoring dashboards, automated cluster management tools, integration tests (create/delete topics per test), CI/CD pipelines.

Tiered storage (Kafka 3.6+ EA, Confluent GA earlier): Separates hot and cold storage. Recent data stays on broker local disks (hot tier — fast). Older data offloaded to cheap object storage like S3 or GCS (cold tier — slow but cheap). Problem it solves: Kafka retention is constrained by broker disk capacity. Long retention (months/years for compliance, replay, audit) requires enormous and expensive local disk capacity. How it works: 1. Records written to local disk (hot tier) as usual 2. After a configurable delay, broker uploads sealed segments to remote storage (S3) 3. Local copies deleted after remote copy confirmed 4. Consumers reading recent data → served from local disk (fast) 5. Consumers reading old data → fetched from S3 (slower, higher latency) 6. If a consumer seeks to offset 6 months ago, broker fetches from S3 transparently Benefits: • 10-100× cheaper storage (S3 vs SSD): $0.023/GB/month (S3) vs ~$0.20+/GB/month (broker SSD) • Unlimited retention: Keep events for years cost-effectively • Broker scaling independent of retention: Add brokers for throughput, not for storage • Disaster recovery: Remote storage is durable (11 9s durability for S3) Trade-offs: • Higher latency for cold reads (S3 round trips) • Network egress costs for remote reads • More complex operational setup Competitors: Confluent Tiered Storage (GA), WarpStream (S3-native Kafka), Redpanda (tiered storage on S3)

Kafka records have two timestamps: CreateTime (producer-assigned, event time) and LogAppendTime (broker-assigned, ingestion time). Controlled by message.timestamp.type on the topic. Event time vs ingestion time vs processing time: • Event time: When the event actually occurred (e.g., user clicked at 10:00:00). Set by producer. • Ingestion time: When Kafka broker received it (may be later if network lag). Set by broker with LogAppendTime. • Processing time: When the consumer processes it (even later if consumer has lag). Why event time matters: Processing time and ingestion time shift with load, restarts, and consumer lag — not reliable for business semantics. "Revenue per minute" must use the minute the transaction occurred, not when we processed it. Kafka Streams time semantics: By default uses event time (record timestamps). Window operations are based on these timestamps. Late arrivals: Records may arrive out of order (network delays, distributed producers). Kafka Streams handles this with: • grace period: How long to wait for late events in a window before finalizing results. • Window grace(Duration.ofSeconds(30)): Accept records up to 30 seconds late. Watermarks: In Flink (used with Kafka source), watermarks signal "no events with timestamp < T will arrive." Allows window finalization without waiting forever for late events. Timestamp extractor (Kafka Streams): Custom implementation to extract event time from record value (e.g., JSON payload timestamp) rather than record header timestamp.

auto.offset.reset: Controls where a consumer group starts reading when there is no committed offset for a partition. When triggered: • New consumer group (never committed for this topic/partition) • Consumer group's committed offset is older than the topic's retention period (offset expired) • Consumer group is reset manually earliest: Start from the oldest available offset. Consumer reads all historical messages since the beginning of retention. "Give me everything available." latest: Start from the latest offset (the end of the log). Consumer only reads messages produced after it first starts. "Start fresh, ignore history." When to use each: • earliest: Audit processing, backfill pipelines, one-time historical analysis, new feature consuming existing event history. • latest: Real-time processing where historical data is irrelevant (live metrics, notifications), consumer should only react to new events. None: Throw OffsetOutOfRangeException if no committed offset found. Fail loudly — forces explicit handling. Use when consumer should never silently skip or replay data. Common mistake: Two different consumer groups on the same topic with different auto.offset.reset values behave very differently when first started. Document clearly. Reset behavior is per partition: If some partitions have valid committed offsets and others don't (new partitions added), auto.offset.reset only applies to the partitions without valid offsets.

Idempotent producer (enable.idempotence=true, default true in Kafka 3.0+): Guarantees that retried produce requests do not result in duplicate records in the partition. Problem without idempotence: Producer sends record → network timeout → producer doesn't know if broker received it → retries → broker may have received the first attempt → duplicate record in the partition. Solution — PID + sequence number: • Each producer is assigned a Producer ID (PID) by the broker at initialization. • Each record includes a monotonically increasing sequence number per (PID, partition). • Broker tracks the last sequence number seen from each (PID, partition). • If broker receives a record with a sequence number ≤ last seen, it discards it as a duplicate. • If sequence number > last + 1 (gap), broker returns OutOfOrderSequenceException (data loss detected). PID lifetime: PID resets on producer restart. After a restart, duplicate detection history is lost — a retry after restart could produce a duplicate. For true exactly-once across restarts, use transactional.id which provides a stable identity. Implied settings: enable.idempotence=true automatically sets: • acks=all (required for idempotence) • max.in.flight.requests.per.connection=5 (maintained with ordering guarantee) • retries=Integer.MAX_VALUE Scope: Idempotence within a single producer session on a single partition. Not across producers or across restarts (without transactional.id).

Kafka supports several messaging patterns through topic and consumer group configuration. 1. Point-to-Point (Queue): One producer, one consumer group. Each message consumed by exactly one consumer. Achieved by having all consumers in the same group. Standard task distribution pattern — like a work queue. 2. Publish-Subscribe (Fanout): One producer, multiple independent consumer groups. Each group gets all messages. Achieved by having different group IDs. OrderPlaced event → Payment Service (group-1) AND Email Service (group-2) AND Analytics (group-3) all get every event independently. 3. Competing consumers within a group: Multiple consumers in one group. Each partition served by one consumer. Horizontal scale-out of a single logical consumer. The standard scaling pattern. 4. Request-Reply: Requester sends to request-topic with replyTopic header and correlationId. Responder processes and sends to reply-topic. ReplyingKafkaTemplate matches by correlationId. Synchronous-like over async infrastructure. 5. Event streaming (log replay): Consumer seeks to beginning and replays all events. New service bootstraps its state by consuming full topic history. Unique to Kafka (not possible with traditional queues that delete after consume). 6. CQRS with projections: Command side writes events to Kafka. Query side(s) each maintain their own read model by consuming the event stream. Multiple different projections from one event source. 7. Stream processing: One topic feeds a Kafka Streams or Flink application that transforms/enriches and outputs to another topic. Continuous processing pipeline.

Low-latency Kafka — minimizing time from producer.send() to consumer receiving the record. Producer tuning: • linger.ms=0 (default): Send immediately, don't wait for batch to fill. • acks=1: Don't wait for ISR replication (risk: leader crash = data loss). • compression.type=none or lz4: Compression adds CPU time; LZ4 is fastest if you need it. • batch.size small: Smaller batches sent sooner. Broker tuning: • num.replica.fetchers: Increase for faster follower replication. • replica.fetch.min.bytes=1: Don't wait to accumulate before replication. • socket.send.buffer.bytes and socket.receive.buffer.bytes: Match OS TCP buffer sizes. • log.flush.interval.messages and log.flush.interval.ms: Don't force frequent fsyncs — let OS handle (default, relies on replication for durability). Consumer tuning: • fetch.min.bytes=1: Return data immediately, don't wait for min bytes threshold. • fetch.max.wait.ms=0: Don't wait at all if no data. • max.poll.records=1: Process one record at a time (lowest per-record latency, lower throughput). • enable.auto.commit=false + manual commit: Avoid commit delay from auto-commit interval. Infrastructure: • Same AZ: Co-locate producers and brokers. Co-locate brokers and consumers. Network round trip matters. • SSD: Faster fsync, lower I/O latency. • Dedicated broker network interface: No shared bandwidth with other workloads. Typical achievable latency: Sub-10ms end-to-end with tuning.

Producer partitioner: Determines which partition a record is sent to. Default behavior (DefaultPartitioner, Kafka 2.4+): • If record has a key: hash(key) % numPartitions → same key always goes to same partition (sticky by key). Consistent routing enables ordering per key. • If record has no key: Sticky partitioner — accumulate records in one batch per partition, then switch. Reduces number of partitions touched per batch, improving compression and network efficiency. Older behavior (before 2.4): Round-robin for null-key records — spread across all partitions per record. Created many small batches, less efficient. Custom partitioner: ```java public class RegionPartitioner implements Partitioner { public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { String region = ((Order) value).getRegion(); int numPartitions = cluster.partitionCountForTopic(topic); if ("US".equals(region)) return 0; if ("EU".equals(region)) return 1; return 2; } } // Configure: props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RegionPartitioner.class); ``` UniformPartitioner (Kafka 3.3+): New default. Ensures more uniform distribution than StickyPartitioner when many keys map to the same partition due to hash collisions. Warning: Custom partitioners that create hot partitions (most traffic to one partition) limit consumer parallelism. Design for even distribution.

Rack awareness: Distribute partition replicas across different physical racks, availability zones, or data centers to survive rack-level failures. Problem without rack awareness: All replicas of a partition might land on brokers in the same rack. A power failure affecting that rack takes down all replicas → partition unavailable even with replication.factor=3. Configuration: • Each broker sets broker.rack=us-east-1a (or rack ID) • When creating topics (or auto-assigning replicas), Kafka spreads replicas across different rack IDs Algorithm: Kafka assigns replicas across brokers ensuring no two replicas of the same partition are on brokers with the same rack.id value. With 3 AZs and replication.factor=3: one replica per AZ — survives one full AZ failure. Cloud deployments: For AWS, set broker.rack to the EC2 availability zone (us-east-1a, us-east-1b, us-east-1c). Deploy brokers evenly across AZs. Verification: kafka-topics.sh --describe shows the partition-to-broker mapping. Verify replicas are on different rack IDs. Client rack awareness: Consumers can prefer fetching from replicas in the same rack (client.rack=us-east-1a) to reduce cross-AZ network costs. The broker serves the consumer from the closest in-sync replica. Configured with replica.selector.class=RackAwareReplicaSelector on brokers. Cost optimization: In cloud environments, cross-AZ traffic costs money. Consumer rack awareness with replica.selector.class keeps reads within the same AZ.

Schema migration challenge: Kafka events are immutable and retained. Old events with old schemas will always exist alongside new events with new schemas. Strategy 1 — Additive changes only (preferred): Only add optional fields. Never remove or rename. Old consumers ignore new fields (forward compatible). New consumers handle missing fields with defaults (backward compatible). Schema registry enforces BACKWARD compatibility mode. Strategy 2 — Event upcasting: Old events remain in old format. When consuming, apply an upcaster function that transforms old schema to new schema before business logic. Chain upcasters for multiple versions: v1 → v2 → v3. ```java if (event.version == 1) event = upcastV1ToV2(event); if (event.version == 2) event = upcastV2ToV3(event); process(event); // always processes v3 ``` Strategy 3 — New topic for breaking changes: Producer writes to new topic (orders-v2). Old consumers continue reading orders-v1. Migration consumers read both topics during transition. When all consumers migrated, orders-v1 is decommissioned. Strategy 4 — Versioned event types: Include version in event type: com.example.OrderPlaced.v1 vs com.example.OrderPlaced.v2. Consumers subscribe to specific versions. Separate deserialization per version. Best practice: Use Avro with Schema Registry in BACKWARD_TRANSITIVE compatibility mode. This enforces that every new schema version can read ALL previous versions — the safest evolution path.

Kafka quotas: Rate limits applied to producers and consumers at the broker level to prevent noisy neighbors from overwhelming shared Kafka infrastructure. Quota types: • Producer byte-rate quota: Max bytes per second a producer client can produce to all topics combined. • Consumer byte-rate quota: Max bytes per second a consumer client can fetch. • Request rate quota: Max fraction of time (0.0-1.0) the broker spends on requests from a client. Limits CPU usage per client. Configuring quotas: ```bash # Set 10MB/s producer quota for user "order-service" kafka-configs.sh --bootstrap-server localhost:9092 \ --alter --add-config 'producer_byte_rate=10485760' \ --entity-type users --entity-name order-service # Set 5MB/s consumer quota for client-id "analytics-consumer" kafka-configs.sh \ --alter --add-config 'consumer_byte_rate=5242880' \ --entity-type clients --entity-name analytics-consumer ``` Quota enforcement: When a client exceeds its quota, the broker calculates a throttle delay (in ms) and adds it to the response. Client-side SDK respects this delay by pausing before the next request — automatic, transparent throttling. Quota entities: Can be applied to user (SASL principal), client.id, or combination. More specific entity overrides more general: user+client-id > user > client-id > default. Use cases: Prevent a batch job from saturating Kafka bandwidth affecting real-time services. Limit test environments from impacting production clusters.

Tombstone: A Kafka record with a non-null key and a null value on a compacted topic. It signals that the entry for this key should be deleted. How tombstones work with compaction: 1. Produce a tombstone: producer.send(new ProducerRecord<>("user-profiles", userId, null)) 2. The tombstone is written to the partition like any record 3. Consumer sees the tombstone: value is null → interpret as deletion 4. During log compaction: the log cleaner retains the tombstone for delete.retention.ms (default 86400000 = 24 hours) 5. After delete.retention.ms: the tombstone itself is removed Why retain tombstones temporarily: Consumers reading at different positions need to see the deletion signal. If the tombstone were immediately deleted, a slow consumer that hasn't seen it yet would miss the deletion. Consumer handling: ```java for (ConsumerRecord<String, UserProfile> record : records) { if (record.value() == null) { // Tombstone — delete this key from local state cache.remove(record.key()); } else { cache.put(record.key(), record.value()); } } ``` Kafka Streams KTable: Automatically handles tombstones — null value means deletion from the table. Downstream joins and aggregations react accordingly. Pitfall: Produce a tombstone before the compaction cleanup removes the previous record. Tombstone must be produced to the same partition as the original (same key → same partition via partitioner).

Spring Kafka provides @KafkaListener for declarative consumer configuration. Dependency: spring-kafka autoconfigured with spring-boot-starter. Basic listener: ```java @Component public class OrderConsumer { @KafkaListener(topics = "orders", groupId = "order-processor") public void handleOrder(ConsumerRecord<String, Order> record) { log.info("Received order: {} offset: {}", record.value(), record.offset()); orderService.process(record.value()); } } ``` With manual acknowledgment: ```java @KafkaListener(topics = "orders") public void handleOrder(Order order, Acknowledgment ack) { try { orderService.process(order); ack.acknowledge(); // commit offset only on success } catch (RetryableException e) { // don't ack — will be redelivered } } ``` Error handling and DLQ: ```java @Bean public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) { DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template, (r, e) -> new TopicPartition(r.topic() + ".DLT", r.partition())); return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3)); } ``` Configuration in application.yml: ```yaml spring: kafka: consumer: bootstrap-servers: localhost:9092 group-id: order-processor auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring.json.trusted.packages: "com.example.orders" listener: ack-mode: MANUAL ```

Offset commit: Persisting the consumer's current position (how far it has read) so it can resume from that point after a restart or rebalance. What gets committed: offset + 1 (the next record to read, not the last read record). This is a convention — CommitSync({tp: OffsetAndMetadata(record.offset() + 1)}). Commit strategies and their semantics: Auto commit (enable.auto.commit=true): Commits polled offsets every auto.commit.interval.ms (5s default). Risk: records polled but not processed are committed → on crash, lost. Provides at-most-once semantics in failure scenarios. Manual commit after processing: ```java for (ConsumerRecord<K, V> r : records) process(r); consumer.commitSync(); ``` At-least-once: If crash between process and commit, records redelivered on restart. Manual commit before processing: ```java consumer.commitSync(); for (ConsumerRecord<K, V> r : records) process(r); ``` At-most-once: Committed before processed. On crash, records skipped. Per-record commit: ```java for (ConsumerRecord<K, V> r : records) { process(r); consumer.commitSync(Map.of(tp, new OffsetAndMetadata(r.offset() + 1))); } ``` Higher durability (commits after each record) but lower throughput. Rebalance-safe commit: Commit in onPartitionsRevoked() callback before rebalance reassigns partitions. Transactional commit: sendOffsetsToTransaction() — atomic with producer transaction. Exactly-once within Kafka.

Kafka topic naming conventions — no enforced standard, but these patterns are widely adopted: Common patterns: • kebab-case: user-events, order-placed, payment-processed. Most common. Kafka CLI friendly. • dot-notation (domain-driven): com.example.orders.placed, io.company.payments.v2. Mirrors Java package naming. Groups related topics. • prefix by team/domain: payments.transactions, orders.fulfillment, users.registrations. • env prefix: prod.orders, staging.orders — separates environments on shared clusters (not recommended for production — use separate clusters instead). Include these components: • Entity/domain: orders, users, payments • Event/action: created, updated, cancelled • Version (if needed): .v2 suffix or /v2 directory convention Examples: • order-created • payment-processed.v2 • user-profile-updated • inventory-reserved • shipment-dispatched.DLT (dead letter topic convention) Avoid: • Underscores: Some metrics systems have issues with dots and underscores mixed. Dots are used in JMX metrics — avoid if possible to prevent metric name conflicts. Hyphens are safe. • Very long names: Hard to read in monitoring dashboards. • Generic names (events, messages): Too vague, doesn't convey purpose. • Including sensitive info in topic names: Names are visible in cluster metadata. Versioning convention: topics ending in .DLT = dead letter topic. topics ending in .v2 = schema version 2.

Kafka comes with built-in performance testing tools: Producer performance test: ```bash kafka-producer-perf-test.sh \ --topic benchmark-topic \ --num-records 10000000 \ --record-size 1024 \ --throughput -1 \ --producer-props bootstrap.servers=localhost:9092 acks=1 compression.type=lz4 ``` --throughput -1: No rate limit (max throughput). Set to 100000 to test at 100K records/sec. Consumer performance test: ```bash kafka-consumer-perf-test.sh \ --topic benchmark-topic \ --messages 10000000 \ --bootstrap-server localhost:9092 ``` Key metrics to measure: • MB/sec throughput • Records/sec • Average latency (ms) and P99 latency • Producer: request-latency-avg, batch-size-avg, compression-rate End-to-end latency test: OpenMessaging Benchmark (OMB): Industry-standard benchmark for messaging systems. Tests real-world producer + consumer scenarios. Results comparable across systems (Kafka vs Pulsar vs RabbitMQ). Capacity planning benchmark: Test with production-like message size, compression, replication factor, and ACK settings. Synthetic benchmarks with tiny messages or acks=0 are misleading. Disk throughput baseline: fio or dd to measure raw disk I/O capacity before blaming Kafka. Network: iperf3 between broker nodes and client machines. Kafka is network-intensive — verify 10Gbps+ between brokers.

Both are distributed messaging/streaming platforms, but with different architectural philosophies. Kafka: • Architecture: Brokers store data locally on disk. Brokers are coupled (compute + storage together). • Scaling: Scale by adding brokers. Partition rebalancing required when adding brokers. • Tiered storage: Available in newer versions (3.6+, Confluent GA earlier). • Protocol: Custom binary TCP protocol. • Ecosystem: Kafka Streams, Kafka Connect, MirrorMaker, large ecosystem. • Maturity: Very mature. Battle-tested at massive scale (LinkedIn, Netflix, Uber). Apache Pulsar: • Architecture: Compute-storage separation. Brokers are stateless. BookKeeper handles storage (ledger segments). Scale compute (brokers) and storage (bookies) independently. • Scaling: Add stateless brokers without data rebalancing. Add BookKeeper nodes for storage. • Multi-tenancy: Native tenant/namespace/topic hierarchy. Quota and isolation per tenant built-in. • Geo-replication: Built-in, easier to configure than Kafka MirrorMaker. • Functions: Pulsar Functions (lightweight serverless compute) built in, similar to Kafka Streams. • Protocol: AMQP, MQTT, Kafka protocol (via Kafka-on-Pulsar), native Pulsar protocol. Choose Kafka: Larger ecosystem, more tooling, better documentation, proven at hyper-scale. Most engineers already know it. Choose Pulsar: Need compute-storage separation for independent scaling, strong multi-tenancy, easier geo-replication.

Spring Kafka KafkaTemplate for type-safe, managed Kafka producer. Dependency: spring-kafka (included in spring-boot-starter). Configuration: ```yaml spring: kafka: producer: bootstrap-servers: localhost:9092 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer acks: all retries: 3 properties: enable.idempotence: true spring.json.add.type.headers: false ``` Producing messages: ```java @Service public class OrderEventPublisher { private final KafkaTemplate<String, OrderEvent> kafkaTemplate; public void publish(OrderEvent event) { // Fire and forget kafkaTemplate.send("orders", event.getOrderId(), event); // With callback kafkaTemplate.send("orders", event.getOrderId(), event) .addCallback( result -> log.info("Sent offset: {}", result.getRecordMetadata().offset()), ex -> log.error("Failed to send: {}", event, ex) ); // With CompletableFuture (Spring Kafka 3.0+) CompletableFuture<SendResult<String, OrderEvent>> future = kafkaTemplate.send("orders", event.getOrderId(), event); future.whenComplete((result, ex) -> { if (ex == null) log.info("Sent: {}", result.getRecordMetadata()); else log.error("Failed", ex); }); } } ``` Transactional producer: ```java @Transactional("kafkaTransactionManager") public void publishInTransaction(OrderEvent event) { kafkaTemplate.send("orders", event); kafkaTemplate.send("audit-log", event); // Both sent atomically } ```

Kafka Connector: A reusable, configurable component that integrates Kafka with external systems. Part of Kafka Connect framework. No custom producer/consumer code needed — declare what to connect and how. Types: • Source connector: Pulls data from an external system and produces to Kafka. Example: Debezium (DB → Kafka), S3 source (S3 → Kafka), REST API source. • Sink connector: Consumes from Kafka and writes to an external system. Example: Elasticsearch sink, S3 sink, JDBC sink, BigQuery sink. Connector architecture: • Connector: Top-level configuration and task coordinator. • Task: Unit of work that does actual data movement. Multiple tasks run in parallel for throughput. • Worker: Kafka Connect process that hosts connectors and tasks. Can be standalone (single process) or distributed (cluster of workers). Deploying a connector (REST API): ```bash curl -X POST http://kafka-connect:8083/connectors \ -H "Content-Type: application/json" \ -d '{ "name": "elasticsearch-sink", "config": { "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector", "tasks.max": "3", "topics": "orders,products", "connection.url": "http://elasticsearch:9200", "type.name": "_doc", "key.ignore": "true" } }' ``` Ecosystem: 200+ connectors on Confluent Hub (official marketplace). Popular: Debezium, S3, Elasticsearch, JDBC, MongoDB, Snowflake, Google BigQuery, Salesforce. Scaling: Increase tasks.max for more parallelism. Distributed Connect cluster handles failover automatically.

Fetch protocol: How Kafka consumers retrieve records from brokers. Understanding this enables precise throughput and latency tuning. Fetch request flow: 1. Consumer sends FetchRequest to the leader of each assigned partition. 2. Request includes: topicPartitions, fetchOffset (where to start reading), maxBytes, minBytes, maxWait. 3. Broker reads from log starting at fetchOffset. Returns up to maxBytes of records. 4. If data < minBytes: broker holds request up to maxWait before responding (long polling). 5. Consumer processes returned records, updates local offset, polls again. Key configurations: • fetch.min.bytes (default 1): Broker waits until this many bytes available. Higher = larger batches, more latency. fetch.min.bytes=50000 means broker waits for 50KB before responding. • fetch.max.wait.ms (default 500ms): Max time broker holds an open fetch request. After this, responds even if below fetch.min.bytes. • max.partition.fetch.bytes (default 1MB): Max bytes returned per partition per fetch. Must be >= largest message size. • fetch.max.bytes (default 50MB): Max total bytes returned across all partitions per fetch request. Fetcher thread: Consumer has a background fetcher thread that maintains a queue of pre-fetched records. poll() drains from this queue — enables pipelining of fetch and processing. Fetch from replica: With rack-aware consumer and replica.selector.class on broker, consumer fetches from nearest replica (same AZ) instead of always leader — reduces cross-AZ latency and cost.

Classic (eager) rebalance: When ANY consumer joins or leaves the group, ALL consumers stop, revoke ALL their partitions, and the group coordinator reassigns everything. Full stop-the-world. Simple but wasteful — even unaffected partitions lose their consumers temporarily. Cooperative (incremental) rebalance (Kafka 2.4+): Only the affected partitions are moved. Unaffected consumers continue processing their partitions throughout the rebalance. Protocol: 1. Consumer joins or leaves → coordinator triggers rebalance 2. First rebalance round: All consumers report their current partitions. Coordinator computes which partitions need to move. 3. Coordinator tells only the consumers that need to release some partitions to revoke them. 4. Second round: Those consumers revoke their assigned partitions. Coordinator assigns freed partitions to new owners. 5. Consumers that didn't have to revoke anything NEVER stopped. Result: A 10-consumer group where one consumer leaves → only ~1 partition moved. The other 9 consumers continue processing without pause. Configuration: ```properties partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor ``` StickyAssignor: Preserves assignments from previous rebalance as much as possible. Minimizes partition movement. But still uses eager protocol. CooperativeStickyAssignor: Same stickiness + incremental cooperative rebalance. Best of both worlds. Recommended for production (Kafka 2.4+). Migration: Can migrate from eager to cooperative without full group restart by including both assignors in the config list temporarily.

Log cleaner: Background threads (log.cleaner.threads, default 1) that perform log compaction on topics with cleanup.policy=compact. Compaction goal: For each key in a compacted topic, retain only the latest record. Older records with the same key are removed. Segments: "Clean" (previously compacted), "Dirty" (records added since last compaction). Cleaner targets dirty segments. Cleaning algorithm: 1. Build an offset map: For every key in dirty segments, record its highest offset (the "winning" record). 2. Scan dirty segments again: If a record's offset < the winning offset for its key, it's a duplicate → skip/remove it. 3. Write cleaned records to new segment files. 4. Replace dirty segments with cleaned segments atomically. Cleanable ratio: min.cleanable.dirty.ratio (default 0.5): Compaction runs when dirty ratio > 50% of total log size. Higher ratio = less frequent compaction (larger dirty sections accumulate). Compaction guarantees: • At least the last value for each key is retained (head of log may still be uncompacted) • Ordering within a key is preserved • Compaction is not instantaneous — there is always an uncompacted "tail" (active segment) min.compaction.lag.ms: Records must be at least this old before eligible for compaction (default 0 = compact immediately). max.compaction.lag.ms: Guarantees compaction happens within this time for any dirty record. Compaction + delete: cleanup.policy=compact,delete — compacts AND respects retention period. Old compacted segments are deleted when older than retention.

Systematic approach to debugging Kafka consumer problems: Step 1 — Check consumer lag: ```bash kafka-consumer-groups.sh --bootstrap-server localhost:9092 \ --describe --group my-group # Shows: TOPIC, PARTITION, CURRENT-OFFSET, LOG-END-OFFSET, LAG, CONSUMER-ID ``` High lag = consumer falling behind. Zero lag but not processing = consumer isn't running. Step 2 — Check if consumer is running: Look at CONSUMER-ID column. Empty = consumer not running. Multiple = scaling OK. Step 3 — Check consumer logs: Look for: org.apache.kafka.clients.consumer rebalance events, commit failures, deserialization errors, processing exceptions. Step 4 — Check for poison pill: Consumer repeatedly throws exception on same partition/offset. Offset not advancing. Fix: implement DLQ or seek past the bad message. Step 5 — Check rebalance frequency: Frequent rebalances in logs? → Increase session.timeout.ms. Consumer taking too long between polls? → Increase max.poll.interval.ms or reduce max.poll.records. Step 6 — Metric inspection: • records-lag: Should be near 0 for real-time • fetch-latency-avg: Should be low • commit-latency-avg: Should be consistent Step 7 — Thread dump: If consumer is hung (not calling poll), take a thread dump (jstack PID). Look for consumer thread blocked on processing, DB call, or lock. Step 8 — Reset offsets (last resort): ```bash kafka-consumer-groups.sh --reset-offsets --group my-group \ --topic orders --to-datetime 2024-01-15T10:00:00.000 --execute ```

min.insync.replicas (min.isr): Broker-side configuration (set on the topic or broker default) that specifies the minimum number of in-sync replicas that must acknowledge a produce request when acks=all. Interaction: • Only relevant when acks=all (or acks=-1). Ignored for acks=0 or acks=1. • If ISR count < min.isr → broker returns NotEnoughReplicasException to producer → producer must handle this (retry or fail). • Effectively: acks=all means "wait for min.isr replicas" not "all existing replicas." Typical production config (3 replicas): • replication.factor=3: 3 copies of each partition • min.insync.replicas=2: At least 2 must ack • acks=all: Producer waits for all ISR (which must be ≥ 2) With this config: Can tolerate 1 broker failure (ISR drops to 2, still meets min.isr=2). Produces fail if 2 brokers down simultaneously (ISR=1 < min.isr=2 → NotEnoughReplicasException). This is correct behavior — with only 1 replica, committing is risky. Trade-off: Higher min.isr = better durability but lower availability. If min.isr = replication.factor (e.g., min.isr=3, RF=3), ALL replicas must be in sync for produces to succeed — very brittle (any broker failure blocks writes). Common mistake: Setting min.insync.replicas but using acks=1 — min.isr has no effect. Must use acks=all. Monitor: ISRShrinkRateAndTimeMs JMX metric. Alert if ISR shrinks frequently — indicates unhealthy replicas.

Kafka's role in data lakehouse: The real-time ingestion layer that captures events from operational systems and delivers them to the lakehouse storage layer. Typical data lakehouse with Kafka: Ingestion: Operational systems → Kafka topics (via application events or Debezium CDC). Kafka acts as the real-time event bus. Stream processing: Flink or Kafka Streams reads from Kafka, performs enrichment/transformation/aggregation, writes to Kafka output topics or directly to lakehouse storage. Storage: Apache Iceberg, Apache Hudi, or Delta Lake on S3/ADLS/GCS. These table formats support ACID transactions, schema evolution, and time travel on object storage. Flink or Spark writes structured streaming output in Iceberg/Hudi/Delta format. Query layer: Presto, Trino, Apache Spark SQL, or Athena queries the data lake tables with SQL. Combines recent streaming data + historical batch data. Kafka → Iceberg example with Flink: • Kafka source → Flink job → Iceberg sink (writes Parquet files to S3 with Iceberg metadata) • Trino queries Iceberg table = reads all historical + recent data with full ACID guarantees Why this matters: • Eliminates the batch/streaming divide (Lambda architecture complexity) • Kafka provides real-time freshness; lakehouse provides query power • One storage layer serves both streaming (Kafka) and batch (Spark/SQL) access patterns • Kafka Streams can materialize aggregations to Iceberg for operational dashboards

Log flush: Writing dirty pages from OS page cache to disk (fsync). Controls when Kafka's data is durably written to disk beyond the OS buffer. Flush configurations: • log.flush.interval.messages: Flush after every N messages (per partition). Default: Long.MAX_VALUE (virtually never — rely on replication). • log.flush.interval.ms: Flush every N milliseconds. Default: Long.MAX_VALUE. Kafka's durability philosophy: Don't rely on disk flush for durability — rely on REPLICATION. With replication.factor=3 and acks=all, data exists on 3 brokers' page caches before the producer gets an ack. Even if all 3 lose power simultaneously (rare), you need 3 simultaneous machine failures before losing committed data. Risks of frequent flushing: • fsync is expensive — serializes I/O, limits throughput significantly • Kafka's sequential write performance advantage disappears • Negates the benefit of OS page cache buffering When to consider explicit flushing: • Single broker setup (no replication): Flush more frequently as failover to replica isn't available. • Extreme durability requirements where replication isn't considered sufficient. • Testing environment where you want predictable disk state. After crash without recent flush: Broker recovers by replaying redo log (segment files are consistent) + fetches missed records from replicas. This is the designed recovery path — not fsync-based crash safety. Recommendation: Leave flush at defaults (rely on replication). Add disk monitoring and set up RAID or NVMe SSDs if disk hardware failure is a concern.

Kafka-based Saga: Use Kafka topics as the communication channel between saga steps. Each step listens to a command/event topic and publishes results. Orchestrated Saga with Kafka: Saga orchestrator publishes commands to service-specific topics: ``` ordersaga-commands → [PaymentService, InventoryService, ShippingService] ``` Each service processes its command and publishes result: ``` payment-results → SagaOrchestrator inventory-results → SagaOrchestrator ``` Orchestrator state machine: ```java @KafkaListener(topics = "payment-results") public void handlePaymentResult(PaymentResult result) { OrderSaga saga = sagaRepository.findById(result.getSagaId()); if (result.isSuccess()) { saga.setState(SagaState.PAYMENT_DONE); kafkaTemplate.send("inventory-commands", new ReserveInventory(saga.getOrderId(), saga.getItems())); } else { saga.setState(SagaState.FAILED); // no compensation needed if payment failed } sagaRepository.save(saga); } ``` Compensation: On failure, orchestrator publishes compensation commands: ```java kafkaTemplate.send("payment-commands", new RefundPayment(sagaId, amount)); ``` Durability: Kafka retains all command/result events. On orchestrator restart, it can replay from committed offset. Saga state is stored in DB — survives restarts. Idempotency: Each service must handle duplicate commands (Kafka at-least-once). Use event_id for deduplication. Alternative: Use Temporal for complex sagas — it wraps this Kafka coordination in a durable workflow engine.

These are the same partitions — there is no separate "consumer group partition." The distinction is about how partitions relate to consumer groups. Partitions (of a topic): Fixed physical divisions of a topic's data. Set at topic creation. Each partition is an ordered log. Example: orders topic with 12 partitions. Consumer group membership: A consumer group subscribes to a topic. The group coordinator assigns partitions from that topic to consumers in the group. Each partition goes to exactly ONE consumer in the group. Assignment example — 12 partitions, 3 consumers in group-A: • Consumer-1: partitions 0,1,2,3 • Consumer-2: partitions 4,5,6,7 • Consumer-3: partitions 8,9,10,11 Multiple groups — same partitions, independent consumption: • Group-A (payment-service): all 12 partitions assigned to its consumers • Group-B (email-service): all 12 partitions assigned to its consumers independently Both groups read all messages, progress independently. Scaling limits: • More consumers than partitions in a group → excess consumers idle • More partitions → more parallelism possible (add more consumers to match) Consumer group rebalance: When consumer group membership changes, the partition-to-consumer assignment is recomputed. Partitions don't change — only their assignment within the group. Practical implication: Size your partition count based on target peak consumer parallelism. You can always add consumers (up to partition count) but partition count changes are harder.

ksqlDB: A database purpose-built for stream processing using SQL syntax. Runs on top of Kafka. Allows querying, filtering, aggregating, and joining Kafka topics with SQL-like statements — without writing Java code. Core concepts: • Stream: Unbounded sequence of events (maps to a Kafka topic). Each row is an independent event. CREATE STREAM. • Table: Current state per key (maps to a compacted Kafka topic). Latest value per key wins. CREATE TABLE. • Persistent query: Continuously running SQL that processes incoming events and writes results to a new topic/table. • Pull query: Synchronous point-in-time query against a materialized table (like a SELECT against a DB). For request-response patterns. Examples: ```sql -- Create a stream from a Kafka topic CREATE STREAM orders (id VARCHAR KEY, amount DOUBLE, region VARCHAR) WITH (kafka_topic='orders', value_format='json'); -- Aggregate: orders per minute per region CREATE TABLE orders_per_minute AS SELECT region, COUNT(*) as order_count FROM orders WINDOW TUMBLING (SIZE 1 MINUTE) GROUP BY region EMIT CHANGES; -- Filter: high-value orders to new topic CREATE STREAM high_value_orders AS SELECT * FROM orders WHERE amount > 1000; ``` Deployment: ksqlDB server cluster. REST API and CLI for management. Scales horizontally. Use cases: Real-time dashboards, anomaly detection rules, enrichment pipelines, data preparation for downstream sinks — all without custom Kafka Streams code.

Default limits: • message.max.bytes (broker): 1MB default. Largest message the broker will accept. • max.request.size (producer): 1MB default. Largest request the producer will send. • max.partition.fetch.bytes (consumer): 1MB default. Max bytes fetched per partition per request. All three must be consistent — broker limit and producer/consumer limits must be aligned. Problems with large messages in Kafka: • Memory pressure: Large messages held in buffers consume more broker and consumer memory. • Replication: Large messages are replicated as-is — consume follower fetch capacity. • Consumer processing: Large messages slow down the poll loop. • GC: Large byte arrays on JVM heap cause GC pressure. Options for handling large payloads: 1. Increase limits (simplest): Set message.max.bytes, max.request.size, max.partition.fetch.bytes all to the same larger value (e.g., 10MB). Only feasible for occasional large messages — not for consistently large payloads. 2. Compression: Enable ZSTD/LZ4 compression. Large text/JSON payloads often compress 5-10×. Stays within 1MB limit while carrying more data. 3. Claim Check pattern (recommended for large payloads): Store large payload in S3. Send only the S3 reference in Kafka. Consumer fetches from S3 on demand. Decouples message size from Kafka constraints entirely. 4. Split large events: Break a large batch event into multiple individual events. Natural Kafka pattern. 5. Chunking: Split payload into fixed-size chunks, each a Kafka message with chunk index metadata. Consumer reassembles. Complex — use Claim Check instead.

Producer callbacks: Asynchronous notification when a produce request completes (success or failure). Called on the I/O thread after the broker responds. Using callbacks: ```java producer.send(record, (metadata, exception) -> { if (exception == null) { // Success log.info("Sent to topic={} partition={} offset={}", metadata.topic(), metadata.partition(), metadata.offset()); } else { // Failure log.error("Failed to send record: {}", record, exception); // Handle: retry, DLQ, alert } }); ``` Callback semantics: • Called asynchronously on the Kafka I/O thread — do NOT block in the callback (no DB calls, no heavy computation). Blocking the I/O thread stalls ALL sends. • Called in partition-order — callbacks for the same partition are called in the order records were sent. • Called even for retriable errors that were retried internally (only called once, after final result). Common patterns: • Log success with offset for auditing • Track send metrics (success count, failure count) • On failure: publish to DLQ, trigger alert, update DB status • On success: update DB status to "sent to Kafka" CompletableFuture approach (Spring Kafka 3.0+): ```java kafkaTemplate.send("orders", key, value) .whenComplete((result, ex) -> { if (ex != null) handleFailure(ex); else log.info("offset: {}", result.getRecordMetadata().offset()); }); ``` Not to confuse with: producer.send().get() — this blocks the calling thread until complete (synchronous). Defeats async purpose. Use only in tests.

These are fundamentally different concepts — the word "replication" means different things in each context. Topic replication (data replication): Copies of a partition's data across multiple brokers. Controlled by replication.factor. Provides fault tolerance — if one broker dies, another has the full data. Example: orders partition-0 with replication.factor=3: One leader + two followers. All three have identical data. Producers/consumers use the leader. If leader dies, a follower is elected leader automatically. Consumer group replication (HA for consumers): Running multiple consumer instances in the same group. Each partition assigned to one consumer at a time. If a consumer dies, its partitions are reassigned to surviving consumers during rebalance. Example: 3 consumers in group payment-service for orders topic (12 partitions). Consumer-1 dies → rebalance → Consumer-2 and Consumer-3 take over its partitions. No data loss (Kafka retains data). Processing resumes from committed offset. They work together: • Topic replication ensures data isn't lost if a broker fails • Consumer group ensures processing continues if a consumer instance fails • Both needed for a highly available, fault-tolerant Kafka pipeline Another analogy: Topic replication = RAID for your data. Consumer group = having multiple workers — if one worker quits, others pick up the work. No overlap: Topic replication is broker-side. Consumer group HA is client-side. Separate configuration, separate failure domains.

Multi-datacenter (multi-region) Kafka: Replicate data between geographically separated Kafka clusters. Approaches: 1. Active-Passive (DR setup): One primary cluster in DC1 handles all writes. MirrorMaker 2 (or Confluent Replicator) replicates all topics to DC2 passively. On disaster, fail over to DC2. Consumer groups may need offset translation (MM2 provides this). 2. Active-Active: Both clusters accept writes. Bidirectional replication via MM2. Risk: concurrent writes to the same key produce conflicts — handle with last-write-wins (timestamp) or event merging logic. Topics prefixed by DC to avoid duplication loops (DC1 replicates dc1.orders, DC2 replicates dc2.orders — each only consumes the other's prefix). 3. Hub and Spoke: Multiple regional clusters replicate to a central cluster. Central cluster aggregates all events for cross-region analytics. Regional clusters handle local producers/consumers at low latency. MirrorMaker 2 configuration: ```properties clusters=dc1, dc2 dc1.bootstrap.servers=dc1-kafka:9092 dc2.bootstrap.servers=dc2-kafka:9092 dc1->dc2.enabled=true dc1->dc2.topics=orders,payments ``` Challenges: • Network latency: Cross-DC replication adds latency. Don't make cross-DC Kafka calls in critical user request paths. • Offset management: Offsets in DC1 ≠ offsets in DC2. MM2 tracks the mapping. On failover, consumers use offset translation. • Throughput cost: Replicating all data × network bandwidth × egress cost. • Consistency: Active-active has no strong consistency. Design for eventual consistency. Confluentβ Cluster Linking: More efficient than MM2 — reads directly from source cluster without intermediate consumers.

Leader epoch: A monotonically increasing counter that tracks the current generation of a partition's leader. Incremented every time a new leader is elected for a partition. Purpose: Detect and resolve data divergence after leader failures and prevent "zombie" leaders from corrupting data. Problem it solves: Before leader epochs, a follower that became the new leader might have truncated records that other followers had — causing data divergence. With leader epochs, followers can verify they're synchronized with the current legitimate leader, not a stale one. How it works: • Each message batch includes the leader epoch when it was written. • On follower startup or reconnection, it asks the leader for the offset of the end of the current leader epoch's log. • Follower compares its own log against the epoch boundary. If it has records from a higher epoch (zombie scenario), it truncates to be consistent with the current leader. • Followers can also detect they've been replicated stale data and reject it. Fencing zombie producers: Transactional producers include the leader epoch. If a broker receives a produce request from an old producer epoch (transactional.id was assigned a new epoch — "zombie" producer), it rejects the request. Prevents a slow producer from overwriting committed records. Leader epoch vs offset high watermark: High watermark was the original mechanism for follower catch-up. Leader epochs provide a more precise, epoch-scoped boundary that handles edge cases the HW mechanism didn't handle correctly.

group.id: The consumer group identifier. All consumer instances with the same group.id form a single consumer group and share partition assignments. Behavior based on group.id: Same group.id → competing consumers (work distribution): • 3 instances with group.id="payment-processor" • Kafka assigns partitions among them: instance-1 gets partitions 0-3, instance-2 gets 4-7, instance-3 gets 8-11 • Each message processed by exactly ONE instance • Horizontal scaling of processing Different group.id → independent consumers (fan-out): • group.id="payment-processor" gets all messages • group.id="email-service" independently gets all messages • group.id="analytics" independently gets all messages • Same message processed by ALL consumer groups Null group.id: Possible in Kafka — consumer without group. Must manually assign partitions (consumer.assign()). No group membership, no rebalancing, no coordinator. Used for specific offset management or admin tooling. Never use in normal application consumers. Choosing group.id: • One group per logical consumer application/service • Use meaningful names: "order-service-payment-processor" not "group-1" • Include service name for debugging: "notification-service-order-events" Group cleanup: Offsets retained for offsets.retention.minutes after the group has no active consumers. After this, auto.offset.reset applies when the group is restarted.

High watermark (HW): The offset up to which all in-sync replicas have replicated records. Consumers can only read records up to the high watermark — records above it are not yet confirmed as replicated. Why HW exists: Prevent consumers from reading data that might be lost if the leader fails before replication. Example: • Leader has offset 0-9 (10 records) • Follower-1 has replicated up to offset 7 • Follower-2 has replicated up to offset 8 • ISR = [leader, follower-1, follower-2] • High watermark = 7 (lowest of all ISR members) • Consumer can only read offsets 0-7 • If leader fails now, follower-1 or follower-2 becomes leader. Worst case: offsets 8-9 never reach new leader. HW prevents consumers from seeing data that may be lost. HW advancement: Followers send FetchRequest to leader. Leader tracks the fetchOffset of each follower. HW = min(all ISR fetchOffsets). As followers catch up, HW advances. Log End Offset (LEO): The next offset to be written. Always > HW unless no in-flight records. Consumer sees: Records up to HW. "Uncommitted" records (HW < offset ≤ LEO) invisible to consumers (in read_committed mode they're invisible; in read_uncommitted they may be visible but could be rolled back). Leader epoch + HW: Together they determine the safe read boundary. HW is partition-level; leader epoch provides generation context.

Simplest local Kafka setup using Docker Compose: ```yaml # docker-compose.yml services: zookeeper: image: confluentinc/cp-zookeeper:7.5.0 environment: ZOOKEEPER_CLIENT_PORT: 2181 kafka: image: confluentinc/cp-kafka:7.5.0 depends_on: [zookeeper] ports: - "9092:9092" environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true" kafka-ui: image: provectuslabs/kafka-ui:latest ports: - "8080:8080" environment: KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092 ``` KRaft mode (no ZooKeeper, Kafka 3.4+): ```yaml kafka: image: confluentinc/cp-kafka:7.5.0 environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk # must be base64 UUID ``` Testcontainers (integration testing): ```java @Testcontainers class KafkaIntegrationTest { @Container KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.0")); @DynamicPropertySource static void kafkaProperties(DynamicPropertyRegistry r) { r.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers); } } ```

Two separate timers control consumer liveness detection: heartbeat.interval.ms (default 3000ms): How often the consumer sends a heartbeat to the group coordinator. The heartbeat is a lightweight "I'm alive" signal sent by a dedicated heartbeat thread (separate from the poll thread). This thread runs independently of processing. session.timeout.ms (default 45000ms): If the coordinator doesn't receive a heartbeat within this window, it declares the consumer dead and triggers a rebalance. Must be >= 3 × heartbeat.interval.ms. Kafka recommends 3 heartbeats per session timeout. Two ways a consumer can "miss" heartbeats: 1. Consumer process crashed: Heartbeat thread stops. After session.timeout.ms, coordinator detects death. 2. Consumer alive but poll() not called within max.poll.interval.ms (default 5 minutes): The consumer is alive (heartbeats succeed) but not making progress. Coordinator forces it to leave the group. This handles "zombie" consumers that are alive but stuck processing. Configuration guidance: • Decrease session.timeout.ms (e.g., 15s) for faster failure detection — tolerate latency spikes less. • Increase max.poll.interval.ms if processing takes > 5 minutes (large batch processing, slow DB writes). • heartbeat.interval.ms = session.timeout.ms / 3 as a rule. Common mistake: Slow processing in the poll loop delays the next poll() → exceeds max.poll.interval.ms → rebalance → consumer rejoins → same partition assigned → infinite rebalance loop. Fix: reduce max.poll.records or increase max.poll.interval.ms.

Changelog topic: An internal Kafka topic that Kafka Streams uses to back its stateful stores. Every state update (insert/update/delete to a RocksDB state store) is mirrored to the changelog topic. Purpose — fault tolerance and recovery: • If a Kafka Streams instance crashes, its local RocksDB state store is lost. • On restart, Kafka Streams reads the changelog topic from the beginning (or from a recent snapshot offset) to rebuild the state store before resuming processing. • This ensures stateful aggregations survive instance failures. Naming convention: [application-id]-[store-name]-changelog. Example: order-analytics-order-counts-changelog. Compaction: Changelog topics use log compaction (cleanup.policy=compact). Only latest value per key retained — efficient restoration (no need to replay full history, just latest state). Standby replicas (num.standby.replicas): • num.standby.replicas=1: One additional Kafka Streams task per state store partition pre-fetches the changelog and maintains a warm copy of the state. • On failure, the standby immediately takes over with a nearly up-to-date state — minimal restore time. • Trade-off: Extra compute and Kafka fetch traffic for standby tasks. Interactive queries: Kafka Streams exposes state stores for query via a REST endpoint pattern. Streams RPC layer allows querying any instance's state (routing to the instance with the relevant partition). Deleting changelog topics: When you delete a Kafka Streams application's topics (kafka-streams-application.sh --cleanup), changelog topics are also cleaned up.

Running out of disk space on a Kafka broker is a critical incident. The consequences and recovery steps: What happens: • New incoming messages fail to write → IOError → broker stops accepting produces for affected partitions. • Producers receive errors (CONNECTION_RESET, LEADER_NOT_AVAILABLE) and retry. • Followers can't replicate → partitions become under-replicated. • In severe cases, the broker process may crash. Why it happens: • Insufficient retention management: too many topics, too long retention, high throughput. • Log compaction not keeping up with writes. • Unexpected traffic spike. • Temp files, GC logs, heap dumps consuming disk. Immediate remediation: 1. Identify which directories are full: df -h, du -sh /kafka/data/topics/* 2. Reduce retention temporarily: kafka-configs.sh --alter --add-config retention.ms=3600000 --entity-name high-volume-topic (1 hour) to trigger immediate segment deletion. 3. Or: Manually delete old log segments (DANGEROUS — must stop the broker first, delete sealed segments, restart). 4. Expand disk: Add/mount more storage. Move Kafka data directory to larger volume (requires broker restart). 5. Add a new broker and reassign partitions to spread load. Prevention: • Alert at 70-75% disk utilization (leave buffer for compaction temp files). • Use tiered storage for long-retention topics. • Monitor disk usage per topic: du -sh /kafka/data/[topic-partition]/ • Set segment retention appropriately — don't default to "keep everything forever."

Kafka supports two types of configuration: static (in server.properties, requires restart) and dynamic (updated at runtime via AdminClient or CLI, takes effect immediately). Dynamic configuration: Some configurations can be changed without broker restart. Stored in ZooKeeper (classic mode) or the metadata quorum (KRaft mode) and propagated to brokers. Updating topic config dynamically: ```bash # Change retention to 24 hours kafka-configs.sh --bootstrap-server localhost:9092 \ --alter \ --entity-type topics \ --entity-name orders \ --add-config retention.ms=86400000 # Remove a config override (revert to broker default) kafka-configs.sh --alter --entity-type topics --entity-name orders \ --delete-config retention.ms # Describe current config kafka-configs.sh --describe --entity-type topics --entity-name orders ``` Updating broker config dynamically: ```bash # Change log cleaner threads across all brokers kafka-configs.sh --bootstrap-server localhost:9092 \ --alter --entity-type brokers --entity-default \ --add-config log.cleaner.threads=4 # Change for a specific broker kafka-configs.sh --alter --entity-type brokers --entity-name 1 \ --add-config log.retention.ms=604800000 ``` Dynamically configurable examples: retention.ms, max.message.bytes, compression.type (topic-level). log.cleaner.threads, num.recovery.threads.per.data.dir (broker-level). NOT dynamically configurable (requires restart): listeners, log.dirs, num.partitions, zookeeper.connect. Verification: After changing, verify with --describe and monitor broker logs for application of the change.

Seek API: Allows consumers to manually position their offset to any position in a partition, overriding the committed offset. Enables replay, skip, and point-in-time recovery. Seek methods: ```java consumer.seek(topicPartition, offset); // seek to specific offset consumer.seekToBeginning(partitions); // seek to earliest available offset consumer.seekToEnd(partitions); // seek to latest offset (skip all current) ``` Usage patterns: Replay from beginning: Re-process all messages after a bug fix. ```java consumer.subscribe(topics, new ConsumerRebalanceListener() { public void onPartitionsAssigned(Collection<TopicPartition> partitions) { consumer.seekToBeginning(partitions); } }); ``` Seek to timestamp: Find the first offset with timestamp >= target. ```java Map<TopicPartition, Long> query = Map.of(tp, targetTimestamp); Map<TopicPartition, OffsetAndTimestamp> result = consumer.offsetsForTimes(query); consumer.seek(tp, result.get(tp).offset()); ``` Skip a poison pill: Seek past a bad message. ```java consumer.seek(record.partition(), record.offset() + 1); ``` When seek takes effect: After the next poll() call. The seek is buffered — doesn't take effect immediately. Caution: Seeking to an offset before the retention window throws OffsetOutOfRangeException. Handle with auto.offset.reset=earliest or explicit fallback logic. Spring Kafka: @KafkaListener with seekToBeginning=true, or inject KafkaListenerEndpointRegistry + seekToBeginning() via the listener container.