{"id":417567,"date":"2024-06-30T01:38:38","date_gmt":"2024-06-30T01:38:38","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=417567"},"modified":"-0001-11-30T00:00:00","modified_gmt":"2024-06-30T01:38:38","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=417567","title":{"rendered":"Synchronous Request-Response using REST and Apache Kafka<\/span>"},"content":{"rendered":"
<\/div>\n
\n
\n
\n
\n

On one of the Innotech\u2019s projects, we received a task to convert asynchronous requests into synchronous ones. Basically, the purpose was to integrate REST and Apache Kafka into the same request.<\/p>\n

In more detail, we have two services that communicate with each other. Let\u2019s call them A and B. Service A receives a request from a consumer for data that is stored in service B. Thus, service A sends a request for data to B in REST and waits for the response of this request in Kafka. The user is waiting for data until this response is received.<\/p>\n

It appears to be a common problem, and the solution should be in Google or Stack Overflow. We managed to find the solution of a similar problem only in an existing library connecting two servers to Kafka. So the problem was solved using Java, Spring framework and a little IT-wit.<\/p>\n

Problem Statement<\/h2>\n

Before beginning implementation, we need to state the problem and understand what we have and what we want to achieve.<\/p>\n

We have a Client<\/em> service that will have one end-point. An end-point, in turn, will take a normal string, send it to the second service, and wait for the response in Kafka.<\/p>\n

And we have the second service \u2014\u00a0Server<\/em>. It will receive a REST string from the Client<\/em> service, convert it to UpperCase and return in response by Kafka.<\/p>\n

The Client<\/em> service awaits for the response from the Server<\/em>.\u00a0The user will wait for the result until the response is delivered. Waiting for a response may take a long time. To avoid too high of latency, it is necessary to provide an interrupt timeout, e.g. 10 seconds. But this condition should not be necessary; we should be able to wait indefinitely for a response. But definitely not waiting for 7.5 million years, otherwise the response could be disappointing.<\/p>\n

This is an example of how it works. The consumer sends a string to the Client<\/em> service, e.g. \u201cabc123\u201d, and until the response from the Server<\/em> service is delivered, they will be kept hanging. The response which will be returned to the consumer from the Server<\/em> should be \u201cABC123\u201d. If the wait time exceeds the timeout value, an HTTP code with a 504 (Gateway Timeout) error will be returned instead of the response.<\/p>\n

<\/figcaption><\/figure>\n

Client Implementation<\/h2>\n

I will only describe the main points. If you need the whole code, there is a link to the repository below.<\/p>\n

SenderReceiver Implementation<\/strong><\/p>\n

The logic core is a\u00a0SenderReceiver<\/code> class implementation, which is responsible for the process’s rest state until it receives data from the outside.<\/p>\n

This class consists of two main methods:\u00a0a receive()<\/code>\u00a0method is responsible for resting until data is received, and a\u00a0send()<\/code> method is responsible for receiving data and waking up the\u00a0receive()<\/code> method. To understand the current state of the thread, 1) whether resting or waiting (the receive()<\/code> method), or 2) receiving and waking up (the send()<\/code> method), we will create a\u00a0boolean<\/code> flag called\u00a0transfer<\/code>\u00a0and initialize it using the true<\/code> value.<\/p>\n

