http://jira.mfd.msk:8088/browse/CLS-290 Добавил CORRELATION_ID header в сообщения кафки, для использования Spring ReplyingKafkaTemplate

platform-parent/platform-messaging/pom.xml

@@ -49,5 +49,9 @@
             <artifactId>platform-enum</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
     </dependencies>
 </project>

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/BaseRequest.java

@@ -1,5 +1,6 @@
 package ru.spcex.clearing.platform.messaging.domain;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import ru.spcex.platform.classes.base.interfaces.WithId;
 
@@ -26,6 +27,9 @@ public class BaseRequest<T> implements WithId {
     @JsonProperty
     private Long userId;
 
+    @JsonIgnore
+    private byte[] correlationId;
+
     @Override
     public Long getId() {
         return id;
@@ -58,4 +62,12 @@ public class BaseRequest<T> implements WithId {
     public void setUserId(Long userId) {
         this.userId = userId;
     }
+
+    public byte[] getCorrelationId() {
+        return correlationId;
+    }
+
+    public void setCorrelationId(byte[] correlationId) {
+        this.correlationId = correlationId;
+    }
 }

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java

@@ -9,13 +9,16 @@ import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.header.Header;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.kafka.support.KafkaHeaders;
 import ru.spcex.clearing.platform.messaging.domain.ActionType;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
 import ru.spcex.clearing.platform.messaging.domain.Consts;
 import ru.spcex.clearing.platform.messaging.logic.functional.BuilderConsumerStep;
 import ru.spcex.clearing.platform.messaging.logic.functional.ConsumerSpecificClass;
+import ru.spcex.clearing.platform.messaging.service.sender.CorrelationHeader;
 import ru.spcex.platform.utils.enumeration.EnumMessage;
 import ru.spcex.platform.utils.enumeration.IMessageResolver;
 import ru.spcex.platform.utils.log.ExceptionUtils;
@@ -80,19 +83,24 @@ public class QueueConsumer implements AutoCloseable {
                     consumer.subscribe(callbacks.keySet());
                 }
                 Object o = null;
+                Header correlationId = null;
                 int lastErrors = 0;
                 while (!closed.get()) {
                     try {
                         ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
                         for (ConsumerRecord<String, Object> next : records) {
+                            correlationId = next.headers().lastHeader(KafkaHeaders.CORRELATION_ID);
                             ConsumerSpecificClass<?> callback = callbacks.get(next.topic());
                             Class<?> clazz = callback.getClazz();
                             JavaType payloadType = json.getTypeFactory().constructParametricType(BaseRequest.class, clazz);
                             o = json.readValue((String) next.value(), payloadType);
+                            if (correlationId != null) {
+                                ((BaseRequest<?>) o).setCorrelationId(correlationId.value());
+                            }
                             if (needsProcessing(next.topic(), (BaseRequest<?>) o)) {
                                 Object topicResponse = callback.acceptRaw(o);
                                 if (producer != null) {
-                                    sendResponse((BaseRequest<?>) o, topicResponse);
+                                    sendResponse((BaseRequest<?>) o, topicResponse, correlationId);
                                 }
                             }
                         }
@@ -100,7 +108,7 @@ public class QueueConsumer implements AutoCloseable {
                     } catch (Throwable e) {
                         log.error(ExceptionUtils.getStackTrace(e));
                         if (producer != null && o != null) {
-                            sendErrorResponse((BaseRequest<?>) o);
+                            sendErrorResponse((BaseRequest<?>) o, correlationId);
                         }
                         if (lastErrors++ > 20) {
                             log.warn("Too many error at row, {}. Sleep.", lastErrors);
@@ -124,7 +132,7 @@ public class QueueConsumer implements AutoCloseable {
     }
 
     //мб перенести в другой класс
-    private void sendErrorResponse(BaseRequest<?> o) {
+    private void sendErrorResponse(BaseRequest<?> o, Header correlationId) {
         outputExecutor.submit(() -> {
             try {
                 Future<RecordMetadata> send;
@@ -136,7 +144,11 @@ public class QueueConsumer implements AutoCloseable {
                 req.setRequestPayload(success);
                 req.setId(o.getId());
                 req.setActionType(ActionType.SYSTEM);
-                send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, req));
+                ProducerRecord<String, Object> respRec = new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, req);
+                if (correlationId != null) {
+                    respRec.headers().add(KafkaHeaders.CORRELATION_ID, correlationId.value());
+                }
+                send = producer.send(respRec);
                 send.get();
             } catch (Exception e) {
                 if (e instanceof InterruptedException) {
@@ -147,7 +159,7 @@ public class QueueConsumer implements AutoCloseable {
         });
     }
 
-    private void sendResponse(BaseRequest<?> o, Object response) {
+    private void sendResponse(BaseRequest<?> o, Object response, Header correlationId) {
         outputExecutor.submit(() -> {
             try {
                 Future<RecordMetadata> send;
@@ -163,7 +175,11 @@ public class QueueConsumer implements AutoCloseable {
                     success.setStatus(Status.Success);
                     req.setRequestPayload(success);
                 }
-                send = producer.send(new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, req));
+                ProducerRecord<String, Object> respRec = new ProducerRecord<>(Consts.REQUEST_INFO_UPDATE, req);
+                if (correlationId != null) {
+                    respRec.headers().add(new CorrelationHeader(correlationId.value().clone()));
+                }
+                send = producer.send(respRec);
                 send.get();
             } catch (Exception e) {
                 log.error(ExceptionUtils.getStackTrace(e));

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/CorrelationHeader.java

@@ -0,0 +1,23 @@
+package ru.spcex.clearing.platform.messaging.service.sender;
+
+import org.apache.kafka.common.header.Header;
+import org.springframework.kafka.support.KafkaHeaders;
+
+public class CorrelationHeader implements Header {
+
+    private final byte[] correlationId;
+
+    public CorrelationHeader(byte[] correlationId) {
+        this.correlationId = correlationId;
+    }
+
+    @Override
+    public String key() {
+        return KafkaHeaders.CORRELATION_ID;
+    }
+
+    @Override
+    public byte[] value() {
+        return correlationId;
+    }
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSender.java

@@ -34,14 +34,17 @@ public class KafkaSender {
         return new KafkaSenderBuilderImpl();
     }
 
-
+    public Long sendRequestToQueue(String destination, Object requestPayload, byte[] correlationId) {
-    public Long sendRequestToQueue(String destination, Object requestPayload) {
         BaseRequest<Object> request = new BaseRequest<>();
         request.setId(idGenerator.get());
         request.setActionType(ActionType.SYSTEM);
         request.setRequestPayload(requestPayload);
         //сохраняет данные о запросе в хранилище
         saveRequestToStorage(destination, request);
+        ProducerRecord<Object, BaseRequest<Object>> prdRec = new ProducerRecord<>(destination, request);
+        if (correlationId != null) {
+            prdRec.headers().add(new CorrelationHeader(correlationId.clone()));
+        }
         Future<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, request));
         try {
             send.get();
@@ -55,6 +58,11 @@ public class KafkaSender {
         return request.getId();
     }
 
+
+    public Long sendRequestToQueue(String destination, Object requestPayload) {
+        return sendRequestToQueue(destination, requestPayload, null);
+    }
+
     private void saveRequestToStorage(String destination, BaseRequest<Object> request) {
         if (!saveRequestInfo) {
             return;