Apache Kafka Producers are going to write data to topics and topics are made of partitions. Now the producers in Kafka will automatically know to which broker and partition to write based on your message and in case there is a Kafka broker failure in your cluster the producers will automatically recover from it which makes Kafka resilient and which makes Kafka so good and used today. But how to create a safe producer in Apache Kafka? So there are two answers to this question according to Kafka Version you are using.
Prerequisites:
- Apache Kafka – Producer Acknowledgement and min.insync.replicas
- Apache Kafka – Producer Retry and max.in.flight.requests.per.connection
- Apache Kafka – Idempotent Producer
Kafka Version < 0.11
If you are using Kafka Version < 0.11 then you have to set the following producer properties in order to make your producer safe.
- acks = all (producer level) : It will ensure that data is properly replicated before an ack is received
- min.insync.replicas = 2 (broker/topic level) : It will ensure that two brokers in In-Sync Replica at least have the data after an ack
- retries = Integer.MAX_VALUE (producer level) : It will ensure that transient errors are retried indefinitely
- max.inflight.requests = 1 (producer level) : It will ensure that only one request is tried at any time, preventing message re-ordering in case of retries
Kafka Version >= 0.11
If you are using Kafka Version >= 0.11 then you have to set the following producer properties in order to make your producer safe.
- enable idempotence = true (producer level)
- acks = all (producer level) : It will ensure that data is properly replicated before an ack is received
- min.insync.replicas = 2 (broker/topic level) : It will ensure that two brokers in In-Sync Replica at least have the data after an ack
- retries = Integer.MAX_VALUE (producer level) : It will ensure that transient errors are retried indefinitely
- max.inflight.requests = 5 (producer level) : Default
Note: Running a "safe producer" might impact throughput and latency, always test it for your use cases.
Create Safe Producer using Java
To create a safe producer using java we have to set these additional properties in the code
Java
properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true" ); properties.setProperty(ProducerConfig.ACKS_CONFIG, "all" ); properties.setProperty(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE)); properties.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5" ); |
Example Project
In this example, we are going to discuss the step-by-step implementation of how to Create a Safe Apache Kafka Producer using Java.
Step by Step Implementation
Step 1: Create a New Apache Kafka Project in IntelliJ
To create a new Apache Kafka Project in IntelliJ using Java and Maven please refer to How to Create an Apache Kafka Project in IntelliJ using Java and Maven.
Step 2: Install and Run Apache Kafka
To Install and Run Apache Kafka in your local system please refer to How to Install and Run Apache Kafka.
Step 3: Create Producer using Java
First, we have to create Producer Properties. And to create Producer Properties refer to the below code snippet
Create Producer Properties:
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());
Add Additional Properties for Creating Safe Producer:
properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); properties.setProperty(ProducerConfig.ACKS_CONFIG, "all"); properties.setProperty(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE)); properties.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
Create the Producer:
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
Create a Producer Record:
ProducerRecord<String, String> record = new ProducerRecord<>("first_neveropen_topic", "hello_neveropen");
Send data asynchronously:
producer.send(record);
Flush and Close the Producer:
producer.flush(); producer.close();
Below is the complete code. Comments are added inside the code to understand the code in more detail.
Java
package basics; 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; import java.util.Properties; public class KafkaProducerDemo { public static void main(String[] args) { String bootstrapServer = "127.0.0.1:9092" ; // Create Producer Properties 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.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE)); properties.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5" ); // Create the Producer KafkaProducer<String, String> producer = new KafkaProducer<>(properties); // Create a Producer Record ProducerRecord<String, String> record = new ProducerRecord<>( "first_gfg_topic" , "hello_neveropen!!" ); // Send Record producer.send(record); // Flush and Close the Producer producer.flush(); producer.close(); } } |
Step 4: Run the Application
Now run the application and below is the output. Please have a look at the bold properties in the below output console.
[main] INFO org.apache.kafka.clients.producer.ProducerConfig - ProducerConfig values: acks = all batch.size = 16384 bootstrap.servers = [127.0.0.1:9092] buffer.memory = 33554432 client.dns.lookup = use_all_dns_ips client.id = producer-1 compression.type = none connections.max.idle.ms = 540000 delivery.timeout.ms = 120000 enable.idempotence = true interceptor.classes = [] internal.auto.downgrade.txn.commit = false key.serializer = class org.apache.kafka.common.serialization.StringSerializer linger.ms = 0 max.block.ms = 60000 max.in.flight.requests.per.connection = 5 max.request.size = 1048576 metadata.max.age.ms = 300000 metadata.max.idle.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner receive.buffer.bytes = 32768 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retries = 2147483647 retry.backoff.ms = 100 sasl.client.callback.handler.class = null sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0.8 sasl.login.callback.handler.class = null sasl.login.class = null sasl.login.refresh.buffer.seconds = 300 sasl.login.refresh.min.period.seconds = 60 sasl.login.refresh.window.factor = 0.8 sasl.login.refresh.window.jitter = 0.05 sasl.mechanism = GSSAPI security.protocol = PLAINTEXT security.providers = null send.buffer.bytes = 131072 socket.connection.setup.timeout.max.ms = 30000 socket.connection.setup.timeout.ms = 10000 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.3] ssl.endpoint.identification.algorithm = https ssl.engine.factory.class = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.certificate.chain = null ssl.keystore.key = null ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLSv1.3 ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.certificates = null ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS transaction.timeout.ms = 60000 transactional.id = null value.serializer = class org.apache.kafka.common.serialization.StringSerializer [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 2.8.0 [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: ebb1d6e21cc92130 [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1674753797898 [kafka-producer-network-thread | producer-1] INFO org.apache.kafka.clients.Metadata - [Producer clientId=producer-1] Cluster ID: OIx0v3RmSd2y0zKUaBM7-Q [main] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-1] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms. [main] INFO org.apache.kafka.common.metrics.Metrics - Metrics scheduler closed [main] INFO org.apache.kafka.common.metrics.Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter [main] INFO org.apache.kafka.common.metrics.Metrics - Metrics reporters closed [main] INFO org.apache.kafka.common.utils.AppInfoParser - App info kafka.producer for producer-1 unregistered Process finished with exit code 0
And you can see the message in the Kafka consumer console. Run this command
kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic first_gfg_topic
Output: