DEV Community: Dejan Maric

Kafka Consumer – Committing consumer group offset

Dejan Maric — Sun, 26 Jan 2025 11:19:56 +0000

Consumer group offset – an introduction

What is a consumer group offset?

It’s a way for a consumer to keep track of where it is in the topic. For each topic and partition the consumer group will have one number representing the latest consumed record from Kafka.

When we wrote about Kafka topic we mentioned how each record in a partition has an offset – a sequential ID that marks its position within the partition and uniquely identifies it. The same concept is being used here. A consumer group will have one offset for each partition it is consuming. So if we have a consumer group that is reading from a topic with two partitions (let’s call them P1 and P2), the consumer group will have an offset for P1 – the last record read from P1 partition – and an offset for P2 – again, the last record read from P2 partition.

An example of a current offset (position) for a consumer group 1

There are two ways to keep track of consumer group offsets:

Storing them in Kafka, or
Saving them in external storage (e.g. in a file or database)

This blog post covers the first approach – storing the offset in Kafka. The blog post describing the second approach is coming soon.

Where is offset stored in Kafka?

When using Kafka to store offsets, these are saved in a topic called __consumer_offsets (yes, those are two underscores at the beginning). Kafka client library provides built-in support for storing and retrieving offsets, ensuring that consumers resume from the last stored position after a restart.

What happens if there’s no stored offset for the consumer?

If no offset is found, the consumer follows the auto.offset.reset configuration to decide what to do next.

Consumer behaviour, depending on the auto.offset.reset:

latest (default) – Jump to the end of a topic and wait for new messages. None of the existing messages will be read
earliest – Jump to the start of the topic (will read all existing messages)
none – Throws an exception if no existing offset is found

Auto vs Manual Commit

What is an auto-commit?

This is a feature in Kafka client that simplifies the offset management by periodically committing the consumer offset to the Kafka cluster. The client is doing this in a separate thread and by default it is committing the offset every 5 seconds without the explicit action from us. This means that you can simply start consuming records and when your client restarts it will continue from the latest committed offset without you needing to do anything about it. This comes with some downsides we’ll mention soon.

The use of auto-commit is controlled by the enable.auto.commit property, which defaults to true when group.id property is set.

while (true) {
  records = consumer.poll(timeout);
  // process the records
}

Downsides of using auto-commit

Based on how the auto-commit works, there is room for losing data when using it. Here’s how that might happen.

Since the client commits offsets in a separate thread, it does not track where your application is in processing polled records. In addition, since Kafka client is often polling multiple records at a time, the following scenario could happen:

The client polls 10 records
You start processing records and on the e.g. 3rd record the background thread commits the offset. It will commit the offset for the entire poll, which means that it will mark you’re currently on the 10th record.
If your application now fails before processing all 10 records, you have effectively lost data because the next time your application starts it will receive the info it is on the 10th record and the next record to consume is the 11th record.

Simple way to avoid data loss is to only commit offsets after you have successfully processed them, which is why we have manual commit.

Manual commit

Manual offset commit allows explicit control over when offsets are recorded. By disabling auto-commit (enable.auto.commit=false), the application must explicitly commit offsets using the commitSync() or commitAsync() methods. This level of control allows offsets to be committed only after specific conditions are met, such as successfully processing a batch of records.

while (true) {
  records = consumer.poll(timeout);
  // process the records
  //when all the polled records are consumed, commit the offset
  consumer.commitSync(); // or consumer.commitAsync()
}

When to rely on auto-commit?

If your application can tolerate:

reprocessing the same data from time to time (because it’s designed as idempotent) and
losing a few messages here and there

then this is a good feature to leave enabled. It reduces the client’s complexity.

Otherwise, use manual commit.

Synchronous vs Asynchronous Offset Commit

After choosing manual commit, the next decision is whether to commit offsets synchronously or asynchronously.

If you want to be absolutely sure the offset is committed before continuing your work, use commitSync(). Be aware that this will block your thread until the commit is confirmed by Kafka server or an exception is thrown.

Async offset commit has two variants, both of which are non-blocking:

commitAsync() which ignores any exceptions thrown and
commitAsync(OffsetCommitCallback callback) where you can provide a callback that will handle success or exceptions (eg timeout happened, or consumer group rebalance is in progress).

Conclusion

Consumer group offset is a fundamental aspect of how Kafka ensures reliable message consumption. It determine where a consumer resumes reading after a restart and play a critical role in data consistency.

While auto-commit simplifies offset management, it comes with the risk of data loss if records are not processed before a commit occurs. Manual offset management, on the other hand, provides greater control, ensuring that offsets are only committed after records are fully processed.

Choosing between synchronous and asynchronous commits depends on your application’s requirements—commitSync() guarantees persistence at the cost of blocking, while commitAsync() offers non-blocking flexibility with potential trade-offs.

Understanding these offset management techniques allows developers to design Kafka consumers that balance performance, reliability, and fault tolerance, ensuring data integrity in real-world applications.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

Photo credit: @kencheungphoto

Unit testing Kafka producer using MockProducer

Dejan Maric — Tue, 11 Apr 2023 10:00:00 +0000

Sometimes your Kafka producer code is doing things that need to be properly validated and of course, we developers resort to writing a test. If the functionality we want to test is nicely encapsulated we can do that using a unit test. Kafka helps us with that by providing a mock implementation of Producer<> interface called, you guessed it, MockProducer.

Preparation for test

TransactionProcessor class below is our class under test. It has a process(Transaction) method that receives a Transaction object which in our example only contains userId and amount properties. Depending on the amount the processor will decide to which topic to write the object. If the amount is above 100.000 it will use the transactions_high_prio topic. Otherwise, it will write a transaction to the transactions_regular_prio topic.

public class TransactionProcessor {
    public static final double HIGH_PRIORITY_THRESHOLD = 100.000;
    private final Producer<String, String> kafkaProducer;
    private final String highPrioTopic;
    private final String regularPrioTopic;

    private final Gson gson = new Gson();

    public TransactionProcessor(Producer<String, String> kafkaProducer, String highPrioTopic, String regularPrioTopic) {
        this.kafkaProducer = kafkaProducer;
        this.highPrioTopic = highPrioTopic;
        this.regularPrioTopic = regularPrioTopic;
    }

    public void process(Transaction transaction){
        String selectedTopic = regularPrioTopic;
        if (transaction.getAmount() >= HIGH_PRIORITY_THRESHOLD) {
            selectedTopic = highPrioTopic;
        }
        String transactionJson = gson.toJson(transaction);
        ProducerRecord<String, String> record = 
            new ProducerRecord<>(selectedTopic, transaction.getUserId(), transactionJson);
        kafkaProducer.send(record);
    }
}

And Transaction class looks like this:

public class Transaction {
    private String userId;
    private double amount;
    //removed for brevity
}

An important thing to notice here is that TransactionProcessor uses the Producer interface, not the implementation (which is the KafkaProducer class). This will make it possible to unit test our adapter using the MockProducer.

MockProducer in action

Ok, now onto the test class. TransactionProcessorTest creates an instance of the MockProducer that we will provide to the TransactionProcessor.

class TransactionProcessorTest {

    private static final String HIGH_PRIO_TOPIC = "transactions_high_prio";
    private static final String REGULAR_PRIO_TOPIC = "transactions_regular_prio";
    MockProducer<String, String> mockProducer = 
        new MockProducer<>(true, new StringSerializer(), new StringSerializer());

MockProducer constructor takes a couple of parameters, namely key and value serializers, in our case StringSerializer(s). The first parameter, autocomplete, is a boolean that tells MockProducer to automatically complete all requests immediately. In regular testing, you want to set this parameter to true so that messages are immediately ‘sent’. If you set it to false you will need to explicitly call completeNext() and errorNext(RuntimeException) methods after calling the send() method. You would want to do this to e.g. test the error handling in your producer (by providing the exception you want to handle as the parameter to the errorNext method).

After we’ve created the MockProducer, we create the instance of the class we wish to test.

TransactionProcessor processor = 
    new TransactionProcessor(mockProducer, HIGH_PRIO_TOPIC, REGULAR_PRIO_TOPIC);

Now it's time to test whether the selection of topics based on amount is correct. We will create two Transaction objects, the first one with a low amount and the second one with an amount higher than our threshold (which is 100.000).

