This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 708a656e368b CAMEL-25915: camel-ftp - Reconnect after a failed poll
with sendNoop=false and a custom poll strategy (#28232)
708a656e368b is described below
commit 708a656e368b131183251aefefb84611f262defe
Author: Johan Almqvist <[email protected]>
AuthorDate: Wed Oct 21 21:07:00 2026 +0200
CAMEL-25915: camel-ftp - Reconnect after a failed poll with sendNoop=false
and a custom poll strategy (#28232)
With sendNoop=false, an ftp or ftps consumer whose poll failed on a
connection that the server had closed relied on the default poll
strategy to disconnect. With a custom pollStrategy nothing disconnected,
and every following poll failed. A failure caused by an I/O error that
is thrown out of the poll now also makes the consumer send a noop
before the next poll, as a failure while processing a file already did.
The upgrade guide and the setSendNoop Javadoc now say that with
sendNoop=false a mina-sftp connection that dropped silently is found by
the next poll, and point to serverAliveInterval.
Claude-Session: https://claude.ai/code/session_01SonJJYXyt16gNRQkSu72AH
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../file/remote/RemoteFileConfiguration.java | 4 +-
.../component/file/remote/RemoteFileConsumer.java | 12 +++++
.../remote/RemoteFileConsumerSendNoopTest.java | 52 ++++++++++++++++++++--
.../integration/FtpConsumerSendNoopFalseIT.java | 46 ++++++++++++++++++-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 5 ++-
5 files changed, 112 insertions(+), 7 deletions(-)
diff --git
a/components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileConfiguration.java
b/components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileConfiguration.java
index fca7581f7fa0..e53f504e120d 100644
---
a/components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileConfiguration.java
+++
b/components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileConfiguration.java
@@ -418,7 +418,9 @@ public abstract class RemoteFileConfiguration extends
GenericFileConfiguration {
* before the next poll, and connects again if the noop fails. If the
server closes idle connections and the delay
* between polls is longer than its idle timeout, every other poll then
fails, so keep this option on for such a
* consumer. For sftp and mina-sftp the client notices a closed
connection, but not one that dropped silently. For
- * sftp the noop is a keep-alive message that does not wait for a reply.
+ * sftp the noop is a keep-alive message that does not wait for a reply,
so it does not find such a connection
+ * either. For mina-sftp such a connection is then found by the next poll,
which fails before the consumer connects
+ * again. Set serverAliveInterval to let the client close such a
connection itself.
*/
public void setSendNoop(boolean sendNoop) {
this.sendNoop = sendNoop;
diff --git
a/components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
b/components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
index fede2bf00662..7f7d6d565a46 100644
---
a/components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
+++
b/components/camel-ftp-common/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
@@ -283,6 +283,18 @@ public abstract class RemoteFileConsumer<T> extends
GenericFileConsumer<T> {
}
}
+ @Override
+ protected int poll(Exchange dynamic) throws Exception {
+ try {
+ return super.poll(dynamic);
+ } catch (Exception e) {
+ // a failure thrown out of the poll, such as on a connection that
the server has closed, goes to the poll
+ // strategy, and only the default poll strategy disconnects
+ sendNoopOnNextPollIfIOFailure(e);
+ throw e;
+ }
+ }
+
@Override
protected void handleException(String message, Exchange exchange,
Throwable t) {
// a failure while retrieving or processing a file is not thrown out
of the poll, so the poll strategy does not
diff --git
a/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/RemoteFileConsumerSendNoopTest.java
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/RemoteFileConsumerSendNoopTest.java
index 4b1151e1f6d3..c17192411fac 100644
---
a/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/RemoteFileConsumerSendNoopTest.java
+++
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/RemoteFileConsumerSendNoopTest.java
@@ -20,12 +20,14 @@ import java.io.IOException;
import java.net.SocketException;
import org.apache.camel.component.file.GenericFileOperationFailedException;
+import org.apache.camel.component.file.GenericFileProcessStrategy;
import org.apache.camel.test.junit6.CamelTestSupport;
import org.apache.commons.net.ftp.FTPFile;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -38,8 +40,9 @@ import static org.mockito.Mockito.when;
/**
* Unit tests for the connection check that {@link
RemoteFileConsumer#connectIfNecessary()} does before each poll: by
* default it sends a noop, and with sendNoop=false it only checks the
connection state of the client. With
- * sendNoop=false the consumer sends a noop before the next poll after an I/O
failure, also when the ftp consumer
- * ignores the failure as a file that does not exist, unless it disconnects
before that poll.
+ * sendNoop=false the consumer sends a noop before the next poll after an I/O
failure, whether the failure is thrown out
+ * of the poll or happens while processing a file, also when the ftp consumer
ignores the failure as a file that does
+ * not exist, unless it disconnects before that poll.
*/
class RemoteFileConsumerSendNoopTest extends CamelTestSupport {
@@ -52,13 +55,13 @@ class RemoteFileConsumerSendNoopTest extends
CamelTestSupport {
RemoteFileOperations<SftpRemoteFile> sftpOperations =
mock(RemoteFileOperations.class);
operations = sftpOperations;
return new SftpConsumer(sftpEndpoint, exchange -> {
- }, sftpOperations, null);
+ }, sftpOperations, mock(GenericFileProcessStrategy.class));
}
// the ftp consumer (also used by ftps) needs the ftp specific
operations
FtpOperations ftpOperations = mock(FtpOperations.class);
operations = ftpOperations;
return new FtpConsumer((RemoteFileEndpoint<FTPFile>) endpoint,
exchange -> {
- }, ftpOperations, null);
+ }, ftpOperations, mock(GenericFileProcessStrategy.class));
}
/**
@@ -138,6 +141,47 @@ class RemoteFileConsumerSendNoopTest extends
CamelTestSupport {
verify(operations, times(1)).connect(any(), any());
}
+ @ParameterizedTest
+ @ValueSource(strings = { "ftp", "ftps", "sftp" })
+ void testSendNoopFalseSendsNoopAfterPollFailure(String scheme) {
+ RemoteFileConsumer<?> consumer = createLoggedInConsumer(scheme +
"://localhost/inbox?sendNoop=false");
+ // the client still regards itself as connected, as an ftp client does
after the server closed the connection
+ when(operations.isConnected()).thenReturn(true);
+ when(operations.sendNoop()).thenReturn(false);
+ // so the first command of the poll fails
+ when(operations.getCurrentDirectory()).thenThrow(
+ new GenericFileOperationFailedException(
+ "Cannot get current directory", new
SocketException("Connection reset")));
+
+ // the failure is thrown out of the poll, and a custom poll strategy
does not disconnect
+ assertThrows(GenericFileOperationFailedException.class, () ->
consumer.poll(null));
+ verify(operations, never()).sendNoop();
+ verify(operations, never()).forceDisconnect();
+
+ // the next poll sends a noop, which fails, so it connects again
+ consumer.connectIfNecessary();
+
+ verify(operations, times(1)).sendNoop();
+ verify(operations, times(1)).connect(any(), any());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "ftp", "ftps", "sftp" })
+ void testSendNoopFalseSendsNoNoopAfterOtherPollFailure(String scheme) {
+ RemoteFileConsumer<?> consumer = createLoggedInConsumer(scheme +
"://localhost/inbox?sendNoop=false");
+ when(operations.isConnected()).thenReturn(true);
+ when(operations.getCurrentDirectory())
+ .thenThrow(new GenericFileOperationFailedException("Cannot get
current directory"));
+
+ assertThrows(GenericFileOperationFailedException.class, () ->
consumer.poll(null));
+
+ consumer.connectIfNecessary();
+
+ verify(operations, times(2)).isConnected();
+ verify(operations, never()).sendNoop();
+ verify(operations, never()).connect(any(), any());
+ }
+
@ParameterizedTest
@ValueSource(strings = { "ftp", "ftps", "sftp" })
void testSendNoopFalseKeepsConnectionWhenNoopAfterIOFailureSucceeds(String
scheme) {
diff --git
a/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/integration/FtpConsumerSendNoopFalseIT.java
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/integration/FtpConsumerSendNoopFalseIT.java
index 3a30777c49d2..7403476167bb 100644
---
a/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/integration/FtpConsumerSendNoopFalseIT.java
+++
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/integration/FtpConsumerSendNoopFalseIT.java
@@ -21,10 +21,12 @@ import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.camel.BindToRegistry;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.file.remote.BaseServerTestSupport;
import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.DefaultPollingConsumerPollStrategy;
import org.apache.camel.test.infra.ftp.services.embedded.EmbeddedConfiguration;
import org.apache.camel.test.infra.ftp.services.embedded.FtpEmbeddedService;
import org.apache.ftpserver.FtpServerFactory;
@@ -43,13 +45,18 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* The consumer counterpart of {@link FtpProducerSendNoopFalseIT}: with
sendNoop=false the consumer does not send a NOOP
* command before each poll, and stays logged in between polls, and it
connects again after the server has closed the
- * connection.
+ * connection, also with a poll strategy that does not disconnect.
*/
public class FtpConsumerSendNoopFalseIT extends BaseServerTestSupport {
private final AtomicInteger noops = new AtomicInteger();
private final AtomicInteger logins = new AtomicInteger();
private final AtomicInteger retrieves = new AtomicInteger();
+ private final AtomicInteger pwds = new AtomicInteger();
+
+ // a poll strategy that does not disconnect when a poll fails, unlike the
default one of the ftp consumer
+ @BindToRegistry("myPollStrategy")
+ private final DefaultPollingConsumerPollStrategy pollStrategy = new
DefaultPollingConsumerPollStrategy();
@RegisterExtension
public FtpEmbeddedService service = new FtpEmbeddedService() {
@@ -67,6 +74,8 @@ public class FtpConsumerSendNoopFalseIT extends
BaseServerTestSupport {
logins.incrementAndGet();
} else if ("RETR".equalsIgnoreCase(request.getCommand())) {
retrieves.incrementAndGet();
+ } else if ("PWD".equalsIgnoreCase(request.getCommand())) {
+ pwds.incrementAndGet();
}
return super.beforeCommand(session, request);
}
@@ -143,6 +152,36 @@ public class FtpConsumerSendNoopFalseIT extends
BaseServerTestSupport {
assertEquals(2, logins.get(), "The consumer should connect again after
the server closed the connection");
}
+ @Test
+ public void
testSendNoopFalseConnectsAgainAfterServerClosedConnectionWithCustomPollStrategy()
throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("one");
+
+ template.sendBodyAndHeader("file://" + service.getFtpRootDir() +
"/pollstrategy", "one", Exchange.FILE_NAME,
+ "hello.txt");
+ context.getRouteController().startRoute("recoverPollStrategy");
+
+ MockEndpoint.assertIsSatisfied(context);
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .until(() ->
!service.ftpFile("pollstrategy/hello.txt").toFile().exists());
+ // and the consumer has polled again, as each poll starts with a PWD
command (stepwise=true)
+ int pwdsAfterDelete = pwds.get();
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> pwds.get()
> pwdsAfterDelete + 1);
+
+ // the server closes the connection between polls. The client does not
notice, so the first command of the next
+ // poll fails, and the failure is thrown out of the poll to a poll
strategy that does not disconnect
+ service.disconnectAllSessions();
+
+ mock.reset();
+ mock.expectedBodiesReceived("two");
+
+ template.sendBodyAndHeader("file://" + service.getFtpRootDir() +
"/pollstrategy", "two", Exchange.FILE_NAME,
+ "hello.txt");
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertEquals(2, logins.get(), "The consumer should connect again after
the server closed the connection");
+ }
+
private void consumeFileAndPollAgain(String routeId) throws Exception {
template.sendBodyAndHeader("file://" + service.getFtpRootDir() +
"/sendnoop", "Hello World", Exchange.FILE_NAME,
"hello.txt");
@@ -174,6 +213,11 @@ public class FtpConsumerSendNoopFalseIT extends
BaseServerTestSupport {
from(getRecoverUrl() +
"&ignoreFileNotFoundOrPermissionError=true")
.routeId("recoverIgnoreFileNotFound").autoStartup(false)
.convertBodyTo(String.class).to("mock:result");
+
+
from("ftp://admin@localhost:{{ftp.server.port}}/pollstrategy?password=admin&delete=true&delay=100";
+ + "&sendNoop=false&pollStrategy=#myPollStrategy")
+ .routeId("recoverPollStrategy").autoStartup(false)
+ .convertBodyTo(String.class).to("mock:result");
}
};
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 94e1f7e5073f..b992efb54f7e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -2274,4 +2274,7 @@ consumer still sends a noop before it renames or deletes
a file on the server, a
unless `noop=true`.
For sftp and mina-sftp the client notices a closed connection, but not one
that dropped silently. For sftp the noop is
-a keep-alive message that does not wait for a reply, so it does not find such
a connection either.
+a keep-alive message that does not wait for a reply, so it does not find such
a connection either. For mina-sftp the
+noop is a request that waits for the reply, so with `sendNoop=false` such a
connection is found by the next poll,
+which fails and is logged before the consumer connects again. For sftp and
mina-sftp, set `serverAliveInterval` to let
+the client close such a connection itself, so that the consumer connects again
before the poll.
