Spring Integration Error Handling with Email Once at end

Question

How do I handle errors thrown in the below SpringIntegrationFlow just by logging the error and continuing with the next record. As well should be able to send an email only once irrespective of 1 or more records failed. Is there something equivalent of Spring Batch's StepListener? Thanks

return IntegrationFlows.from(jdbcMessageSource(), p -> p.poller(pollerSpec()))
                       .enrichHeaders(Collections.singletonMap(MessageHeaders.ERROR_CHANNEL, appErrorChannel()))
                       .split()
                       .channel(c -> c.executor(Executors.newCachedThreadPool()))
                       .transform(transformer, "transform")
                       .enrichHeaders(headerEnricherSpec -> headerEnricherSpec.header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE))
                       .handle(Http.outboundGateway(url)
                                   .httpMethod(HttpMethod.POST)
                                   .expectedResponseType(String.class)
                                   .requestFactory(requestFactory))
                       .get();

@Bean
public MessageChannel appErrorChannel() {

    return new DirectChannel();
}

@Bean("appErrorFlow")
public IntegrationFlow appErrorFlow() {

    // @formatter:off
    return IntegrationFlows.from(appErrorChannel())
                            .log(Level.ERROR, message -> " Failed Message " + message.getPayload())
                            .aggregate(a -> a.correlationStrategy(m -> !m.getHeaders().isEmpty()))
                            .handle(Http.outboundGateway(mailURL)
                                        .httpMethod(HttpMethod.POST))
                            .get();
    // @formatter:on
}

Exception:

2022-01-13 06:41:31,702 [Worker-1                  ] WARN  o.s.i.c.MessagePublishingErrorHandler - 005006f2dee5b673 Error message was not delivered.
org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'unknown.channel.name'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=ErrorMessage [payload=*******Removed On Purpose*******]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:76)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317)
    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.channel.MessagePublishingErrorHandler.handleError(MessagePublishingErrorHandler.java:96)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:60)
    at org.springframework.cloud.sleuth.instrument.async.TraceRunnable.run(TraceRunnable.java:64)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
    at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
    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)
    ... 11 common frames omitted

user_2756547 · Answer

I believe the closest equivalent for StepListener is a ChannelInterceptor.

See IntegrationFlowDefinition.intercept() operator to place in between your endpoints in that flow.

However according your splitter configuration, it doesn't seem like you are going to lose anything and errors are just going to be logged. That c.executor() does the trick to shift the work for each record into a thread pool and its error cannot effect the rest of records.

Anyway it sounds like better for you to have a custom error channel be set into headers just before that split() - .enrichHeaders(Collections.singletonMap(MessageHeaders.ERROR_CHANNEL, myErrorChannel())).

Than you have an aggregator subscribed to this channel where you aggregate errors for the batch and emit a message to send an email. This is going to happen really only once per batch or none at all if no errors.

Spring Integration Error Handling with Email Once at end

1 Answers