Simple Kafka Producer and Consumer using Java

📨 Producing and Consuming Messages in Apache Kafka using Java

Apache Kafka is a high-performance, distributed event streaming platform that enables applications to publish and subscribe to data streams in real time.
In this post, we’ll walk through how to set up Kafka locally, and create a simple Java producer and consumer to exchange messages.

---

⚙️ Step 1 — Download Apache Kafka

Download the latest stable version of Apache Kafka from the official website:

👉 https://kafka.apache.org/downloads

After downloading, extract the .tgz file:

tar -xzf kafka_2.13-3.7.0.tgz
cd kafka_2.13-3.7.0

---

🚀 Step 2 — Start Zookeeper and Kafka Server

Kafka requires Zookeeper to manage its brokers (in older versions).
Start both Zookeeper and Kafka services using the default configuration files.

# Start Zookeeper
./bin/zookeeper-server-start.sh config/zookeeper.properties

# Start Kafka broker
./bin/kafka-server-start.sh config/server.properties

✅ Once both are running, Kafka is ready to accept producer and consumer connections.

---

📦 Step 3 — Create a Maven Project

Create a new Java Maven project and add the following dependency to your pom.xml file:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>0.10.1.0</version>
</dependency>

💡 Note: This example uses an older Kafka client (0.10.x).
For newer versions, use org.apache.kafka:kafka-clients (post-0.11), but the logic remains similar.

---

🧠 Step 4 — Write Java Code for Producer and Consumer

Below is a simple example that sends a message to a Kafka topic (order) and then consumes it.

🧩 MyKafkaConsumer.java

package com.vinod.test;

import java.util.*;
import kafka.consumer.*;
import kafka.producer.*;
import kafka.javaapi.producer.Producer;
import kafka.javaapi.consumer.ConsumerConnector;

/**
 * Simple example to produce and consume messages using Apache Kafka.
 *
 * <p>This example demonstrates:
 * <ul>
 *   <li>Creating a Kafka producer to send messages to a topic.</li>
 *   <li>Creating a Kafka consumer to read messages from that topic.</li>
 * </ul>
 *
 * @author vinod
 */
public class MyKafkaConsumer {
    private ConsumerConnector consumer;

    /** 
     * Method to create a Kafka producer and send a message.
     */
    public void testProducer() {
        Properties properties = new Properties();
        properties.put("metadata.broker.list", "localhost:9092");
        properties.put("serializer.class", "kafka.serializer.StringEncoder");

        ProducerConfig producerConfig = new ProducerConfig(properties);
        Producer<String, String> producer = new Producer<>(producerConfig);

        KeyedMessage<String, String> message =
            new KeyedMessage<>("order", "Sending Customer Order, please process");

        producer.send(message);
        System.out.println("✅ Message sent successfully to 'order' topic!");
    }

    /**
     * Method to create a Kafka consumer and read messages from the topic.
     */
    public void testConsumer() {
        String topic = "order";
        Properties props = new Properties();
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "vinod");
        props.put("zookeeper.session.timeout.ms", "5000");
        props.put("zookeeper.sync.time.ms", "250");
        props.put("auto.commit.interval.ms", "1000");

        consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));

        Map<String, Integer> topicCount = new HashMap<>();
        topicCount.put(topic, 1);

        Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams =
            consumer.createMessageStreams(topicCount);

        List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(topic);

        for (KafkaStream<byte[], byte[]> stream : streams) {
            ConsumerIterator<byte[], byte[]> it = stream.iterator();
            while (it.hasNext()) {
                String message = new String(it.next().message());
                System.out.println("📩 Message from topic '" + topic + "': " + message);
            }
        }

        if (consumer != null) {
            consumer.shutdown();
        }
    }

    public static void main(String[] args) {
        MyKafkaConsumer app = new MyKafkaConsumer();
        app.testProducer();  // Send message
        app.testConsumer();  // Read message
    }
}

---

🧾 Step 5 — Example Output

When you run the program, the following output will appear on your console:

✅ Message sent successfully to 'order' topic!
📩 Message from topic 'order': Sending Customer Order, please process

---

📘 Step 6 — Download Full Example

You can download or clone the working example from GitHub:

🔗 https://github.com/kkvinodkumaran/kafka

---

🧠 How It Works — High-Level Flow

Here’s the conceptual flow of how this example operates:

[Producer] → (Message) → [Kafka Broker] → (Stored in topic: order) → [Consumer]

• Producer connects to the broker and publishes the message.

• Broker stores the message in the specified topic partition.

• Consumer subscribes to that topic and retrieves messages sequentially.

---

🧩 Key Takeaways

Concept	Description
Producer	Sends data to Kafka topics
Consumer	Reads data from Kafka topics
Broker	Kafka server that stores and delivers messages
Topic	Named channel for message streams
Zookeeper	Manages broker metadata (for older Kafka versions)

---

Reference: Kafka