java - 如何修复Camel route 的 “java.nio.charset.MalformedInputException: Input length = 1”？

Question

我有一个 Camel 路由，它会轮询 ftp 服务器上的 zip 文件，从 zip 中提取 csv 文件，然后将 csv 的每一行作为 json 正文发送一个 http 请求。有些 csv 文件大小 <10MB，有些文件大小约为 500MB。问题是我无法完全处理较小的文件，发送了一些行然后我得到 MalformedInputException 并开始回滚。

到目前为止，我发现解决此问题的唯一方法是在拆分之前convertBodyTo(String.class)。这解决了较小文件的问题，但是较大的文件会导致堆空间错误，因为整个文件都会加载到内存中。

    DataFormat dataFormat = BindyCsvDataFormat(myModel.class);

    from("ftp://" + username + "@" + ftpUrl + "?password=" + password + "&noop=True" +
            "&binary=True&sortBy=reverse:file:modified&antInclude=" + filePrefix + "*" +
            "&eagerMaxMessagesPerPoll=false&maxMessagesPerPoll=1")
            .streamCaching()
            .unmarshal().zipFile()
            .convertBodyTo(String.class) // works for small files with this line, but not for large
            .split(body().tokenize("\r\n", 1, true))
            .streaming()
            .unmarshal(dataFormat)
            .marshal().json(JsonLibrary.XStream)
            .to(outputUrl)
            .end();

Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.RuntimeCamelException: Scanner aborted because of an IOException!
    at org.apache.camel.processor.Splitter$SplitterIterable$1.hasNext(Splitter.java:175) ~[camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.MulticastProcessor$MulticastState.run(MulticastProcessor.java:304) ~[camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$3.run(DefaultReactiveExecutor.java:116) ~[camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:185) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:59) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:87) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:228) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:454) [camel-file-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:223) [camel-file-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:186) [camel-file-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:172) [camel-support-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:99) [camel-support-3.0.0-RC1.jar:3.0.0-RC1]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
    at java.lang.Thread.run(Thread.java:834) [?:?]
Caused by: java.nio.charset.MalformedInputException: Input length = 1
    at java.nio.charset.CoderResult.throwException(CoderResult.java:274) ~[?:?]
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:339) ~[?:?]
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178) ~[?:?]
    at java.io.InputStreamReader.read(InputStreamReader.java:185) ~[?:?]
    at java.io.Reader.read(Reader.java:189) ~[?:?]
    at org.apache.camel.util.Scanner.readMore(Scanner.java:162) ~[camel-util-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.util.Scanner.next(Scanner.java:136) ~[camel-util-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.util.Scanner.next(Scanner.java:44) ~[camel-util-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.support.GroupTokenIterator.doNext(GroupTokenIterator.java:122) ~[camel-support-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.support.GroupTokenIterator.next(GroupTokenIterator.java:112) ~[camel-support-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.Splitter$SplitterIterable$1.next(Splitter.java:182) ~[camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.Splitter$SplitterIterable$1.next(Splitter.java:158) ~[camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.MulticastProcessor$MulticastState.run(MulticastProcessor.java:303) ~[camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    ... 16 more

java.nio.charset.MalformedInputException: Input length = 1
    at java.nio.charset.CoderResult.throwException(CoderResult.java:274) ~[?:?]
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:339) ~[?:?]
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178) ~[?:?]
    at java.io.InputStreamReader.read(InputStreamReader.java:185) ~[?:?]
    at java.io.Reader.read(Reader.java:189) ~[?:?]
    at org.apache.camel.util.Scanner.readMore(Scanner.java:162) ~[camel-util-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.util.Scanner.next(Scanner.java:136) ~[camel-util-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.util.Scanner.next(Scanner.java:44) ~[camel-util-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.support.GroupTokenIterator.doNext(GroupTokenIterator.java:122) ~[camel-support-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.support.GroupTokenIterator.next(GroupTokenIterator.java:112) ~[camel-support-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.Splitter$SplitterIterable$1.next(Splitter.java:182) ~[camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.Splitter$SplitterIterable$1.next(Splitter.java:158) ~[camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.MulticastProcessor$MulticastState.run(MulticastProcessor.java:303) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$3.run(DefaultReactiveExecutor.java:116) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:185) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:59) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:87) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:228) [camel-base-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.component.file.GenericFileConsumer.processExchange(GenericFileConsumer.java:454) [camel-file-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.component.file.GenericFileConsumer.processBatch(GenericFileConsumer.java:223) [camel-file-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:186) [camel-file-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:172) [camel-support-3.0.0-RC1.jar:3.0.0-RC1]
    at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:99) [camel-support-3.0.0-RC1.jar:3.0.0-RC1]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
    at java.lang.Thread.run(Thread.java:834) [?:?]```

Answer 1

错误消息指出编码问题。两种不同类型的文件是否具有不同的编码？

如果是，您将必须以某种方式确定文件的编码。或者至少知道您获得的不同类型并将正确的编码分配给文件。

如果您知道编码，则可以按照 @bedla 的建议将其设置为 Exchange 属性。

因此，如果小文件是 ISO 编码的，您可以尝试

.setProperty(Exchange.CHARSET_NAME, constant("ISO-8859-1"))