Aug
10
Apache Kafka – Simple Producer Example ”; Previous Next Let us create an application for publishing and consuming messages using a Java client. Kafka producer client consists of the following API’s. KafkaProducer API Let us understand the most important set of Kafka producer API in this section. The central part of the KafkaProducer API is KafkaProducer class. The KafkaProducer class provides an option to connect a Kafka broker in its constructor with the following methods. KafkaProducer class provides send method to send messages asynchronously to a topic. The signature of send() is as follows producer.send(new ProducerRecord<byte[],byte[]>(topic, partition, key1, value1) , callback); ProducerRecord − The producer manages a buffer of records waiting to be sent. Callback − A user-supplied callback to execute when the record has been acknowl-edged by the server (null indicates no callback). KafkaProducer class provides a flush method to ensure all previously sent messages have been actually completed. Syntax of the flush method is as follows − public void flush() KafkaProducer class provides partitionFor method, which helps in getting the partition metadata for a given topic. This can be used for custom partitioning. The signature of this method is as follows − public Map metrics() It returns the map of internal metrics maintained by the producer. public void close() − KafkaProducer class provides close method blocks until all previously sent requests are completed. Producer API The central part of the Producer API is Producer class. Producer class provides an option to connect Kafka broker in its constructor by the following methods. The Producer Class The producer class provides send method to send messages to either single or multiple topics using the following signatures. public void send(KeyedMessaget<k,v> message) – sends the data to a single topic,par-titioned by key using either sync or async producer. public void send(List<KeyedMessage<k,v>>messages) – sends data to multiple topics. Properties prop = new Properties(); prop.put(producer.type,”async”) ProducerConfig config = new ProducerConfig(prop); There are two types of producers – Sync and Async. The same API configuration applies to Sync producer as well. The difference between them is a sync producer sends messages directly, but sends messages in background. Async producer is preferred when you want a higher throughput. In the previous releases like 0.8, an async producer does not have a callback for send() to register error handlers. This is available only in the current release of 0.9. public void close() Producer class provides close method to close the producer pool connections to all Kafka bro-kers. Configuration Settings The Producer API’s main configuration settings are listed in the following table for better under-standing − S.No Configuration Settings and Description 1 client.id identifies producer application 2 producer.type either sync or async 3 acks The acks config controls the criteria under producer requests are con-sidered complete. 4 retries If producer request fails, then automatically retry with specific value. 5 bootstrap.servers bootstrapping list of brokers. 6 linger.ms if you want to reduce the number of requests you can set linger.ms to something greater than some value. 7 key.serializer Key for the serializer interface. 8 value.serializer value for the serializer interface. 9 batch.size Buffer size. 10 buffer.memory controls the total amount of memory available to the producer for buff-ering. ProducerRecord API ProducerRecord is a key/value pair that is sent to Kafka cluster.ProducerRecord class constructor for creating a record with partition, key and value pairs using the following signature. public ProducerRecord (string topic, int partition, k key, v value) Topic − user defined topic name that will appended to record. Partition − partition count Key − The key that will be included in the record. Value − Record contents public ProducerRecord (string topic, k key, v value) ProducerRecord class constructor is used to create a record with key, value pairs and without partition. Topic − Create a topic to assign record. Key − key for the record. Value − record contents. public ProducerRecord (string topic, v value) ProducerRecord class creates a record without partition and key. Topic − create a topic. Value − record contents. The ProducerRecord class methods are listed in the following table − S.No Class Methods and Description 1 public string topic() Topic will append to the record. 2 public K key() Key that will be included in the record. If no such key, null will be re-turned here. 3 public V value() Record contents. 4 partition() Partition count for the record SimpleProducer application Before creating the application, first start ZooKeeper and Kafka broker then create your own topic in Kafka broker using create topic command. After that create a java class named Sim-pleProducer.java and type in the following coding. //import util.properties packages import java.util.Properties; //import simple producer packages import org.apache.kafka.clients.producer.Producer; //import KafkaProducer packages import org.apache.kafka.clients.producer.KafkaProducer; //import ProducerRecord packages import org.apache.kafka.clients.producer.ProducerRecord; //Create java class named “SimpleProducer” public class SimpleProducer { public static void main(String[] args) throws Exception{ // Check arguments length value if(args.length == 0){ System.out.println(“Enter