Kafka - sending the reply exactly to the sender

Kafka

Messaging Systems

Information Technology

Data Streaming

Software Development

Kafka - sending the reply exactly to the sender

Master System Design with Codemia

Enhance your system design skills with over 120 practice problems, detailed solutions, and hands-on exercises.

Start Practicing Learn More

Apache Kafka is a distributed event streaming platform capable of handling trillions of events a day. Initially conceived as a messaging queue, Kafka is based on an abstraction of a distributed commit log. Since being created and open-sourced by LinkedIn in 2011, Kafka has quickly evolved from messaging queue to a full-fledged event streaming platform.

What is Kafka?

Kafka is a system optimized for generating, storing, and processing streams of records. It allows for the decoupling of data streams and systems, providing high-throughput, fault-tolerant services that are necessary in modern data-driven applications.

Core Components of Kafka:

Producer: Responsible for publishing records into Kafka topics.
Consumer: Reads and processes records from topics.
Broker: A Kafka server that stores data and serves clients.
Topic: A category or feed name to which records are published.
Partition: Topics may be split into partitions for better data management and scalability.

Sending Messages Exactly to the Sender

One vital aspect of Kafka is managing how messages are communicated between producers and consumers. In some scenarios, especially in applications like request-response cycles, it's crucial to send the response exactly back to the sender (producer).

For implementing a scenario where the Kafka producer receives messages relevant only to it, typically a correlation ID can be used. The correlation ID ensures that the message's response is delivered to the exact request initiator.

Technical Example:

Let’s assume a scenario where a service (Producer A) sends a request message to Kafka, which is processed by another service (Service B), and the result should be sent back specifically to Producer A. Here’s how you could set it up:

Producer A sends a message to a common topic, including a unique correlation ID and reply-to topic.

java

1    Properties props = new Properties();
2    props.put("bootstrap.servers", "localhost:9092");
3    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
4    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
5    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
6
7    String correlationId = UUID.randomUUID().toString();
8    String replyToTopic = "producerA_responses";
9    ProducerRecord<String, String> record = new ProducerRecord<>("requestTopic", correlationId, "My request data" + "||" + replyToTopic);
10    producer.send(record);
11    producer.close();

Service B consumes the message, processes it, and sends a response to the reply-to topic mentioned in the message with the same correlation ID.

java

1    Properties consumerProps = new Properties();
2    consumerProps.put("bootstrap.servers", "localhost:9092");
3    consumerProps.put("group.id", "service-b");
4    consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
5    consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
6    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
7    consumer.subscribe(Arrays.asList("requestTopic"));
8
9    while (true) {
10        ConsumerRecords<String, String> records = consumer.poll(100);
11        for (ConsumerRecord<String, String> record : records) {
12            String[] parts = record.value().split("\\|\\|");
13            String request = parts[0];
14            String replyTopic = parts[1];
15
16            // Process request and generate a response
17            String response = "Processed " + request;
18            KafkaProducer<String, String> responseProducer = new KafkaProducer<>(props);
19            ProducerRecord<String, String> responseRecord = new ProducerRecord<>(replyTopic, record.key(), response);
20            responseProducer.send(responseRecord);
21            responseProducer.close();
22        }
23    }

Producer A listens on producerA_responses for messages with the matching correlation ID.

java

1    consumerProps.put("group.id", "producer-a");
2    KafkaConsumer<String, String> responseConsumer = new KafkaConsumer<>(consumerProps);
3    responseConsumer.subscribe(Arrays.asList("producerA_responses"));
4
5    while (true {
6        ConsumerRecords<String, String> records = responseConsumer.poll(100);
7        for (ConsumerRecord<String, String> record : records) {
8            if (record.key().equals(correlationId)) {
9                // Handle the response
10                break;
11            }
12        }
13    }

Summary Table:

Component	Role in Kafka	Example
Producer	Sends records to Kafka topics	KafkaProducer sends data to "requestTopic"
Consumer	Reads records from topics	KafkaConsumer reads from "requestTopic"
Broker	Manages storage and processing of records in a cluster	Handles records, ensures replication and fault tolerance
Topic	Categorizes records	"requestTopic" and "producerA_responses"
Partition	Splits topic for scalability and fault tolerance	Topic "requestTopic" may have multiple partitions

This approach ensures that messages are not only scalable and performant but also directed correctly, coupling the advantages of Kafka's distributed system with the precision of direct messaging.