    @Test
    public void testPrioritySelection(){
        Double lowAmount = 50.2d;
        Double highAmount = 250000d;
        Transaction regularPrioTransaction = new Transaction("user1", lowAmount);
        processor.process(regularPrioTransaction);
        Transaction highPrioTransaction = new Transaction("user2", highAmount);
        processor.process(highPrioTransaction);

        assertThat(mockProducer.history()).hasSize(2);

        ProducerRecord<String, String> regTransactionRecord = mockProducer.history().get(0);
        assertThat(regTransactionRecord.value()).contains(lowAmount.toString());
        assertThat(regTransactionRecord.topic()).isEqualTo(REGULAR_PRIO_TOPIC);

        ProducerRecord<String, String> highTransactionRecord = mockProducer.history().get(1);
        assertThat(highTransactionRecord.value()).contains(highAmount.toString());
        assertThat(highTransactionRecord.topic()).isEqualTo(HIGH_PRIO_TOPIC);
    }

After calling processor.process(…) method twice we want to validate that there are two records sent to Kafka. For that, we use MockProducer#history() method which returns the list of records that the producer received to send to Kafka. We fetch each record from the history to ensure it is ‘sent’ to the proper topic.

Code on Github

All code examples from this blog post are available on Coding Harbour’s GitHub.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

Photo credit: @paulschnuerle

How to use PEM certificates with Apache Kafka

Dejan Maric — Thu, 06 May 2021 20:07:18 +0000

It’s been a long waiting but it’s finally here: starting with Apache Kafka 2.7 it is now possible to use TLS certificates in PEM format with brokers and java clients.

You might wonder – why does it matter?

PEM is a scheme for encoding x509 certificates and private keys as Base64 ASCII strings. This makes it easier to handle your certificates. You can simply provide keys and certificates to the app as string parameters (e.g. through environment variables). This is especially useful if your applications are running in containers, where mounting files to containers makes the deployment pipeline a bit more complex. In this post, I want to show you two ways to use PEM certificates in Kafka.

Providing certificates as strings

Brokers and CLI tools

You can add certificates directly to the configuration file of your clients or brokers. If you’re providing them as single-line strings, you must transform the original multiline format to a single line by adding the line feed characters ( \n ) at the end of each line. Here’s how the SSL section of the properties file should look:

security.protocol=SSL
ssl.keystore.type=PEM
ssl.keystore.certificate.chain=-----BEGIN CERTIFICATE-----\nMIIDZjC...\n-----END CERTIFICATE-----
ssl.keystore.key=-----BEGIN ENCRYPTED PRIVATE KEY-----\n...\n-----END ENCRYPTED PRIVATE KEY-----
ssl.key.password=<private_key_password>
ssl.truststore.type=PEM
ssl.truststore.certificates=-----BEGIN CERTIFICATE-----\nMICC...\n-----END CERTIFICATE-----

Note that ssl.keystore.certificate.chain needs to contain your signed certificate as well as all the intermediary CA certificates. For more details on this see the Common gotchas section below.

Your private key goes into ssl.keystore.key field, while the password for the private key (if you use one) goes to ssl.key.password field.

Java clients

Java clients use exactly the same properties, but constants help with readability:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
...omitted other producer configs...
//SSL configs
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
properties.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PEM");
properties.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "<certificate_chain_string>");
properties.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "<private_key_string>");
// key password is needed if the private key is encrypted
properties.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "<private_key_password>");
properties.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PEM");
properties.put(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, "<trusted_certificate>");

producer = new KafkaProducer<>(properties);

Providing certificates as files

If you already use mTLS authentication towards Kafka, then the easiest way to transition towards PEM certificates is to use them as files, replacing the java keystore and truststore you use today. This approach makes it easy to transition from PKCS12 files to PEM files.

Brokers and CLI tools

Here’s how the ssl section of the properties file should look:

security.protocol=SSL
ssl.keystore.type=PEM
ssl.keystore.location=/path/to/file/containing/certificate/chain
ssl.key.password=<private_key_password>
ssl.truststore.type=PEM
ssl.truststore.location=/path/to/truststore/certificate

ssl.keystore.type and ssl.truststore.type properties tell Kafka in which format we are providing the certificates and the truststore.

Next, ssl.keystore.location points to a file that should contain the following:

your private key
your signed certificate
as well as any intermediary CA certificates

For more details about the certificate chain see the Common gotchas section below.

You will need to set the ssl.key.password if your private key is encrypted (which I hope it is). Make sure not to provide the ssl.keystore.password otherwise you’ll get an error.

Java clients

Again Java clients use the same properties, but here we’re using the constants provided by the Kafka client library:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
...omitted other producer configs...
//SSL configs
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
properties.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PEM");
properties.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "/path/to/file/containing/certificate/chain");
// key password is needed if the private key is encrypted
properties.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "<private_key_password>");
properties.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PEM");
properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/path/to/truststore/certificate");

producer = new KafkaProducer<>(properties);

Common gotchas when setting up a certificate chain

If your private key is encrypted (which it should always be), you need to convert it from PKCS#1 to PKCS#8 format for java/kafka to be able to read it properly
If you want to provide the PEM certificate as a one-line string, make sure to add the line feed characters at the end of each line ( \n ). Otherwise, the certificate will be considered invalid.
The certificate chain has to include your certificate together with all the intermediary CA certificates that signed it, in that order. So for example, if your certificate was signed by certificate A which was signed by cert B which was signed by the root certificate, your certificate chain has to include: your certificate, certificate A and certificate B, in that order. Do note that the root certificate should not be in the chain.
Certificate order in your certificate chain is important (see point 3)

Example of Kafka SSL setup with PEM certificates

Testing an SSL setup of your clients is not simple, because setting up a Kafka cluster with SSL authentication is not a straightforward process. This is why I created a docker-compose project with a single zookeeper and broker, enabled with SSL authentication. This project used many ideas from the excellent cp-demo project by Confluent.

To use the project, clone the docker-compose repository, and navigate to the kafka-ssl folder.

git clone https://github.com/codingharbour/kafka-docker-compose.git
cd kafka-ssl

Running the start-cluster.sh script will generate the self-signed root certificate. The script will use it to sign all other certificates. It will also generate and sign the certificates for the broker and zookeeper and certificates for one producer and consumer. After this, the script will start the cluster using docker-compose.

Don’t have docker-compose? Check: how to install docker-compose

In addition, the startup script will generate producer.properties and consumer.properties files you can use with kafka-console-* tools.

The consumer.properties file is an example of how to use PEM certificates as strings. The producer.properties, on the other hand, uses certificates stored in PEM files. This way you can see and test both approaches described in this blog post.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour

Photo credit: FLY:D

How to use Protobuf with Apache Kafka and Schema Registry

Dejan Maric — Sun, 05 Jul 2020 08:00:20 +0000

Full guide on working with Protobuf in Apache Kafka

Since Confluent Platform version 5.5 Avro is no longer the only schema in town. Protobuf and JSON schemas are now supported as the first-class citizens in Confluent universe. But before I go on explaining how to use Protobuf with Kafka, let’s answer one often asked question…

Why do we need schemas

When applications communicate through a pub-sub system, they exchange messages and those messages need to be understood and agreed upon by all the participants in the communication. Additionally, you would like to detect and prevent changes to the message format that would make messages unreadable for some of the participants.

That’s where a schema comes in – it represents a contract between the participants in communication, just like an API represents a contract between a service and its consumers. And just as REST APIs can be described using OpenAPI (Swagger) so the messages in Kafka can be described using Avro, Protobuf or Avro schemas.

Schemas describe the structure of the data by:

specifying which fields are in the message
specifying the data type for each field and whether the field is mandatory or not

In addition, together with Schema Registry, schemas prevent a producer from sending poison messages – malformed data that consumers cannot interpret. Schema Registry will detect if breaking changes are about to be introduced by the producer and can be configured to reject such changes. An example of a breaking change would be deleting a mandatory field from the schema.

Introduction to Protobuf

Similar to Apache Avro, Protobuf is a method of serializing structured data. A message format is defined in a .proto file and you can generate code from it in many languages including Java, Python, C++, C#, Go and Ruby. Unlike Avro, Protobuf does not serialize schema with the message. So, in order to deserialize the message, you need the schema in the consumer.

Here’s an example of a Protobuf schema containing one message type:

syntax = "proto3";

package com.codingharbour.protobuf;

message SimpleMessage {
    string content = 1;
    string date_time = 2;
}

In the first line, we define that we’re using protobuf version 3. Our message type called SimpleMessage defines two string fields: content and date_time. Each field is assigned a so-called field number, which has to be unique in a message type. These numbers identify the fields when the message is serialized to the Protobuf binary format. Google suggests using numbers 1 through 15 for most frequently used fields because it takes one byte to encode them.

Protobuf supports common scalar types like string, int32, int64 (long), double, bool etc. For the full list of all scalar types in Protobuf check the Protobuf documentation.

Besides scalar types, it is possible to use complex data types. Below we see two schemas, Order and Product, where Order can contain zero, one or more Products:

message Order {
    int64 order_id = 1;
    int64 date_time = 2;
    repeated Product product = 3;
}

message Product {
    int32 product_id = 1;
    string name = 2;
    string description = 3;
}

Now, let’s see how these schemas end up in the Schema Registry.

Schema Registry and Protobuf

Schema Registry is a service for storing a versioned history of schemas used in Kafka. It also supports the evolution of schemas in a way that doesn’t break producers or consumers. Until recently Schema Registry supported only Avro schemas, but since Confluent Platform 5.5 the support has been extended to Protobuf and JSON schemas.

If you worked with Avro and Kafka before, this section will not contain any surprises. Like with Avro, Schema Registry provides a serializer and deserializer for Protobuf, called KafkaProtobufSerializer and KafkaProtobufDeserializer.

