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

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 idempotent producers for exactly-once semantics within a single producer session. Idempotent producers prevent duplicate messages when network errors cause retries.

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 tracks producer state per topic-partition, storing the last sequence number for each (producerId, epoch) pair:

  • Deduplication: Duplicate retries are detected by their sequence numbers and acknowledged without re-appending

  • Persistent state: Producer sequence state survives broker restarts by persisting to Redis, so a client retry after a broker bounce is still detected as a duplicate

  • Bounded memory: Idle producer IDs are automatically expired to prevent unbounded memory growth in high-churn environments

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)

Set via environment variable:

export KORVET_BROKER_PRODUCER_ID_EXPIRATION=2d

Or in application.yml:

korvet:
  broker:
    producer-id-expiration: 2d
The expiration interval matches Kafka’s producer.id.expiration.ms semantics. Evicting an idle producer means very late retries lose duplicate detection, which is acceptable in most production scenarios.

Next Steps