How to build your first Apache KafkaConsumer application

Question:

How do I get started in building my first Kafka consumer application?

Edit this page

Example use case:

You'd like to integrate an Apache KafkaConsumer in your event-driven application, but you're not sure where to start. In this tutorial you'll build a small application reading records from Kafka with a KafkaConsumer. You can use the code in this tutorial as an example of how to use an Apache Kafka consumer

Code example:

Try it

1
Initialize the project

To get started, make a new directory anywhere you’d like for this project:

mkdir kafka-consumer-application && cd kafka-consumer-application

2
Get Confluent Platform

Next, create the following docker-compose.yml file to obtain Confluent Platform:

---
version: '2'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:5.5.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-enterprise-kafka:5.5.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TOOLS_LOG4J_LOGLEVEL: ERROR
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:9092
      CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'

And launch it by running:

docker-compose up -d

3
Create a topic

In this step we’re going to create a topic for use during this tutorial.

But first, you’re going to open a shell on the broker docker container.

Open a new terminal and window then run this command:

docker-compose exec broker bash

Now use the following command to create the topic:

kafka-topics --create --topic input-topic --bootstrap-server broker:9092 --replication-factor 1 --partitions 1

Keep this terminal window open as you’ll need to run a console-producer in a few steps.

4
Configure the project

Create the following Gradle build file, named build.gradle for the project:

buildscript {
    repositories {
        jcenter()
    }
    dependencies {
        classpath "com.commercehub.gradle.plugin:gradle-avro-plugin:0.15.1"
        classpath "com.github.jengelman.gradle.plugins:shadow:4.0.2"
    }
}

plugins {
    id "java"
    id "com.google.cloud.tools.jib" version "1.1.1"
    id "idea"
    id "eclipse"
}

sourceCompatibility = "1.8"
targetCompatibility = "1.8"
version = "0.0.1"

repositories {
    jcenter()

    maven {
        url "https://packages.confluent.io/maven"
    }
}

apply plugin: "com.github.johnrengelman.shadow"

dependencies {
    implementation "org.slf4j:slf4j-simple:1.7.26"
    implementation "org.apache.kafka:kafka-streams:2.5.0"
    testImplementation "junit:junit:4.12"
    testImplementation 'org.hamcrest:hamcrest:2.2'
}

test {
    testLogging {
        outputs.upToDateWhen { false }
        showStandardStreams = true
        exceptionFormat = "full"
    }
}

jar {
  manifest {
    attributes(
      "Class-Path": configurations.compileClasspath.collect { it.getName() }.join(" "),
      "Main-Class": "io.confluent.developer.KafkaConsumerApplication"
    )
  }
}

shadowJar {
    archiveBaseName = "kafka-consumer-application-standalone"
    archiveClassifier = ''
}

And be sure to run the following command to obtain the Gradle wrapper:

gradle wrapper

Next, create a directory for configuration data:

mkdir configuration

Then create a development configuration file at configuration/dev.properties:

# Consumer properties
bootstrap.servers=localhost:29092
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
max.poll.interval.ms=300000
enable.auto.commit=true
auto.offset.reset=earliest
group.id=consumer-application

# Application specific properties
file.path=consumer-records.out
input.topic.name=input-topic
input.topic.partitions=1
input.topic.replication.factor=1

Let’s do a quick overview of some of the more important properties here:

The key.deserializer and value.deserializer properties provide a class implementing the Deserializer interface for converting byte arrays into the expected object type of the key and value respectively.

The max.poll.interval.ms is the maximum amount of time a consumer may take between calls to Consumer.poll(). If a consumer instance takes longer than the specified time, it’s considered non-reponsive and removed from the consumer-group triggering a rebalance.

Setting enable.auto.commit configuration to true enables the Kafka consumer to handle commiting offsets automatically for you. The default setting is true, but it’s included here to make it explicit. When you enable auto commit, you need to ensure you’ve processed all records before the consumer calls poll again. Once there is a subsequent call to poll, all the records returned from the previous call are considered processed and the consumer commits the offsets.

auto.offset.reset - If a consumer instance can’t locate any offsets for its topic-partition assignment(s), it will resume processing from the earliest available offset.

group.id - Kafka uses the concept of a consumer-group which is used to represent a logical single group. A consumer-group can be made up of mulitple members all sharing the same group.id configuration. As members leave or join the consumer-group, the group-coordinator triggres a rebalance which causes topic-partion reassignment among active members of the group.

5
Create the Kafka Consumer Application

Create a directory for the Java files in this project:

mkdir -p src/main/java/io/confluent/developer

To complete this tutorial, you’ll build a main application class and a helper class

