#java #spring #spring-boot #rabbitmq #spring-cloud-stream
#java #весна #весенняя загрузка #rabbitmq #spring-cloud-stream
Вопрос:
У меня есть приложение, в котором я должен переместить сообщение в DLQ (очередь мертвых писем) в конкретном случае, но при первом сообщении генерируется исключение, но оно игнорирует мое значение max-attempts: 2, и я вижу свое сообщение об исключении 4 раза, и сообщение не ушлов DLQ.
Как упоминалось в документах spring cloud stream, сообщение должно отправляться в DLQ при возникновении исключения. Когда я пытаюсь отправить другое сообщение, я продолжаю получать другое исключение в журналах консоли, пока не перезапущу приложение, чтобы снова получить первое сообщение, но теперь сообщение отправляется в DLQ.
Я попытался использовать .onErrorContinue, в этом случае мое приложение не прерывается при первом сообщении, но сообщения не отправляются в DLQ.
Редактировать:
Существует открытый инцидент в репозитории spring-cloud-stream для этой проблемы https://github.com/spring-cloud/spring-cloud-stream/issues/1922
Если у кого-нибудь есть какой-либо обходной путь для повторной попытки, очередь мертвых писем внутри реактивной функции, пожалуйста, предоставьте ответ.
Мое приложение
@Bean
public Function<Flux<String>, Flux<?>> processor() {
return PaymentDetailsFlux -> PaymentDetailsFlux
// .retryWhen(Retry.backoff(3, Duration.ofMillis(100000)))
.flatMap(
paymentDetails -> {
throw new RuntimeException("intentional");
/* extracting payload from input */
// Map<Object, Object> payload = new HashMap<>();
// try {
// ObjectMapper jsonMapper = new ObjectMapper();
// payload = jsonMapper.readValue(paymentDetails, Map.class);
// } catch (JsonProcessingException e) {
// log.error("error while serializing input payload: " paymentDetails);
// e.printStackTrace();
// }
// String message_id = (String) payload.get("message_id");
// log.info("<" message_id ">" " Received message from payment-supplier");
// log.info("<" message_id ">" " message payload: " paymentDetails);
//
// /* DLQ if rule not found */
// if(!isValidateMAP(payload)){
// throw new RuntimeException("intentional");
//// return Flux.error(new RuntimeException("<" message_id "> " errMessage));
// }
// return Mono.just(paymentDetails);
}
);
// .onErrorContinue((throwable, o) -> {
// log.error(throwable.getMessage());
// });
}
my application.yml
spring:
application:
name: payment-supplier
profiles:
active: dev
cloud:
function:
definition: processor;ruleProcessor
stream:
bindings:
processor-in-0:
destination: processor-in #input topic name
processor-in-0.group: processor #queue name
processor-out-0:
destination: rule-in #output topic name
ruleProcessor-in-0:
destination: ruleProcessor-in #input topic name
ruleProcessor-in-0.group: processor #queue name
#ruleProcessor-out-0: #StreamBridge
default-binder: rabbit
#Defining DLQ - Dead Letter Queue
rabbit:
bindings:
processor-in-0:
consumer:
auto-bind-dlq: true
republish-to-dlq: true
max-attempts: 2
requeueRejected: true
backOffInitialInterval: 900000
backOffMaxInterval: 900000
producer:
autoBindDlq: true
republishToDlq: true
requeueRejected: true
maxAttempts: 2
backOffInitialInterval: 900000
backOffMaxInterval: 900000
#Defining DLQ - Dead Letter Queue
# rabbit:
default:
consumer:
auto-bind-dlq: true
republishToDlq : true
maxAttempts : 3
requeueRejected : true
# backOffInitialInterval: 900000
# backOffMaxInterval: 900000
producer:
auto-bind-dlq: true
republishToDlq : true
maxAttempts : 3
requeueRejected : true
# backOffInitialInterval: 900000
# backOffMaxInterval: 900000
Первое сообщение
2021-10-18 13:12:40.704 ERROR 6704 --- [-in.processor-1] c.f.c.c.BeanFactoryAwareFunctionRegistry : Failed to invoke function 'processor'
java.lang.RuntimeException: intentional
at com.gaic.paymentprocessor.service.PaymentProcessorService.lambda$processor$0(PaymentProcessorService.java:50) ~[classes/:na]
springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.javaspringframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:374) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329) ~[spring-retry-1.3.1.jar:na]
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225) ~[spring-retry-1.3.1.jar:na]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:370) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(Aborg.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195) ~[spring-rabbit-2.3.10.jar:2.3.10]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
2021-10-18 13:12:40.704 ERROR 6704 --- [-in.processor-1] c.f.c.c.BeanFactoryAwareFunctionRegistry : Failed to convert output
java.lang.RuntimeException: intentional
at com.gaic.paymentprocessor.service.PaymentProcessorService.lambda$processor$0(PaymentProcessorService.java:50) ~[classes/:na]
springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1554) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:968) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1289) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195) ~[spring-rabbit-2.3.10.jar:2.3.10]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
2021-10-18 13:12:40.704 ERROR 6704 --- [-in.processor-1] onfiguration$FunctionToDestinationBinder : Failure was detected during execution of the reactive function 'processor'
2021-10-18 13:12:40.712 ERROR 6704 --- [-in.processor-1] reactor.core.publisher.Operators : Operator called default onErrorDropped
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.RuntimeException: intentional
Caused by: java.lang.RuntimeException: intentional
at com.gaic.paymentprocessor.service.PaymentProcessorService.lambda$processor$0(PaymentProcessorService.java:50) ~[classes/:na]
at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:386) ~[reactor-core-3.4.10.jar:3.4.10]
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120) ~[reactor-core-3.4.10.jar:3.4.10]
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200) ~[reactor-core-3.4.10.jar:3.4.10]
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120) ~[reactor-core-3.4.10.jar:3.4.10]org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72) ~[spring-integration-core-5.5.4.jar:5.5.4]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317) ~[spring-integration-core-5.5.4.jar:5.5.4]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272) ~[spring-integration-core-5.5.4.jar:5.5.4]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) ~[spring-messaging-5.3.10.jar:5.3.10]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-5.3.10.jar:5.3.10]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.3.10.jar:5.3.10]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) ~[spring-messaging-5.3.10.jar:5.3.10]
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:208) ~[spring-integration-core-5.5.4.jar:5.5.4]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1300(AmqpInboundChannelAdapter.java:69) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:374) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329) ~[spring-retry-1.3.1.jar:na]
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225) ~[spring-retry-1.3.1.jar:na]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:370) ~[spring-integration-amqp-5.5.4.jar:5.5.4]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1656) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1575) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1563) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1554) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:968) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1289) ~[spring-rabbit-2.3.10.jar:2.3.10]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195) ~[spring-rabbit-2.3.10.jar:2.3.10]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
java.lang.RuntimeException: intentional
at com.gaic.paymentprocessor.service.PaymentProcessorService.lambda$processor$0(PaymentProcessorService.java:50)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:386)
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200)
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:120)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200)
at reactor.core.publisher.UnicastManySinkNoBackpressure.tryEmitNext(UnicastManySinkNoBackpressure.java:120)
at reactor.core.publisher.SinkManySerialized.tryEmitNext(SinkManySerialized.java:100)
at org.springframework.integration.util.IntegrationReactiveUtils.lambda$adaptSubscribableChannelToPublisher$8(IntegrationReactiveUtils.java:142)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:208)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1300(AmqpInboundChannelAdapter.java:69)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:374)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:370)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1656)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1575)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1563)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1554)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:968)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1289)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195)
at java.base/java.lang.Thread.run(Thread.java:834)
Журналы второго сообщения
2021-10-18 12:57:25.422 ERROR 9928 --- [-in.processor-1] o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'payment-supplier.processor-in-0'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[1330], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=processor-in, amqp_deliveryTag=2, deliveryAttempt=3, amqp_consumerQueue=processor-in.processor, amqp_redelivered=false, id=368d916d-af66-24c1-59c1-491d5dbc6dfa, amqp_consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, sourceData=(Body:'[B@1d04b729(byte[1330])' MessageProperties [headers={}, contentLength=0, receivedDeliveryMode=PERSISTENT, redelivered=false, receivedExchange=processor-in, receivedRoutingKey=, deliveryTag=2, consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, consumerQueue=processor-in.processor]), contentType=application/json, timestamp=1634542042374}], failedMessage=GenericMessage [payload=byte[1330], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=processor-in, amqp_deliveryTag=2, deliveryAttempt=3, amqp_consumerQueue=processor-in.processor, amqp_redelivered=false, id=368d916d-af66-24c1-59c1-491d5dbc6dfa, amqp_consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, sourceData=(Body:'[B@1d04b729(byte[1330])' MessageProperties [headers={}, contentLength=0, receivedDeliveryMode=PERSISTENT, redelivered=false, receivedExchange=processor-in, receivedRoutingKey=, deliveryTag=2, consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, consumerQueue=processor-in.processor]), contentType=application/json, timestamp=1634542042374}]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:76)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:208)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1300(AmqpInboundChannelAdapter.java:69)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:374)
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329)
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:225)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:370)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1656)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1575)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1563)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1554)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1498)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:968)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1289)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1195)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[1330], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=processor-in, amqp_deliveryTag=2, deliveryAttempt=3, amqp_consumerQueue=processor-in.processor, amqp_redelivered=false, id=368d916d-af66-24c1-59c1-491d5dbc6dfa, amqp_consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, sourceData=(Body:'[B@1d04b729(byte[1330])' MessageProperties [headers={}, contentLength=0, receivedDeliveryMode=PERSISTENT, redelivered=false, receivedExchange=processor-in, receivedRoutingKey=, deliveryTag=2, consumerTag=amq.ctag-fibQI2iMkG5_giNWmMMwlA, consumerQueue=processor-in.processor]), contentType=application/json, timestamp=1634542042374}]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:139)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
... 23 more
Ответ №1:
Я ответил на вопрос в Github (https://github.com/spring-cloud/spring-cloud-stream/issues/1922 ), вот ответ
Привет @vishalmamidi,
Я реализовал свою собственную версию DLQ с помощью StreamBridge, в принципе, вам нужно реализовать onErrorNext и отправить ваше сообщение в DLQ через StreamBridge, а затем вернуться Mono.empty() , просто имейте в виду, что вам нужно использовать onErrorStop() для предотвращения любого странного поведения с вашими ошибками up / down stream, чтобы убедиться, что ошибка переходит внет onErrorNext onErrorContinue . Вот фрагмент кода, который я использовал
@Bean
@ConditionalOnProperty("app.reader.writerTest.enabled", havingValue = "true", matchIfMissing = true)
fun writerTest(
streamBridge: StreamBridge
) = Function<Flux<Message<TestReaderRequest>>, Mono<Void>> { item ->
item
.map {
println("TEST: $it")
it
}
.flatMap { request ->
Mono
.just("Service Test")
.map {
throw Exception("Testing DLQ")
}
.handleDlq(request, streamBridge, logger, 1)
}
.then()
}
fun <T> Mono<T>.handleDlq(request: Message<*>, streamBridge: StreamBridge, logger: Logger, retryThreshold: Int = 1) =
onErrorResume { exception -> handleDlq<T>(request, retryThreshold, streamBridge, logger, exception) }
.onErrorStop()
fun <T> Flux<T>.handleDlq(request: Message<*>, streamBridge: StreamBridge, logger: Logger, retryThreshold: Int = 1) =
onErrorResume { exception -> handleDlq<T>(request, retryThreshold, streamBridge, logger, exception) }
.onErrorStop()
private fun <T> handleDlq(
request: Message<*>,
retryThreshold: Int,
streamBridge: StreamBridge,
logger: Logger,
exception: Throwable
): Mono<T> {
logger.error(
"Unable to finish the operation of request: $request due to: ",
exception
)
val receivedTopic = request.headers[RECEIVED_TOPIC]
val attempts = request.headers[DELIVERY_ATTEMPT, AtomicInteger::class.java] ?: AtomicInteger(1)
val topic = if (attempts.get() < retryThreshold) {
attempts.incrementAndGet()
logger.info("About to retry the message of $request for topic: $receivedTopic, attempt #${attempts.get()} from $retryThreshold")
receivedTopic.toString()
} else {
// TODO: needs to find the dlq name from application.yml
val dlq = receivedTopic.toString() "-dlq"
logger.info("About to send the message of $request to dlq: $dlq after retrying ${attempts.get()} time(s)")
dlq
}
val message = MessageBuilder
.fromMessage(request)
.setHeaders(buildRetryOrDlqMessageHeaders(request, attempts, topic, exception))
.build()
streamBridge.send(topic, message)
return Mono.empty()
}
private fun buildRetryOrDlqMessageHeaders(
request: Message<*>,
attempts: AtomicInteger,
topic: String,
exception: Throwable
): MessageHeaderAccessor {
val messageHeaders = MessageHeaderAccessor(request)
messageHeaders.setHeader("spring.cloud.stream.sendto.destination", topic)
messageHeaders.setHeader(DELIVERY_ATTEMPT, attempts)
messageHeaders.setHeader(DLT_ORIGINAL_TOPIC, request.headers[RECEIVED_TOPIC])
messageHeaders.setHeader(DLT_EXCEPTION_FQCN, exception.javaClass.name)
messageHeaders.setHeader(DLT_EXCEPTION_MESSAGE, exception.message)
messageHeaders.setHeader(DLT_EXCEPTION_STACKTRACE, exception.stackTraceToString())
return messageHeaders
}
Дайте мне знать, если у вас есть какие-либо вопросы, надеюсь, это поможет