CAMEL-24190: Allow stream-download on-completion handover to async ro…

components/camel-file/pom.xml

@@ -49,11 +49,23 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-core-languages</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.camel</groupId>
             <artifactId>camel-core-engine</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <version>${mockito-version}</version>
+            <scope>test</scope>
+        </dependency>
 
     </dependencies>
 </project>

components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileEndpoint.java

@@ -1625,6 +1625,16 @@ public void setSynchronous(boolean synchronous) {
         this.synchronous = synchronous;
     }
 
+    /**
+     * Whether the consumer streams file content as an {@link java.io.InputStream} body instead of loading it into
+     * memory first. Remote file components (FTP, SFTP, SMB, etc.) override this when {@code streamDownload=true}.
+     *
+     * @since 4.22
+     */
+    public boolean isStreamDownload() {
+        return false;
+    }
+
     public String getChecksumFileAlgorithm() {
         return checksumFileAlgorithm;
     }

components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java

@@ -19,7 +19,7 @@
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePropertyKey;
 import org.apache.camel.spi.ExceptionHandler;
-import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.SynchronizationVetoable;
 import org.apache.camel.support.LoggingExceptionHandler;
 import org.apache.camel.util.StringHelper;
 import org.slf4j.Logger;
@@ -31,7 +31,7 @@
  * The work is for example to move the processed file into a backup folder, delete the file or in case of processing
  * failure do a rollback.
  */
