Kafka Producer Using Java

Implementation of Kafka Producer using Java

package com.abc.demo;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

public class KafkaProducerTest {

 public static void main(String[] args) throws InterruptedException, ExecutionException{
  //Create producer property
  String bootstrapServer = "localhost:9092";
  Properties properties = new Properties();
  properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
  properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  
  //Create safe producer
  properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
  properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
  properties.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
  properties.setProperty(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE));
  
  //High throughput producer (at the expense of a bit of latency and CPU usage)
  properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
  properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "20"); //20ms wait time
  properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, Integer.toString(32*1024)); //32KB batch size
  
  //Create producer
  KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
  
  //create a producer record
  ProducerRecord<String, String> record = new ProducerRecord<>("topicName", "firstRecord");
  //create producer record with key
  //new ProducerRecord<>("topicName", "MessageKey", "Message");
  //create producer record with key and partition number
  //new ProducerRecord<>("topicName", 1 /*partition number*/, "MessageKey", "Message");
  
  //send data - asynchronous
  //without callback
  //producer.send(record);
  //with callback
  producer.send(record, (recordMetadata, exception) -> {
   if(exception == null){
    System.out.println(recordMetadata.topic() + "+" + recordMetadata.partition() + "+" + recordMetadata.offset());
   }else{
    System.err.println(exception.getMessage());
   }
  });
  
  //send data - synchronous
  //without callback
  //producer.send(record).get(); //.get() make it synchronous call
  
  //flush data
  producer.flush();
  
  //flush and close producer
  producer.close();
 }
}

send() Is Asynchronous by Default

producer.send() doesn’t wait for the broker to acknowledge the record — it hands the ProducerRecord to an internal buffer and a background I/O thread does the actual network work, returning immediately with a Future<RecordMetadata>. That gives you three ways to use it, in increasing order of overhead:

• Fire-and-forget — producer.send(record) and ignore the return value. Fastest, but you won’t know if the write failed.
• Asynchronous with a callback — as in the example above, producer.send(record, callback). The callback runs on the producer’s I/O thread once the send completes (or fails), so you get delivery confirmation without blocking the calling thread.
• Synchronous — producer.send(record).get(). Calling .get() on the returned Future blocks the calling thread until the broker responds, turning the async call into a request/response round trip. This throws away most of the throughput benefit of batching and should be reserved for cases where you genuinely need to know the outcome before moving on.

Building a “Safe” Producer

The block of properties in the example — ENABLE_IDEMPOTENCE_CONFIG, ACKS_CONFIG, RETRIES_CONFIG, and MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION — together configure what’s commonly called a safe producer, one that avoids both message loss and duplicate/out-of-order writes under retries:

• acks=all tells the leader broker to wait until the record has been replicated to all in-sync replicas before acknowledging it, so a single broker crash right after the write can’t lose the record.
• enable.idempotence=true has the producer tag each record with a sequence number per partition; the broker uses that to detect and drop duplicate retries, so a retried send can’t create a duplicate message the way a plain retry otherwise could.
• retries set high (Integer.MAX_VALUE here) means transient errors get retried rather than surfaced as a failed send.
• max.in.flight.requests.per.connection caps how many unacknowledged requests can be outstanding to a broker at once. With idempotence enabled, Kafka allows this to be as high as 5 while still preserving per-partition ordering (the sequence numbers let the broker put retried batches back in order); without idempotence, anything above 1 risks a retried batch landing after a later one and reordering your records.

One naming quirk worth knowing if you go looking at ProducerConfig’s source: almost every constant there ends in _CONFIG (ACKS_CONFIG, RETRIES_CONFIG, BATCH_SIZE_CONFIG…) — except MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, which doesn’t. It’s easy to typo this one by reflexively adding _CONFIG, which won’t compile.

As of Kafka 3.0, enable.idempotence defaults to true and acks defaults to all for newly created producers (KIP-679), so a modern client gets much of this safety even without setting these properties explicitly — but pinning them yourself keeps the behavior explicit and version-independent.

Retries Are Time-Bounded by delivery.timeout.ms

Setting retries to Integer.MAX_VALUE doesn’t mean the producer retries forever. The actual ceiling is ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG ("delivery.timeout.ms", default 120000ms / 2 minutes) — the total time budget from when send() is called to when the record is either acknowledged or given up on, covering all retries and backoffs in between. Once that window elapses, the send fails (surfaced via the callback’s exception argument, or thrown from .get() for a synchronous call) regardless of how many retries are left.

Partitioning: Where Does a Record End Up?

Which partition a ProducerRecord lands in depends on what you construct it with:

• Explicit partition number — new ProducerRecord<>("topicName", 1, "MessageKey", "Message") goes exactly there, no further logic involved.
• Key, no partition — the default partitioner hashes the key so that every record with the same key always lands on the same partition, which is what lets you preserve per-key ordering (e.g. all events for one user processed in order).
• No key, no partition — records are distributed across partitions using a sticky partitioner: it batches all records without a key onto one partition at a time and switches partitions when a batch fills or linger.ms elapses, rather than choosing a new partition per record. This favors larger, more efficient batches over perfectly even per-record distribution.

Flushing and Closing

producer.flush() blocks until every record sent so far has completed (delivered or failed) — useful before shutting down, or between logically distinct groups of sends when you need a completion checkpoint. producer.close() does an implicit flush before releasing the producer’s network resources, so calling both back-to-back as in the example is redundant but harmless. For a bounded shutdown, producer.close(Duration) lets you cap how long close() will wait for in-flight sends before forcing the connection closed.