From 3931d453da4757f797ae06e74a2768199ca80c37 Mon Sep 17 00:00:00 2001 From: ialbert Date: Thu, 11 May 2023 12:18:48 +0300 Subject: [PATCH] =?UTF-8?q?http://jira.mfd.msk:8088/browse/CLS-290=20?= =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D0=BB=20CORRELATION=5FID?= =?UTF-8?q?=20header=20=D0=B2=20=D1=81=D0=BE=D0=BE=D0=B1=D1=89=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D1=8F=20=D0=BA=D0=B0=D1=84=D0=BA=D0=B8,=20=D0=B4=D0=BB?= =?UTF-8?q?=D1=8F=20=D0=B8=D1=81=D0=BF=D0=BE=D0=BB=D1=8C=D0=B7=D0=BE=D0=B2?= =?UTF-8?q?=D0=B0=D0=BD=D0=B8=D1=8F=20Spring=20ReplyingKafkaTemplate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- platform-parent/platform-messaging/pom.xml | 4 +++ .../messaging/domain/BaseRequest.java | 12 ++++++++ .../messaging/service/QueueConsumer.java | 28 +++++++++++++++---- .../service/sender/CorrelationHeader.java | 23 +++++++++++++++ .../messaging/service/sender/KafkaSender.java | 12 ++++++-- 5 files changed, 71 insertions(+), 8 deletions(-) create mode 100644 platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/CorrelationHeader.java diff --git a/platform-parent/platform-messaging/pom.xml b/platform-parent/platform-messaging/pom.xml index 40ee2a9a9..0de27f80b 100644 --- a/platform-parent/platform-messaging/pom.xml +++ b/platform-parent/platform-messaging/pom.xml @@ -49,5 +49,9 @@ platform-enum provided + + org.springframework.kafka + spring-kafka + \ No newline at end of file diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/BaseRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/BaseRequest.java index 5672500aa..8fd44bef4 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/BaseRequest.java +++ b/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 implements WithId { @JsonProperty private Long userId; + @JsonIgnore + private byte[] correlationId; + @Override public Long getId() { return id; @@ -58,4 +62,12 @@ public class BaseRequest implements WithId { public void setUserId(Long userId) { this.userId = userId; } + + public byte[] getCorrelationId() { + return correlationId; + } + + public void setCorrelationId(byte[] correlationId) { + this.correlationId = correlationId; + } } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java index 0b26d5c03..307432a21 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java +++ b/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 records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS)); for (ConsumerRecord 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 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 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 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 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)); diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/CorrelationHeader.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/CorrelationHeader.java new file mode 100644 index 000000000..83f0c7c8f --- /dev/null +++ b/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; + } +} diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSender.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSender.java index 613966cc8..35617813f 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/sender/KafkaSender.java +++ b/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) { + public Long sendRequestToQueue(String destination, Object requestPayload, byte[] correlationId) { BaseRequest request = new BaseRequest<>(); request.setId(idGenerator.get()); request.setActionType(ActionType.SYSTEM); request.setRequestPayload(requestPayload); //сохраняет данные о запросе в хранилище saveRequestToStorage(destination, request); + ProducerRecord> prdRec = new ProducerRecord<>(destination, request); + if (correlationId != null) { + prdRec.headers().add(new CorrelationHeader(correlationId.clone())); + } Future 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 request) { if (!saveRequestInfo) { return;