The job of this serializer is to convert the Java object to a protobuf binary format before the producer writes the message to Kafka.

The additional job of the serialiser is to check whether the protobuf schema exists in the Schema Registry. If not, it will write the schema to Schema Registry and it will write the schema id to the message (at the beginning of the message). Then, when the Kafka record reaches the consumer, the consumer will use KafkaProtobufDeserializer to fetch the schema from the Schema Registry based on the schema id from the message. Once the schema is fetched, the KafkaProtobufDeserializer will use it to deserialize the message. This way the consumer doesn’t need to know the schema in advance to be able to consume messages from Kafka.

Registering and using Protobuf schema

This is why, when using KafkaProtobuf(De)Serializer in a producer or a consumer, we need to provide the URL of the Schema Registry.

Code generation in Java

Ok, now we know how a protobuf schema looks and we know how it ends up in Schema Registry. Let’s see now how we use protobuf schemas from Java.

The first thing that you need is a protobuf-java library. In these examples, I’m using maven, so let’s add the maven dependency:

<dependencies>
    <dependency>
        <groupId>com.google.protobuf</groupId>
        <artifactId>protobuf-java</artifactId>
        <version>3.12.2</version>
    </dependency>
</dependencies>

The next thing you want to do is use the protoc compiler to generate Java code from . proto files. But we’re not going to invite the compiler manually, we’ll use a maven plugin called protoc-jar-maven-plugin :

<plugin>
    <groupId>com.github.os72</groupId>
    <artifactId>protoc-jar-maven-plugin</artifactId>
    <version>3.11.4</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>run</goal>
            </goals>
            <configuration>
                <inputDirectories>
                    <include>${project.basedir}/src/main/protobuf</include>
                </inputDirectories>
                <outputTargets>
                    <outputTarget>
                        <type>java</type>
                        <addSources>main</addSources>
                        <outputDirectory>${project.basedir}/target/generated-sources/protobuf</outputDirectory>
                    </outputTarget>
                </outputTargets>
            </configuration>
        </execution>
    </executions>
</plugin>

The protobuf classes will be generated during the generate-sources phase. The plugin will look for proto files in the src/main/protobuf folder and the generated code will be created in the target/generated-sources/protobuf folder.

To generate the class in the target folder run:

mvn clean generate-sources

Note: All code examples from this blog post are available on Coding Harbour's GitHub.

Ok, now that we have our class generated, let’s send it to Kafka using the new Protobuf serializer.

Running a local Kafka cluster

Before we get started, let’s boot up a local Kafka cluster with the Schema Registry, so we can try our out code right away. We will run our cluster using docker-compose.

Don’t have docker-compose? Check: how to install docker-compose

I’ve prepared a docker-compose file with one Zookeeper, one Kafka broker and the Schema Registry. You can grab it from https://github.com/codingharbour/kafka-docker-compose

Navigate to single-node-avro-kafka folder and run:

docker-compose up -d

The output should look similar to this:

Starting sn-zookeeper ... done
Starting sn-kafka     ... done
Starting sn-schema-registry ... done

Your local Kafka cluster is now ready to be used. By running docker-compose ps , we can see that the Kafka broker is available on port 9092, while the Schema Registry runs on port 8081. Make a note of that, because we’ll need it soon.

$ docker-compose ps
       Name                    Command            State                     Ports
----------------------------------------------------------------------------------------------------
sn-kafka             /etc/confluent/docker/run   Up      0.0.0.0:9092->9092/tcp
sn-schema-registry   /etc/confluent/docker/run   Up      0.0.0.0:8081->8081/tcp
sn-zookeeper         /etc/confluent/docker/run   Up      0.0.0.0:2181->2181/tcp, 2888/tcp, 3888/tcp

Writing a Protobuf Producer

With Kafka cluster up and running is now time to create a Java producer that will send our SimpleMessage to Kafka. First, let’s prepare the configuration for the Producer:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaProtobufSerializer.class);
properties.put(KafkaProtobufSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

Producer<String, SimpleMessage> producer = new KafkaProducer<>(properties);

Notice that we are using KafkaProtobufSerializer as the value serializer class. This is the new serializer available in Confluent Platform since version 5.5. It works similarly to KafkaAvroSerializer: when publishing messages it will check with Schema Registry if the schema is available there. If the schema is not yet registered, it will write it to Schema Registry and then publish the message to Kafka. For this to work, the serializer needs the URL of the Schema Registry and in our case, that’s http://localhost:8081.

Next, we prepare the KafkaRecord, using the SimpleMessage class generated from the protobuf schema:

SimpleMessage simpleMessage = SimpleMessage.newBuilder()
        .setContent("Hello world")
        .setDateTime(Instant.now().toString())
        .build();

ProducerRecord<String, SimpleMessage> record
                = new ProducerRecord<>("protobuf-topic", null, simpleMessage);

This record will be written to the topic called protobuf-topic. The last thing to do is to write the record to Kafka:

producer.send(record);
producer.flush();
producer.close();

Usually, you wouldn’t call flush() method, but since our application will be stopped after this, we need to ensure the message is written to Kafka before that happens.

Writing a Protobuf Consumer

We said that the consumer doesn’t need to know the schema in advance to be able to deserialize the message, thanks to Schema Registry. But, having the schema available in advance allows us to generate the Java class out of it and use the class in our code. This helps with code readability and makes a code strongly typed.

Here’s how to do it. First, you will generate a java class(es) as explained in Code generation in Java section. Next, we prepare the configuration for the Kafka consumer:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "protobuf-consumer-group");      
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

Here we’re defining a broker URL, consumer group of our consumer and telling the consumer that we’ll handle offset commits ourselves.

Next, we define deserializer for the messages:

properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaProtobufDeserializer.class);
properties.put(KafkaProtobufDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
properties.put(KafkaProtobufDeserializerConfig.SPECIFIC_PROTOBUF_VALUE_TYPE, SimpleMessage.class.getName());

We use string deserializer for the key, but for the value, we’re using the new KafkaProtobufDeserializer. For the protobuf deserializer, we need to provide the Schema Registry URL, as we did for the serializer above.

The last line is the most important. It tells the deserializer to which class to deserializer the record values. In our case, it’s the SimpleMessage class (the one we generated from the protobuf schema using the protobuf maven plugin).

Now we’re ready to create our consumer and subscribe it to protobuf-topic :

KafkaConsumer<String, SimpleMessage> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Collections.singleton("protobuf-topic"));

And then we poll Kafka for records and print them to the console:

while (true) {
    ConsumerRecords<String, SimpleMessage> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, SimpleMessage> record : records) {
        System.out.println("Message content: " + record.value().getContent());
        System.out.println("Message time: " + record.value().getDateTime());
    }
    consumer.commitAsync();
}

Here we’re consuming a batch of records and just printing the content to the console.

Remember when we configured the consumer to let us handle committing offsets by setting ENABLE_AUTO_COMMIT_CONFIG to false? That’s what we’re doing in the last line: only after we’ve fully processed the current group of records will we commit the consumer offset.

That’s all there is to writing a simple protobuf consumer. Let’s now check one more variant.

Generic Protobuf Consumer

What if you want to handle messages in a generic way in your consumer, without generating a Java class from a protobuf schema? Well, you can use an instance of DynamicMessage class from protobuf library. DynamicMessage has a reflective API, so you can navigate through message fields and read their values. Here’s how you can do it…

First, let’s configure the consumer. Its configuration is very similar to the previous example:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "generic-protobuf-consumer-group");      
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaProtobufDeserializer.class);
properties.put(KafkaProtobufDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

The only thing missing is the SPECIFIC_PROTOBUF_VALUE_TYPE configuration. Since we want to handle messages in a generic way, we don’t need this configuration.

Now we’re ready to create our consumer and subscribe it to protobuf-topic topic, as in the previous example:

KafkaConsumer<String, SimpleMessage> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Collections.singleton("protobuf-topic"));

And then we poll Kafka for records and print them to the console:

while (true) {
    ConsumerRecords<String, DynamicMessage> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, DynamicMessage> record : records) {
        for (FieldDescriptor field : record.value().getAllFields().keySet()) {
            System.out.println(field.getName() + ": " + record.value().getField(field));
        }
    }
    consumer.commitAsync();
}

Without SPECIFIC_PROTOBUF_VALUE_TYPE configured in our consumer, the consumer will always return the instance of DynamicMessage in the record’s value. Then we use the DynamicMessage.getAllFields() method to obtain the list of FieldDescriptors. Once we have all the descriptors we can simply iterate through them and print the value of each field.

Check out the JavaDoc to find out more about DynamicMessage.

That wraps our Kafka Protobuf guide. Now you’re ready to start writing producers and consumers that send Protobuf messages to Apache Kafka with help of Schema Registry.

All the code from this blog post is available on Coding Harbour's GitHub.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

Photo credit: Joshua Woroniecki

Learn how to use Kafkacat – the most versatile Kafka CLI client

Dejan Maric — Mon, 04 May 2020 18:47:42 +0000

Kafkacat is an awesome tool and today I want to show you how easy it is to use it and what are some of the cool things you can do with it.

All the features explained below are available in version 1.5.0.

Looking for a quick Kafkacat reference? Download the Kafkacat 1-page cheatsheet

Installing Kafkacat

Kafkacat is available from Homebrew (latest version) and some Linux repositories, but it is possible that Linux repos don’t contain the latest version. If that’s the case, you can always run the latest kafkacat from docker.

The basics

Kafkacat is a command-line tool for producing and consuming Kafka messages. In addition, you can view metadata about the cluster or topics.

Kafkacat has quite a few parameters and it might look scary learning them all, yet (most of) the parameters make sense and are easy to remember. Let’s start with the most important: modes. When making a call to Kafkacat, you’ll always use it in one of the four modes it has. All the modes use capital letter:

-P = Produce data
-C = Consume data
-L = List metadata
-Q = Query

The next most important option is the broker list (-b) and after that, it’s usually topic (-t).

So you can almost write your command like a story. The following command:

kafkacat -C -b localhost:9092 -t topic1 -o beginning

could be read as: I want to Consume from broker localhost:9092 and topic topic1 with offset set to the beginning.

Ok, now that I have hopefully convinced you that all those cryptical parameters make sense, let’s look at how to use Kafkacat to achieve some common tasks.

Producing data (-P)

What do we need so we could produce data? At a minimum, you need a broker and a topic you want to write to.

Produce values

kafkacat -P -b localhost:9092 -t topic1

Default message separator is Enter. Type your messages, and separate them with Enter.

Producing keys and values

If you want to produce messages with key, you need to specify the Key delimiter (-K). Let’s use a colon to separate the key and the message in the input:

kafkacat -P -b localhost:9092 -t topic1 -K :
key3:message3
key4:message4

Notice that parameter uses capital K.

Produce messages with headers

If you want to add headers to the messages, add them using -H parameter, in a key=value format:

kafkacat -P -b localhost:9092 \
-t topic1 \
-H appName=kafkacat -H appId=1

As you see, additional headers are added by repeating -H flag. Note that all the messages produced will have the two headers specified with -H flag.

Produce data from a file

If you want to produce data using a file, use the option -l (as in: fi*l*e)… I did say that most of the parameters are easy to remember :). Let’s say we have a file called data.txt containing key-value pairs, separated by a colon:

key1:message1
key2:message2
key3:message3

So the command would be:

kafkacat -P -b localhost:9092 -t topic1 -K: -l data.txt

Produce message with compression

Using a (-z) parameter you can specify message compression:

kafkacat -P -b localhost:9092 -t topic1 -z snappy

Supported values are: snappy, gzip and lz4.

Consuming data (-C)

Simple consumer

Consume all the messages from a topic

kafkacat -C -b localhost:9092 -t topic1

Note that, unlike kafka-console-consumer, kafkacat will consume the messages from the beginning of the topic by default. This approach makes more sense to me, but YMMV.

Consume X messages

You can control how many messages will be consumed using the count parameter (-c, lowercase).

kafkacat -C -b localhost:9092 -t topic1 -c 5

Consuming from an offset

If you want to read data from a particular offset, you can use the -o parameter. The offset parameter is very versatile. You can:

Consume messages from the beginning or end

kafkacat -C -b localhost:9092 -t topic1 -o beginning

Use constants beginning or end to tell kafkacat where to begin the consumption.

Consume from a given offset

kafkacat -C -b localhost:9092 -t topic1 -o 123

Use an absolute value for the offset and Kafkacat will start consuming from the given offset. If you don’t specify the partition to consume, Kafkacat will consume all the partitions from the given offset.

Consume last X messages in a partition(s)

kafkacat -C -b localhost:9092 -t topic1 -o -10

We do this by using a negative offset value.

Consume based on a timestamp

It is possible to start consuming after a given timestamp in milliseconds using the format -o s@start_timestamp. Technically this is consuming based on an offset, the difference is that kafkacat figures out the offset for you based on the provided timestamp(s).

kafkacat -C -b localhost:9092 -t topic1 -o s@start_timestamp

You can also stop consuming when a given timestamp is reached using:

kafkacat -C -b localhost:9092 -t topic1 -o e@end_timestamp

This is very useful when you are debugging an error that occurred, you have the timestamp of the error, but you want to check how the message looked. Then, combining the start and end offset, you can narrow down your search:

kafkacat -C -b localhost:9092 -t topic1 -o s@start_timestamp -o e@end_timestamp

Formatting the output

By default, Kafkacat will print out only the message payload (value of the Kafka record), but you can print anything you’re interested in. To define the custom output, specify (-f) flag, as in format, followed by a format string. Here’s an example that prints a string with key and value of the message:

kafkacat -C -b localhost:9092 -t topic1 \
 -f 'Key is %k, and message payload is: %s \n'

%k and %s are format string tokens. The output might be something like this:

Key is key3, and message payload is: message3
Key is key4, and message payload is: message4

So what can you print out using format string?

Topic (%t),
partition (%p)
offset (%o)
timestamp (%T)
message key (%k)
message value (%s)
message headers (%h)
key length (%K)
value length (%S)

As you’ve seen above, you can use newline (\n \r) or tab characters(\t) in the format string as well.

Serdes

If messages are not written as strings, you need to configure a proper serde for keys and values using -s parameter.

For example, if both key and value are 32-bit integers, you would read it using:

kafkacat -C -b localhost:9092 -t topic1 -s i

You can specify separately serde for the key and value using:

kafkacat -C -b localhost:9092 -t topic1 -s key=i -s value=s

You will find the list of all the serdes in a kafkacat help (kafkacat -h).

Avro serde

Avro messages are a bit special since they require a schema registry. But Kafkacat has you covered there as well. Use (-r) to specify the schema registry URL:

kafkacat -C -b localhost:9092 \
-t avro-topic \
-s key=s -s value=avro \
-r http://localhost:8081

In the example above, we’re are reading messages from a topic where keys are strings, but values are Avro.

List metadata (-L)

Listing metadata gives you info about topics: how many partitions it has, which broker is a leader for a partition as well as the list of in-sync replicas (isr).

Metadata for all topics

kafkacat -L -b localhost:9092

Simply calling -L with no other parameters will display the metadata for all the topics in the cluster.

Metadata for a given topic

If you want to see metadata for just one topic, specify it using (-t) parameter:

kafkacat -L -b localhost:9092 -t topic1

Query mode (-Q)

If you want to find an offset of a Kafka record based on a timestamp, Query mode can help with that. Just specify the topic, partition and a timestamp:

kafkacat -b localhost:9092 -Q -t topic1:1:1588534509794

Is this all?

I'm glad that you asked because it's not :) I have created a 1-page Kafkacat cheatsheet for you to download. Grab it here.

Guide to Apache Avro and Kafka

Dejan Maric — Tue, 14 Apr 2020 17:27:50 +0000

Photo credit: @bosco_shots

Or, how to produce and consume Kafka records using Avro serialization in Java.

So far we’ve seen how to produce and consume simple String records using Java and console tools. In this post, I would like to show you how to send and read Avro messages from Java using the kafka-clients library.

If you’re new to Avro, I have written a full post about why you should consider Avro serialization for Kafka messages, so check it out to learn more.

Running a Kafka cluster locally

To test the producers and consumers, let’s run a Kafka cluster locally, consisting of one broker, one zookeeper and a Schema Registry.

To simplify our job, we will run these servers as Docker containers, using docker-compose.

Don’t have docker-compose? Check: how to install docker-compose

I’ve prepared a docker-compose file which you can grab from Coding Harbour’s GitHub:

git clone https://github.com/codingharbour/kafka-docker-compose.git

Once you have the project, navigate to a folder called single-node-avro-kafka and start the Kafka cluster:

docker-compose up -d

The output should look something like this:

$ docker-compose up -d
Starting sn-zookeeper ... done
Starting sn-kafka ... done
Starting sn-schema-registry ... done

$ docker-compose ps
       Name                    Command            State                     Ports
----------------------------------------------------------------------------------------------------
sn-kafka             /etc/confluent/docker/run   Up      0.0.0.0:9092->9092/tcp
sn-schema-registry   /etc/confluent/docker/run   Up      0.0.0.0:8081->8081/tcp
sn-zookeeper         /etc/confluent/docker/run   Up      0.0.0.0:2181->2181/tcp, 2888/tcp, 3888/tcp

Defining the Avro Schema

Let’s create a schema for the messages we’ll be sending through Kafka. We’ll call our message: SimpleMessage, and it will have two fields:

content – a string field, holding the message we want to send and
date_time – human-readable date-time showing when the message was sent

Avro schemas are written in a JSON format, so our SimpleMessage schema will look like this:

{
    "type": "record",
    "name": "SimpleMessage",
    "namespace": "com.codingharbour.avro",
    "fields": [
        {"name": "content", "type":"string", "doc": "Message content"},
        {"name": "date_time", "type":"string", "doc": "Datetime when the message was generated"}
    ]
}

