clearing-service kafka queue-consumer fix consumer timeout
This commit is contained in:
ialbert
parent
a8ad3eb474
commit
07c5dba41e
2 changed files with 271 additions and 2 deletions
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<String, ConsumerSpecificClass<?>> callbacks;
|
||||
private final Logger log = LoggerFactory.getLogger(getClass());
|
||||
private final AtomicBoolean closed = new AtomicBoolean(false);
|
||||
private final Consumer<String, Object> consumer;
|
||||
private final ExecutorService inputExecutor;
|
||||
private final ExecutorService processExecutor;
|
||||
private final ExecutorService outputExecutor;
|
||||
private final ObjectMapper json;
|
||||
protected boolean supportStartOffsetTimeWindow;
|
||||
private Producer<String, Object> producer;
|
||||
|
||||
public QueueConsumerV2(Consumer<String, Object> 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<String, Object> kafkaQueue, Producer<String, Object> 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<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
|
||||
for (ConsumerRecord<String, Object> 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<RecordMetadata> send;
|
||||
//default response
|
||||
BaseRequest<RequestInfoUpdate> 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<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) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
log.error(ExceptionUtils.getStackTrace(e));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
protected void sendResponse(BaseRequest<?> o, Object response, Header correlationId) {
|
||||
outputExecutor.submit(() -> {
|
||||
try {
|
||||
Future<RecordMetadata> send;
|
||||
BaseRequest<Object> 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<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));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
protected <T> BuilderConsumerStep<T> callback(Class<T> clazz) {
|
||||
return ConsumerSpecificClass.build(clazz);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Валидация запроса
|
||||
*
|
||||
* @param <R> Класс проверяемого запроса
|
||||
* @return null если ошибок нет
|
||||
*/
|
||||
@Deprecated
|
||||
public <R> RequestInfoUpdate validate(BaseRequest<R> userRequest,
|
||||
Function<R, IValidator> validatorBuilder,
|
||||
IMessageResolver messageResolver) {
|
||||
if (validatorBuilder != null) {
|
||||
R req = userRequest.getRequestPayload();
|
||||
IValidator validator = validatorBuilder.apply(req);
|
||||
Optional<EnumMessage> 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;
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue