Apache Camel File组件聚合后异常无法移动文件到.error目录问题

需求 #

当输入目录下同时存在pdf和xml两个配对文件时，将二者移动到输出目录；如果其中一个文件缺失，重试数次后将已存在的文件移动到.error目录。

问题现象 #

File组件的错误处理逻辑受聚合器影响出现不符合预期的表现：

• 在聚合之前抛出异常时，行为符合预期，对应代码中的(1)位置，文件会被移动到.error目录
• 在聚合之后抛出异常时，File组件会将文件移动到.done文件夹而不是预期的.error文件夹，对应代码中的(2)位置

疑问 #

1. 需要补充什么配置或逻辑，才能让(2)位置的异常触发文件移动到.error目录？
2. 是否可以不通过抛出异常实现相同需求？尝试过聚合器的discardOnCompletionTimeout参数，但没有达到预期效果。

测试代码 #

import org.apache.camel.Exchange;
import org.apache.camel.Expression;
import org.apache.camel.RoutesBuilder;
import org.apache.camel.builder.AggregationStrategies;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.junit5.CamelTestSupport;
import org.apache.commons.io.FilenameUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;

public class SimpleMockTest extends CamelTestSupport {
    public static final int COMPLETION_TIMEOUT = 1000;

    @TempDir
    Path tempDir;

    @Test
    public void exceptionAfterAggregation() throws Exception {
        final MockEndpoint mockEndpoint = getMockEndpoint("mock:result");
        mockEndpoint.setResultWaitTime(COMPLETION_TIMEOUT * 3);
        mockEndpoint.expectedFileExists(tempDir.resolve(".done/willMoveToDone.xml"));
        mockEndpoint.expectedFileExists(tempDir.resolve(".done/willMoveToDone.pdf"));
        mockEndpoint.expectedFileExists(tempDir.resolve(".error/shouldMoveToError.xml")); // this fails

        createFile("willMoveToDone.xml", "<xml>1</xml>");
        createFile("willMoveToDone.pdf", "pdfFileContent");
        createFile("shouldMoveToError.xml", "<xml>2</xml>");

        mockEndpoint.assertIsSatisfied();
    }

    @Override
    protected RoutesBuilder createRouteBuilder() {
        return new RouteBuilder() {
            @Override
            public void configure() {
                from("file://" + tempDir + "?include=.*\\.(xml|pdf)&moveFailed=.error&move=.done")
//                        .process(exchange -> {
//                            throw new RuntimeException("Files would be moved to .error as expected");     // (1)
//                        })
                        .aggregate(CORRELATION_EXPRESSION, AggregationStrategies.flexible().accumulateInCollection(ArrayList.class))
                        .completionTimeout(COMPLETION_TIMEOUT)
                        .completionSize(2)

                        .choice()
                            .when(simple("${body.size()} != 2"))
                                .throwException(RuntimeException.class, "Why doesn't the File component move this file to .error?")     // (2)
                        .otherwise()
                            .to("mock:result")
                        .routeId("fileAggregationRoute");
            }

        };
    }

    private final Expression CORRELATION_EXPRESSION = new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            final String fileName = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
            final String correlationExpression = FilenameUtils.getBaseName(fileName);
            return exchange.getContext().getTypeConverter().convertTo(
                    type,
                    correlationExpression
            );
        }
    };

    private void createFile(String filename, String content) throws IOException {
        final File xmlFile = tempDir.resolve(filename).toFile();
        Files.write(xmlFile.toPath(), content.getBytes(StandardCharsets.UTF_8));
    }
}

pom.xml配置 #

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.mycompany.app</groupId>
  <artifactId>camel-playground</artifactId>
  <version>1.0-SNAPSHOT</version>

  <name>my-app</name>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>11</maven.compiler.source>
    <maven.compiler.target>11</maven.compiler.target>
  </properties>

  <dependencyManagement>
    <dependencies>
      <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-bom</artifactId>
        <version>3.11.3</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>
    </dependencies>
  </dependencyManagement>

  <dependencies>
    <dependency>
      <groupId>org.apache.camel</groupId>
      <artifactId>camel-core</artifactId>
    </dependency>
    <dependency>
      <groupId>org.apache.camel</groupId>
      <artifactId>camel-test-junit5</artifactId>
    </dependency>
    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>1.7.32</version>
    </dependency>
    <dependency>
      <groupId>commons-io</groupId>
      <artifactId>commons-io</artifactId>
      <version>2.11.0</version>
    </dependency>
  </dependencies>
</project>

解决方案 #

问题1修复方法 #

出现该现象的原因是Camel聚合器默认使用独立的异步处理单元，原始文件消费的工作单元（UnitOfWork）在进入聚合器后就已标记完成，File组件的moveFailed逻辑仅会在原始消费的工作单元标记为失败时触发，聚合后抛出的异常属于新的工作单元，不会被原始File消费逻辑捕获。
只需在聚合器配置中添加shareUnitOfWork()参数，让聚合后的异常可以传播回原始的消费工作单元，即可触发moveFailed逻辑，修改后的聚合配置如下：

.aggregate(CORRELATION_EXPRESSION, AggregationStrategies.flexible().accumulateInCollection(ArrayList.class))
.shareUnitOfWork() // 新增该行配置
.completionTimeout(COMPLETION_TIMEOUT)
.completionSize(2)

问题2无异常实现方案 #

可以通过聚合器的超时回调逻辑实现需求，无需抛出异常：

1. 保留completionTimeout配置，设置discardOnCompletionTimeout(false)（默认即为false）
2. 新增onCompletion分支判断聚合结果大小，当大小不足2时，通过自定义处理器读取聚合体内的文件元信息，调用文件操作API手动移动到.error目录即可。
   也可以自定义聚合器的超时处理逻辑，直接在超时回调中处理不完整的配对文件，避免抛出异常。

内容的提问来源于stack exchange，提问作者dev-random