The schema consists of a couple of elements:

Type – Describes a data type of the entire schema. Type ‘record’ means that the schema describes a complex data type, which includes other fields.
Name – The name of the schema. In our case “SimpleMessage”
Namespace – Namespace of the schema that qualifies the name. In our case, the namespace is “com.codingharbour.avro”
List of fields – One or more fields that are in this complex data type

Each field in a schema is a JSON object with multiple attributes:

name – name of the field
type – data type of the field. Avro supports primitive types like int, string, bytes etc, and complex types like record, enum, etc
doc – Documentation for the given field
default – the default value for the field, used by the consumer to populate the value when the field is missing from the message.

For more info on Avro data types and schema check the Avro spec.

Schema Registry

As I’ve mentioned in the previous post, every Avro message contains the schema used to serialize it. But sending thousands or millions of messages per second with the same schema is a huge waste of bandwidth and storage space. That’s where the Schema Registry, KafkaAvroSerializer and KafkaAvroDeserializer come into play.

Instead of writing the schema to the message, KafkaAvroSerializer will write the schema to the Schema Registry and it will only write the schema id to the message. Then, when the Kafka record reaches the consumer, the consumer will use KafkaAvroDeserializer to fetch the schema from the Schema Registry based on the schema id from the message. Once the schema is fetched, the KafkaAvroDeserializer can deserialize the message.

Registering and using Avro schema

This is why, when using KafkaAvro(De)Serializer in a producer or a consumer, we need to provide the URL of the schema registry. Remember that our Schema Registry runs on port 8081.

Here’s a snippet from our producer:

properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
properties.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

If we don’t specify the URL, the (de)serializer will complain when we try to send/read a message.

Ok, the next thing is to see how an Avro schema gets translated into a Java object.

Avro record in Java

Note : do not confuse an Avro record with a Kafka record. Each Avro schema describes one or more Avro records. An Avro record is a complex data type in Avro, consisting of other fields, with their own data types (primitive or complex). Kafka record, on the other hand, consists of a key and a value and each of them can have separate serialization. Meaning, e.g. that Kafka key may be one Avro record, while a Kafka value is another Avro record (if we choose to use Avro serialization for both the key and the value).

When it comes to representing an Avro record in Java, Avro library provides two interfaces: GenericRecord or SpecificRecord. Let’s see what the difference is and when to use which.

An instance of a GenericRecord allows us to access the schema fields either by index or by name, as seen below:

GenericRecord record = ... ; //obtain a generic record
//accessing the field by name
record.put("date_time", "2020-01-01 12:45:00")
String dateTime = (String) record.get("date_time"); 

//accessing the field by index
record.put(0, "this is message number 1");
String content = (String) record.get(0);

Using a GenericRecord is ideal when a schema is not known in advance or when you want to handle multiple schemas with the same code (e.g. in a Kafka Connector). The drawback of GenericRecord is the lack of type-safety. GenericRecord’s put and get methods work with Object.

SpecificRecord is an interface from the Avro library that allows us to use an Avro record as a POJO. This is done by generating a Java class (or classes) from the schema, by using avro-maven-plugin. The generated class will implement the SpecificRecord interface, as seen below.

/* Class generated by avro-maven-plugin*/
public class SimpleMessage extends org.apache.avro.specific.SpecificRecordBase 
implements org.apache.avro.specific.SpecificRecord {
   //content removed for brevity
}

//using the SpecificRecord by using the actual implementation
SimpleMessage simpleMessage = new SimpleMessage();
simpleMessage.setContent("Hello world");

The drawback of SpecificRecord is that you need to generate a class for each schema you plan to use, in advance. Which again means you need the Avro schema in advance, to be able to generate the Java class.

Producing Avro messages using GenericRecord

First, we prepare the properties the producer needs. We specify our brokers, serializers for the key and the value, as well as the URL for the Schema Registry. Then we instantiate the Kafka producer:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
properties.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

Producer<String, GenericRecord> producer = new KafkaProducer<>(properties);

As you see, we are using String serializer for the keys and Avro for values. Notice that the producer expects _GenericRecord_s as the value of the Kafka record.

The next step is to create an instance of an Avro record based on our schema:

//avro schema
String simpleMessageSchema =
        "{" +
        " \"type\": \"record\"," +
        " \"name\": \"SimpleMessage\"," +
        " \"namespace\": \"com.codingharbour.avro\"," +
        " \"fields\": [" +
        " {\"name\": \"content\", \"type\": \"string\", \"doc\": \"Message content\"}," +
        " {\"name\": \"date_time\", \"type\": \"string\", \"doc\": \"Datetime when the message\"}" +
        " ]" +
        "}";

//parse the schema
Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(simpleMessageSchema);

//prepare the avro record
GenericRecord avroRecord = new GenericData.Record(schema);
avroRecord.put("content", "Hello world");
avroRecord.put("date_time", Instant.now().toString());

Here, we specified the schema directly in the code. By parsing the schema we get a Schema object, which we use to instantiate a new GenericRecord. Finally, we set the record’s fields by name, using the put method.

The last thing to do is create a Kafka record with null key and Avro record in the value and write it to a topic called avro-topic:

//prepare the kafka record
ProducerRecord<String, GenericRecord> record = new ProducerRecord<>("avro-topic", null, avroRecord);

producer.send(record);
//ensures record is sent before closing the producer
producer.flush(); 

producer.close();

Producing Avro messages using SpecificRecord

Another way to produce the same record as above is to use the SpecificRecord interface. We will generate a Java cclass from the Avro schema using the avro-maven-plugin. We’ll add the plugin to our pom.xml:

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.9.2</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
                <outputDirectory>${project.basedir}/target/generated-sources/avro/</outputDirectory>
                <stringType>String</stringType>
            </configuration>
        </execution>
    </executions>
</plugin>

Avro plugin is configured above to generate classes based on schemas in the src/main/avro folder and to store the classes in the target/generated-sources/avro/.

If you check the src/main/avro folder, you will see the Avro schema for our SimpleMessage. It’s the same schema we used in the GenericRecord example above. When you execute mvn compile, the SimpleMessage class will be generated in the target folder.

Then we’ll define properties for the Kafka producer, same as in the GenericRecord example:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
properties.put("schema.registry.url", "http://localhost:8081");

Producer<String, SpecificRecord> producer = new KafkaProducer<>(properties);

The only difference compared to the GenericRecord example is the type for the value of the Kafka record, which is now SpecificRecord.

Next, we create the instance of the SimpleMessage:

//create the specific record
SimpleMessage simpleMessage = new SimpleMessage();
simpleMessage.setContent("Hello world");
simpleMessage.setDateTime(Instant.now().toString());

And lastly, we create a Kafka record and write it to the “avro-topic” topic:

ProducerRecord<String, SpecificRecord> record = new ProducerRecord<>("avro-topic", null, simpleMessage);

producer.send(record);
//ensures record is sent before closing the producer
producer.flush();

producer.close();

Note that both producers above have written to a topic called ‘avro-topic’. So we now have two records to consume. Let’s see how we can create the consumers.

Consuming Avro messages using GenericRecord

The consumer that uses GenericRecord, does not need a schema nor a Java class generated from the schema. All the data will be obtained by the deserializer from the schema registry.

First, we’ll create properties for the consumer and istantiate it:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "generic-record-consumer-group");
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);

properties.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(properties);

Then we’ll subscribe our consumer to the ‘avro-topic’ topic and start listening for records:

consumer.subscribe(Collections.singleton("avro-topic"));

//poll the record from the topic
while (true) {
    ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, GenericRecord> record : records) {
        System.out.println("Message content: " + record.value().get("content"));
        System.out.println("Message time: " + record.value().get("date_time"));
    }
    consumer.commitAsync();
}

Here we get the field values by name, using the Object get(String key) method of GenericRecord.

Consuming Avro messages using SpecificRecord

Last thing to show is how to consume Avro kafka record, which is automatically casted into proper Java class, generated from Avro schema.

As before, we’ll start with preparing properties for the consumer and instantiating it:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "specific-record-consumer-group");
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);

properties.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
properties.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true); //ensures records are properly converted

KafkaConsumer<String, SimpleMessage> consumer = new KafkaConsumer<>(properties);

Everything is the same as with the previous consumer, except the third line from the bottom. Let’s look at it again:

properties.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);

This line is necessary if you want your Avro records to be properly converted into the excepted Java class (in our case, this is SimpleMessage).

Now, all we have to do is subscribe our consumer to the topic and start consuming:

consumer.subscribe(Collections.singleton("avro-topic"));

//poll the record from the topic
while (true) {
    ConsumerRecords<String, SimpleMessage> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, SimpleMessage> record : records) {
        System.out.println("Message content: " + record.value().getContent()); //1
        System.out.println("Message time: " + record.value().getDateTime()); //2
    }
    consumer.commitAsync();
}

You see above lines marked 1 and 2 how the fields of SimpleMessage are accessed using proper getter methods.

