This version is still in development and is not considered stable yet. For the latest stable version, please use Korvet 0.20!

Producing Messages

This guide shows how to produce messages to Korvet using Kafka clients.

Using kafka-console-producer

The simplest way to produce messages:

kafka-console-producer --bootstrap-server localhost:9092 --topic my-topic

Type messages and press Enter to send each one.

Java Producer

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

ProducerRecord<String, String> record =
    new ProducerRecord<>("my-topic", "key", "value");

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.println("Sent to partition " + metadata.partition() +
                         " at offset " + metadata.offset());
    } else {
        exception.printStackTrace();
    }
});

producer.close();

Python Producer

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')

producer.send('my-topic', b'Hello, Korvet!')
producer.flush()

Message Format

Messages consist of:

  • Key (optional): Used for partitioning

  • Value: The message payload

  • Headers (optional): Key-value metadata

  • Timestamp: Automatically set if not provided

Partitioning

Messages are distributed across partitions based on:

  • Key hash: If a key is provided, messages with the same key go to the same partition

  • Round-robin: If no key is provided, messages are distributed evenly

Idempotent Producers

Korvet supports Kafka idempotent producers. They prevent duplicate appends when a lost response causes a producer to retry a batch.

Enabling Idempotence

Enable idempotence in your Kafka producer configuration:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("enable.idempotence", "true");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

How It Works

Korvet validates producer sequences in memory to reject gaps with OUT_OF_ORDER_SEQUENCE_NUMBER. Korvet attaches (producerId, epoch-sequence) to every idempotent XADD. Redis returns the original stream ID for a retry, so Korvet returns the original Kafka base offset without maintaining a separate producer-state hash. Redis 8.6 or later is required for idempotent produce. Korvet can start against an older Redis version, but idempotent produce requests fail when Redis rejects the unsupported XADD IDMP command.

In-memory state retains metadata for the five most recent batches, matching Kafka’s in-flight batch window, and is expired after an idle interval.

Configuration

Property Description Default

korvet.broker.producer-id-expiration

Time after which an idle producer’s dedup state is evicted. Very late retries from evicted producers will not be detected as duplicates.

1d (24 hours)

korvet.broker.idempotence.window

How long Redis retains native idempotency identifiers.

5m

korvet.broker.idempotence.max-records-per-producer

Maximum identifiers retained per producer and physical stream.

2000

Set via environment variable:

export KORVET_BROKER_PRODUCER_ID_EXPIRATION=2d

Or in application.yml:

korvet:
  broker:
    producer-id-expiration: 2d
    idempotence:
      window: 5m
      max-records-per-producer: 2000

Size max-records-per-producer to hold at least the largest retried batch plus records from up to five later in-flight batches. Changing the window or maximum clears a stream’s native IDMP map when that stream is next configured.

Native IDMP state is scoped to a physical Redis stream. A retry that crosses a tiered-storage segment roll is protected only while the broker’s in-memory batch metadata remains warm. A batch larger than max-records-per-producer can evict its own earliest identifiers.

Next Steps