Apache Kafka Consumers and Producers with Java and Python

We will see the most basic examples of Kafka Consumers and Producers with Python and Java, although these examples can be extended quite a bit further (for example, with Streams, or batches or bulks of data).

Python

We can use programming languages to create Consumers and Producers for our Kafka (or a third party’s) with Python. To do so, we can use the “Kafka-python” library (its documentation is at kafka-python (documentación))

pip install kafka-python

Producer in Python

To produce messages, we will use “KafkaProducer”, which we import first (KafkaProducer documentation at KafkaProducer (kafka-python docs) ), and in it we define in “bootstrap_servers” a String with the Kafka servers and their port (Kafka’s default port is “9092”). Then we use the “send” method to send whatever message we want to a topic (it has to be sent as a byte array, which in Python can be serialized as easily as adding a “b” in front of the String or using the “encode” method). In the following example, I connect to my server at the IP “27.0.173.161” and the Kafka broker port “9092” to send three messages:

from kafka import KafkaProducer

servidores_bootstrap = '27.0.173.161:9092'
topic = 'miNuevoTopic'

productor = KafkaProducer(bootstrap_servers=servidores_bootstrap)

productor.send(topic, b'Un mensaje desde Python')
productor.send(topic, b'Otro mensaje desde Python')
productor.send(topic, b'Un tercer mensaje desde Python')

Note about keys: we can extend the send method to add a message key to each message, for example (both the key and the value must be byte arrays):

productor.send('miNuevoTopic', key=b'claveDelMensaje', value=b'Un mensaje')

Note if we have several servers: In a production environment this will be the norm, so if you have several servers, put them separated by commas so that all of them are taken into account (although it actually connects to the first one that succeeds and then Zookeeper handles the rest).

Consumer in Python

For its part, to consume messages, we import and use “KafkaConsumer” (KafkaConsumer documentation at KafkaConsumer (kafka-python docs) ). Just as with “KafkaProducer”, we define in “bootstrap_servers” the Kafka servers we will consume messages from. In the following example I create in the variable “consumidor” a KafkaConsumer object, which is an iterator, so I can loop over it with a “for” loop that returns messages “msg” (which hold all the information, from the value we produced earlier to the topic it comes from, etc.), and from which I can get the value entered earlier by calling “value” (a byte array is received, which in Python can be deserialized with the “decode” method):

from kafka import KafkaConsumer

servidores_bootstrap = '27.0.173.161:9092'
topic = 'miNuevoTopic'

consumidor = KafkaConsumer(topic, bootstrap_servers=servidores_bootstrap)

for msg in consumidor:
    print(msg.value)

This code prints:

b'Un mensaje desde Python'
b'Otro mensaje desde Python'
b'Un tercer mensaje desde Python'

Note about the “consumer group”: When initializing the “KafkaConsumer”, we can pass it a keyword argument “group_id” (which is None by default):

consumidor = KafkaConsumer(topic, bootstrap_servers=servidores_bootstrap, group_id='idGrupoConsumidor')

Java