There you have it, two ways to produce and two ways to consume Kafka Avro records. Hope this was helpful.

As always, the code from this blog post is available on CodingHarbour’s github repo

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

Why use Avro data format with Apache Kafka?

Dejan Maric — Sat, 21 Mar 2020 18:27:05 +0000

Avro is an open-source binary serialization format. But why use it with Kafka? Why not send JSON or XML messages? What benefits does it give us? This is what we’ll be exploring today.

Serialization

It’s important to understand that records in a topic are just arrays of bytes. Kafka broker doesn’t care about the type of data we’re sending. To make those byte arrays useful in our applications, producers and consumers must know how to interpret them. Are we sending CSV data or JSON or XML? Producers and consumers use (de)serialization to transform data to byte arrays and back. Yet, that’s only the beginning.

Schema

Whenever we use data as an integration mechanism, the schema of the data becomes very important, because it becomes the contract between producers and consumers. The teams behind producers and consumers need to agree on the following:

Which fields are in a record
Which fields are optional and which are mandatory
Where and how should this be documented
Which default values should a consumer use for the fields that are missing from the record
How should changes to the schema be handled

Data format

Now, you could use JSON with a JSON schema or use XML with an XSD schema to describe the message format. But there’s one downside with these: messages in these formats often use more space to convey the same information due to the nature of JSON and XML.

So is there a better way?

Yes. You could use Apache Avro. Avro is a data serialization format that is developed under the Apache umbrella and is suggested to be used for Kafka messages by the creators of Apache Kafka themselves. Why?

By serializing your data in Avro format, you get the following benefits:

Avro relies on a schema. This means every field is properly described and documented
Avro data format is a compact binary format, so it takes less space both on a wire and on a disk
It has support for a variety of programming languages
in Avro, every message contains the schema used to serialize it. That means that when you’re reading messages, you always know how to deserialize them, even if the schema has changed

Yet, there’s one thing that makes Avro not ideal for usage in Kafka, at least not out-of-the-box, because…

Every Avro message contains the schema used to serialize the message

Think about this for a moment: if you plan on sending millions of messages a day to Kafka, it’s a terrible waste of bandwidth and storage space to send the same schema information over and over again.

So, the way to overcome this is to…

Separate the schema from the message

That’s where a Schema Registry comes into play. Schema Registry is developed by Confluent, a company behind Apache Kafka, and it provides a RESTful interface for storing and receiving Avro schemas.

Instead of sending the schema inside a Kafka record, a producer starts by checking whether schema already exists in the Schema Registry. If not, it will write the schema there (step 1 below). Then the producer will obtain the id of the schema (step 2) and send that id inside the record (step 3), saving a lot of space this way. The consumer will read the message (step 4) and then contact the Schema Registry with the schema id from the record to get the full schema (step 5) and cache it locally.

Registering and using an Avro schema

Ok, that seems nice. But that’s not the only place where Schema Registry helps. Let’s see what happens when you want to…

Evolve the schema

Imagine yourself in a situation where, 2 years after releasing the producer application, you decide to change the format of the message in a way that, in your opinion, doesn’t break the compatibility and thus should not affect the consumers. Now you have 2 options:

you can either be a nice person, find all the consumers, check whether the suggested changes will affect them and if so, ask them to change
or you can do the change and wait for the mob with torches, shovels, and rakes to come your way

Assuming you chose to avoid the option that would put your picture on every train station with a monetary reward under it, what is the probable outcome of the option number one?

Those who worked in a corporate environment know the answer: a lot of planning, budgeting, negotiating and sometimes even postponing because there are more pressing issues. Getting 10s or 100s of consumers to perform the upgrade before you can continue with the changes in the producer is a sure way to an asylum.

Apache Avro and Schema Registry are coming to the rescue once again. Schema Registry allows us to enforce the rules for validating a schema compatibility when the schema is modified. If a new message breaks the schema compatibility, a producer will reject to write the message.

This way your consumers are protected from e.g. someone suddenly changing the data type of the field from a long to a string or removing the mandatory field.

At the same time, it gives you clear guidelines about which changes to the schema are allowed. So no longer weeks or months of coordinating the change just to remove an optional field in a message.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

What is a consumer group in Kafka?

Dejan Maric — Wed, 04 Mar 2020 16:10:37 +0000

Photo credit: Stefan Vladimirov

When sending messages using a messaging system, you typically have two scenarios you want to achieve. Either you want to:

send a message to a targeted group of consumers (which might be just one consumer) or
broadcast the message to all the consumers

Kafka allows you to achieve both of these scenarios by using consumer groups.

Consumer group

A consumer group is a group of consumers (I guess you didn’t see this coming?) that share the same group id. When a topic is consumed by consumers in the same group, every record will be delivered to only one consumer. As the official documentation states: “If all the consumer instances have the same consumer group, then the records will effectively be load-balanced over the consumer instances.”

This way you can ensure parallel processing of records from a topic and be sure that your consumers won’t be stepping on each other toes.

How does Kafka achieve this?

Each topic consists of one or more partitions. When a new consumer is started it will join a consumer group (this happens under the hood) and Kafka will then ensure that each partition is consumed by only one consumer from that group.

So, if you have a topic with two partitions and only one consumer in a group, that consumer would consume records from both partitions.

A single consumer in a consumer group

After another consumer joins the same group, each consumer would continue consuming only one partition.

Multiple consumers in one consumer group

Does it mean if I want to have more than one consumer (from the same group) reading from one topic I need to have more than one partition?

That is correct. If you have more consumers in a group than you have partitions, extra consumers will sit idle, since all the partitions are taken. If you know that you will need many consumers to parallelize the processing, then plan accordingly with the number of partitions.

Additional consumers in a group sit idly

When we talked about topics and partitions, I mentioned that a partition is a unit of parallelism from the consumer’s perspective. Now you know the reason – there is a direct link between the number of partitions and number of consumers from a group reading in parallel.

What if I want to consume the same record from multiple consumers?

That is also possible. You can have many consumers reading the same records from the topic, as long as they all have different group ids.

Multiple consumers reading the same records from the topic

An example to recap

Let’s illustrate what we’ve been talking about with an example.

Let’s say we’re building an online store and it consists of few microservices that are sending events to each other: payment service, shipping service, and notification service. Once the payment service processes the payment it will send an event PaymentProcessed as a record on Kafka topic. Then we want both the shipping service and notification service to consume this record. The shipping service needs the record in order to start the shipping process, while the notification service wants to receive this record so it could send an email to the customer saying ‘Your payment has been received‘. In this case, we want the PaymentProcessed record to be broadcasted to all the consumers.

Microservices communicating using Kafka

Yet, if we have multiple instances of the consuming services, we always want exactly one of the instances to process each record. For example, we wouldn’t want multiple instances of the notification service to process the PaymentProcessed record and send multiple ‘Your payment has been received’ emails to the customer. Nor would we want multiple instances of shipping service to receive the same PaymentProcessed record and start the shipment process multiple times, potentially losing us money.

To ensure the record reaches both the shipping and the notification service but only once, we would put all the payment service instances in one consumer group and put all the notification service instances in another consumer group.

Each Kafka record will reach one shipping and one notification service

This ensures that all the records are always read by both shipping_group and notification_group, but within those groups, one record will always go to only one instance. That’s what consumer groups enable us to do.

A consumer group and record offset

If you remember when we talked about topics, we said that each record is uniquely identified by an offset in the partition. These offsets are used to track which record has been consumed by which consumer group.

Kafka employs an approach of ‘a dumb pipeline, smart clients’ meaning that Kafka brokers don’t know anything about consumer offsets. The consumers themselves are in charge of tracking which records have been consumed. Once the consumer reads the record it will store this offset in a special Kafka topic called __consumer_offsets (yes, those are two underscores at the beginning). When a consumer stores the offset in this topic we’re saying that it’s committing the offset.

This enables consumers to always know which record should be consumed next from a given partition. Since the consumer offset is stored in Kafka, it means that the position of the consumer group is maintained even after restarts.

An example of consumer offsets

In the topic post, I also mentioned that records remain in the topic even after being consumed. This allows multiple consumers to consume the same message, but it also allows one more thing: the same consumer can re-consume the records it already read, by simply rewinding its consumer offset. This is very useful when you e.g. had a bug in your consumer and want to re-read the records after fixing the bug.

And there you have it, Kafka consumer groups in a nutshell.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

How to delete records from a Kafka topic

Dejan Maric — Sat, 29 Feb 2020 16:41:08 +0000

Photo credit: Adli Wahid

Every now and then I get a request from my colleagues who would like to delete some or all the records from a Kafka topic. The request usually comes after someone has produced the wrong data in a test topic while playing around or due to a bug in the producer code. Or simply because they want a clean slate.

Whatever the reason, today I’ll show you a few ways to delete some or all the records from a Kafka topic.

It should go without saying that you should use your best judgment and check (at least) twice before using the methods described below in a production environment.

Kafka-delete-records

This command is available as part of Kafka CLI tools. It requires two parameters:

