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

Apache Camel Kafka Component Example

Apache camel API has the inbuilt kafka component and it is very simple to create producer, consumer and process messages. Here is one simple Kafka producer and consumer example using Apache camel and Kafka.

Steps

1) Download Apache kafka from https://kafka.apache.org/downloads

2) Extract the tar and start the zookeeper and kafka 

 ./zookeeper-server-start.sh ../config/zookeeper.properties

 ./kafka-server-start.sh ../config/server.properties

3. Create a Maven project and add below dependencies

<dependencies>
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-core</artifactId>
			<version>2.16.3</version>
		</dependency>
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-kafka</artifactId>
			<version>2.16.3</version>
		</dependency>
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-jetty</artifactId>
			<version>2.16.3</version>
		</dependency>
		<dependency>
			<groupId>org.scala-lang</groupId>
			<artifactId>scala-library</artifactId>
			<version>2.11.0</version>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-api</artifactId>
			<version>1.6.6</version>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-log4j12</artifactId>
			<version>1.6.6</version>
		</dependency>
		<dependency>
			<groupId>log4j</groupId>
			<artifactId>log4j</artifactId>
			<version>1.2.17</version>
		</dependency>
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-test</artifactId>
			<version>2.17.3</version>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>com.fasterxml.jackson.core</groupId>
			<artifactId>jackson-databind</artifactId>
		</dependency>
	</dependencies>

4. Create a camel main class to add camel routes and its processors.

package com.vinod.test;

import org.apache.camel.Exchange;
import org.apache.camel.Main;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;

public class CamelKafkaConsumerTest {

    public static void main(String... args) throws Exception {
	Main main = new Main();
	main.enableHangupSupport();
	main.addRouteBuilder(new MyCamelJettyBuilder());
	main.run(args);
    }
}

class MyCamelJettyBuilder extends RouteBuilder {
    String topicName = "topic=test";
    String kafkaServer = "kafka:localhost:9092";
    String zooKeeperHost = "zookeeperHost=localhost&zookeeperPort=2181";
    String serializerClass = "serializerClass=kafka.serializer.StringEncoder";
    String autoOffsetOption = "autoOffsetReset=smallest";
    String groupId = "groupId=testingvinod";

    String toKafka = new StringBuilder().append(kafkaServer).append("?").append(
	    topicName).append("&").append(zooKeeperHost).append("&").append(
		    serializerClass).toString();

    String fromKafka = new StringBuilder().append(toKafka).append("&").append(
	    autoOffsetOption).append("&").append(groupId).toString();

    public void configure() {
	from("jetty:http://localhost:8182/mytestservice").process(
		new Processor() {
		    public void process(Exchange exchange) throws Exception {
			String message = exchange.getIn().getBody(String.class);
			exchange.getIn().setBody(message, String.class);
			exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY,
				0);
			exchange.getIn().setHeader(KafkaConstants.KEY, "1");
		    }
		}).to(toKafka);
	from(fromKafka).process(new Processor() {
	    public void process(Exchange exchange) throws Exception {
		if (exchange.getIn() != null) {
		    Message message = exchange.getIn();
		    String data = message.getBody(String.class);
		    System.out.println("Data =" + data.toString());
		}
	    }
	});

    }
}

In this example exposed a jetty end point to give the input message instead of static message, once started the above program use any rest client to test the program.

5) Send message to Kafka topic using our service .. http://localhost:8182/mytestservice

now we can see the output which we sent and consumed from Kafka in the console

Data  =Test my message by Vinod

6) Download Example

https://github.com/kkvinodkumaran/camel

Reference: Apache Camel, Kafka

Event driven programming using Spring Boot and Reactor Example

Core concepts

Reactor:

Reactor is a framework to make event driven programming much easier and it is based on Reactor Design Pattern. Reactor is good for asynchronous applications on the JVM, it is an event gateway where event consumers are registered with a notification key.

Selector:

Selector is an abstraction to find consumer by invoking event.

Consumers and Event:

Consumers and Events as core module, Consumer is event consumer which needs to be notified for the event.

Producer:

Producer produces the Events and publish.

