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