Kafka Producers & Consumers

Producer send() flow:
  1. Serialize key and value (using configured serializer)
  2. Determine target partition:
     - If key is set: hash(key) % numPartitions
     - If key is null: sticky partitioner (batch to one, then rotate)
     - If partition is explicitly set: use that partition
  3. Add record to internal batch buffer (per partition)
  4. Background sender thread ships batches to brokers
  5. Broker acknowledges (based on acks config)
  6. On failure: retry up to retries times with retry.backoff.ms delay

Key producer configs:
  bootstrap.servers    = "broker1:9092,broker2:9092,broker3:9092"
  key.serializer       = StringSerializer (or Avro, Protobuf, JSON)
  value.serializer     = StringSerializer
  acks                 = "all"
  retries              = 2147483647 (Integer.MAX_VALUE — effectively infinite)
  delivery.timeout.ms  = 120000 (2 min — outer bound on retries)
  enable.idempotence   = true

# How batching works:
#   Producer accumulates records in per-partition buffers
#   A batch is sent when EITHER condition is met:
#     1. batch.size bytes accumulated (default: 16 KB)
#     2. linger.ms time elapsed since first record in batch (default: 0 ms)

# Throughput-optimized settings:
batch.size = 65536          # 64 KB batches
linger.ms = 20              # Wait up to 20ms to fill batch
compression.type = lz4      # Compress batches before sending

# Latency-optimized settings:
batch.size = 16384          # 16 KB (default)
linger.ms = 0               # Send immediately (no waiting)
compression.type = none     # No compression overhead

# The trade-off:
#   Higher linger.ms → larger batches → better throughput, higher latency
#   Lower linger.ms  → smaller batches → lower latency, more network calls

# Enable idempotency (default since Kafka 3.0):
enable.idempotence = true

# How it works:
#   1. Producer gets a unique Producer ID (PID) on initialization
#   2. Each record gets a sequence number (per partition, monotonically increasing)
#   3. Broker tracks: (PID, partition) → last sequence number
#   4. If broker receives a record with sequence ≤ last seen → duplicate, discard
#   5. If broker receives a record with sequence > last + 1 → out of order, reject

# What this guarantees:
#   ✓ Exactly-once delivery from producer to broker (per partition)
#   ✓ No duplicates from retries
#   ✓ In-order delivery within a partition

# What this requires:
#   acks = all (enforced automatically)
#   retries > 0 (enforced automatically)
#   max.in.flight.requests.per.connection ≤ 5 (enforced automatically)

# What this does NOT cover:
#   ✗ Cross-partition atomicity (use transactions for that)
#   ✗ Consumer-side deduplication (consumer may process twice on rebalance)

Topic: "orders" (6 partitions)

Consumer Group "order-processing" (3 consumers):
  Consumer A → [P0, P1]    (handles 2 partitions)
  Consumer B → [P2, P3]    (handles 2 partitions)
  Consumer C → [P4, P5]    (handles 2 partitions)

Consumer Group "analytics" (2 consumers):
  Consumer X → [P0, P1, P2]  (handles 3 partitions)
  Consumer Y → [P3, P4, P5]  (handles 3 partitions)

Key rules:
  1. Each partition → exactly ONE consumer in a group (no sharing)
  2. One consumer can handle MULTIPLE partitions
  3. If consumers > partitions → some consumers are idle
  4. If consumers < partitions → some consumers handle multiple
  5. Different groups are completely independent (different offsets)
  6. Max parallelism = number of partitions

Triggers:
  - Consumer joins the group (new instance deployed)
  - Consumer leaves gracefully (shutdown, close())
  - Consumer crashes (missed heartbeats → session.timeout.ms exceeded)
  - Consumer takes too long to process (max.poll.interval.ms exceeded)
  - Partition count changes (topic altered)

Classic (Eager) Rebalance:
  1. Group Coordinator detects membership change
  2. ALL consumers revoke ALL their partitions (stop processing)
  3. Group Leader computes new assignment
  4. All consumers receive new assignment and resume
  Problem: ENTIRE group stops processing during rebalance

Cooperative (Incremental) Rebalance:
  1. Group Coordinator detects membership change
  2. Only affected partitions are revoked from current owners
  3. Revoked partitions are assigned to new owners
  4. Non-affected consumers continue processing uninterrupted
  Benefit: Minimal disruption — most consumers keep working

# Pattern 1: Auto-commit (default, dangerous)
enable.auto.commit = true
auto.commit.interval.ms = 5000

# Problem: commits happen on a timer, not after processing
# If consumer crashes after auto-commit but before processing → data loss
# If consumer crashes after processing but before auto-commit → duplicates

# Pattern 2: Manual commit after processing (recommended)
enable.auto.commit = false

while (true) {
  records = consumer.poll(Duration.ofMillis(1000))
  for (record in records) {
    process(record)                    // Process first
  }
  consumer.commitSync()                // Then commit (blocks until confirmed)
}

# Pattern 3: Manual async commit (higher throughput)
enable.auto.commit = false

while (true) {
  records = consumer.poll(Duration.ofMillis(1000))
  for (record in records) {
    process(record)
  }
  consumer.commitAsync((offsets, exception) -> {
    if (exception != null) log.error("Commit failed", exception)
  })
}

# Pattern 4: Commit per partition (fine-grained)
# Useful when processing is slow and you want progress per partition
for (partition in records.partitions()) {
  partitionRecords = records.records(partition)
  for (record in partitionRecords) {
    process(record)
  }
  lastOffset = partitionRecords.last().offset()
  consumer.commitSync(Map.of(partition, new OffsetAndMetadata(lastOffset + 1)))
}

# Transactional producer + consumer for exactly-once processing:

# Producer config:
transactional.id = "order-processor-1"  # Unique per instance
enable.idempotence = true               # Required for transactions

# Consumer config:
isolation.level = "read_committed"      # Only see committed transaction records
enable.auto.commit = false              # Offsets committed within transaction

# The read-process-write loop:
producer.initTransactions()

while (true) {
  records = consumer.poll(Duration.ofMillis(1000))

  producer.beginTransaction()
  try {
    for (record in records) {
      result = transform(record)
      producer.send(new ProducerRecord("output-topic", result))
    }
    // Commit consumer offsets within the same transaction
    producer.sendOffsetsToTransaction(currentOffsets, consumerGroupId)
    producer.commitTransaction()
  } catch (exception) {
    producer.abortTransaction()
    // Offsets not committed → records will be re-consumed
  }
}

# What this guarantees:
#   ✓ Output records + offset commit are atomic
#   ✓ Either both happen or neither happens
#   ✓ No duplicates in output topic, no records skipped

# What this does NOT guarantee:
#   ✗ Exactly-once to external systems (DB writes, API calls)
#   ✗ For external side effects, make them idempotent