Apache Kafka: A Primer with Java Examples

Introduction

Apache Kafka has emerged as a powerhouse for real-time data streaming, messaging, and event processing in distributed systems. Its ability to handle massive volumes of data with high throughput and reliability makes it a go-to choice for a variety of use cases, including:

• Real-time analytics: Process data streams to make quick decisions.
• Log aggregation: Collect logs from across systems for centralized monitoring.
• Microservices communication: Facilitate communication between decoupled services.
• IoT data pipelines: Ingest and process sensor data from numerous devices.

Kafka Core Concepts

Before diving into code, let’s get familiar with Kafka’s terminology:

• Topics: Logical streams of data categorized by name.
• Producers: Applications that write data (messages) to Kafka topics.
• Consumers: Applications that read data from Kafka topics.
• Brokers: Kafka servers that manage the storage and replication of data.
• Clusters: A group of Kafka brokers working together.

Setting Up a Development Environment

1. Download Kafka: Head over to and grab the latest binary release.
2. Start Kafka:

• • Unzip the downloaded file.
  • From the extracted directory, run the Zookeeper server (necessary for Kafka coordination): bin/zookeeper-server-start.sh config/zookeeper.properties
  • In a separate terminal, run the Kafka broker: bin/kafka-server-start.sh config/server.properties

1. Java Dependencies: Include the Kafka-clients dependency in your project (Maven example):
2. XML
3. <dependency>
4.     <groupId>org.apache.kafka</groupId>
5.     <artifactId>kafka-clients</artifactId>
6.     <version>3.3.1</version> </dependency>
7. Use code 
8. content_copy

Simple Java Kafka Producer

Java

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class MyKafkaProducer {

    public static void main(String[] args) {

        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”);

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {

            for (int i = 0; i < 10; i++) {

                String key = “key-” + i;

                String value = “message-” + i;

                ProducerRecord<String, String> record = new ProducerRecord<>(“my-topic”, key, value);

                producer.send(record);

            }

            System.out.println(“Messages sent successfully!”);

        } 

    }

}

Use code 

content_copy

Explanation

• Properties: Configuration for connecting to your Kafka cluster (bootstrap. servers), and defining how to serialize message keys and values.
• KafkaProducer: The core class for sending messages.
• ProducerRecord: A key-value pair representing a single message.

Simple Java Kafka Consumer

Java

import org.apache.kafka.clients.consumer.ConsumerConfig;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;

import java.util.Collections;

import java.util.Properties;

public class MyKafkaConsumer {

    public static void main(String[] args) {

        // … (Configuration similar to producer) …

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {

            consumer.subscribe(Collections.singletonList(“my-topic”));

            while (true) {

                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                records.forEach(record -> System.out.println(“Key: ” + record.key() + “, Value: ” + record.value()));

            }

        } 

    }

}

Use code