Producer: Send Messages

This page describes how Korvet handles Kafka ProduceRequest operations.

This page is implementation-oriented and focuses on Kafka-to-Redis mapping details.

Producer Sends Messages

Kafka Wire Protocol

ProduceRequest (API Key: 0)

ProduceRequest {
  transactional_id: null
  acks: 1                    // 0=no ack, 1=leader ack, -1=all replicas
  timeout_ms: 30000
  topics: [
    {
      name: "orders"
      partitions: [
        {
          index: 0
          records: [
            {
              offset: null   // assigned by broker
              timestamp: 1234567890000
              key: "order-123"
              value: {"product": "widget", "quantity": 5}
              headers: [
                {key: "source", value: "web"},
                {key: "version", value: "1.0"}
              ]
            },
            {
              offset: null
              timestamp: 1234567891000
              key: "order-124"
              value: {"product": "gadget", "quantity": 3}
              headers: []
            }
          ]
        }
      ]
    }
  ]
}

ProduceResponse (API Key: 0)

ProduceResponse {
  topics: [
    {
      name: "orders"
      partitions: [
        {
          index: 0
          error_code: 0        // NONE
          base_offset: 42      // first offset assigned
          log_append_time: 1234567890123  // broker append time (ms); -1 unless topic uses LogAppendTime
          log_start_offset: 0
        }
      ]
    }
  ]
  throttle_time_ms: 0
}

Sequence Diagram

Redis Commands Detail

Produce handling translates records into one or more XADD operations on the target stream key. The returned Redis entry IDs are converted to Kafka offsets using the topic’s configured offset mapper.

Implementation Notes

Pipelining: Batch multiple XADD commands for better throughput
Chunked Writes: Large batches may be split into smaller chunks before writing.
Idempotent Retry Handling: Duplicate producer batches can be detected and skipped before storage writes.
Offset Encoding: Kafka offsets are derived from Redis entry IDs using the topic-specific sequenceBits configuration.
Retention: Retention trimming is not applied at produce time. A leader-locked background storage worker enforces retention by issuing XTRIM against partition streams.

Kafka ProduceRequest

ProduceRequest {
  topics: [
    {
      name: "orders"
      partitions: [
        {
          index: 0
          records: [
            { key: "order-123", value: "...", timestamp: 1234567890, headers: {...} }
            { key: "order-124", value: "...", timestamp: 1234567891, headers: {...} }
          ]
        }
      ]
    }
  ]
}

Redis Commands

# The entry ID is broker-assigned (not *), and the record is broken out into
# value/key/headers/timestamp fields (fields absent for null components).
XADD korvet:storage:local:orders:0 <assigned-id> value <value-bytes> key <key-bytes> timestamp <ms>
XADD korvet:storage:local:orders:0 <assigned-id> value <value-bytes> timestamp <ms>

Response Mapping

Redis returns entry IDs: ["1234567890123-0", "1234567890123-1"]
Encode entry IDs as Kafka offsets: offset = (timestamp << sequenceBits) | sequence
Return base_offset in ProduceResponse

Offset-to-Entry-ID Mapping

# No mapping storage needed - conversion is stateless
# Entry ID to Kafka offset:
#   offset = (timestamp << sequenceBits) | sequence
# Kafka offset to Entry ID:
#   timestamp = offset >> sequenceBits
#   sequence = offset & ((1 << sequenceBits) - 1)
#   entry_id = "{timestamp}-{sequence}"