diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java index 561d7b160..e5001fcba 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java @@ -28,7 +28,7 @@ import ru.spcex.clearing.platform.messaging.domain.cud.registry.RegistryReturnDe import ru.spcex.clearing.platform.messaging.domain.cud.registry.RegistrySplitDepositActionRequest; import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; import ru.spcex.clearing.platform.messaging.domain.cud.utilities.STradesImportedRequest; -import ru.spcex.clearing.platform.messaging.service.QueueConsumer; +import ru.spcex.clearing.platform.messaging.service.QueueConsumerV2; import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate; import ru.spcex.clearing.platform.messaging.service.Status; import ru.spcex.clearing.service.executors.Sdf06Executor; @@ -53,7 +53,7 @@ import ru.spcex.platform.utils.enumeration.IMessageResolver; import ru.spcex.platform.utils.error.ValidationException; @Service -public class EventsReceiver extends QueueConsumer implements InitializingBean { +public class EventsReceiver extends QueueConsumerV2 implements InitializingBean { private final Logger log = LoggerFactory.getLogger(getClass()); private final Environment environment; private final IMessageResolver errorResolver; diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumerV2.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumerV2.java new file mode 100644 index 000000000..38d084745 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumerV2.java @@ -0,0 +1,269 @@ +package ru.spcex.clearing.platform.messaging.service; + +import java.time.Duration; +import java.time.temporal.ChronoUnit; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import com.fasterxml.jackson.databind.JavaType; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +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; +import ru.spcex.platform.utils.validation.IValidator; + +/** + * утилитный класс для обработки сообщений из очереди + */ +public class QueueConsumerV2 implements AutoCloseable { + protected final Map> callbacks; + private final Logger log = LoggerFactory.getLogger(getClass()); + private final AtomicBoolean closed = new AtomicBoolean(false); + private final Consumer consumer; + private final ExecutorService inputExecutor; + private final ExecutorService processExecutor; + private final ExecutorService outputExecutor; + private final ObjectMapper json; + protected boolean supportStartOffsetTimeWindow; + private Producer producer; + + public QueueConsumerV2(Consumer kafkaQueue) { + this.consumer = kafkaQueue; + this.callbacks = new HashMap<>(); + this.inputExecutor = Executors.newSingleThreadExecutor(); + this.processExecutor = Executors.newSingleThreadExecutor(); + this.outputExecutor = Executors.newSingleThreadExecutor(); + this.json = new ObjectMapper(); + this.supportStartOffsetTimeWindow = false; + if (producer == null) { + log.debug("For {} kafka producer not set. Did not send reply.", getClass().getName()); + } + } + + /** + * используем этот конструктор, если хотим класть + * в кафку "ответ" - информацию о статусе обработки команд + * + * @param kafkaQueue + * @param kafkaResponseQueue + */ + public QueueConsumerV2(Consumer kafkaQueue, Producer kafkaResponseQueue) { + this(kafkaQueue); + this.producer = kafkaResponseQueue; + if (producer == null) { + log.warn("For {} kafka producer not set. Did not send reply.", getClass().getName()); + } + } + + protected boolean needsProcessing(String topicName, BaseRequest request) { + return true; + } + + private final AtomicInteger lastErrors = new AtomicInteger(0); + public void init() { + final Exception debugCallerStacktrace = new Exception("init call"); + inputExecutor.submit(() -> { + try { + log.debug("Using kafka consumer {} for subscribe on \"{}\"", consumer, callbacks.keySet()); + if (supportStartOffsetTimeWindow) { + consumer.subscribe(callbacks.keySet(), new OffsetChanger(consumer, callbacks.keySet())); + } else { + consumer.subscribe(callbacks.keySet()); + } + Object o = null; + Header correlationId = null; + while (!closed.get()) { + String lastTopic = null; + try { + if (this.lastErrors.get() > 20) { + log.warn("Too many error at row, {}. Sleep.", lastErrors); + try { + Thread.sleep(1000L); + } catch (InterruptedException e){ + log.info("Thread interrupted. {}", ExceptionUtils.getStackTrace(e)); + break; + } + } + ConsumerRecords records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS)); + for (ConsumerRecord next : records) { + lastTopic = next.topic(); + 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 fO = o; + Header fCI = correlationId; + String fLastTopic = lastTopic; + CompletableFuture + .supplyAsync(() -> callback.acceptRaw(fO), processExecutor) + .thenAccept(topicResponse -> { + this.lastErrors.set(0); + if (producer != null) { + sendResponse((BaseRequest) fO, topicResponse, fCI); + } + } + ) + .exceptionally(e -> { + processError(e, fO, fLastTopic, fCI, debugCallerStacktrace); + return null; + }); + } + } + } catch (Throwable e) { + processError(e, o, lastTopic, correlationId, debugCallerStacktrace); + } + } + } catch (WakeupException e) { + if (!closed.get()) throw e; + } catch (Throwable e) { + log.error("QueueConsumer error: {}; main thread stacktrace: {}", + ExceptionUtils.getStackTrace(e), ExceptionUtils.getStackTrace(debugCallerStacktrace)); + } finally { + consumer.close(); + } + }); + } + + private void processError(Throwable e, Object o, String lastTopic, Header correlationId, Exception debugCallerStacktrace) { + log.error("Listener {} last topic \"{}\", error: {}", + QueueConsumerV2.this.getClass().getName(), lastTopic, ExceptionUtils.getStackTrace(e)); + if (lastTopic == null) { + log.debug("main thread stacktrace: {}", ExceptionUtils.getStackTrace(debugCallerStacktrace)); + } + if (producer != null && o != null) { + sendErrorResponse((BaseRequest) o, correlationId); + } + lastErrors.incrementAndGet(); + } + + //мб перенести в другой класс + private void sendErrorResponse(BaseRequest o, Header correlationId) { + outputExecutor.submit(() -> { + try { + Future send; + //default response + BaseRequest req = new BaseRequest<>(); + RequestInfoUpdate success = new RequestInfoUpdate(); + success.setId(o.getId()); + success.setStatus(Status.Error); + req.setRequestPayload(success); + req.setId(o.getId()); + req.setActionType(ActionType.SYSTEM); + 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) { + Thread.currentThread().interrupt(); + } + log.error(ExceptionUtils.getStackTrace(e)); + } + }); + } + + protected void sendResponse(BaseRequest o, Object response, Header correlationId) { + outputExecutor.submit(() -> { + try { + Future send; + BaseRequest req = new BaseRequest<>(); + req.setId(o.getId()); + req.setActionType(ActionType.SYSTEM); + if (response != null) { + req.setRequestPayload(response); + } else { + //default response + RequestInfoUpdate success = new RequestInfoUpdate(); + success.setId(o.getId()); + success.setStatus(Status.Success); + req.setRequestPayload(success); + } + 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)); + } + }); + } + + protected BuilderConsumerStep callback(Class clazz) { + return ConsumerSpecificClass.build(clazz); + } + + + /** + * Валидация запроса + * + * @param Класс проверяемого запроса + * @return null если ошибок нет + */ + @Deprecated + public RequestInfoUpdate validate(BaseRequest userRequest, + Function validatorBuilder, + IMessageResolver messageResolver) { + if (validatorBuilder != null) { + R req = userRequest.getRequestPayload(); + IValidator validator = validatorBuilder.apply(req); + Optional validationError = validator.tillFirstError(); + if (validationError.isPresent()) { + String errorMsg = messageResolver.resolve(validationError.get()); + log.warn("validation error for {} error={}, id={}: {}", req.getClass().getSimpleName(), validationError.get().getSubject(), userRequest.getId(), errorMsg); + return new RequestInfoUpdate() + .setId(userRequest.getId()) + .setStatus(Status.Error) + .setMessage(errorMsg); + } + } + return null; + } + + @Override + public void close() { + log.debug("Closing queue consumer {}", getClass().getSimpleName()); + closed.set(true); + consumer.wakeup(); + inputExecutor.shutdown(); + processExecutor.shutdown(); + outputExecutor.shutdown(); + } + + // нужен для тестов поскольку у каждого сервиса свой consumer + public Consumer getConsumer() { + return consumer; + } +}