# Broker / topic config
replication.factor=3
min.insync.replicas=2   # at least leader + 1 follower must confirm

# Producer config
acks=all                # wait for full ISR acknowledgement

Topic: order.events — 3 partitions

Consumer Group: fulfillment-service (3 consumers) — OPTIMAL
  Consumer 0 → Partition 0
  Consumer 1 → Partition 1
  Consumer 2 → Partition 2

Consumer Group: fulfillment-service (5 consumers) — WASTEFUL
  Consumer 0 → Partition 0
  Consumer 1 → Partition 1
  Consumer 2 → Partition 2
  Consumer 3 → ∅ (idle — no partition)
  Consumer 4 → ∅ (idle — no partition)

// ✅ Key-based routing — all events for the same order land on the same partition
await producer.send({
  topic: 'order.events',
  messages: [{
    key: orderId,  // CRC32(key) % partitionCount → deterministic partition
    value: JSON.stringify({ type: 'OrderPlaced', orderId, userId }),
  }],
})

// ❌ Null key → round-robin partitioning → no ordering guarantee per entity
await producer.send({
  topic: 'system.metrics',
  messages: [{ key: null, value: JSON.stringify(metric) }], // ordering irrelevant for metrics
})

// ✅ Kafka 2.4+ — only revoke partitions that actually need to move
const consumer = kafka.consumer({
  groupId: 'fulfillment-service',
  partitionAssigners: [PartitionAssigners.cooperativeStickyAssignor],
})

session.timeout.ms=45000       # declare consumer dead after 45s without heartbeat
heartbeat.interval.ms=3000     # send heartbeat every 3s
max.poll.interval.ms=300000    # rebalance triggered if poll() not called within 5min
                               # increase for slow message processing (DB writes, API calls)

// ❌ At-most-once — data loss possible on crash or broker failure
await consumer.run({
  autoCommit: true,        // offset committed on poll — BEFORE processing
  eachMessage: async ({ message }) => {
    await riskyProcess(message) // if this throws: message LOST (offset already committed)
  }
})

// ✅ At-least-once — no data loss, but duplicates possible; consumer must be idempotent
await consumer.run({
  autoCommit: false,
  eachMessage: async ({ topic, partition, message }) => {
    await process(message)   // process FIRST
    await consumer.commitOffsets([{  // THEN commit offset
      topic, partition,
      offset: (Number(message.offset) + 1).toString()
    }])
    // Crash between process() and commitOffsets() → redelivery on restart → duplicate
  }
})

// ✅ Exactly-once — atomic commit of output + input offset in one transaction
const producer = kafka.producer({
  idempotent: true,
  transactionalId: 'order-processor-p0', // unique per producer instance
  acks: 'all',
})

await producer.transaction(async (txn) => {
  const results = await processIncomingBatch(batch)
  await txn.send({ topic: 'processed.orders', messages: results })
  await txn.sendOffsets({
    consumerGroupId: 'order-processor',
    topics: [{ topic: 'raw.orders', partitions: [{ partition: 0, offset: '42' }] }]
  })
  // Either both commit atomically or neither commits
})

# High-durability exactly-once producer configuration
acks=all
enable.idempotence=true
retries=2147483647                          # unlimited — safe with idempotence
max.in.flight.requests.per.connection=5    # safe with idempotence; without: must be 1
batch.size=16384                           # 16KB — increase for higher throughput
linger.ms=5                               # wait up to 5ms for batch to fill
compression.type=snappy
transactional.id=my-service-p0            # unique per producer instance
transaction.timeout.ms=60000

# Monitor consumer lag per partition
kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --describe --group fulfillment-service

# TOPIC         PARTITION  LOG-END-OFFSET  CURRENT-OFFSET  LAG
# order.events  0          10450           10432           18
# order.events  1          9910            9871            39
# order.events  2          11203           11203           0    ← healthy

// ✅ Dead-letter queue — skip poison pills without blocking the partition
await consumer.run({
  autoCommit: false,
  eachMessage: async ({ topic, partition, message }) => {
    const retries = Number(message.headers?.['x-retry-count'] ?? 0)
    try {
      await processMessage(message)
      await commitOffset(topic, partition, message.offset)
    } catch (err) {
      if (retries >= 3) {
        await dlqProducer.send({
          topic: `${topic}.dlq`,
          messages: [{ key: message.key, value: message.value,
            headers: { 'x-original-topic': topic, 'x-error': err.message } }]
        })
        await commitOffset(topic, partition, message.offset) // skip and continue
      } else {
        throw err // retry framework increments x-retry-count
      }
    }
  }
})