First, you’ll create the main application,KafkaConsumerApplication, which is the focal point of this tutorial; consuming records from a Kafka topic.

Let’s go over some of the key parts of the KafkaConsumerApplication starting with the constructor:

KafkaConsumerApplication constructor
public KafkaConsumerApplication(final Consumer<String, String> consumer,
                                final ConsumerRecordsHandler<String, String> recordsHandler) { (1)
    this.consumer = consumer;
    this.recordsHandler = recordsHandler;
}
1 Here you’re suppling instances of the Consumer and ConsumerRecordsHandler via constructor parameters.

By using interfaces vs. concreate implentations you can more easily test the KafkaConsumerApplication class by swapping in a MockConsumer for the test. We’ll cover testing in an upcoming section. Also, interfaces make it simple to change ConsumerRecord handling at run-time.

In this tutorial you’ll inject the dependencies in the KafkaConsumerApplication.main() method, but in practice you may want to use a dependency injection framework library, such as the Spring Framework.

Next, let’s review the KafkaConsumerApplication.runConsumer() method, which provides the core functionality of this tutorial.

KafkaConsumerApplication.process
  public void runConsume(final Properties consumerProps) {
    try {
      consumer.subscribe(Collections.singletonList(consumerProps.getProperty("input.topic.name"))); (1)
      while (keepConsuming) { (2)
        final ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));  (3)
        recordsHandler.process(consumerRecords); (4)
      }
    } finally {
      consumer.close(); (5)
    }
  }
1 Subscribing to the Kafka topic.
2 Using an instance variable keepConsuming to run the Kafka consumer indefinitely. The KafkaConsumerApplication.shutdown() method sets keepConsuming to false.
3 Polling for new records, waiting at most one second for new records. The Consumer.poll() method may return zero results. The consumer is expected to call poll() again within five minutes, from the max.poll.interval.ms config described in step three, "Configure the project".
4 Handing off the polled ConsumerRecords to the ConsumerRecordsHandler interface.
5 Closing the consumer is essential to prevent resource leaking, hence the finally block.

Now go ahead and create the src/main/java/io/confluent/developer/KafkaConsumerApplication.java file:

package io.confluent.developer;


import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class KafkaConsumerApplication {

  private volatile boolean keepConsuming = true;
  private ConsumerRecordsHandler<String, String> recordsHandler;
  private Consumer<String, String> consumer;

  public KafkaConsumerApplication(final Consumer<String, String> consumer,
                                  final ConsumerRecordsHandler<String, String> recordsHandler) {
    this.consumer = consumer;
    this.recordsHandler = recordsHandler;
  }

  public void runConsume(final Properties consumerProps) {
    try {
      consumer.subscribe(Collections.singletonList(consumerProps.getProperty("input.topic.name")));
      while (keepConsuming) {
        final ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
        recordsHandler.process(consumerRecords);
      }
    } finally {
      consumer.close();
    }
  }

  public void shutdown() {
    keepConsuming = false;
  }

  public static Properties loadProperties(String fileName) throws IOException {
    final Properties props = new Properties();
    final FileInputStream input = new FileInputStream(fileName);
    props.load(input);
    input.close();
    return props;
  }

  public static void main(String[] args) throws Exception {

    if (args.length < 1) {
      throw new IllegalArgumentException(
          "This program takes one argument: the path to an environment configuration file.");
    }

    final Properties consumerAppProps = KafkaConsumerApplication.loadProperties(args[0]);
    final String filePath = consumerAppProps.getProperty("file.path");
    final Consumer<String, String> consumer = new KafkaConsumer<>(consumerAppProps);
    final ConsumerRecordsHandler<String, String> recordsHandler = new FileWritingRecordsHandler(Paths.get(filePath));
    final KafkaConsumerApplication consumerApplication = new KafkaConsumerApplication(consumer, recordsHandler);

    Runtime.getRuntime().addShutdownHook(new Thread(consumerApplication::shutdown));

    consumerApplication.runConsume(consumerAppProps);
  }

}

6
Create supporting classes

To complete this tutorial, you’ll need to also create an interface for a helper class.

First create the interface at src/main/java/io/confluent/developer/ConsumerRecordsHandler.java

package io.confluent.developer;

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

public interface ConsumerRecordsHandler<K, V> {
   void process(ConsumerRecords<K, V> consumerRecords);
}

Using an interface will make it easier to change how you want to work with ConsumerRecords without having to modify all of your existing code.

Next you’ll create an implementation of the ConsumerRecordsHandler interface named FileWritingRecordsHandler, but before you do that, let’s take a peek under the hood to understand how the helper class works.

The FileWritingRecordsHandler is a simple class that writes values of consumed records to a file, it’s worth a quick review of the process method:

