Mastering Kafka Producer API: Tips, Configurations, and Common Pitfalls

Kafka Overview

Kafka is a mature distributed messaging queue system that has been in production for nearly a decade. It is widely used in big‑data stream processing, log handling, and many other scenarios, often serving as a critical component in modern data pipelines.

Standard Producer API Overview

The producer API is the primary way to send messages to Kafka. The following diagram illustrates the data flow between a client and the Kafka cluster.

Client : Any process that connects to a Kafka broker to write messages.

Broker : A physical server that is part of the Kafka cluster.

Topic : An abstract category that groups messages; each topic contains one or more partitions.

Partition : A log that stores messages for a topic; partitions are independent and each has its own offset.

Replicas : Copies of a partition stored on multiple brokers; one replica acts as the leader.

The producer API guarantees at‑least‑once and at‑most‑once delivery semantics. Exactly‑once is not provided by the basic API.

At‑least‑once: messages may be duplicated; consumers must handle deduplication.

At‑most‑once: messages may be lost; used when performance is prioritized over reliability.

Exactly‑once: not supported by the standard producer API; requires transactional messaging.

Example: creating a topic with two partitions and two replicas.

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 \
    --replication-factor 2 --partitions 2 --topic test

Java producer example (basic configuration and send loop).

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 3);
props.put("retry.backoff.ms", 2000);
props.put("compression.type", "lz4");
props.put("batch.size", 16384);
props.put("linger.ms", 200);
props.put("max.request.size", 1048576);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("request.timeout.ms", 10000);
props.put("max.block.ms", 30000);
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 1000; i++) {
    Future<RecordMetadata> future = producer.send(new ProducerRecord<>("test", UUID.randomUUID().toString()));
    System.out.println("produce offset:" + future.get().offset());
}
producer.close();

Client workflow details:

Connect to brokers using bootstrap.servers; retries on failure.

Send ApiVersions request to discover supported API versions.

Request Metadata for the target topic, obtain partition leaders and replica information.

Determine the target partition (hash of the key) and enqueue the record.

Check max.request.size and batch constraints ( batch.size, linger.ms).

Send Produce request, handling retries according to retries and retry.backoff.ms.

Configuration highlights: acks: all (strongest), 1 (balance), or 0 (at‑most‑once). retries and retry.backoff.ms control retry behavior. compression.type selects compression algorithm (e.g., lz4). request.timeout.ms and max.block.ms define client‑side timeouts.

Idempotent Producer Overview

Enabling idempotence ( enable.idempotence=true) provides exactly‑once semantics for a single partition. The client automatically forces acks=all, sets a high retries value, and limits max.inflight.requests.per.connection to ≤ 5.

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("enable.idempotence", true);
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 1000; i++) {
    Future<RecordMetadata> future = producer.send(new ProducerRecord<>("test", UUID.randomUUID().toString()));
    System.out.println("produce offset:" + future.get().offset());
}
producer.close();

When idempotence is enabled, the client sends an InitProducerId request to obtain a PID, attaches a monotonically increasing sequence number to each record, and the broker discards duplicates based on PID and sequence.

Transactional Messaging Overview

Transactional messaging extends exactly‑once guarantees across multiple partitions and requires an idempotent producer.

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("enable.idempotence", "true");
props.put("transactional.id", "testtrans-1");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record0);
    producer.send(record1);
    producer.sendOffsetsToTxn(...);
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    producer.close();
} catch (KafkaException e) {
    producer.abortTransaction();
}

Workflow:

Call initTransactions() → client requests InitProducerId with a transactional ID.

Begin a transaction, send records, optionally send consumer offsets, then commit or abort.

The broker tracks epochs to prevent stale producers from writing after a failure.

Conclusion

The article reviewed Kafka producer usage from a client perspective, highlighted important configurations, explained idempotent and transactional producers, and provided practical code snippets to help developers troubleshoot and optimize their Kafka publishing pipelines.