Kafka producer using Java
In this article, we will learn how to write Kafka producer in Java language. We will create a test topic then we will build our producer and set its properties. After that, we will start sending messages to that producer. We will also see how we can use Keys to send messages to the same partition in a particular topic.
Creating a topic
In Kafka, Producers send records to the topic. So we need one topic to which our producer sends records. You can use this code to create a new topic in the Kafka server. In the next steps, it is assumed that you have created the topic with name myTopic and we are going to use that to send messages from the producer.
Kafka Producer
Now we can start creating our own Kafka producer in Java. Let us see, how to set up Kafka Producer and set its properties.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; 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 FirstKafkaProducer { public static void main (String args[]){ String bootstrapServer = "localhost:9092"; String keySerializer = StringSerializer.class.getName(); String valueSerializer = StringSerializer.class.getName(); String producerId = "simpleProducer"; int retries = 2; String topicName = "myTopic"; Properties properties = new Properties(); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServer); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,keySerializer); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,valueSerializer); properties.put(ProducerConfig.CLIENT_ID_CONFIG, producerId); properties.put(ProducerConfig.RETRIES_CONFIG,retries); } } |
In the above code, we can see that we have set up properties for the producer. We need to pass the list of the bootstrap server so that the producer can connect to them and start sending messages. In this example, we are using a producer which uses a key as well as messages in String format so that we are using String Serializer. CLIENT_ID_CONFIG property, we are setting a simple name to our producer in the Kafka server. In case of failure to send a message to Kafka topic, we want to try sending that message again. By setting RETRIES_CONFIG property, we can guarantee that in case of failure this producer will try sending that message two more times. You can learn more about Kafka Producers and their configuration here.
Sending messages
We need to create ProducerRecord to send messages to Kafka topic.
|
1 2 3 4 |
Producer producer = new KafkaProducer(properties); ProducerRecord producerRecord = new ProducerRecord(topicName,"Hello World !!!"); producer.send(producerRecord); producer.close(); |
Here we have created producer record and we are sending our first message to Kafka topic. You can notice that we have not passed any key in this producer record. In the absence of the key, this message will be sent to any partitions on that topic. If we want to send the same type of messages to the same partitions on that topic then we need to pass key with that message.
|
1 |
ProducerRecord producerRecord = new ProducerRecord(topicName,"1","Hello Again !!"); |
All messages with key = 1 will be assigned to the same partition of this topic. Send function of producer returns record metadata which has information to which partition our message went. We can read that using the following code.
|
1 |
System.out.println(producer.send(producerRecord).get().partition()); |
Conclusion
In this tutorial, we have created one simple Kafka producer in Java and understand its different configurations. you can download the code used in this article at git repository. In the next article, We will learn how to write consumer using Java and read messages from Kafka topics. Till then, Happy learning!!
