Kafka Producer APIs

Kafka has provided you with a rich set of APIs to create applications to interact with it. We will go through Producer API details and understand its uses.

Creating a Kafka producer involves the following steps:

1. Required configuration.
2. Creating a producer object.
3. Setting up a producer record.
4. Creating a custom partition if required.
5. Additional configuration.

Required configuration: In most applications, we first start with creating the initial configuration without which we cannot run the application. The following are three mandatory configuration parameters:

• bootstrap.servers: This contains a list of Kafka brokers addresses. The address is specified in terms of hostname:port. We can specify one or more broker detail, but we recommend that you provide at least two so that if one broker goes down, producer can use the other one.

• key.serializer: The message is sent to Kafka brokers in the form of a key-value pair. Brokers expect this key-value to be in byte arrays. So we need to tell producer which serializer class is to be used to convert this key-value object to a byte array. This property is set to tell the producer which class to use to serialize the key of the message.

Kafka provides us with three inbuilt serializer classes: ByteArraySerializer, StringSerializer, and IntegerSerializer. All these classes are present in the org.apache.kafka.common.serialization package and implement the serializer interface.

• value.serializer: This is similar to the key.serializer property, but this property tells the producer which class to use in order to serialize the value. You can implement your own serialize class and assign to this property.

Let's see how we do it in a programming context.

Here is how Java works for Producer APIs:

Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "broker1:port,broker2:port");
producerProps.put("key.serializer",
 "org.apache.kafka.common.serialization.StringSerializer");
 producerProps.put("value.serializer",
 "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<String,String>(producerProps);

The Producer API in Scala:

 val producerProps = new Properties()
 producerProps.put("bootstrap.servers", "broker1:port,broker2:port");

 producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
 producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

 val producer = new KafkaProducer[String, String](producerProps)

The preceding code contains three specific points:

• Properties object: We start with creating a property object; this object contains the put method that is used to put the configuration key-value pair in place
  
• Serializer class: We will use StringSerializer for both key and value as our key and value will be of the string type
• Producer object: We create a producer object by passing the configuration object to it, which provides the producer with specific information about broker servers, serializer classes, and other configurations that we will see later