a bootstrap server and
a JSON file, describing which records should be deleted.

The command allows you to delete all the records from the beginning of a partition, until the specified offset.

NOTE: It is not possible to delete records in the middle of the topic.

The JSON file specifies one or more partitions from which we want to remove the records. Let’s create delete-records.json file as below:

{
    "partitions": [
        {
            "topic": "my-topic",
            "partition": 0,
            "offset": 3
        }
    ],
    "version": 1
}

Here we’ve specified that for the partition 0 of the topic “my-topic” we want to delete all the records from the beginning until offset 3.

Now we’re ready to delete records. Execute:

kafka-delete-records --bootstrap-server localhost:9092 \
--offset-json-file delete-records.json

After the command finishes the start offset for the partition 0 will be 3.

Deleting all the records in a topic

NOTE: This will not work for compacted topics

If you want to prune all the messages, another option is to reduce the retention of the topic to a small value (e.g. 100ms), wait for the brokers to remove all the records from the topic and then set the topic retention to its original value. Here’s how to do it.

First, set the retention.ms to 100 milliseconds.

kafka-configs --zookeeper localhost:2181 \
--entity-type topics \
--entity-name my-topic \
--alter --add-config retention.ms=100

Then, wait for the brokers to remove the messages with expired retention (that is, all of them). To know if the process is finished, check whether the start offset and end offset are the same. This means there are no more records available on the topic. Depending on your setup, it might take few minutes for Kafka to clean up the topic, so keep checking the start offset.

Use the GetOffsetShell class to check the beginning and ending offset of a topic’s partitions. To check the end offset set parameter time to value -1:

kafka-run-class kafka.tools.GetOffsetShell \
--broker-list localhost:9092 \
--topic my-topic \
--time -1

To check the start offset, use --time -2

kafka-run-class kafka.tools.GetOffsetShell \
--broker-list localhost:9092 \
--topic my-topic \
--time -2

Once the topic has been purged, return the retention.ms to its original value:

kafka-configs --zookeeper localhost:2181 \
--entity-type topics \
--entity-name my-topic \
--alter --add-config retention.ms=<ORIGINAL VALUE>

Delete a topic and create it again

Not as elegant as the previous two approaches, yet it might be an easier solution in some cases (e.g. if topic creation is scripted).

kafka-topics --bootstrap-server localhost:9092 \
--topic my-topic \
--delete

Then create it again:

kafka-topics --bootstrap-server localhost:9092 \
--topic my-topic \
--create \
--partitions <number_of_partitions> \
--replication-factor <replication_factor>

Few things to be aware of when using this approach

Make sure the deletion of topics is enabled in your cluster. Set delete.topic.enable=true. From Kafka 1.0.0 this property is true by default.

Make sure all consumers have stopped consuming the data from the topic you want to delete. Otherwise, they will throw errors like:

Received unknown topic or partition error in fetch for partition my-topic-0

Error while fetching metadata with correlation id 123 : {my-topic=LEADER_NOT_AVAILABLE}

One more thing that might happen if you have consumers up and running is that the topic will get auto-created if the cluster-wide property auto.create.topics.enable is true (and by default it is). Not bad per se, but it will use a default number of partitions (1) and a replication factor (1), which might not be what you wanted.

Moral of the story is – make sure to stop your consumers before using this approach 🙂

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

How to create a Kafka consumer in Java

Dejan Maric — Sun, 24 Nov 2019 00:31:57 +0000

Photo credit: Kazuky Akayashi

In this post, I’ll show you how to consume Kafka records in Java. We’ll read data from a topic called java_topic. To test how our consumer is working, we’ll produce data using the Kafka CLI tool. And all this in under 5 minutes, so let’s jump right in.

Running a Kafka cluster locally

We’ll use Docker Compose to run our local Kafka cluster. It will consist of one Zookeeper instance and one Kafka broker.

Don’t have docker-compose? Check: how to install docker-compose

I’ve prepared a docker-compose file which you can grab from Coding Harbour’s GitHub:

git clone https://github.com/codingharbour/kafka-docker-compose.git

Once you have the project, navigate to a folder called single-node-kafka and start the Kafka cluster:

docker-compose up -d

The output should look something like this:

Creating network "single-node-kafka_default" with the default driver
Creating sn-zookeeper ... done
Creating sn-kafka     ... done

Your local Kafka cluster is now ready to be used. By running docker-compose ps , we can see that the Kafka broker is available on port 9092. Make a note of that, because we’ll need it soon.

$ docker-compose ps
    Name                Command            State              Ports
-------------------------------------------------------------------------------
sn-kafka       /etc/confluent/docker/run   Up      0.0.0.0:9092->9092/tcp
sn-zookeeper   /etc/confluent/docker/run   Up      2181/tcp, 2888/tcp, 3888/tcp

Dependencies

To be able to consumer records from Kafka we need the Kafka client library. Add this dependency to your pom.xml:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.1</version>
</dependency>

Check the latest version of kafka-clients library at maven repository. At the time of writing it was 2.4.1.

Create a Kafka topic

To create a topic we’ll use a Kafka CLI tool called kafka-topics , that comes bundled with Kafka binaries. In our case, it means the tool is available in the docker container named sn-kafka.

First, open your favourite terminal and connect to the running Kafka container:

docker exec -it sn-kafka /bin/bash

Now that we’re inside the container where we have access to Kafka CLI tools, let’s create our topic, called java_topic:

kafka-topics --bootstrap-server localhost:9092 \
--create --topic java_topic \
--partitions 1 \
--replication-factor 1

Creating a Kafka consumer

There are a couple of properties we need to set up for Kafka consumer to work properly:

Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-first-consumer-group");
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

Consumer<String, String> consumer = new KafkaConsumer<>(properties);

First, we specify the URL where our Kafka broker is running, by configuring BOOTSTRAP_SERVERS_CONFIG property. In cases when you have more than one broker (which is the usual scenario), you would specify them in a comma-separated string, e.g:

"serverA:9092,serverB:9092,serverC:9092"

Second, the consumer needs to know how to deserialize the records from a byte array. Kafka brokers are agnostic of the data types and are treating records as byte arrays. This means that producers need to know how to serialize data into byte arrays and consumers need to know how to deserialize it back.

Each Kafka record consists of a key and a value. These can potentially be of different types, so the consumer needs to know which deserializer to use for key and which one to use for value. That’s where KEY_DESERIALIZER_CLASS_CONFIG and VALUE_DESERIALIZER_CLASS_CONFIG come into play. In our case, we’re using String for both key and value.

The next property is a consumer group id. It’s a unique identifier of a group of consumers and is used, among other things, to track which records have been consumed by the consumers in this group. If you want to learn more about consumer groups, check this post.

The property AUTO_OFFSET_RESET_CONFIG, tells the consumer where it should start consuming messages the first time it starts. By default, a consumer will only consume messages that arrive to the topic after the consumer is started for the first time. By setting the value to “earliest” we tell the consumer to read all the records that already exist in the topic.

By default, Kafka consumer commits the offset periodically. Last property, ENABLE_AUTO_COMMIT_CONFIG, tells the consumer that we’ll handle committing the offset in the code. I’ll show you how to do it soon.

Consuming the records

Now that we have our consumer configured and created, it’s time to consume some data. In Kafka producers push the data to topics and consumers are frequently polling the topic(s) to check for new records. So the usual way is to poll for new records in an endless while loop and once there are new records, to process them. But before we can poll topic for records, we need to subscribe our consumer to one or more topics:

consumer.subscribe(Collections.singleton("java_topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, String> record : records) {
        System.out.println("Message received: " + record.value());
    }
    consumer.commitAsync();
}

consumer.poll(Duration) will return immediately if there are available records. Otherwise, it will block until either a record is available or timeout has expired. If timeout expired, the poll method will return an empty record set.

The record processing would happen in the for loop. We are simply printing the value of the record to system out.

Remember how we told the consumer library that we’ll be committing the offset manually? That’s what the line consumer.commitAsync() is doing. You want to commit the offset after you’ve finished processing the message, not before.

Now that we have our consumer in place, let’s produce some messages to test our consumer.

Producing messages for our consumer

To produce the messages to this topic, we’ll use a CLI tool that comes bundled with Kafka binaries called kafka-console-producer. This is a CLI tool that is part of Apache Kafka binaries and you can download it from the official website. Yet, since we’re using Kafka’s docker image, the CLI tools are already available in the Kafka broker’s container. To be able to use the tool we first need to connect to the container called sn-kafka:

docker exec -it sn-kafka /bin/bash

Now we can produce some messages. If you check the code in the previous section, you will see that our consumer is waiting for the messages to arrive on a topic called java_topic , so let’s produce to that topic:

kafka-console-producer --broker-list localhost:9092 --topic java_topic
>

Producer tool will display the prompt, showing us that it is waiting for the message to send. Type the message and send it by pressing Enter.

Run the consumer

Now that the record is ready in Kafka topic, we can start our consumer and check the output. The message we sent should quickly appear. You can go back to the terminal where kafka-console-producer is running and send a few more messages, then check the consumer to see the messages consumed.