private boolean transfer = true;<\/code><\/pre>\n
Now let\u2019s implement\u00a0the receive()<\/code> method:<\/p>\n
public synchronized void receive() {     while (transfer) {         if (timeout != 0 && start.before(new Date(System.currentTimeMillis() - timeout))) {             timeoutException = new TimeoutException();             Thread.currentThread().interrupt();             return;         }         try {             wait(timeout);         } catch (InterruptedException e) {             Thread.currentThread().interrupt();         }     }     Thread.currentThread().interrupt(); }<\/code><\/pre>\n
Since initially the flag\u00a0transfer = true<\/code>, the process goes into a standby mode until it is woken up by the\u00a0send()<\/code> method, or milliseconds set in timeout<\/code> elapse.<\/p>\n
If the\u00a0timeout<\/code>\u00a0value is different from 0, it means that the thread interrupt mechanism by runtime has been set. To do this, we must compare the current time offset by this value with the start time of the thread. If this time is exceeded, we save\u00a0TimeoutException()<\/code>\u00a0into a variable which is initially null (we\u2019ll need it later), and terminate the thread. If the\u00a0transfer<\/code>\u00a0flag becomes\u00a0false<\/code>\u00a0and the thread has not finished waiting using a timeout, we will simply exit the loop and terminate the thread.<\/p>\n
You can see that\u00a0the timeout<\/code>\u00a0with a value of 0 is unique. If it is 0, the thread will wait indefinitely to wake up. After it is woken up and\u00a0transfer<\/code> changes\u00a0to a false<\/code> value, the thread will terminate.<\/p>\n
Now, let\u2019s implement the\u00a0send()<\/code> method:<\/p>\n
public synchronized void send(final T data) {     transfer = false;     this.data = data;     notifyAll(); }<\/code><\/pre>\n
You can see that the\u00a0send()<\/code> method is quite simple. It receives data from outside, changes our\u00a0transfer<\/code>\u00a0flag to\u00a0false<\/code>, saves the data, and wakes up all the threads that are hanging in\u00a0wait()<\/code>.<\/p>\n
In fact, there is another important auxiliary method called\u00a0getData()<\/code> which either returns the data stored in\u00a0data<\/code>, or generates an error if the relevant variable is not null.<\/p>\n
public T getData() throws TimeoutException {     if (Objects.nonNull(timeoutException))         throw timeoutException;     return data; }<\/code><\/pre>\n
SenderReceiverMap Implementation<\/strong><\/p>\n
After we have implemented the mechanism of waiting for a response and waking up once it is delivered, we need to implement a class that will store a set of waits. This is necessary in order to link requests from the Client <\/em>service with responses from the Server<\/em> service. This way, a lot of users can \u201cpull\u201d our end-point simultaneously, and we will return each user the data they requested without getting mixed up. Let\u2019s call this class\u00a0SenderReceiverMap<\/code>.<\/p>\n
Obviously, the responses from the Server<\/em> can be delivered in any order \u2013 it depends on the processing time of a particular request.\u00a0For example, the processing time of one request may be 5 seconds, and of another may be 3 seconds. To be able to look for connections between the thread and the user\u2019s request, we need to label them uniquely. To do this, we must enter an\u00a0id<\/code> of the request. To perform the search quickly, we need to use Map. Since we are working with threads, we need to use safe collections. Thus, we will get the following:<\/p>\n
private final ConcurrentMap<T, SenderReceiver> senderReceiverConcurrentMap;<\/code><\/pre>\n
As you can probably guess,\u00a0T<\/code>\u00a0is an id type. It can be anything, such as an Integer, String, or UUID. I prefer UUID.<\/p>\n
To add a new wait, we need to implement a method which will receive a pre-generated request id<\/code>\u00a0and which we will additionally transfer<\/code> to the\u00a0Server<\/em>\u00a0(more on this later) and\u00a0timeout.<\/p>\n
public Thread add(T id, Long timeout) {     SenderReceiver<V> responseWait = new SenderReceiver<V>(timeout);     senderReceiverConcurrentMap.put(id, responseWait);     Runnable task = responseWait::receive;     return new Thread(task); }<\/code><\/pre>\n
V<\/code>\u00a0is a type of the message data. In our case, it is String. You can see that we create a SenderReceiver<\/code>\u00a0and add it to the collection with a relevant request id. Then we create a new\u00a0Thread()<\/code>\u00a0and immediately return it, so that we can pause it until the data is delivered in the\u00a0send()<\/code> method of the\u00a0SenderReceiver<\/code> class.<\/p>\n
We will also need methods that can return the required SenderReceiver<\/code>\u00a0from the\u00a0SenderReceiverConcurrentMap<\/code>, and check if there is\u00a0id<\/code>\u00a0in the\u00a0SenderReceiverConcurrentMap<\/code>, and a method to remove a request from the\u00a0SenderReceiverConcurrentMap<\/code>. Review this implementation below, since there is nothing to comment on here.<\/p>\n
public SenderReceiver<V> get(T id) {     return senderReceiverConcurrentMap.get(id); } public Boolean containsKey(T id) { return senderReceiverConcurrentMap.containsKey(id); } public SenderReceiver remove(T id) { return senderReceiverConcurrentMap.remove(id); }<\/code><\/pre>\n
Implementing a Request for Data from the Server Service<\/strong><\/p>\n
Now we will implement a request to the Server<\/em> service via REST, and add its id to the collection together with the wait for the result.<\/p>\n
public String get(String text) throws TimeoutException {     UUID requestId = UUID.randomUUID();     while (senderReceiverMap.containsKey(requestId)) {         requestId = UUID.randomUUID();     }      String responseFromServer = this.sendText(requestId, text);     System.out.println(\"REST response from server: \" + responseFromServer);      Thread thread = senderReceiverMap.add(requestId, timeout);     thread.start();     try {         thread.join();     } catch (InterruptedException e) {         e.printStackTrace();     }     String responseKafka;     try {         responseKafka = senderReceiverMap.get(requestId).getData();     } catch (TimeoutException e) {         throw e;     } finally {         senderReceiverMap.remove(requestId);     }     return responseKafka; }<\/code><\/pre>\n
As you can see, we receive a string from the user in this method. Next, we create a request id<\/code> and send it together with the received string. We add the request together with its\u00a0id to the collection of requests, and run the resulting\u00a0Thread<\/code>. At this step, the process will freeze until someone \u201cpulls\u201d the send()<\/code>\u00a0method from the\u00a0SenderReceiver<\/code> object, which can be found by a particular request\u00a0id<\/code>.<\/p>\n
If someone calls the\u00a0send()\u00a0method and sends data to it, or the thread itself terminates via\u00a0timeout<\/code>, the method will continue to work. It calls\u00a0senderReceiverMap.get(requestId).getData()<\/code> which will return either\u00a0TimeoutException<\/code> or the data sent to the\u00a0send()<\/code> method. All we have to do is to remove the already-processed request from the collection and return the data we received.<\/p>\n
KafkaListener Implementation<\/strong><\/p>\n
The information explained above makes clear that we only have to call the\u00a0send()<\/code>\u00a0method from a particular SenderReceiver<\/code> wait object that can be easily found by its\u00a0id<\/code>. Since we synchronize requests with\u00a0Kafka<\/code>, we should call this method when we receive the data from it.<\/p>\n
@KafkaListener(topics = \"${kafka.topic}\", groupId = \"${kafka.groupId}\") public void listenGroupFoo(ConsumerRecord<String, KafkaMessage<String>> record) {     UUID rqId = this.getRqId(record.headers());     if (senderReceiverMap.containsKey(rqId)) {         SenderReceiver<String> stringSenderReceiver = senderReceiverMap.get(rqId);         stringSenderReceiver.send(record.value().getData());     } }<\/code><\/pre>\n
There is nothing complicated here either: from\u00a0headers<\/code>\u00a0we obtain a request\u00a0id\u00a0required to obtain the relevant\u00a0SenderReceiver<\/code>\u00a0from Map, which we will use to call the\u00a0send()<\/code> method.<\/p>\n
As you can see, there is no need to use Kafka for synchronization; we can use any other thread as long as we have a request\u00a0id<\/code>\u00a0and the data we want to return. For example, you could use a different message broker, or a totally different REST request. This is a flexible solution, because all you need to do is to call\u00a0send()<\/code>.<\/p>\n
Server Service Implementation<\/h2>\n
This is really very simple. Just don\u2019t forget to change the\u00a0port<\/code> on which the Server<\/em> service will run to ensure it will run simultaneously with the Client<\/em> service. To do this, set the following parameter in the\u00a0application.properties<\/code>:<\/p>\n
server.port=8888<\/code><\/pre>\n
End-Point Implementation<\/strong><\/p>\n
The\u00a0Server<\/em>\u00a0service must have an end-point that receives data from the\u00a0Client<\/em>\u00a0service and sends the result to Kafka linked with the request\u00a0id<\/code>.<\/p>\n
@PostMapping(\"\/test\") public String test(@RequestBody RequestDto request) {      Runnable runnable =             () -> {                 System.out.println(\"Start requestId: \" + request.getRequestId() + \"   text: \" + request.getData());                  try {                     int sleepMs = ThreadLocalRandom.current().nextInt(0, 10000 + 1);                     System.out.println(\"RequestId: \" + request.getRequestId() + \" sleep: \" + sleepMs + \"ms\");                     Thread.sleep(sleepMs);                 } catch (InterruptedException e) {                     e.printStackTrace();                 }                  kafkaMessageSender.send(request.getRequestId(), new KafkaMessage<>(request.getData().toUpperCase()));                  System.out.println(\"End requestId: \" + request.getRequestId());             };     Thread thread = new Thread(runnable);     thread.start();      return \"Ok!\"; }<\/code><\/pre>\n
This end-point takes a structure which contains\u00a0requestId<\/code>\u00a0\u2014 a request\u00a0id<\/code>\u00a0and\u00a0data<\/code>\u00a0\u2014 the data needed to be processed. This is a string in our case. This end-point creates a separate thread, which doesn\u2019t wait for a thread to terminate, but immediately returns an \u201cOk!\u201d<\/code> string via REST in response. In the thread, we emulate how hard the Server<\/em> works. To do this, we randomly generate the number of milliseconds during which the thread will fall asleep, and after it wakes up it will send data (an uppercase string) to Kafka.<\/p>\n
Implementation of Sending a Message in Kafka<\/strong><\/p>\n
Sending in Kafka looks like this:<\/p>\n
public void send(final UUID requestId, final KafkaMessage<String> message) {     ProducerRecord<String, KafkaMessage<String>> record = new ProducerRecord<>(topic, message);     record.headers().add(new RecordHeader(RQ_ID, requestId.toString().getBytes()));     ListenableFuture<SendResult<String, KafkaMessage<String>>> future = kafkaTemplate.send(record);     future.addCallback((success) -> { }, System.out::println );     kafkaTemplate.flush(); }<\/code><\/pre>\n
In the\u00a0headers<\/code>, we set a new RQ_ID<\/code> header where we save a request id<\/code>, and then we just call\u00a0send()<\/code> where we send the processed data.<\/p>\n
This where the Server<\/em> service stops.<\/p>\n
Conclusions<\/h2>\n
In fact, you don\u2019t have to use REST+Kafka in this solution. As you can see, the solution is universal, and this implementation can easily be changed to fit any interaction, whether it is REST+REST, or Kafka+Kafka, or even pigeon mail.<\/p>\n
You can find a case study\u00a0here<\/a>.<\/p>\n<\/p>\n<\/div>\n<\/div>\n<\/div>\n
<\/div>\n

 \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438  https:\/\/habr.com\/ru\/articles\/694292\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"
<\/div>\n
\n
\n
\n
\n
On one of the Innotech\u2019s projects, we received a task to convert asynchronous requests into synchronous ones. Basically, the purpose was to integrate REST and Apache Kafka into the same request.<\/p>\n
In more detail, we have two services that communicate with each other. Let\u2019s call them A and B. Service A receives a request from a consumer for data that is stored in service B. Thus, service A sends a request for data to B in REST and waits for the response of this request in Kafka. The user is waiting for data until this response is received.<\/p>\n
It appears to be a common problem, and the solution should be in Google or Stack Overflow. We managed to find the solution of a similar problem only in an existing library connecting two servers to Kafka. So the problem was solved using Java, Spring framework and a little IT-wit.<\/p>\n
Problem Statement<\/h2>\n
Before beginning implementation, we need to state the problem and understand what we have and what we want to achieve.<\/p>\n
We have a Client<\/em> service that will have one end-point. An end-point, in turn, will take a normal string, send it to the second service, and wait for the response in Kafka.<\/p>\n
And we have the second service \u2014\u00a0Server<\/em>. It will receive a REST string from the Client<\/em> service, convert it to UpperCase and return in response by Kafka.<\/p>\n
The Client<\/em> service awaits for the response from the Server<\/em>.\u00a0The user will wait for the result until the response is delivered. Waiting for a response may take a long time. To avoid too high of latency, it is necessary to provide an interrupt timeout, e.g. 10 seconds. But this condition should not be necessary; we should be able to wait indefinitely for a response. But definitely not waiting for 7.5 million years, otherwise the response could be disappointing.<\/p>\n
This is an example of how it works. The consumer sends a string to the Client<\/em> service, e.g. \u201cabc123\u201d, and until the response from the Server<\/em> service is delivered, they will be kept hanging. The response which will be returned to the consumer from the Server<\/em> should be \u201cABC123\u201d. If the wait time exceeds the timeout value, an HTTP code with a 504 (Gateway Timeout) error will be returned instead of the response.<\/p>\n
<\/figcaption><\/figure>\n
Client Implementation<\/h2>\n
I will only describe the main points. If you need the whole code, there is a link to the repository below.<\/p>\n
SenderReceiver Implementation<\/strong><\/p>\n
The logic core is a\u00a0SenderReceiver<\/code> class implementation, which is responsible for the process’s rest state until it receives data from the outside.<\/p>\n
This class consists of two main methods:\u00a0a receive()<\/code>\u00a0method is responsible for resting until data is received, and a\u00a0send()<\/code> method is responsible for receiving data and waking up the\u00a0receive()<\/code> method. To understand the current state of the thread, 1) whether resting or waiting (the receive()<\/code> method), or 2) receiving and waking up (the send()<\/code> method), we will create a\u00a0boolean<\/code> flag called\u00a0transfer<\/code>\u00a0and initialize it using the true<\/code> value.<\/p>\n
private boolean transfer = true;<\/code><\/pre>\n
Now let\u2019s implement\u00a0the receive()<\/code> method:<\/p>\n
public synchronized void receive() {     while (transfer) {         if (timeout != 0 && start.before(new Date(System.currentTimeMillis() - timeout))) {             timeoutException = new TimeoutException();             Thread.currentThread().interrupt();             return;         }         try {             wait(timeout);         } catch (InterruptedException e) {             Thread.currentThread().interrupt();         }     }     Thread.currentThread().interrupt(); }<\/code><\/pre>\n
Since initially the flag\u00a0transfer = true<\/code>, the process goes into a standby mode until it is woken up by the\u00a0send()<\/code> method, or milliseconds set in timeout<\/code> elapse.<\/p>\n
If the\u00a0timeout<\/code>\u00a0value is different from 0, it means that the thread interrupt mechanism by runtime has been set. To do this, we must compare the current time offset by this value with the start time of the thread. If this time is exceeded, we save\u00a0TimeoutException()<\/code>\u00a0into a variable which is initially null (we\u2019ll need it later), and terminate the thread. If the\u00a0transfer<\/code>\u00a0flag becomes\u00a0false<\/code>\u00a0and the thread has not finished waiting using a timeout, we will simply exit the loop and terminate the thread.<\/p>\n
You can see that\u00a0the timeout<\/code>\u00a0with a value of 0 is unique. If it is 0, the thread will wait indefinitely to wake up. After it is woken up and\u00a0transfer<\/code> changes\u00a0to a false<\/code> value, the thread will terminate.<\/p>\n
Now, let\u2019s implement the\u00a0send()<\/code> method:<\/p>\n
public synchronized void send(final T data) {     transfer = false;     this.data = data;     notifyAll(); }<\/code><\/pre>\n
You can see that the\u00a0send()<\/code> method is quite simple. It receives data from outside, changes our\u00a0transfer<\/code>\u00a0flag to\u00a0false<\/code>, saves the data, and wakes up all the threads that are hanging in\u00a0wait()<\/code>.<\/p>\n
In fact, there is another important auxiliary method called\u00a0getData()<\/code> which either returns the data stored in\u00a0data<\/code>, or generates an error if the relevant variable is not null.<\/p>\n
public T getData() throws TimeoutException {     if (Objects.nonNull(timeoutException))         throw timeoutException;     return data; }<\/code><\/pre>\n
SenderReceiverMap Implementation<\/strong><\/p>\n
After we have implemented the mechanism of waiting for a response and waking up once it is delivered, we need to implement a class that will store a set of waits. This is necessary in order to link requests from the Client <\/em>service with responses from the Server<\/em> service. This way, a lot of users can \u201cpull\u201d our end-point simultaneously, and we will return each user the data they requested without getting mixed up. Let\u2019s call this class\u00a0SenderReceiverMap<\/code>.<\/p>\n
Obviously, the responses from the Server<\/em> can be delivered in any order \u2013 it depends on the processing time of a particular request.\u00a0For example, the processing time of one request may be 5 seconds, and of another may be 3 seconds. To be able to look for connections between the thread and the user\u2019s request, we need to label them uniquely. To do this, we must enter an\u00a0id<\/code> of the request. To perform the search quickly, we need to use Map. Since we are working with threads, we need to use safe collections. Thus, we will get the following:<\/p>\n
private final ConcurrentMap<T, SenderReceiver> senderReceiverConcurrentMap;<\/code><\/pre>\n
As you can probably guess,\u00a0T<\/code>\u00a0is an id type. It can be anything, such as an Integer, String, or UUID. I prefer UUID.<\/p>\n
To add a new wait, we need to implement a method which will receive a pre-generated request id<\/code>\u00a0and which we will additionally transfer<\/code> to the\u00a0Server<\/em>\u00a0(more on this later) and\u00a0timeout.<\/p>\n
public Thread add(T id, Long timeout) {     SenderReceiver<V> responseWait = new SenderReceiver<V>(timeout);     senderReceiverConcurrentMap.put(id, responseWait);     Runnable task = responseWait::receive;     return new Thread(task); }<\/code><\/pre>\n
V<\/code>\u00a0is a type of the message data. In our case, it is String. You can see that we create a SenderReceiver<\/code>\u00a0and add it to the collection with a relevant request id. Then we create a new\u00a0Thread()<\/code>\u00a0and immediately return it, so that we can pause it until the data is delivered in the\u00a0send()<\/code> method of the\u00a0SenderReceiver<\/code> class.<\/p>\n
We will also need methods that can return the required SenderReceiver<\/code>\u00a0from the\u00a0SenderReceiverConcurrentMap<\/code>, and check if there is\u00a0id<\/code>\u00a0in the\u00a0SenderReceiverConcurrentMap<\/code>, and a method to remove a request from the\u00a0SenderReceiverConcurrentMap<\/code>. Review this implementation below, since there is nothing to comment on here.<\/p>\n
public SenderReceiver<V> get(T id) {     return senderReceiverConcurrentMap.get(id); } public Boolean containsKey(T id) { return senderReceiverConcurrentMap.containsKey(id); } public SenderReceiver remove(T id) { return senderReceiverConcurrentMap.remove(id); }<\/code><\/pre>\n
Implementing a Request for Data from the Server Service<\/strong><\/p>\n
Now we will implement a request to the Server<\/em> service via REST, and add its id to the collection together with the wait for the result.<\/p>\n
public String get(String text) throws TimeoutException {     UUID requestId = UUID.randomUUID();     while (senderReceiverMap.containsKey(requestId)) {         requestId = UUID.randomUUID();     }      String responseFromServer = this.sendText(requestId, text);     System.out.println(\"REST response from server: \" + responseFromServer);      Thread thread = senderReceiverMap.add(requestId, timeout);     thread.start();     try {         thread.join();     } catch (InterruptedException e) {         e.printStackTrace();     }     String responseKafka;     try {         responseKafka = senderReceiverMap.get(requestId).getData();     } catch (TimeoutException e) {         throw e;     } finally {         senderReceiverMap.remove(requestId);     }     return responseKafka; }<\/code><\/pre>\n
As you can see, we receive a string from the user in this method. Next, we create a request id<\/code> and send it together with the received string. We add the request together with its\u00a0id to the collection of requests, and run the resulting\u00a0Thread<\/code>. At this step, the process will freeze until someone \u201cpulls\u201d the send()<\/code>\u00a0method from the\u00a0SenderReceiver<\/code> object, which can be found by a particular request\u00a0id<\/code>.<\/p>\n
If someone calls the\u00a0send()\u00a0method and sends data to it, or the thread itself terminates via\u00a0timeout<\/code>, the method will continue to work. It calls\u00a0senderReceiverMap.get(requestId).getData()<\/code> which will return either\u00a0TimeoutException<\/code> or the data sent to the\u00a0send()<\/code> method. All we have to do is to remove the already-processed request from the collection and return the data we received.<\/p>\n
KafkaListener Implementation<\/strong><\/p>\n
The information explained above makes clear that we only have to call the\u00a0send()<\/code>\u00a0method from a particular SenderReceiver<\/code> wait object that can be easily found by its\u00a0id<\/code>. Since we synchronize requests with\u00a0Kafka<\/code>, we should call this method when we receive the data from it.<\/p>\n
@KafkaListener(topics = \"${kafka.topic}\", groupId = \"${kafka.groupId}\") public void listenGroupFoo(ConsumerRecord<String, KafkaMessage<String>> record) {     UUID rqId = this.getRqId(record.headers());     if (senderReceiverMap.containsKey(rqId)) {         SenderReceiver<String> stringSenderReceiver = senderReceiverMap.get(rqId);         stringSenderReceiver.send(record.value().getData());     } }<\/code><\/pre>\n
There is nothing complicated here either: from\u00a0<\/code><\/p>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-417567","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/417567","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=417567"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/417567\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=417567"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=417567"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=417567"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}