java - 捕获 Spring Integration DSL 上的错误

Question

我们有一个连接到 GCP Pubsub 的 Spring Integration DSL 管道，并且一切正常:使用 Function 实现和 的集合，按照管道中的定义接收和处理数据.handle().

我们遇到的问题(以及为什么我在引号中使用“work”)是，在某些处理程序中，当在配套数据库中找不到某些数据时，我们会引发 IllegalStateException，这迫使数据被重新处理(在此过程中，另一个服务可能会完成配套数据库，然后功能将开始工作)。此异常从未在任何地方显示。

我们 try catch errorHandler 的内容，但我们确实找不到以编程方式执行此操作的正确方法(无 XML)。

我们的函数有这样的东西:

Record record = recordRepository.findById(incomingData).orElseThrow(() -> new IllegalStateException("Missing information: " + incomingData));

此 IllegalStateException 没有出现在日志中的任何位置。

此外，也许值得一提的是，我们的 channel 定义为

    @Bean
    public DirectChannel cardInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public PubSubInboundChannelAdapter cardChannelAdapter(
            @Qualifier("cardInputChannel") MessageChannel inputChannel,
            PubSubTemplate pubSubTemplate) {
        PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubSubTemplate, SUBSCRIPTION_NAME);
        adapter.setOutputChannel(inputChannel);
        adapter.setAckMode(AckMode.AUTO);
        adapter.setPayloadType(CardDto.class);
        return adapter;
    }

Answer 1

我对适配器不熟悉，但我只是查看了代码，看起来他们只是拒绝消息并且不记录任何内容。

您可以将建议添加到处理程序的端点以捕获并记录异常

    .handle(..., e -> e.advice(exceptionLoggingAdvice)


@Bean
public MethodInterceptor exceptionLoggingAdvice() {
    return invocation -> {
        try {
            return invocation.proceed();
        }
        catch (Exception thrown) {
            // log it
            throw thrown;
        }
    }
}

编辑

@SpringBootApplication
public class So57224614Application {

    public static void main(String[] args) {
        SpringApplication.run(So57224614Application.class, args);
    }

    @Bean
    public IntegrationFlow flow(MethodInterceptor myAdvice) {
        return IntegrationFlows.from(() -> "foo", endpoint -> endpoint.poller(Pollers.fixedDelay(5000)))
                .handle("crasher", "crash", endpoint -> endpoint.advice(myAdvice))
                .get();
    }

    @Bean
    public MethodInterceptor myAdvice() {
        return invocation -> {
            try {
                return invocation.proceed();
            }
            catch (Exception e) {
                System.out.println("Failed with " + e.getMessage());
                throw e;
            }
        };
    }


}

@Component
class Crasher {

    public void crash(Message<?> msg) {
        throw new RuntimeException("test");
    }

}

和

Failed with nested exception is java.lang.RuntimeException: test