Spring Integration - custom errorChannel - регистрируется только первое исключение

Question

Это продолжение предыдущего вопроса (требования приведены в исходном вопросе).

Spring Integration - Filter - Отправка сообщений в другую конечную точку

Моя проблема заключается в том, что если во входном файле имеется более одной ошибки, регистрируется только первая ошибка.Последующие ошибки не регистрируются.

Модифицированный код:

@Configuration
public class CreateUserConfiguration {
    @Bean
    public IntegrationFlow createUser() {
        return IntegrationFlows.from(Files.inboundAdapter(new File(INPUT_DIR)))
                .enrichHeaders(h -> h.header("errorChannel", "exceptionChannel", true))
                .transform(csvToUserBeanTransformer, "convertCsvToUserBean")
                .split(userBeanSplitter, "splitUserBeans")
                .wireTap(flow -> flow.<UserBean>filter(userBean -> !userBean.getStatus().equalsIgnoreCase("SUCCESS")).channel("errorSummaryReportGenerationChannel"))
                .transform(userBeanToJSONTransformer, "convertUserBeanToJSON")
                .handle(Files.outboundAdapter(new File(OUTPUT_SUCCESS_DIRECTORY)))
                .get();
    }

    @Bean
    public IntegrationFlow logErrorSummary() {
        return IntegrationFlows.from("errorSummaryReportGenerationChannel")
                .handle((p,h) -> {
                    return ((UserBean)(p)).getUserID() + "\t" + ((UserBean)(p)).getStatus();
                })
                .transform(Transformers.objectToString())
                .handle(Files.outboundAdapter(new File(OUTPUT_FAILED_REPORT_FILE_NAME)))
                .get();
    }

    @Bean
    public IntegrationFlow logError() {
        return IntegrationFlows.from("exceptionChannel")
                .enrichHeaders(h -> h.headerExpression("errorFileName", "payload.failedMessage.headers.fileName"))
                .wireTap(flow -> flow.handle(msg -> System.out.println("Received on exceptionChannel " + msg.getHeaders().get("errorFileName"))))
                .transform(Transformers.objectToString())
                .handle(Files.outboundAdapter(new File(generateOutputDirectory(OUTPUT_FAILED_DIRECTORY))).autoCreateDirectory(true).fileExistsMode(FileExistsMode.APPEND).fileNameExpression("getHeaders().get(\"errorFileName\")+'.json'"))
                .get();
    }

    @Bean(name = "exceptionChannel")
    MessageChannel exceptionChannel() {
        return MessageChannels.executor(new SimpleAsyncTaskExecutor()).get();
    }

    @Bean(name="errorSummaryReportGenerationChannel")
    MessageChannel errorSummaryReportGenerationChannel() {
        return DirectChannel();
    }
}

Чего я ожидаю:

In errorSummaryReport-

B123  ERROR, FREQUENCY
C123  FREQUENCY_DETAIL

В OUTPUT_FAILED_DIRECTORY -

B123.json -> stacktrace of error
C123.json -> stacktrace of error

ЧТО Я ВИДЮ: (отсутствует информация C123)

В ошибкеSummaryReport -

B123  ERROR, FREQUENCY

In OUTPUT_FAILED_DIRECTORY -

B123.json -> stacktrace of error

Answer 1

Проблема возникает из .split(userBeanSplitter, "splitUserBeans").

Я бы сказал, что она полностью аналогична той, что мы делаем в простой Java с циклом for.Итак, если метод в цикле выдает исключение, вы действительно выходите из цикла, и следующий элемент больше не будет обрабатываться.

Чтобы исправить вашу проблему, вам нужно добавить .channel(c -> c.executor(myExecutor())) вобрабатывать разделенные элементы параллельно и обрабатывать ошибки в отдельном потоке.Таким образом, цикл в разделителе не будет затронут.