Use Apache Kafka with Java

In this example we will be using the official Java client maintained by the Apache Kafka team. A list of alternative Java clients can be found here.

Dependencies

Add the kafka_2.12 package to your application. This package is available in maven:

Using client ⇆ broker encryption (SSL)

If you have chosen to enable client ⇆ broker encryption, see here for information on the certificates required to establish an SSL connection to your cluster.

Producing Messages

Client Configuration

Before creating a Kafka producer client, you first need to define the configuration properties for the producer client to use. In this example we provide only the required properties for the producer client. See here for the full list of configuration options.

Create a new file named producer.properties, ensuring the password, truststore location, and bootstrap servers list are correct:

Note: If your cluster does not have client ⇆ broker encryption enabled your file should look like this:

Make sure the password and bootstrap servers list are correct.

Note: To connect to your Kafka cluster over the private network, use port 9093 instead of 9092.

Java Code

Now that we have a properties file we can create a Kafka producer.

First, load the properties:

Once we’ve loaded the producer properties, we can create the producer itself:

Before we can send a message we first need to create a ProducerRecord object:

Now that we have a ProducerRecord object, sending it is trivial:

Note: We use the producer’s flush method here to ensure the message gets sent before the program exits. In normal operation the producer will send messages in batches when it has either accumulated a certain number of messages, or has waited a certain amount of time.

Consuming Messages

Client Configuration

As in the producer example, before creating a Kafka consumer client, you first need to define the configuration properties for the consumer client to use. In this example we provide only the required properties for the consumer client. See here for the full list of configuration options.

Create a new file named consumer.properties, ensuring the password, truststore location, and bootstrap servers list are correct:

Note: If your cluster does not have client ⇆ broker encryption enabled your file should look like this:

Make sure the password and bootstrap servers list are correct.

Note: To connect to your Kafka cluster over the private network, use port 9093 instead of 9092.

Java Code

Now that we have a properties file we can create a Kafka producer.

First, load the properties:

Once we’ve setup the consumer properties, we can create the consumer itself:

Before we can consume messages, we need to subscribe to the topics we wish to receive messages from:

Now we are ready to consume messages from Kafka. To consume a single batch of messages, we use the consumer’s poll method:

Combined with a loop, we can continually consume messages from Kafka as they are produced:

Putting Them Together

Now that we have a consumer and producer set up, it’s time to combine them.

Start the consumer

Start the consumer before starting the producer because by default consumers only consume messages that were produced after the consumer started.

Start the producer

Now that the consumer is setup and ready to consume messages, you can now start your producer.

If the consumer and producer are setup correctly the consumer should output the message sent by the producer shortly after it was produced: