// ❌ Scaling beyond partition count — consumers 7–12 are idle, wasting memory
const consumer = kafka.consumer({ groupId: 'fulfillment-service' })
// Topic has 6 partitions. Running 12 consumer instances: 6 active, 6 idle.
// Additional instances cannot help. They simply join the group and get no assignment.

// ✅ Correct scale-out: consumers = partitions (or less)
// If you need more parallelism: increase partition count first, then scale consumers

// ❌ Naive partition count increase breaks per-entity ordering
// Before: 4 partitions → all events for order-42 always land on partition 1 (hash("order-42") % 4 = 1)
// After:  8 partitions → hash("order-42") % 8 = 5 — different partition, different consumer
// Existing consumers processing order-42 events are now split across partition 1 AND 5
// → Out-of-order processing possible during the transition window

// ✅ Key-based partitioning preserves ordering within an entity's lifecycle
await producer.send({
  topic: 'orders.created',
  messages: [{
    key:   order.id,           // deterministic: same orderId → same partition
    value: JSON.stringify(order),
  }]
})
// Adding partitions: plan a cutover window where in-flight orders drain before the split takes effect

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

# Output:
# GROUP               TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# fulfillment-service orders.created  0          4,120,441       4,122,831       2390 consumer-1
# fulfillment-service orders.created  1          4,118,882       4,122,832       3950 consumer-2
# fulfillment-service orders.created  2          4,119,991       4,122,830       2839 consumer-3

// ✅ Programmatic lag monitoring — emit to metrics on every batch
const consumer = kafka.consumer({ groupId: 'fulfillment-service' })

await consumer.run({
  eachBatch: async ({ batch, resolveOffset, heartbeat }) => {
    const lagAtBatchStart = Number(batch.highWatermark) - Number(batch.lastOffset())
    metrics.gauge('consumer.lag', lagAtBatchStart, {
      topic:     batch.topic,
      partition: batch.partition.toString(),
      group:     'fulfillment-service',
    })

    for (const message of batch.messages) {
      await processOrder(JSON.parse(message.value!.toString()))
      resolveOffset(message.offset)
      await heartbeat()   // prevents session timeout during long batch processing
    }
  }
})

// ❌ Default max.poll.records = 500 — floods slow downstream
const consumer = kafka.consumer({
  groupId: 'fulfillment-service',
  maxInFlightRequests: 1,
})
// Each poll returns up to 500 messages. If downstream DB takes 50ms each:
// 500 × 50ms = 25 seconds per batch → heartbeat timeout → rebalance

// ✅ Reduce batch size to match downstream throughput
const consumer = kafka.consumer({
  groupId: 'fulfillment-service',
  maxInFlightRequests: 1,
})
await consumer.connect()

await consumer.run({
  eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning }) => {
    for (const message of batch.messages) {
      if (!isRunning()) break

      const start = Date.now()
      await fulfillmentDb.insert(JSON.parse(message.value!.toString()))
      const duration = Date.now() - start

      // Dynamic backpressure: if downstream is slow, yield before next message
      if (duration > 100) {
        await new Promise(resolve => setTimeout(resolve, duration * 0.5))
      }

      resolveOffset(message.offset)
      await heartbeat()
    }
  }
})

// ✅ Prefetch creates a natural backpressure ceiling at the broker level
const channel = await connection.createChannel()
await channel.prefetch(5)   // broker holds back until < 5 unacked messages

await channel.consume('orders.created', async (msg) => {
  if (!msg) return
  try {
    await fulfillmentDb.insert(JSON.parse(msg.content.toString()))
    channel.ack(msg)    // releases one prefetch slot — broker sends next message
  } catch (err) {
    // Downstream is unhealthy — nack without requeue
    // Broker delivers to DLX, consumer gets a break
    channel.nack(msg, false, false)
  }
}, { noAck: false })

// ✅ Enable cooperative sticky rebalancing — only moved partitions pause
import { Kafka, PartitionAssigners } from 'kafkajs'

const consumer = kafka.consumer({
  groupId: 'fulfillment-service',
  partitionAssigners: [PartitionAssigners.roundRobin],
  // kafkajs uses cooperative protocol when broker supports it (Kafka 2.4+)
  sessionTimeout:        30_000,   // 30s — how long before broker considers consumer dead
  heartbeatInterval:      3_000,   // 3s — must be << sessionTimeout
  maxWaitTimeInMs:        5_000,
})

consumer.on(consumer.events.GROUP_JOIN, ({ payload }) => {
  console.log('Assigned partitions:', payload.memberAssignment)
})

consumer.on(consumer.events.STOP, () => {
  // Consumer stopped — partitions released back to group coordinator
})

// ✅ Tuning to prevent rebalance storms during slow processing
const consumer = kafka.consumer({
  groupId: 'fulfillment-service',
  sessionTimeout:      45_000,   // How long broker waits before declaring consumer dead
                                  // Increase if consumers are slow but healthy
  heartbeatInterval:    5_000,   // Must be < sessionTimeout / 3
  maxWaitTimeInMs:     10_000,   // Max time broker blocks waiting for new messages

  // kafkajs eachBatch: always call heartbeat() inside the loop
  // eachMessage: kafkajs calls heartbeat automatically between messages
})

// Key insight: sessionTimeout vs max.poll.interval.ms (Java client only)
// In kafkajs: sessionTimeout covers both — the heartbeat loop is internal
// In Java client:
//   sessionTimeout    = heartbeat timeout (background thread)
//   max.poll.interval.ms = time between poll() calls (processing timeout)
//   Increase max.poll.interval.ms for slow processors, not sessionTimeout

// Per-message: simple, lower throughput, automatic heartbeat (kafkajs)
await consumer.run({
  eachMessage: async ({ message }) => {
    await processOrder(JSON.parse(message.value!.toString()))
    // kafkajs sends heartbeat automatically between eachMessage invocations
    // Safe for processing < sessionTimeout / 2 per message
  }
})

// Batch: higher throughput, manual heartbeat required, enables bulk DB writes
await consumer.run({
  eachBatch: async ({ batch, resolveOffset, heartbeat, commitOffsetsIfNecessary }) => {
    // Bulk insert: one DB round-trip for the entire batch vs one per message
    const records = batch.messages.map(m => JSON.parse(m.value!.toString()))
    await fulfillmentDb.bulkInsert(records)

    // Commit the last offset in the batch
    resolveOffset(batch.messages[batch.messages.length - 1].offset)
    await commitOffsetsIfNecessary()
    await heartbeat()
  }
})