Working with Kafka in Java: Libraries, Configuration, and Code Examples

Working with Kafka in Java: Libraries, Configuration, and Code Examples

Apache Kafka is a distributed streaming platform widely used for building high-throughput messaging systems, log collection, real-time event processing, and microservice integration. In the Java ecosystem, Kafka holds a special place thanks to the official Kafka Clients library and a variety of additional tools. In this article, we will thoroughly examine how to work with Kafka in Java, which libraries to use, and provide practical code examples.

Key Libraries for Working with Kafka in Java

For integrating Java applications with Apache Kafka, there are several key libraries:

• Kafka Clients — the official library from Apache, providing APIs for producers and consumers. This is the minimal and most reliable way to interact with brokers.
• Spring Kafka — an extension for the Spring Framework that simplifies configuring and using Kafka through annotations and templates (KafkaTemplate, @KafkaListener).
• Kafka Streams — a library for building stream processing applications in Java, without the need for Spark or Flink. It allows filtering, aggregation, and joining of streams.
• Akka Streams Kafka (Alpakka) — a reactive library for working with Kafka based on Akka Streams, suitable for asynchronous and non-blocking applications.

In this guide, we will focus on the official Kafka Clients library, as it is the foundation for all other tools.

Adding Dependencies (Maven)

To get started, add the following dependency to your pom.xml:

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

If you are using Gradle, add:

implementation 'org.apache.kafka:kafka-clients:3.7.0'

Configuring the Producer

The producer sends messages to Kafka topics. Let's look at a simple example of creating and sending a message.

import org.apache.kafka.clients.producer.*;import java.util.Properties;

public class SimpleProducer {    public static void main(String[] args) {        // Configure producer properties        Properties props = new Properties();        props.put("bootstrap.servers", "localhost:9092");        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");        props.put("acks", "all"); // Wait for acknowledgment from all replicas        props.put("retries", 3);  // Number of retry attempts on error

        // Create the producer        Producer<String, String> producer = new KafkaProducer<>(props);

        // Send a message        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "Hello, Kafka from Java!");        producer.send(record, (metadata, exception) -> {            if (exception == null) {                System.out.println("Message sent to topic: " + metadata.topic() +                                   ", partition: " + metadata.partition() +                                   ", offset: " + metadata.offset());            } else {                System.err.println("Send error: " + exception.getMessage());            }        });

        // Close the producer        producer.close();    }}

In this example, we set the broker address (localhost:9092), serializers for the key and value (in this