1. Introduction
Apache Kafka is an open-source and distributed stream processing system that is fault-tolerant and provides high throughput. Kafka is basically a messaging system that implements a publisher-subscriber model. The messaging, storage, and stream processing capabilities of Kafka allow us to store and analyze real-time data streams at scale.
In this tutorial, we’ll first look at the significance of a key in a Kafka message. We’ll then learn how we can publish messages with a key to a Kafka topic.
2. Significance of a Key in a Kafka Message
As we know, Kafka effectively stores a stream of records in the order in which we generate the records.
When we publish a message to a Kafka topic, it’s distributed among the available partitions in a round-robin fashion. Hence, within a Kafka topic, the order of messages is guaranteed within a partition but not across partitions.
When we publish messages with a key to a Kafka topic, all messages with the same key are guaranteed to be stored in the same partition by Kafka. Thus, keys in Kafka messages are useful if we want to maintain order for messages having the same key.
To summarize, keys aren’t mandatory as a part of sending messages to Kafka. Basically, if we wish to maintain a strict order of messages with the same key, then we should definitely be using keys with messages. For all other cases, having null keys will provide a better distribution of messages amongst the partitions.
Next, let’s straightaway deep dive into some of the implementation code having Kafka messages with a key.
3. Setup
Before we begin, let’s first initialize a Kafka cluster, set up the dependencies, and initialize a connection with the Kafka cluster.
Kafka’s Java library provides easy-to-use Producer and Consumer API that we can use to publish and consume messages from Kafka.
3.1. Dependencies
Firstly, let’s add the Maven dependency for Kafka Clients Java library to our project’s pom.xml file:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.4.0</version>
</dependency>
3.2. Cluster and Topic Initialization
Secondly, we’ll need a running Kafka cluster to which we can connect and perform various Kafka operations. The guide assumes that a Kafka cluster is running on our local system with the default configurations.
Lastly, we’ll create a Kafka topic with multiple partitions that we can use to publish and consume messages. Referring to our Kafka Topic Creation guide, let’s create a topic named “baeldung“:
Properties adminProperties = new Properties();
adminProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
Admin admin = Admin.create(adminProperties);
Here, we created an instance of Kafka Admin with the basic configurations defined by the Properties instance. Next, we’ll use this Admin instance to create a topic named “baeldung” with five partitions:
admin.createTopics(Collections.singleton(new NewTopic("baeldung", 5, (short) 1)));
Now that we have the Kafka cluster setup initialized with a topic, let’s publish some messages with a key.
4. Publishing Messages With a Key
To demonstrate our coding examples, we’ll first create an instance of KafkaProducer with some basic producer properties defined by the Properties instance. Next, we’ll use the created KafkaProducer instance to publish messages with a key and verify the topic partition.
Let’s deep dive into each of these steps in detail.
4.1. Initialize Producer
First, let’s create a new Properties instance that holds the producer’s properties to connect to our local broker:
Properties producerProperties = new Properties();
producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
Further, let’s create an instance of KafkaProducer using the created producer’s Properties instance:
KafkaProducer <String, String> producer = new KafkaProducer<>(producerProperties);
The KafkaProducer class’s constructor accepts a Properties object (or a Map) and returns an instance of KafkaProducer.
4.2. Publish Messages
The Kafka Publisher API provides multiple constructors to create an instance of ProducerRecord with a key. We use the ProducerRecord<K,V>(String topic, K key, V value) constructor to create a message with a key:
ProducerRecord<String, String> record = new ProducerRecord<>("baeldung", "message-key", "Hello World");
Here, we created an instance of ProducerRecord for the “baeldung” topic with a key.
Now, let’s publish a few messages to the Kafka topic and verify the partitions:
for (int i = 1; i <= 10; i++) {
ProducerRecord<String, String> record = new ProducerRecord<>("baeldung", "message-key", String.valueOf(i));
Future<RecordMetadata> future = producer.send(record);
RecordMetadata metadata = future.get();
logger.info(String.valueOf(metadata.partition()));
}
We used the KafkaProducer.send(ProducerRecord<String, String> record) method to publish a message to Kafka. The method returns an instance of Future of the type RecordMetadata. We then use the blocking call to Future<RecordMetadata>.get() method that returns an instance of RecordMetadata when the message is published.
Next, we use the RecordMetadata.partition() method and fetch the partition of the messages.
The above code snippet produces the following logged result:
1
1
1
1
1
1
1
1
1
1
Using this, we verified that the messages we published with the same key are published to the same partition.
5. Conclusion
In this article, we learned the significance of a key in a Kafka message.
We first saw how we can publish a message with a key to a topic. We then discussed how we can verify that messages with the same key are published to the same partition.
As always, the complete code for all the examples is available over on GitHub.
