如何解决Camel-HDFS在Windows递归上传时的路径非法字符问题

我用Apache Camel编写路由，将Windows本地文件递归上传到HDFS，代码如下：

public class FileRouteBuilder extends EndpointRouteBuilder {

    @Override
    public void configure() throws Exception {
        from(file("C://Users/pcn/Desktop/test").noop(true).recursive(true))
                .process(new FileProcessor())
                .to(hdfs("localhost:9000/2209212/"))
                .log(LoggingLevel.DEBUG, "completed");
    }
}

开启recursive选项后，出现路径非法字符错误，错误栈如下：

java.lang.IllegalArgumentException: Illegal character in path at index 33: hdfs://localhost:9000/220922/rec\test4.txt
    at java.base/java.net.URI.create(URI.java:883)
    at org.apache.camel.component.hdfs.HdfsInfoFactory.newFileSystem(HdfsInfoFactory.java:102)
    at org.apache.camel.component.hdfs.HdfsInfoFactory.newHdfsInfoWithoutAuth(HdfsInfoFactory.java:63)
    at org.apache.camel.component.hdfs.HdfsInfoFactory.newHdfsInfoWithoutAuth(HdfsInfoFactory.java:41)
    at org.apache.camel.component.hdfs.HdfsOutputStream.createOutputStream(HdfsOutputStream.java:50)
    at org.apache.camel.component.hdfs.HdfsProducer.doProcess(HdfsProducer.java:205)
    at org.apache.camel.component.hdfs.HdfsProducer.process(HdfsProducer.java:188)
    at org.apache.camel.support.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:66)
    at org.apache.camel.processor.SendDynamicProcessor.lambda$process$0(SendDynamicProcessor.java:197)
    at org.apache.camel.support.cache.DefaultProducerCache.doInAsyncProducer(DefaultProducerCache.java:318)
    at org.apache.camel.processor.SendDynamicProcessor.process(SendDynamicProcessor.java:182)
    at org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$SimpleTask.run(RedeliveryErrorHandler.java:469)
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:187)
    at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:64)
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:184)
    at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:398)
    at org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:492)
    at org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:245)
    at org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:206)
    at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:197)
    at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:111)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)

我尝试用自定义Processor替换路径中的反斜杠，但无效：

@Service
@Slf4j
public class FileProcessor implements Processor {

    @Override
    public void process(Exchange exchange) throws Exception {
        GenericFile body = exchange.getIn().getBody(GenericFile.class);
        body.setEndpointPath(exchange.getIn().getBody(GenericFile.class).getEndpointPath().replace("\/", "/"));
        body.setRelativeFilePath(exchange.getIn().getBody(GenericFile.class).getRelativeFilePath().replace("\/", "/"));
        body.setAbsoluteFilePath(exchange.getIn().getBody(GenericFile.class).getAbsoluteFilePath().replace("\/", "/"));

        exchange.getIn().setBody(body);

    }
}

也试过设置属性并用toD，结果一样。请问该怎么修复？

方法1：配置File组件的路径分隔符 #

直接在File组件中指定pathSeparator为正斜杠，从根源上让递归扫描生成的相对路径统一使用正斜杠，无需额外处理：

from(file("C://Users/pcn/Desktop/test")
        .noop(true)
        .recursive(true)
        .pathSeparator("/")) // 强制使用正斜杠作为路径分隔符
    .to(hdfs("localhost:9000/2209212/"))
    .log(LoggingLevel.DEBUG, "completed");

方法2：修改Processor的路径处理逻辑 #

之前的Processor无效是因为没有修改HDFS组件实际依赖的Exchange属性。HDFS组件构建目标路径时，会读取CamelFileName这类属性，而非GenericFile的路径字段。修改后的Processor如下：

@Service
@Slf4j
public class FileProcessor implements Processor {

    @Override
    public void process(Exchange exchange) throws Exception {
        String relativePath = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
        if (relativePath != null) {
            // 替换Windows反斜杠为正斜杠
            String normalizedPath = relativePath.replace("\\", "/");
            exchange.getIn().setHeader(Exchange.FILE_NAME, normalizedPath);
        }
    }
}

方法3：用toD动态构建目标路径 #

如果需要更灵活的路径控制，可通过toD动态拼接路径并替换反斜杠：

from(file("C://Users/pcn/Desktop/test").noop(true).recursive(true))
    .toD("hdfs://localhost:9000/2209212/${header.CamelFileName.replace('\\\\', '/')}")
    .log(LoggingLevel.DEBUG, "completed");

注意这里需要转义反斜杠，用\\\\表示单个反斜杠。

内容的提问来源于stack exchange，提问作者mokpo