-public class GenericFileOnCompletion<T> implements Synchronization {
+public class GenericFileOnCompletion<T> implements SynchronizationVetoable {
 
     private static final Logger LOG = LoggerFactory.getLogger(GenericFileOnCompletion.class);
     private final GenericFileEndpoint<T> endpoint;
@@ -65,6 +65,16 @@ public void onFailure(Exchange exchange) {
         onCompletion(exchange);
     }
 
+    @Override
+    public boolean allowHandover() {
+        return endpoint.isStreamDownload();
+    }
+
+    @Override
+    public void beforeHandover(Exchange target) {
+        // no state transformation needed; the stream is already on the exchange body
+    }
+
     public ExceptionHandler getExceptionHandler() {
         return exceptionHandler;
     }

components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileOnCompletionHandoverTest.java

@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.file;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePropertyKey;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.engine.DefaultUnitOfWork;
+import org.apache.camel.spi.SynchronizationVetoable;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.ExchangeHelper;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class GenericFileOnCompletionHandoverTest {
+
+    @Mock
+    private GenericFileEndpoint<Object> endpoint;
+
+    @Mock
+    private GenericFileOperations<Object> operations;
+
+    @Mock
+    private GenericFileProcessStrategy<Object> processStrategy;
+
+    private CamelContext context;
+    private GenericFile<Object> testFile;
+
+    @BeforeEach
+    void setUp() throws Exception {
+        context = new DefaultCamelContext();
+        context.start();
+        testFile = genericFile();
+        when(endpoint.getCamelContext()).thenReturn(context);
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        context.stop();
+    }
+
+    @Test
+    void implementsSynchronizationVetoable() {
+        GenericFileOnCompletion<Object> completion
+                = new GenericFileOnCompletion<>(endpoint, operations, processStrategy, testFile, "/data/file.txt");
+        assertInstanceOf(SynchronizationVetoable.class, completion);
+    }
+
+    @Test
+    void allowHandoverWhenStreamDownloadEnabled() {
+        assertTrue(createCompletion(true).allowHandover());
+    }
+
+    @Test
+    void denyHandoverWhenStreamDownloadDisabled() {
+        assertFalse(createCompletion(false).allowHandover());
+    }
+
+    @Test
+    void unitOfWorkHandoversCompletionWhenStreamDownload() throws Exception {
+        when(endpoint.isIdempotent()).thenReturn(false);
+        when(endpoint.getInProgressRepository()).thenReturn(new MemoryIdempotentRepository());
+
+        Exchange source = new DefaultExchange(context);
+        DefaultUnitOfWork sourceUow = createUnitOfWork(source);
+        source.getExchangeExtension().setUnitOfWork(sourceUow);
+
+        GenericFileOnCompletion<Object> completion = createCompletion(true);
+        source.getExchangeExtension().addOnCompletion(completion);
+
+        Exchange target = new DefaultExchange(context);
+        DefaultUnitOfWork targetUow = createUnitOfWork(target);
+        target.getExchangeExtension().setUnitOfWork(targetUow);
+
+        sourceUow.handoverSynchronization(target);
+
+        sourceUow.done(source);
+        verify(processStrategy, never()).commit(operations, endpoint, source, testFile);
+
+        targetUow.done(target);
+        verify(processStrategy).commit(operations, endpoint, target, testFile);
+    }
+
+    @Test
+    void unitOfWorkDoesNotHandoverCompletionWhenNotStreamDownload() throws Exception {
+        when(endpoint.isIdempotent()).thenReturn(false);
+        when(endpoint.getInProgressRepository()).thenReturn(new MemoryIdempotentRepository());
+
+        Exchange source = new DefaultExchange(context);
+        DefaultUnitOfWork sourceUow = createUnitOfWork(source);
+        source.getExchangeExtension().setUnitOfWork(sourceUow);
+
+        GenericFileOnCompletion<Object> completion = createCompletion(false);
+        source.getExchangeExtension().addOnCompletion(completion);
+
+        Exchange target = new DefaultExchange(context);
+        DefaultUnitOfWork targetUow = createUnitOfWork(target);
+        target.getExchangeExtension().setUnitOfWork(targetUow);
+
+        sourceUow.handoverSynchronization(target);
+
+        sourceUow.done(source);
+        verify(processStrategy).commit(operations, endpoint, source, testFile);
+        verify(processStrategy, never()).commit(operations, endpoint, target, testFile);
+    }
+
+    @Test
+    void correlatedCopyHandoversStreamDownloadCompletion() throws Exception {
+        when(endpoint.isIdempotent()).thenReturn(false);
+        when(endpoint.getInProgressRepository()).thenReturn(new MemoryIdempotentRepository());
+
+        Exchange source = new DefaultExchange(context);
+        DefaultUnitOfWork sourceUow = createUnitOfWork(source);
+        source.getExchangeExtension().setUnitOfWork(sourceUow);
+        source.getExchangeExtension().addOnCompletion(createCompletion(true));
+
+        Exchange copy = ExchangeHelper.createCorrelatedCopy(source, true, false);
+        DefaultUnitOfWork copyUow = createUnitOfWork(copy);
+        copy.getExchangeExtension().setUnitOfWork(copyUow);
+
+        sourceUow.done(source);
+        verify(processStrategy, never()).commit(operations, endpoint, source, testFile);
+
+        copyUow.done(copy);
+        verify(processStrategy).commit(operations, endpoint, copy, testFile);
+        assertSame(source.getExchangeId(), copy.getProperty(ExchangePropertyKey.CORRELATION_ID));
+    }
+
+    private DefaultUnitOfWork createUnitOfWork(Exchange exchange) {
+        return new DefaultUnitOfWork(exchange);
+    }
+
+    private GenericFileOnCompletion<Object> createCompletion(boolean streamDownload) {
+        when(endpoint.isStreamDownload()).thenReturn(streamDownload);
+        return new GenericFileOnCompletion<>(endpoint, operations, processStrategy, testFile, "/data/file.txt");
+    }
+
+    private GenericFile<Object> genericFile() {
+        GenericFile<Object> file = new GenericFile<>();
+        file.setFileName("file.txt");
+        file.setAbsolute(true);
+        file.setAbsoluteFilePath("/data/file.txt");
+        return file;
+    }
+}

components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileEndpoint.java

@@ -120,6 +120,11 @@ public String getServiceProtocol() {
         return getConfiguration().getProtocol();
     }
 
+    @Override
+    public boolean isStreamDownload() {
+        return getConfiguration().isStreamDownload();
+    }
+
     @Override
     public Map<String, String> getServiceMetadata() {
         if (getConfiguration().getUsername() != null) {

components/camel-smb/src/main/java/org/apache/camel/component/smb/SmbEndpoint.java

@@ -128,6 +128,11 @@ public SmbConfiguration getConfiguration() {
         return configuration;
     }
 
+    @Override
+    public boolean isStreamDownload() {
+        return configuration.isStreamDownload();
+    }
+
     /**
      * Default Existing File Move Strategy
      *

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc

@@ -298,6 +298,16 @@ and `Files.copy()` instead of `FileChannel`-based streaming. The `forceWrites` f
 The default value has been changed from `true` to `false` to reflect the actual behavior. The option will
 be removed in a future release.
 
+=== camel-file / camel-ftp / camel-sftp / camel-smb - streamDownload with async endpoints
+
+When `streamDownload=true`, file-based consumers now allow their on-completion logic to hand over to downstream
+threads (for example when routing to `seda:` or `jms:`). This keeps the downloaded stream open until the downstream
+route has finished reading it, instead of closing it immediately on the consumer poll thread.
+
+Because handover defers commit/rollback (move, delete, idempotent confirmation, and stream release), file handles and
+remote connections may stay open longer while the downstream queue is busy. This only applies when `streamDownload=true`;
+routes without stream download keep the previous on-completion timing.
+
 === camel-mail - MimeMultipartDataFormat inbound header filtering
 
 When unmarshalling a MIME message with `headersInline=true`, the `mime-multipart` data format now applies a