In Java we need to install the Kafka-Clients dependency (documentation at https://kafka.apache.org/documentation/#producerapi and Kafka Java Client (Confluent Documentation)), which we can find at org.apache.kafka » kafka-clients (Maven Repository) (pick the latest version or the one you want). We can install this dependency with Maven’s Pom.xml (to learn what Maven is, we have a dedicated article https://jarroba.com/maven/, or if you want to set up a quick Maven project from the command line, I leave you the way to do it in a box called “Install Maven on Linux from the terminal and create a quick archetype” in this article about installing Apache Spark) by adding (I will install version 3.1.0; you should change it to the version you need):

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.1.0</version>
</dependency>

Producer in Java

We need to use/import the “org.apache.kafka.clients.producer” package. First we set the properties (using the Properties class) with the servers in BOOTSTRAP_SERVERS_CONFIG, VALUE_SERIALIZER_CLASS_CONFIG and KEY_SERIALIZER_CLASS_CONFIG. In Kafka Producers, every time we send a message the message key is always sent (if we don’t declare the key, a null is sent, which must also be serialized), so you always have to specify the serializers for both the key and the message; for example, if we want to convert a String into a byte array, we can use the Java class “StringSerializer” and assign it both to the VALUE_SERIALIZER_CLASS_CONFIG property (for the messages) and to KEY_SERIALIZER_CLASS_CONFIG (for the message keys). Next, we create a new KafkaProducer with the previous properties and send our messages with the “send” method, which requires a ProducerRecord object where we put the topic and the message to send. At the end, it is important to close the connection with “close”, but before that we send any remaining messages with “flush” (since with “send” messages are kept in a buffer until it reaches a certain size and then they are all sent together as a batch of messages, so before closing the connection, we try to send the messages in the buffer with “flush”).

import java.util.Properties;

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 ProductorKafka {

    public static void main(String[] args) {
        String servidoresBootstrap = "27.0.173.161:9092";
        String topic = "miNuevoTopic";

        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servidoresBootstrap);
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> productor = new KafkaProducer<>(props);
        try {
            productor.send(new ProducerRecord<>(topic, "Un mensaje desde Java"));
            productor.send(new ProducerRecord<>(topic, "Otro mensaje desde Java"));
            productor.send(new ProducerRecord<>(topic, "Un tercer mensaje desde Java"));
        } finally {
            productor.flush();
            productor.close();
        }

    }

}

Note about keys: we can extend the send method to add a message key to each message, for example:

productor.send(new ProducerRecord<>(topic, "clave", "Un mensaje desde Java"));

Consumer in Java

We need to use/import the “org.apache.kafka.clients.consumer” package. First we set the properties (using the Properties class) with the servers in BOOTSTRAP_SERVERS_CONFIG, VALUE_DESERIALIZER_CLASS_CONFIG and KEY_DESERIALIZER_CLASS_CONFIG (Note: Consumers DEserialize, unlike Producers, which serialize). To deserialize we use the Java class “StringDeserializer”. We create a new KafkaConsumer with the previous properties, and we subscribe this consumer to the topic with “subscribe” (you can subscribe to several topics at once, which is why it asks for a collection, even if we pass it a single topic). To fetch the messages we use the “poll” method (you can pass it a timeout; I pass it 0 seconds because in this example I want it to run immediately, but ideally this code should be in an infinite loop so it always fetches messages, waiting a bit between calls so as not to saturate the network and to use fewer resources). Then we create a “for” loop to go through all the messages that “poll” obtained in the previous call. Each message comes wrapped in a “ConsumerRecord” object, and you can get the message String with the “value” method (you can also get the key if you want, with “key”):

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class ConsumidorKafka {

    public static void main(String[] args) {

        String servidoresBootstrap = "27.0.173.161:9092";
        String topic = "miNuevoTopic";

        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servidoresBootstrap);
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        
        KafkaConsumer<String, String> consumidor = new KafkaConsumer<>(props);
        try {
            consumidor.subscribe(Collections.singleton(topic));

            ConsumerRecords<String, String> mensajes = consumidor.poll(Duration.ZERO);

            for (ConsumerRecord<String, String> mensaje: mensajes) {
                System.out.println(mensaje.value());
            }
        } finally {
            consumidor.close();
        }

    }

}

This code prints:

Un mensaje desde Python
Otro mensaje desde Python
Un tercer mensaje desde Python

Note about the “consumer group”: We can extend the properties with a new one whose key is GROUP_ID_CONFIG, like this:

props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "miGrupoConsumidor");

Note if we want to reset the Offset: for testing we will want it to always return data from the beginning, so we set the AUTO_OFFSET_RESET_CONFIG property to “earliest”:

props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

Note for creating a continuous listening service: As mentioned before, it is enough to wrap the “poll” (I made it wait 1 second on each loop iteration before requesting more data) and the “for” that reads the messages in a “while(true)”, for example:

while (true) {
    ConsumerRecords<String, String> mensajes = consumidor.poll(Duration.ofSeconds(1));

     for (ConsumerRecord<String, String> mensaje: mensajes) {
          System.out.println(mensaje.value());
     }
}

References