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

Next, create a directory for configuration data:

mkdir configuration

2
Sign up for Confluent Cloud and provision resources

Sign up for Confluent Cloud, a fully-managed Apache Kafka service. Then provision your resources:

  1. After you log in to Confluent Cloud, click on Add cloud environment and name the environment learn-kafka. Using a new environment keeps your learning resources separate from your other Confluent Cloud resources.

  2. From the Billing & payment section in the Menu, apply the promo code CC100KTS to receive an additional $100 free usage on Confluent Cloud (details).

  3. Click on LEARN and follow the instructions to launch a Kafka cluster and to enable Schema Registry.

Confluent Cloud

3
Create a properties file with Confluent Cloud information

From the Confluent Cloud UI, navigate to your Kafka cluster and click on Clients and then select Java.

Create new credentials for your Kafka cluster and Schema Registry, and then Confluent Cloud will show a configuration similar to below with your new credentials automatically populated (make sure show API keys is checked). Copy and paste it into a configuration/ccloud.properties file on your machine.

# Required connection configs for Kafka producer, consumer, and admin
bootstrap.servers={{ BOOTSTRAP_SERVERS }}
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule   required username='{{ CLUSTER_API_KEY }}'   password='{{ CLUSTER_API_SECRET }}';
sasl.mechanism=PLAIN
# Required for correctness in Apache Kafka clients prior to 2.6
client.dns.lookup=use_all_dns_ips

# Best practice for Kafka producer to prevent data loss
acks=all

# Required connection configs for Confluent Cloud Schema Registry
schema.registry.url={{ SR_URL }}
basic.auth.credentials.source=USER_INFO
basic.auth.user.info={{ SR_API_KEY }}:{{ SR_API_SECRET }}
Do not directly copy and paste the above configuration. You must copy it from the UI so that it includes your Confluent Cloud information and credentials.

4
Download and setup the Confluent Cloud CLI

Instructions for installing Confluent Cloud CLI and configuring it to your Confluent Cloud environment is available from within the Confluent Cloud UI. Navigate to your Kafka cluster, click on the CLI and tools section, and run through the steps in the CCloud CLI tab.

5
Create a topic

In this step we’re going to create a topic for use during this tutorial. Use the following command to create the topic:

ccloud kafka topic create input-topic

6
Configure the project

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

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

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

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

repositories {
    mavenCentral()

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

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

dependencies {
    implementation "org.slf4j:slf4j-simple:1.7.30"
    implementation "org.apache.kafka:kafka-streams:2.7.0"
    testImplementation "junit:junit:4.13.2"
    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

7
Add application and consumer properties

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

# Consumer properties
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

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-responsive and removed from the consumer-group triggering a rebalance.

Setting enable.auto.commit configuration to true enables the Kafka consumer to handle committing 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 multiple members all sharing the same group.id configuration. As members leave or join the consumer-group, the group-coordinator triggers a rebalance which causes topic-partition reassignment among active members of the group.

8
Update the properties file with Confluent Cloud information

Using the command below, append the contents of configuration/ccloud.properties (with your Confluent Cloud configuration) to configuration/dev.properties (with the application properties).

cat configuration/ccloud.properties >> configuration/dev.properties

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-responsive and removed from the consumer-group triggering a rebalance.

Setting enable.auto.commit configuration to true enables the Kafka consumer to handle committing 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 multiple members all sharing the same group.id configuration. As members leave or join the consumer-group, the group-coordinator triggers a rebalance which causes topic-partition reassignment among active members of the group.

9
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 supplying instances of the Consumer and ConsumerRecordsHandler via constructor parameters.

By using interfaces vs. concrete implementations 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);
  }

}

10
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 each record’s value in a List
2 If the List 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);
      }
    }
  }
}

11
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

12
Produce sample data to the input topic

Using a terminal window, run the following command to start a Confluent Cloud CLI producer:

ccloud kafka topic produce input-topic

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

Enter Ctrl+C to exit.

13
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

14
Teardown Confluent Cloud resources

You may try another Kafka tutorial, but if you don’t plan on doing other tutorials, use the Confluent Cloud UI or CLI to destroy all the resources you created. Verify they are destroyed to avoid unexpected charges.

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.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;

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(500);
    addTopicPartitionsAssignmentAndAddConsumerRecords(topic, mockConsumer, topicPartition);
    Thread.sleep(500);
    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.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;

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