// ❌ Default settings — one network round-trip per message
const producer = kafka.producer({
  // linger.ms = 0 (default): send immediately, never wait for batch to fill
  // batch.size = 16384 (default): 16 KB max batch — rarely reached at linger.ms=0
})
// At 1,000 msg/s: 1,000 broker write requests/s
// At 50,000 msg/s: 50,000 broker write requests/s — network / I/O saturated

// ✅ Tuned for throughput — batch messages over 5ms window
const producer = kafka.producer({
  // kafkajs equivalent: configure via KafkaJS producer options
  maxInFlightRequests: 5,
})

// In kafkajs, batching is controlled at the send() level with batch API:
await producer.sendBatch({
  topicMessages: [{
    topic:    'orders.created',
    messages: events.map(e => ({
      key:   e.orderId,
      value: JSON.stringify(e),
    }))
  }]
  // Kafka client groups all messages into one or more RecordBatches
  // based on partition assignment and batch.size
})

// For Java/confluent clients (where linger.ms is explicit):
// props.setProperty("linger.ms", "5");
// props.setProperty("batch.size", String.valueOf(64 * 1024)); // 64 KB

# JMX metric: records-per-request-avg
# > 10: good batching (10+ messages per network request)
# < 2:  poor batching (near one message per request — check linger.ms)

# kafkajs: measure at application level
let batchCount    = 0
let messageCount  = 0

const originalSendBatch = producer.sendBatch.bind(producer)
producer.sendBatch = async (batch) => {
  const count = batch.topicMessages.reduce((sum, t) => sum + t.messages.length, 0)
  batchCount++
  messageCount += count
  metrics.histogram('producer.batch.size', count)
  return originalSendBatch(batch)
}
// Alert if avg batch size < 5 on high-throughput producers

// Compression is set at the producer level and applies per RecordBatch
// KafkaJS: zstd requires the @kafkajs/zstd plugin — not bundled in core
// npm install --save @kafkajs/zstd
import { CompressionTypes, KafkaConfig } from 'kafkajs'
import { ZstdCodec }                     from '@kafkajs/zstd'

// Register the codec once at app startup
CompressionCodecs[CompressionTypes.ZSTD] = ZstdCodec

const producer = kafka.producer({
  createPartitioner: Partitioners.LegacyPartitioner,
})

await producer.send({
  topic:       'orders.created',
  compression: CompressionTypes.ZSTD,   // ✅ modern default (after codec registration)
  messages:    messages,
})

// ❌ Individual message compression — wrong level
// Compression works on batches, not individual messages
// A single 100-byte JSON message compresses poorly — compression overhead > savings

// ✅ Ensure batch is large enough for compression to be effective
// Minimum effective batch for compression: ~10 KB uncompressed
// At linger.ms=5, batch.size=64 KB:
//   100-byte messages: ~640 messages per batch → ~4 KB compressed (zstd)
//   1 KB messages: ~64 messages per batch → ~8 KB compressed
// Both achieve significant compression ratio at reasonable batch sizes

partitions = ceil(target_throughput_MB/s / min(producer_MB/s_per_partition, consumer_MB/s_per_partition))

// Sizing example:
const targetThroughputMBps    = 100   // 100 MB/s peak ingest
const producerMBpsPerPartition = 10   // measured: broker handles 10 MB/s per partition
const consumerMBpsPerPartition = 50   // consumer can process 50 MB/s per partition

const bottleneck   = Math.min(producerMBpsPerPartition, consumerMBpsPerPartition) // 10
const minPartitions = Math.ceil(targetThroughputMBps / bottleneck)                // 10

// Apply headroom multiplier for future growth
const recommendedPartitions = minPartitions * 2  // 20 — room to double throughput without repartition

// acks=0: fire and forget — fastest, zero durability guarantee
await producer.send({
  topic:    'metrics.raw',
  acks:     0,   // producer does not wait for broker acknowledgement
  messages: [{ value: JSON.stringify(metric) }]
})
// Throughput: maximum. Data loss: possible on any broker crash.
// Use only for metrics and telemetry where loss is acceptable.

// acks=1: leader ack — fast, durability until leader crashes
await producer.send({
  topic:    'logs.application',
  acks:     1,   // leader writes to its local log and acks — followers not guaranteed
  messages: [{ value: JSON.stringify(logEntry) }]
})
// Throughput: high. Risk: if leader crashes before ISR replication, message is lost.

// acks=-1 (all): ISR quorum — slower, highest durability
await producer.send({
  topic:    'orders.created',
  acks:     -1,  // all ISR replicas must write before ack is sent
  messages: [{ value: JSON.stringify(order) }]
})
// Throughput: moderate. Risk: lost only if all ISR replicas fail simultaneously.
// Required for at-least-once and exactly-once delivery guarantees.

// ✅ Idempotent producer: prevents duplicate messages from producer retries
// Required for exactly-once and strongly recommended for at-least-once
const producer = kafka.producer({
  idempotent:          true,   // enables producer sequence numbers
  maxInFlightRequests: 5,      // idempotent producers can have up to 5 in-flight requests
  // Kafka enforces: if the same sequence is received twice, the second is a no-op
})
// Effect: producer retries on broker ack timeout no longer produce duplicates
// Cost: none — sequence numbers are broker-side bookkeeping, zero application change