Here is one example which use the Reactor pattern to Message produce and consumes.

1. Create a Maven project and add below dependencies

<properties>
        <java.version>1.8</java.version>
        <version.reactor>2.0.6.RELEASE</version.reactor>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-bus</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>org.reactivestreams</groupId>
            <artifactId>reactive-streams</artifactId>
            <version>1.0.0.final</version>
        </dependency>

        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-stream</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-groovy</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor.spring</groupId>
            <artifactId>reactor-spring-core</artifactId>
            <version>${version.reactor}</version>
        </dependency>
        <dependency>
            <groupId>io.projectreactor.spring</groupId>
            <artifactId>reactor-spring-context</artifactId>
            <version>${version.reactor}</version>
        </dependency>

    </dependencies>

2. Create Customer POJO

package com.vinod.test;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Customer {
    private String name;
    private String address;
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getAddress() {
        return address;
    }
    public void setAddress(String address) {
        this.address = address;
    }
    @Override
    public String toString() {
        return "Customer [name=" + name + ", address=" + address + "]";
    }

}

3. Create a Consumer

package com.vinod.test;
import org.springframework.stereotype.Service;
import reactor.bus.Event;
import reactor.fn.Consumer;

@Service
class CustomerReceiver implements Consumer<Event<Customer>> {

    public void accept(Event<Customer> ev) {
        System.out.println("Customer " + ev.getData());
    }

}
 

4. Create a publisher

package com.vinod.test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.bus.Event;
import reactor.bus.EventBus;

@Service
public class CustomerPublisher {

    @Autowired
    EventBus eventBus;

    public void publishCustomerDetails() throws InterruptedException {
        Customer customer = new Customer();
        customer.setName("vinod");
        customer.setAddress("Sasi area");
        eventBus.notify("customer", Event.wrap(customer));
        System.out.println("Message sent");
    }

}

5. Spring boot Main class

package com.vinod.test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import reactor.Environment;
import reactor.bus.EventBus;

import static reactor.bus.selector.Selectors.$;

@Configuration
@EnableAutoConfiguration
@ComponentScan
public class Application implements CommandLineRunner {

    @Bean
    Environment env() {
        return Environment.initializeIfEmpty().assignErrorJournal();
    }

    @Bean
    EventBus createEventBus(Environment env) {
        return EventBus.create(env, Environment.THREAD_POOL);
    }

    @Autowired
    private EventBus eventBus;
    @Autowired
    private CustomerReceiver customerReceiver;
    @Autowired
    private CustomerPublisher customerPublisher;
    @Autowired
    private AdminReceiver adminReceiver;

    public void run(String... args) throws Exception {
        eventBus.on($("customer"), customerReceiver);
        customerPublisher.publishCustomerDetails();
    }

    public static void main(String[] args) throws InterruptedException {
        SpringApplication.run(Application.class, args);

    }

}

6. Run the program (Application.java)

Here we can see the publisher sends the message and consumer is consuming the same.

Message sent
Customer Customer [name=vinod, address=Sasi area]

7. Download Example

https://github.com/kkvinodkumaran/spring

Java Drools Example

Drools is a open source Business Logic integration Platform based on java from JBoss and RedHat. Here is one simple example to execute business logics based on rules. see more about drools

1. Create a maven project and add below dependencies

 
  <dependencies>
        <dependency>
            <groupId>org.drools</groupId>
            <artifactId>drools-compiler</artifactId>
            <version>6.0.0.CR1</version>
        </dependency>
        <dependency>
            <groupId>org.drools</groupId>
            <artifactId>drools-core</artifactId>
            <version>6.0.0.CR1</version>
        </dependency>
    </dependencies>

2. Create a .drl file to add our business logic (eligibility.drl)

import com.vinod.test.Customer
 
dialect "mvel"
 
rule "Customer eligibility rule"
    when
        $customer : Customer(age>=21)              
    then
        System.out.println($customer.name" Eligibile for benefits");
end

3. Create Customer pojo class

package com.vinod.test;