Source code

Source code for this example is available on Coding Harbour’s GitHub.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

How to create a Kafka producer in Java

Dejan Maric — Sun, 24 Nov 2019 00:28:49 +0000

Photo credit: Dawid Zawiła

In this post, I will show you how to produce messages to Kafka from Java using the kafka-clients library. It takes less than 5 minutes and around 10 lines of code. Don’t believe me? Keep reading.

Running a Kafka cluster locally

To be able to produce messages (known in Kafka world as records ) to Kafka we need a Kafka cluster. At a minimum, a Kafka cluster consists of one Kafka server (called a broker) and it needs at least one Zookeeper node.

To simplify our job, we will run these two servers as Docker containers, using docker-compose.

Don’t have docker-compose? Check: how to install docker-compose

I’ve prepared a docker-compose file which you can grab from Coding Harbour’s GitHub:

git clone https://github.com/codingharbour/kafka-docker-compose.git

Once you have the project, navigate to a folder called single-node-kafka and start the Kafka cluster:

docker-compose up -d

The output should look something like this:

Creating network "single-node-kafka_default" with the default driver
Creating sn-zookeeper ... done
Creating sn-kafka     ... done

Your local Kafka cluster is now ready to be used. By running docker-compose ps , we can see that the Kafka broker is available on port 9092. Make a note of that, because we’ll need it soon.

$ docker-compose ps
    Name                Command            State              Ports
-------------------------------------------------------------------------------
sn-kafka       /etc/confluent/docker/run   Up      0.0.0.0:9092->9092/tcp
sn-zookeeper   /etc/confluent/docker/run   Up      2181/tcp, 2888/tcp, 3888/tcp

Dependencies

To be able to write records to Kafka we need the Kafka client library. Add the dependency to your pom.xml:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.1</version>
</dependency>

Check the latest version of kafka-clients library at maven repository. At the time of writing it was 2.4.1.

Creating the Kafka producer

To create Kafka producer we need to provide it with few mandatory properties. I’ll explain them below:

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

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

First, the producer needs to know how to reach the Kafka broker. We specify the broker’s address by using the ProducerConfig.BOOTSTRAP_SERVERS_CONFIG property. In cases when you have more than one broker (which is always the case in production), you would specify them in a comma-separated string, e.g:

"serverA:9092,serverB:9092,serverC:9092"

Second, the producer needs to know how to serialize the records to a byte array. Kafka brokers are agnostic of the data types we’re sending and are treating every record as an array of bytes. This means that producers need to know how to serialize data into byte arrays and consumers need to know how to deserialize it back.

Each Kafka record consists of a key and a value. These can potentially be of different types, so the producer needs to know which serializer to use for key and which one to use for value. That’s where KEY_SERIALIZER_CLASS_CONFIG and VALUE_SERIALIZER_CLASS_CONFIG come into play. In our example, we’ll be sending strings, so we’re configuring StringSerializer.

Producing the record

Ok, it’s time to produce some data. Our producer code will write one record when we run it and the record will contain a string telling us the current time when the record was created.

String recordValue = "Current time is " + Instant.now().toString();

ProducerRecord<String, String> record = new ProducerRecord<>("javatopic", null, recordValue);

producer.send(record);
producer.flush();

The first parameter we need to set when creating the ProducerRecord is the topic to which we’re writing the record. Then we set the key and the value of the record. The key is optional and in this case, we won’t be setting it. The value is, as mentioned, the string with the timestamp when we created the record. After everything is set, we call the send method of the producer.

The last line forces the producer to write the record to the topic right away. You would expect that a call to producer.send(record) would write data to Kafka immediately, but what happens under the hood is that the records are queued to be sent in batches. This way Kafka producer optimizes for throughput and latency since network operations are expensive. Yet, since we’re only producing a single message before shutting down our application, we want to tell the producer to send it right away.

And that’s it. If you now run the application you will produce the message to Kafka.

Have you done so? Ok, then let’s read the message, to ensure it’s really there 🙂

Check the record is in Kafka topic

We’ll use kafka-console-consumer utility to validate our message is written to the topic.

kafka-console-consumer is a CLI tool that is part of Apache Kafka binaries and you can download it from the official website. Yet, since we’re using Kafka’s docker image, the CLI tools are already available in the Kafka broker’s container. To be able to use the tool we first need to connect to the container called sn-kafka:

docker exec -it sn-kafka /bin/bash

Now, run kafka-console-consumer using the following command:

kafka-console-consumer --bootstrap-server localhost:9092 --topic javatopic --from-beginning

After few moments you should see the message.

Congratulations, you have produced the message to Kafka from java, and it only took few lines of code 🙂

Source code

The entire example is available on Coding Harbour’s github.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at Coding Harbour.

How to get started with Apache Kafka in 5 minutes

Dejan Maric — Sun, 24 Nov 2019 00:13:13 +0000

Photo credit:Jukan Tateisi

In this post, I will show you how to run an Apache Kafka cluster on your machine and how to:

create a topic that will keep the data
produce messages to topic and
consume those records

using only the CLI tools that come with Apache Kafka. All this in 5 minutes.

Ok, are you ready? Let’s get started!

Running a Kafka cluster locally

To be able to produce messages to and consume them from Kafka we need… uhm, the Kafka cluster. At a minimum, a Kafka cluster consists of one Kafka server (called broker) and it needs at least one Zookeeper node.

To simplify our job, we will run these two servers using docker-compose.

Don’t have docker-compose? Check: how to install docker-compose

I’ve prepared a docker-compose file which you can grab from Coding Harbour’s GitHub:

git clone https://github.com/codingharbour/kafka-docker-compose.git

Once you have the project, navigate to a folder called single-node-kafka and start the Kafka cluster:

docker-compose up -d

The output should look something like this:

Creating network "single-node-kafka_default" with the default driver
Creating sn-zookeeper ... done
Creating sn-kafka ... done

Your local Kafka cluster is now ready to be used. By running docker-compose ps , as shown below, we can see that Kafka broker is available on port 9092.

$ docker-compose ps
    Name                Command            State              Ports
-------------------------------------------------------------------------------
sn-kafka       /etc/confluent/docker/run   Up      0.0.0.0:9092->9092/tcp
sn-zookeeper   /etc/confluent/docker/run   Up      2181/tcp, 2888/tcp, 3888/tcp

Create a topic

Now that we have our cluster up and running, we’ll create a topic.

A topic is a way to organize messages. A producer is always sending a message to a particular topic and consumers are always reading messages from a particular topic.

To create a topic we’ll use a Kafka CLI tool called kafka-topics , that comes with Kafka. In our case, it means the tool is available in the docker container named sn-kafka.

First, open your favourite terminal and connect to the running Kafka container:

docker exec -it sn-kafka /bin/bash

Now that we’re inside the container where we have access to Kafka CLI tools, let’s create our topic, called first_topic:

kafka-topics --bootstrap-server localhost:9092 \
--create --topic first_topic \
--partitions 1 \
--replication-factor 1

Check that the topic is crated by listing all the topics:

kafka-topics --bootstrap-server localhost:9092 --list

The output should resemble the one below:

__consumer_offsets
first_topic

Produce messages

Next, let’s produce a message to a Kafka topic we just created. For this, we will use a Kafka command-line tool called kafka-console-producer.

In the same terminal window run kafka-console-producer:

kafka-console-producer --broker-list localhost:9092 --topic first_topic
>

Producer tool will display the prompt, showing us that it is waiting for the message to send. Type the message and send it by pressing Enter.

Congratulations, you have successfully produced the message to the topic called first_topic. Now it’s time to consume that message.

Consuming messages

As with producing, we’ll use the CLI tool already available in the kafka broker’s docker container. The tool is called kafka-console-consumer.

Let’s start by opening a new tab/window of the terminal and connecting to Kafka broker’s container:

docker exec -it sn-kafka /bin/bash

Now, run kafka-console-consumer using the following command:

kafka-console-consumer --bootstrap-server localhost:9092 \
--topic first_topic

Aaaand….nothing happens. You're probably wondering where the message you just produced is. You need to know that unless told otherwise Kafka consumer reads only new messages (a.k.a. those arriving to the topic after the consumer is started).

Let's fix that. Stop the consumer by pressing Ctrl+C.

To read the messages that existed in the topic before we started the consumer, we must add the –from-beginning parameter to the kafka-console-consumer command:

kafka-console-consumer --bootstrap-server localhost:9092 \
--topic first_topic \
--from-beginning

And (after few seconds) there’s our message. If you now go back to the producer window and send a few more messages, you will see them immediately appear in the consumer window.

Congratulations, you’ve now managed to send messages to a Kafka topic called first_topic and to consume messages, both old and new, from the same topic using only CLI tools provided with Kafka.

You can now stop the producer and the consumer by pressing Ctrl+C.

Would you like to learn more about Kafka?

I have created a Kafka mini-course that you can get absolutely free. Sign up for it over at the Coding Harbour.