FileWritingRecordsHandler.process
 @Override
  public void process(final ConsumerRecords<String, String> consumerRecords) {
    final List<String> valueList = new ArrayList<>();
    consumerRecords.forEach(record -> valueList.add(record.value())); (1)
    if (!valueList.isEmpty()) {  (2)
      try {
        Files.write(path, valueList, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.APPEND);  (3)
      } catch (IOException e) {
          throw new RuntimeException(e);
      }
    }
  }
1 Iterate over all of the records and store the value in a List
2 If the isn’t empty let’s do something!
3 Pass the List<String> of records to the Files.write() method

In practice you’re certain to do a more realistic workload.

Now go ahead and create the src/main/java/io/confluent/developer/FileWritingRecordsHandler.java file:

package io.confluent.developer;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.List;
import org.apache.kafka.clients.consumer.ConsumerRecords;

public class FileWritingRecordsHandler implements ConsumerRecordsHandler<String, String> {

  private final Path path;

  public FileWritingRecordsHandler(final Path path) {
    this.path = path;
  }

  @Override
  public void process(final ConsumerRecords<String, String> consumerRecords) {
    final List<String> valueList = new ArrayList<>();
    consumerRecords.forEach(record -> valueList.add(record.value()));
    if (!valueList.isEmpty()) {
      try {
        Files.write(path, valueList, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.APPEND);
      } catch (IOException e) {
          throw new RuntimeException(e);
      }
    }
  }
}

7
Compile and run the Kafka Consumer program

In your terminal, run:

./gradlew shadowJar

Now that you have an uberjar for the KafkaConsumerApplication, you can launch it locally. When you run the following, the prompt won’t return, because the application will run until you exit it. There is always another message to process, so streaming applications don’t exit until you force them.

java -jar build/libs/kafka-consumer-application-standalone-0.0.1.jar configuration/dev.properties

8
Produce sample data to the input topic

Using the terminal window you opened in step three, run the following command to start a console-producer:

kafka-console-producer --topic input-topic --broker-list broker:9092

Each line represents input data for the KafkaConsumer application. To send all of the events below, paste the following into the prompt and press enter:

the quick brown fox
jumped over
the lazy dog
Go to Kafka Summit
All streams lead
to Kafka

9
Inspect the consumed records

Your consumer application should have consumed all the records sent and written them out to a file.

In a new terminal, run this command to print the results to the console:

cat consumer-records.out

You should see something like this:

the quick brown fox
jumped over
the lazy dog
Go to Kafka Summit
All streams lead
to Kafka

Test it

1
Create a test configuration file

First, create a test file at configuration/test.properties:

input.topic.name=input-topic
input.topic.partitions=1
input.topic.replication.factor=1

2
Write a test for the consumer application

Create a directory for the tests to live in:

mkdir -p src/test/java/io/confluent/developer

Testing a Kafka consumer application is not too complicated thanks to the MockConsumer.java. Since the KafkaConsumer is well tested, we don’t need to use a live consumer and Kafka broker. We can simply use mock consumer to process some data you’ll feed into it.

There is only one method in KafkaConsumerApplicationTest annotated with @Test, and that is consumerTest(). This method actually runs your KafkaConsumerApplication with the mock consumer.

Now create the following file at src/test/java/io/confluent/developer/KafkaConsumerApplicationTest.java.

package io.confluent.developer;

import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertThat;

import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.TopicPartition;
import org.junit.Test;


public class KafkaConsumerApplicationTest {

  private final static String TEST_CONFIG_FILE = "configuration/test.properties";

  @Test
  public void consumerTest() throws Exception {

    final Path tempFilePath = Files.createTempFile("test-consumer-output", ".out");
    final ConsumerRecordsHandler<String, String> recordsHandler = new FileWritingRecordsHandler(tempFilePath);
    final Properties testConsumerProps = KafkaConsumerApplication.loadProperties(TEST_CONFIG_FILE);
    final String topic = testConsumerProps.getProperty("input.topic.name");
    final TopicPartition topicPartition = new TopicPartition(topic, 0);
    final MockConsumer<String, String> mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);

    final KafkaConsumerApplication consumerApplication = new KafkaConsumerApplication(mockConsumer, recordsHandler);

    // the KafkaConsumerApplication runs synchronously so the test needs to run
    // the application in its own thread
    new Thread(() -> consumerApplication.runConsume(testConsumerProps)).start();
    Thread.sleep(250);
    addTopicPartitionsAssignmentAndAddConsumerRecords(topic, mockConsumer, topicPartition);
    Thread.sleep(250);
    consumerApplication.shutdown();

    final List<String> expectedWords = Arrays.asList("foo", "bar", "baz");
    List<String> actualRecords = Files.readAllLines(tempFilePath);
    assertThat(actualRecords, equalTo(expectedWords));
  }

  private void addTopicPartitionsAssignmentAndAddConsumerRecords(final String topic,
                                 final MockConsumer<String, String> mockConsumer,
                                 final TopicPartition topicPartition) {

    final Map<TopicPartition, Long> beginningOffsets = new HashMap<>();
    beginningOffsets.put(topicPartition, 0L);
    mockConsumer.rebalance(Collections.singletonList(topicPartition));
    mockConsumer.updateBeginningOffsets(beginningOffsets);
    addConsumerRecords(mockConsumer,topic);
  }

  private void addConsumerRecords(final MockConsumer<String, String> mockConsumer, final String topic) {
    mockConsumer.addRecord(new ConsumerRecord<>(topic, 0, 0, null, "foo"));
    mockConsumer.addRecord(new ConsumerRecord<>(topic, 0, 1, null, "bar"));
    mockConsumer.addRecord(new ConsumerRecord<>(topic, 0, 2, null, "baz"));
  }


}

3
Write a test for the records ConsumerRecordsHandler

Now let’s build a test for the ConsumerRecordsHandler implementation used in your application. Even though we have a test for the KafkaConsumerApplication, it’s important that you can test this helper class in isolation.

Create the following file at src/test/java/io/confluent/developer/FileWritingRecordsHandlerTest.java.

package io.confluent.developer;

import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertThat;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.junit.Test;

public class FileWritingRecordsHandlerTest {

  @Test
  public void testProcess() throws IOException {
    final Path tempFilePath = Files.createTempFile("test-handler", ".out");
    try {
      final ConsumerRecordsHandler<String, String> recordsHandler = new FileWritingRecordsHandler(tempFilePath);
      recordsHandler.process(createConsumerRecords());
      final List<String> expectedWords = Arrays.asList("it's but", "a flesh wound", "come back");
      List<String> actualRecords = Files.readAllLines(tempFilePath);
      assertThat(actualRecords, equalTo(expectedWords));
    } finally {
      Files.deleteIfExists(tempFilePath);
    }
  }


  private ConsumerRecords<String, String> createConsumerRecords() {
    final String topic = "test";
    final int partition = 0;
    final TopicPartition topicPartition = new TopicPartition(topic, partition);
    final List<ConsumerRecord<String, String>> consumerRecordsList = new ArrayList<>();
    consumerRecordsList.add(new ConsumerRecord<>(topic, partition, 0, null, "it's but"));
    consumerRecordsList.add(new ConsumerRecord<>(topic, partition, 0, null, "a flesh wound"));
    consumerRecordsList.add(new ConsumerRecord<>(topic, partition, 0, null, "come back"));
    final Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsMap = new HashMap<>();
    recordsMap.put(topicPartition, consumerRecordsList);

    return new ConsumerRecords<>(recordsMap);
  }
}

4
Invoke the tests

Now run the test, which is as simple as:

./gradlew test

Take it to production

1
Create a production configuration file

First, create a new configuration file at configuration/prod.properties with the following content. Be sure to fill in the addresses of your production hosts and change any other parameters that make sense for your setup.

application.id=kafka-consumer-application
bootstrap.servers=<<FILL ME IN>>
schema.registry.url=<<FILL ME IN>>

example.topic.name=<<FILL ME IN>>
example.topic.partitions=<<FILL ME IN>>
example.topic.replication.factor=<<FILL ME IN>>

2
Build a Docker image

In your terminal, execute the following to invoke the Jib plugin to build an image:

gradle jibDockerBuild --image=io.confluent.developer/kafka-consumer-application-join:0.0.1

3
Launch the container

Finally, launch the container using your preferred container orchestration service. If you want to run it locally, you can execute the following:

docker run -v $PWD/configuration/prod.properties:/config.properties io.confluent.developer/kafka-consumer-application-join:0.0.1 config.properties

Deploy on Confluent Cloud

1
Run your app to Confluent Cloud

Instead of running a local Kafka cluster, you may use Confluent Cloud, a fully-managed Apache Kafka service.

First, create your Kafka cluster in Confluent Cloud. Use the promo code C50INTEG to receive an additional $50 free usage (details).

Next, from the Confluent Cloud UI, click on Tools & client config to get the cluster-specific configurations, e.g. Kafka cluster bootstrap servers and credentials, Confluent Cloud Schema Registry and credentials, etc., and set the appropriate parameters in your client application.

Now you’re all set to your run application locally while your Kafka topics and stream processing is backed to your Confluent Cloud instance.