public class Customer {
    private String name;
    private int age;
    public Customer() {
    }
    public Customer(String name, int age) {
        this.name = name;
        this.age = age;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public int getAge() {
        return age;
    }
    public void setAge(int age) {
        this.age = age;
    }
}
 

4. Create a main class to load and fire rules

package com.vinod.test;

import java.io.File;
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.List;

import org.kie.api.KieServices;
import org.kie.api.builder.KieBuilder;
import org.kie.api.builder.KieFileSystem;
import org.kie.api.builder.KieRepository;
import org.kie.api.io.KieResources;
import org.kie.api.io.Resource;
import org.kie.api.runtime.KieContainer;
import org.kie.api.runtime.KieSession;

public class CustomerEligibilityRuleTest {

    public static void main(String[] args) {
        List<File> rules = new ArrayList<File>();
        File currentdir = new File(System.getProperty("user.dir"));
        File[] listFiles = currentdir.listFiles(new FilenameFilter() {
            public boolean accept(File dir, String name) {
                return (name.endsWith(".drl"));
            }
        });
        for (File listFile : listFiles) {
            System.out.println(listFile.getName());
            rules.add(listFile);
        }
        List<Customer> customers = new ArrayList<Customer>();
        customers.add(new Customer("Vinod k kumaran", 21));
        executeRules(rules, customers);
    }

    public static void executeRules(List<File> files, List<Customer> customers) {
        try {
            KieServices kieServices = KieServices.Factory.get();
            KieResources kieResources = kieServices.getResources();
            KieFileSystem kieFileSystem = kieServices.newKieFileSystem();
            KieRepository kieRepository = kieServices.getRepository();
            for (File ruleFile : files) {
                Resource resource = kieResources.newFileSystemResource(ruleFile);
                kieFileSystem.write(resource);
            }
            KieBuilder kb = kieServices.newKieBuilder(kieFileSystem);
            kb.buildAll();
            KieContainer kContainer = kieServices.newKieContainer(kieRepository.getDefaultReleaseId());
            KieSession kSession = kContainer.newKieSession();
            kSession.addEventListener(new CustomerRuleTracker());
            for (Object customer : customers) {
                kSession.insert(customer);
            }
            kSession.fireAllRules();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 

5. Add rule tracker class to listen all stages of firing rules 

This class is added in the KieSession already in the above main class

package com.vinod.test;

import org.drools.core.event.BeforeActivationFiredEvent;
import org.kie.api.event.rule.AfterMatchFiredEvent;
import org.kie.api.event.rule.BeforeMatchFiredEvent;
import org.kie.api.event.rule.DefaultAgendaEventListener;
import org.kie.api.event.rule.MatchCancelledEvent;
import org.kie.api.event.rule.MatchCreatedEvent;
import org.kie.api.event.rule.MatchEvent;
import org.kie.api.event.rule.RuleFlowGroupActivatedEvent;

public class CustomerRuleTracker extends DefaultAgendaEventListener {

    @Override
    public void afterMatchFired(AfterMatchFiredEvent event) {
        System.out.println("Executing afterMatchFired ");
    }
    @Override
    public void beforeRuleFlowGroupActivated(RuleFlowGroupActivatedEvent event) {
        System.out.println("Executing beforeRuleFlowGroupActivated ");

    }
    @Override
    public void beforeMatchFired(BeforeMatchFiredEvent event) {
        System.out.println("Executing beforeMatchFired ");
    }
    public void beforeActivationFired(BeforeActivationFiredEvent event) {
        System.out.println("Executing beforeActivationFired ");

    }
    @Override
    public void matchCreated(MatchCreatedEvent event) {
        System.out.println("Executing matchCreated ");
        registerActions(event);
    }
    @Override
    public void matchCancelled(MatchCancelledEvent event) {
        System.out.println("Executing matchCancelled");

    }
    private void registerActions(MatchEvent event) {
        System.out.println("Executing registerActions");
    }
}

6. Run the main program.

We can see the customer name + eligible for benefits printed as per our rules..

eligibility.drl

Executing matchCreated
Executing registerActions
Executing beforeMatchFired
Vinod k kumaran Eligibile for benefits
Executing afterMatchFired

7. Download Example

https://github.com/kkvinodkumaran/drools