You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2020/09/21 04:39:10 UTC
[camel] branch camel-2.25.x updated: CAMEL-15557: Stream Cache file
not deleted (#4262)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-2.25.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-2.25.x by this push:
new 76e85ef CAMEL-15557: Stream Cache file not deleted (#4262)
76e85ef is described below
commit 76e85efb1a951d1a1e6852fb77efe93cabeb75be
Author: forsthofer <fo...@users.noreply.github.com>
AuthorDate: Mon Sep 21 06:38:49 2020 +0200
CAMEL-15557: Stream Cache file not deleted (#4262)
If you have a route with a Multicast with parallel processing and a
timeout and a sub-route in the multicast which is creating an
OutputStreamCache before the timeout and is writing to the
OutputStreamCache after the timeout then the created file is never
deleted from the file system.
Co-authored-by: Franz Forsthofer <fr...@sap.com>
---
.../converter/stream/FileInputStreamCache.java | 11 ++++++
.../MulticastParallelTimeoutStreamCachingTest.java | 40 +++++++++++++++++++++-
2 files changed, 50 insertions(+), 1 deletion(-)
diff --git a/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java b/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
index 74eacfe..f50fe0b 100644
--- a/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
+++ b/camel-core/src/main/java/org/apache/camel/converter/stream/FileInputStreamCache.java
@@ -239,6 +239,17 @@ public final class FileInputStreamCache extends InputStream implements StreamCac
if (tempFile != null) {
throw new IllegalStateException("The method 'createOutputStream' can only be called once!");
}
+ if (closedOnCompletion && exchangeCounter.get() == 0) {
+ // exchange was already stopped -> in this case the tempFile would never be deleted.
+ // This can happen when in the splitter or Multi-cast case with parallel processing, the CachedOutputStream is created when the main unit of work
+ // is still active, but has a timeout and after the timeout which stops the unit of work the FileOutputStream is created.
+ // We only can throw here an Exception and inform the user that the processing took longer than the set timeout.
+ String error = "Cannot create a FileOutputStream for Stream Caching, because this FileOutputStream would never be removed from the file system."
+ + " This situation can happen with a Splitter or Multi Cast in parallel processing if there is a timeout set on the Splitter or Multi Cast, "
+ + " and the processing in a sub-branch takes longer than the timeout. Consider to increase the timeout.";
+ LOG.error(error);
+ throw new IOException(error);
+ }
tempFile = FileUtil.createTempFile("cos", ".tmp", strategy.getSpoolDirectory());
LOG.trace("Creating temporary stream cache file: {}", tempFile);
diff --git a/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutStreamCachingTest.java b/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutStreamCachingTest.java
index 747775a..cf690f0 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutStreamCachingTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/MulticastParallelTimeoutStreamCachingTest.java
@@ -19,6 +19,7 @@ package org.apache.camel.processor;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FilterInputStream;
+import java.io.IOException;
import java.nio.charset.StandardCharsets;
import org.apache.camel.ContextTestSupport;
@@ -28,6 +29,7 @@ import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.converter.stream.CachedOutputStream;
import org.junit.Before;
import org.junit.Test;
@@ -39,7 +41,9 @@ public class MulticastParallelTimeoutStreamCachingTest extends ContextTestSuppor
protected Endpoint startEndpoint;
protected MockEndpoint x;
-
+ protected MockEndpoint exception;
+ protected MockEndpoint y;
+
public static void deleteDirectory(File file) {
if (file.isDirectory()) {
File[] files = file.listFiles();
@@ -64,6 +68,15 @@ public class MulticastParallelTimeoutStreamCachingTest extends ContextTestSuppor
File[] files = f.listFiles();
assertEquals(0, files.length);
}
+
+ @Test
+ public void testCreateOutputStreamCacheBeforeTimeoutButWriteToOutputStreamCacheAfterTimeout() throws Exception {
+ exception.expectedMessageCount(1);
+ y.expectedMessageCount(0);
+
+ template.sendBody("direct:b", "testMessage");
+ assertMockEndpointsSatisfied();
+ }
@Before
@Override
@@ -72,6 +85,8 @@ public class MulticastParallelTimeoutStreamCachingTest extends ContextTestSuppor
deleteDirectory(new File(TARGET_MULTICAST_PARALLEL_TIMEOUT_STREAM_CACHING_TEST_CACHE));
x = getMockEndpoint("mock:x");
+ y = getMockEndpoint("mock:y");
+ exception = getMockEndpoint("mock:exception");
}
protected RouteBuilder createRouteBuilder() {
@@ -90,6 +105,23 @@ public class MulticastParallelTimeoutStreamCachingTest extends ContextTestSuppor
});
}
};
+
+ final Processor processor2 = new Processor() {
+ public void process(Exchange exchange) throws IOException {
+ // create first the OutputStreamCache and then sleep
+ CachedOutputStream outputStream = new CachedOutputStream(exchange);
+ try {
+ // sleep for one second so that the write to the CachedOutputStream happens after the main exchange has finished due to timeout on the multicast
+ Thread.sleep(1000l);
+ } catch (InterruptedException e) {
+ throw new IllegalStateException("Unexpected exception", e);
+ }
+ outputStream.write(BODY);
+ Message in = exchange.getIn();
+ // use FilterInputStream to trigger streamcaching
+ in.setBody(outputStream.getInputStream());
+ }
+ };
return new RouteBuilder() {
public void configure() {
@@ -99,10 +131,16 @@ public class MulticastParallelTimeoutStreamCachingTest extends ContextTestSuppor
context.getStreamCachingStrategy().setRemoveSpoolDirectoryWhenStopping(false);
context.getStreamCachingStrategy().setSpoolThreshold(1l);
context.setStreamCaching(true);
+
+ onException(IOException.class).to("mock:exception");
from("direct:a").multicast().timeout(500l).parallelProcessing().to("direct:x");
from("direct:x").process(processor1).to("mock:x");
+
+ from("direct:b").multicast().timeout(500l).parallelProcessing().to("direct:y");
+
+ from("direct:y").process(processor2).to("mock:y");
}
};
}