c.a.storage.common.StorageInputStream : 流已经关闭
c.a.storage.common.StorageInputStream : Stream is already closed
我正在使用 camel-azure-storage-blob 在 Azure Blob 存储中移动文件。我正在尝试执行以下路由配置。
from(azurebloburi)
.process(getHeaders())
.log(LoggingLevel.INFO, log, "After adding Headers =${in.headers})")
.doTry()
.to(myapiendpoint)
.process(moveTheFile(SUCCESS_CONTAINER))
.toD(mysuccesscontaineruri-azure)
.doCatch(Exception.class)
.log(LoggingLevel.ERROR, log,
"Exception - Moving into ERROR.")
.process(moveTheFile(ERROR_CONTAINER))
.convertBodyTo(byte[].class)
.toD(myerrorcontainer-azure)
.end()
我在处理文件 .toD(ERROR_CONTAINER) 之前使用这些 header 设置。
exchange.getIn().setHeader(BlobConstants.BLOB_NAME, "changingBlobName");
exchange.getIn().setHeader(BlobConstants.BLOB_OPERATION,BlobOperationsDefinition.uploadBlockBlob);
exchange.getIn().setHeader(BlobConstants.BLOB_CONTAINER_NAME, "myexistingsourcelistener");
我收到以下错误。任何类型的建议都会有所帮助。
"c.a.storage.common.StorageInputStream : 流已经关闭。"
org.apache.camel.InvalidPayloadException: No body available of type: byte[] but has value: com.azure.storage.blob.specialized.BlobInputStream@5e692e22 of type: com.azure.storage.blob.specialized.BlobInputStream on: Message. Caused by: Error during type conversion from type: com.azure.storage.blob.specialized.BlobInputStream to the required type: byte[] with value com.azure.storage.blob.specialized.BlobInputStream@5e692e22 due to java.lang.RuntimeException: Stream is already closed.. Exchange[E8E38A893F8509F-0000000000000000]. Caused by: [org.apache.camel.TypeConversionException - Error during type conversion from type: com.azure.storage.blob.specialized.BlobInputStream to the required type: byte[] with value com.azure.storage.blob.specialized.BlobInputStream@5e692e22 due to java.lang.RuntimeException: Stream is already closed.]
at org.apache.camel.support.MessageSupport.getMandatoryBody(MessageSupport.java:125) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.support.processor.ConvertBodyProcessor.process(ConvertBodyProcessor.java:106) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.support.processor.ConvertBodyProcessor.process(ConvertBodyProcessor.java:144) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.processor.Pipeline$PipelineTask.run(Pipeline.java:109) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:179) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:64) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:184) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobConsumer.processBatch(BlobConsumer.java:156) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobConsumer.poll(BlobConsumer.java:71) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:192) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:107) ~[camel-support-3.11.3.jar:3.11.3]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) ~[na:na]
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
已编辑
如果我删除 .convertBodyTo(byte[].class) 然后会出现以下错误。
java.lang.RuntimeException: Stream is already closed.
at com.azure.storage.common.StorageInputStream.checkStreamState(StorageInputStream.java:135) ~[azure-storage-common-12.11.1.jar:na]
at com.azure.storage.common.StorageInputStream.readInternal(StorageInputStream.java:311) ~[azure-storage-common-12.11.1.jar:na]
at com.azure.storage.common.StorageInputStream.read(StorageInputStream.java:289) ~[azure-storage-common-12.11.1.jar:na]
at com.azure.storage.common.StorageInputStream.read(StorageInputStream.java:234) ~[azure-storage-common-12.11.1.jar:na]
at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:2314) ~[commons-io-2.6.jar:2.6]
at org.apache.commons.io.IOUtils.copy(IOUtils.java:2270) ~[commons-io-2.6.jar:2.6]
at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:2291) ~[commons-io-2.6.jar:2.6]
at org.apache.commons.io.IOUtils.copy(IOUtils.java:2246) ~[commons-io-2.6.jar:2.6]
at org.apache.commons.io.IOUtils.toByteArray(IOUtils.java:765) ~[commons-io-2.6.jar:2.6]
at org.apache.camel.component.azure.storage.blob.BlobUtils.getInputStreamLength(BlobUtils.java:37) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobStreamAndLength.createBlobStreamAndLengthFromExchangeBody(BlobStreamAndLength.java:50) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.operations.BlobOperations.uploadBlockBlob(BlobOperations.java:181) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobProducer.process(BlobProducer.java:86) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.support.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:66) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.processor.SendDynamicProcessor.lambda$process[=14=](SendDynamicProcessor.java:197) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.support.cache.DefaultProducerCache.doInAsyncProducer(DefaultProducerCache.java:318) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.processor.SendDynamicProcessor.process(SendDynamicProcessor.java:182) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.processor.Pipeline$PipelineTask.run(Pipeline.java:109) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:179) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:64) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:184) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobConsumer.processBatch(BlobConsumer.java:156) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobConsumer.poll(BlobConsumer.java:71) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:192) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:107) ~[camel-support-3.11.3.jar:3.11.3]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) ~[na:na]
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
在处理器 BlobInputStream 是经典 Java InputStream 实现之前使用流缓存。
.log(LoggingLevel.ERROR, log,
"Exception - Moving into ERROR.")
.streamCaching()
.to("direct:test")
from("direct:test")
.process(moveTheFile(ERROR_CONTAINER))
.convertBodyTo(byte[].class)
.toD(myerrorcontainer-azure)
.end()
我正在使用 camel-azure-storage-blob 在 Azure Blob 存储中移动文件。我正在尝试执行以下路由配置。
from(azurebloburi)
.process(getHeaders())
.log(LoggingLevel.INFO, log, "After adding Headers =${in.headers})")
.doTry()
.to(myapiendpoint)
.process(moveTheFile(SUCCESS_CONTAINER))
.toD(mysuccesscontaineruri-azure)
.doCatch(Exception.class)
.log(LoggingLevel.ERROR, log,
"Exception - Moving into ERROR.")
.process(moveTheFile(ERROR_CONTAINER))
.convertBodyTo(byte[].class)
.toD(myerrorcontainer-azure)
.end()
我在处理文件 .toD(ERROR_CONTAINER) 之前使用这些 header 设置。
exchange.getIn().setHeader(BlobConstants.BLOB_NAME, "changingBlobName");
exchange.getIn().setHeader(BlobConstants.BLOB_OPERATION,BlobOperationsDefinition.uploadBlockBlob);
exchange.getIn().setHeader(BlobConstants.BLOB_CONTAINER_NAME, "myexistingsourcelistener");
我收到以下错误。任何类型的建议都会有所帮助。
"c.a.storage.common.StorageInputStream : 流已经关闭。"
org.apache.camel.InvalidPayloadException: No body available of type: byte[] but has value: com.azure.storage.blob.specialized.BlobInputStream@5e692e22 of type: com.azure.storage.blob.specialized.BlobInputStream on: Message. Caused by: Error during type conversion from type: com.azure.storage.blob.specialized.BlobInputStream to the required type: byte[] with value com.azure.storage.blob.specialized.BlobInputStream@5e692e22 due to java.lang.RuntimeException: Stream is already closed.. Exchange[E8E38A893F8509F-0000000000000000]. Caused by: [org.apache.camel.TypeConversionException - Error during type conversion from type: com.azure.storage.blob.specialized.BlobInputStream to the required type: byte[] with value com.azure.storage.blob.specialized.BlobInputStream@5e692e22 due to java.lang.RuntimeException: Stream is already closed.]
at org.apache.camel.support.MessageSupport.getMandatoryBody(MessageSupport.java:125) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.support.processor.ConvertBodyProcessor.process(ConvertBodyProcessor.java:106) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.support.processor.ConvertBodyProcessor.process(ConvertBodyProcessor.java:144) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.processor.Pipeline$PipelineTask.run(Pipeline.java:109) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:179) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:64) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:184) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobConsumer.processBatch(BlobConsumer.java:156) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobConsumer.poll(BlobConsumer.java:71) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:192) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:107) ~[camel-support-3.11.3.jar:3.11.3]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) ~[na:na]
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
已编辑
如果我删除 .convertBodyTo(byte[].class) 然后会出现以下错误。
java.lang.RuntimeException: Stream is already closed.
at com.azure.storage.common.StorageInputStream.checkStreamState(StorageInputStream.java:135) ~[azure-storage-common-12.11.1.jar:na]
at com.azure.storage.common.StorageInputStream.readInternal(StorageInputStream.java:311) ~[azure-storage-common-12.11.1.jar:na]
at com.azure.storage.common.StorageInputStream.read(StorageInputStream.java:289) ~[azure-storage-common-12.11.1.jar:na]
at com.azure.storage.common.StorageInputStream.read(StorageInputStream.java:234) ~[azure-storage-common-12.11.1.jar:na]
at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:2314) ~[commons-io-2.6.jar:2.6]
at org.apache.commons.io.IOUtils.copy(IOUtils.java:2270) ~[commons-io-2.6.jar:2.6]
at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:2291) ~[commons-io-2.6.jar:2.6]
at org.apache.commons.io.IOUtils.copy(IOUtils.java:2246) ~[commons-io-2.6.jar:2.6]
at org.apache.commons.io.IOUtils.toByteArray(IOUtils.java:765) ~[commons-io-2.6.jar:2.6]
at org.apache.camel.component.azure.storage.blob.BlobUtils.getInputStreamLength(BlobUtils.java:37) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobStreamAndLength.createBlobStreamAndLengthFromExchangeBody(BlobStreamAndLength.java:50) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.operations.BlobOperations.uploadBlockBlob(BlobOperations.java:181) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobProducer.process(BlobProducer.java:86) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.support.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:66) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.processor.SendDynamicProcessor.lambda$process[=14=](SendDynamicProcessor.java:197) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.support.cache.DefaultProducerCache.doInAsyncProducer(DefaultProducerCache.java:318) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.processor.SendDynamicProcessor.process(SendDynamicProcessor.java:182) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.processor.Pipeline$PipelineTask.run(Pipeline.java:109) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:179) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:64) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:184) ~[camel-core-processor-3.11.3.jar:3.11.3]
at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobConsumer.processBatch(BlobConsumer.java:156) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.component.azure.storage.blob.BlobConsumer.poll(BlobConsumer.java:71) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:192) ~[camel-support-3.11.3.jar:3.11.3]
at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:107) ~[camel-support-3.11.3.jar:3.11.3]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) ~[na:na]
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
在处理器 BlobInputStream 是经典 Java InputStream 实现之前使用流缓存。
.log(LoggingLevel.ERROR, log,
"Exception - Moving into ERROR.")
.streamCaching()
.to("direct:test")
from("direct:test")
.process(moveTheFile(ERROR_CONTAINER))
.convertBodyTo(byte[].class)
.toD(myerrorcontainer-azure)
.end()