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 d60e0ff5e001 CAMEL-24096: readLock=changed - fix readLockTimeout=0
regression
d60e0ff5e001 is described below
commit d60e0ff5e001dcb1d2996eb89ebafcc1c909a183
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Jul 16 10:17:01 2026 +0200
CAMEL-24096: readLock=changed - fix readLockTimeout=0 regression
Fix readLockTimeout=0 regression in FTP, SFTP, Azure Files and SMB
ChangedExclusiveReadLockStrategy. The CAMEL-17121 Tasks API conversion
(Camel 3.14) broke the documented "0 means forever" contract because
TimeBoundedBudget only recognizes -1 as unlimited. Now uses
withUnlimitedDuration() when timeout <= 0. Also converts SFTP strategy
from old while-loop to Tasks API for consistency, extracts
SftpExclusiveReadLockCheck, and fixes copy-paste task name in Azure Files.
Closes #24755
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../FilesChangedExclusiveReadLockStrategy.java | 14 ++-
.../FtpChangedExclusiveReadLockStrategy.java | 12 +-
.../SftpChangedExclusiveReadLockStrategy.java | 125 ++++----------------
.../strategy/SftpExclusiveReadLockCheck.java | 129 +++++++++++++++++++++
.../FtpChangedReadLockTimeoutZeroTest.java | 109 +++++++++++++++++
.../SmbChangedExclusiveReadLockStrategy.java | 12 +-
6 files changed, 286 insertions(+), 115 deletions(-)
diff --git
a/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/strategy/FilesChangedExclusiveReadLockStrategy.java
b/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/strategy/FilesChangedExclusiveReadLockStrategy.java
index df911f374b88..8bbb17eeeca2 100644
---
a/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/strategy/FilesChangedExclusiveReadLockStrategy.java
+++
b/components/camel-azure/camel-azure-files/src/main/java/org/apache/camel/component/file/azure/strategy/FilesChangedExclusiveReadLockStrategy.java
@@ -55,12 +55,16 @@ public class FilesChangedExclusiveReadLockStrategy
implements GenericFileExclusi
throws Exception {
LOG.trace("Waiting for exclusive read lock to file: {}", file);
+ var budgetBuilder = Budgets.iterationTimeBudget()
+ .withInterval(Duration.ofMillis(checkInterval));
+ if (timeout > 0) {
+ budgetBuilder.withMaxDuration(Duration.ofMillis(timeout));
+ } else {
+ budgetBuilder.withUnlimitedDuration();
+ }
BlockingTask task = Tasks.foregroundTask()
- .withBudget(Budgets.iterationTimeBudget()
- .withMaxDuration(Duration.ofMillis(timeout))
- .withInterval(Duration.ofMillis(checkInterval))
- .build())
- .withName("ftp-acquire-exclusive-read-lock")
+ .withBudget(budgetBuilder.build())
+ .withName("azure-files-acquire-exclusive-read-lock")
.build();
FilesExclusiveReadLockCheck exclusiveReadLockCheck
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/FtpChangedExclusiveReadLockStrategy.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/FtpChangedExclusiveReadLockStrategy.java
index b58469f1eb6b..61ec3a9c489f 100644
---
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/FtpChangedExclusiveReadLockStrategy.java
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/FtpChangedExclusiveReadLockStrategy.java
@@ -54,11 +54,15 @@ public class FtpChangedExclusiveReadLockStrategy implements
GenericFileExclusive
throws Exception {
LOG.trace("Waiting for exclusive read lock to file: {}", file);
+ var budgetBuilder = Budgets.iterationTimeBudget()
+ .withInterval(Duration.ofMillis(checkInterval));
+ if (timeout > 0) {
+ budgetBuilder.withMaxDuration(Duration.ofMillis(timeout));
+ } else {
+ budgetBuilder.withUnlimitedDuration();
+ }
BlockingTask task = Tasks.foregroundTask()
- .withBudget(Budgets.iterationTimeBudget()
- .withMaxDuration(Duration.ofMillis(timeout))
- .withInterval(Duration.ofMillis(checkInterval))
- .build())
+ .withBudget(budgetBuilder.build())
.withName("ftp-acquire-exclusive-read-lock")
.build();
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/SftpChangedExclusiveReadLockStrategy.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/SftpChangedExclusiveReadLockStrategy.java
index b21d6aa77d70..385cd5d5cb36 100644
---
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/SftpChangedExclusiveReadLockStrategy.java
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/SftpChangedExclusiveReadLockStrategy.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.file.remote.strategy;
+import java.time.Duration;
+
import com.jcraft.jsch.ChannelSftp;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
@@ -23,9 +25,10 @@ import org.apache.camel.component.file.GenericFile;
import org.apache.camel.component.file.GenericFileEndpoint;
import org.apache.camel.component.file.GenericFileExclusiveReadLockStrategy;
import org.apache.camel.component.file.GenericFileOperations;
-import org.apache.camel.component.file.remote.SftpRemoteFile;
import org.apache.camel.spi.CamelLogger;
-import org.apache.camel.util.StopWatch;
+import org.apache.camel.support.task.BlockingTask;
+import org.apache.camel.support.task.Tasks;
+import org.apache.camel.support.task.budget.Budgets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,113 +53,31 @@ public class SftpChangedExclusiveReadLockStrategy
implements GenericFileExclusiv
public boolean acquireExclusiveReadLock(
GenericFileOperations<ChannelSftp.LsEntry> operations,
GenericFile<ChannelSftp.LsEntry> file, Exchange exchange)
throws Exception {
- boolean exclusive = false;
-
LOG.trace("Waiting for exclusive read lock to file: {}", file);
- long lastModified = Long.MIN_VALUE;
- long length = Long.MIN_VALUE;
- StopWatch watch = new StopWatch();
- long startTime = System.currentTimeMillis();
-
- while (!exclusive) {
- // timeout check
- if (timeout > 0) {
- long delta = watch.taken();
- if (delta > timeout) {
- CamelLogger.log(LOG, readLockLoggingLevel,
- "Cannot acquire read lock within " + timeout + "
millis. Will skip the file: " + file);
- // we could not get the lock within the timeout period, so
- // return false
- return false;
- }
- }
-
- long newLastModified = 0;
- long newLength = 0;
- // operations.listFiles returns SftpRemoteFile[] so
- // do not use generic in the List files
- Object[] files;
- if (fastExistsCheck) {
- // use the absolute file path to only pickup the file we want
to
- // check, this avoids expensive
- // list operations if we have a lot of files in the directory
- String path = file.getAbsoluteFilePath();
- if (path.equals("/") || path.equals("\\")) {
- // special for root (= home) directory
- LOG.trace("Using fast exists to update file information in
home directory");
- files = operations.listFiles();
- } else {
- LOG.trace("Using fast exists to update file information
for {}", path);
- files = operations.listFiles(path);
- }
- } else {
- String path = file.getParent();
- if (path.equals("/") || path.equals("\\")) {
- // special for root (= home) directory
- LOG.trace(
- "Using full directory listing in home directory to
update file information. Consider enabling fastExistsCheck option.");
- files = operations.listFiles();
- } else {
- LOG.trace(
- "Using full directory listing to update file
information for {}. Consider enabling fastExistsCheck option.",
- path);
- files = operations.listFiles(path);
- }
- }
- LOG.trace("List files {} found {} files",
file.getAbsoluteFilePath(), files.length);
- for (Object f : files) {
- SftpRemoteFile rf = (SftpRemoteFile) f;
- boolean match;
- if (fastExistsCheck) {
- // uses the absolute file path as well
- match = rf.getFilename().equals(file.getAbsoluteFilePath())
- || rf.getFilename().equals(file.getFileNameOnly());
- } else {
- match = rf.getFilename().equals(file.getFileNameOnly());
- }
- if (match) {
- newLastModified = rf.getLastModified();
- newLength = rf.getFileLength();
- }
- }
-
- LOG.trace("Previous last modified: {}, new last modified: {}",
lastModified, newLastModified);
- LOG.trace("Previous length: {}, new length: {}", length,
newLength);
- long newOlderThan = startTime + watch.taken() - minAge;
- LOG.trace("New older than threshold: {}", newOlderThan);
-
- if (newLength >= minLength && (minAge == 0 && newLastModified ==
lastModified && newLength == length
- || minAge != 0 && newLastModified < newOlderThan)) {
- LOG.trace("Read lock acquired.");
- exclusive = true;
- } else {
- // set new base file change information
- lastModified = newLastModified;
- length = newLength;
-
- boolean interrupted = sleep();
- if (interrupted) {
- // we were interrupted while sleeping, we are likely being
- // shutdown so return false
- return false;
- }
- }
+ var budgetBuilder = Budgets.iterationTimeBudget()
+ .withInterval(Duration.ofMillis(checkInterval));
+ if (timeout > 0) {
+ budgetBuilder.withMaxDuration(Duration.ofMillis(timeout));
+ } else {
+ budgetBuilder.withUnlimitedDuration();
}
+ BlockingTask task = Tasks.foregroundTask()
+ .withBudget(budgetBuilder.build())
+ .withName("sftp-acquire-exclusive-read-lock")
+ .build();
- return exclusive;
- }
+ SftpExclusiveReadLockCheck exclusiveReadLockCheck
+ = new SftpExclusiveReadLockCheck(fastExistsCheck, minAge,
minLength);
+
+ if (!task.run(exchange.getContext(), () ->
exclusiveReadLockCheck.tryAcquireExclusiveReadLock(operations, file))) {
+ CamelLogger.log(LOG, readLockLoggingLevel,
+ "Cannot acquire read lock within " + timeout + " millis.
Will skip the file: " + file);
- private boolean sleep() {
- LOG.trace("Exclusive read lock not granted. Sleeping for {} millis.",
checkInterval);
- try {
- Thread.sleep(checkInterval);
return false;
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- LOG.debug("Sleep interrupted while waiting for exclusive read
lock, so breaking out");
- return true;
}
+
+ return true;
}
@Override
diff --git
a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/SftpExclusiveReadLockCheck.java
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/SftpExclusiveReadLockCheck.java
new file mode 100644
index 000000000000..aef6ca884e3e
--- /dev/null
+++
b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/SftpExclusiveReadLockCheck.java
@@ -0,0 +1,129 @@
+/*
+ * 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.remote.strategy;
+
+import com.jcraft.jsch.ChannelSftp;
+import org.apache.camel.component.file.GenericFile;
+import org.apache.camel.component.file.GenericFileOperations;
+import org.apache.camel.component.file.remote.SftpRemoteFile;
+import org.apache.camel.util.StopWatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SftpExclusiveReadLockCheck {
+ private static final Logger LOG =
LoggerFactory.getLogger(SftpExclusiveReadLockCheck.class);
+ private final boolean fastExistsCheck;
+ private final long startTime;
+ private final long minAge;
+ private final long minLength;
+ private final StopWatch watch;
+
+ private long lastModified;
+ private long length;
+
+ public SftpExclusiveReadLockCheck(boolean fastExistsCheck, long minAge,
long minLength) {
+ this.fastExistsCheck = fastExistsCheck;
+ this.startTime = System.currentTimeMillis();
+ this.minAge = minAge;
+ this.minLength = minLength;
+ this.watch = new StopWatch();
+
+ this.lastModified = Long.MIN_VALUE;
+ this.length = Long.MIN_VALUE;
+ }
+
+ public boolean tryAcquireExclusiveReadLock(
+ GenericFileOperations<ChannelSftp.LsEntry> operations,
GenericFile<ChannelSftp.LsEntry> file) {
+ long newLastModified = 0;
+ long newLength = 0;
+
+ Object[] files = listFiles(operations, file);
+
+ LOG.trace("List files {} found {} files", file.getAbsoluteFilePath(),
files.length);
+ for (Object f : files) {
+ SftpRemoteFile rf = (SftpRemoteFile) f;
+ boolean match;
+ if (fastExistsCheck) {
+ match = rf.getFilename().equals(file.getAbsoluteFilePath())
+ || rf.getFilename().equals(file.getFileNameOnly());
+ } else {
+ match = rf.getFilename().equals(file.getFileNameOnly());
+ }
+ if (match) {
+ newLastModified = rf.getLastModified();
+ newLength = rf.getFileLength();
+ }
+ }
+
+ LOG.trace("Previous last modified: {}, new last modified: {}",
lastModified, newLastModified);
+ LOG.trace("Previous length: {}, new length: {}", length, newLength);
+ long newOlderThan = startTime + watch.taken() - minAge;
+ LOG.trace("New older than threshold: {}", newOlderThan);
+
+ if (isReadLockAcquired(lastModified, length, newLastModified,
newLength, newOlderThan)) {
+ LOG.trace("Read lock acquired.");
+ return true;
+ }
+
+ lastModified = newLastModified;
+ length = newLength;
+ return false;
+ }
+
+ private Object[] listFiles(
+ GenericFileOperations<ChannelSftp.LsEntry> operations,
GenericFile<ChannelSftp.LsEntry> file) {
+ if (fastExistsCheck) {
+ return listFilesFast(operations, file);
+ } else {
+ return listFilesByFilter(operations, file);
+ }
+ }
+
+ private Object[] listFilesByFilter(
+ GenericFileOperations<ChannelSftp.LsEntry> operations,
GenericFile<ChannelSftp.LsEntry> file) {
+ String path = file.getParent();
+ if (path.equals("/") || path.equals("\\")) {
+ LOG.trace(
+ "Using full directory listing in home directory to update
file information. Consider enabling fastExistsCheck option.");
+ return operations.listFiles();
+ } else {
+ LOG.trace(
+ "Using full directory listing to update file information
for {}. Consider enabling fastExistsCheck option.",
+ path);
+ return operations.listFiles(path);
+ }
+ }
+
+ private Object[] listFilesFast(
+ GenericFileOperations<ChannelSftp.LsEntry> operations,
GenericFile<ChannelSftp.LsEntry> file) {
+ String path = file.getAbsoluteFilePath();
+ if (path.equals("/") || path.equals("\\")) {
+ LOG.trace("Using fast exists to update file information in home
directory");
+ return operations.listFiles();
+ } else {
+ LOG.trace("Using fast exists to update file information for {}",
path);
+ return operations.listFiles(path);
+ }
+ }
+
+ private boolean isReadLockAcquired(
+ long lastModified, long length, long newLastModified, long
newLength, long newOlderThan) {
+ return newLength >= minLength && (minAge == 0 && newLastModified ==
lastModified && newLength == length
+ || minAge != 0 && newLastModified < newOlderThan);
+ }
+}
diff --git
a/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/strategy/FtpChangedReadLockTimeoutZeroTest.java
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/strategy/FtpChangedReadLockTimeoutZeroTest.java
new file mode 100644
index 000000000000..5e7bcf8e1e4e
--- /dev/null
+++
b/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/strategy/FtpChangedReadLockTimeoutZeroTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.remote.strategy;
+
+import java.time.Duration;
+
+import org.apache.camel.support.task.BlockingTask;
+import org.apache.camel.support.task.Tasks;
+import org.apache.camel.support.task.budget.Budgets;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests that readLockTimeout=0 (documented as "forever") does not cause the
Tasks budget to expire immediately.
+ * Regression test for CAMEL-24096.
+ */
+class FtpChangedReadLockTimeoutZeroTest {
+
+ @Test
+ void timeoutZeroShouldNotExpireImmediately() {
+ long timeout = 0;
+ long checkInterval = 100;
+
+ var budgetBuilder = Budgets.iterationTimeBudget()
+ .withInterval(Duration.ofMillis(checkInterval));
+ if (timeout > 0) {
+ budgetBuilder.withMaxDuration(Duration.ofMillis(timeout));
+ } else {
+ budgetBuilder.withUnlimitedDuration();
+ }
+ BlockingTask task = Tasks.foregroundTask()
+ .withBudget(budgetBuilder.build())
+ .withName("ftp-acquire-exclusive-read-lock")
+ .build();
+
+ int[] counter = { 0 };
+ boolean result = task.run(null, () -> {
+ counter[0]++;
+ return counter[0] >= 2;
+ });
+
+ assertTrue(result, "Task with timeout=0 should have acquired the
lock");
+ assertTrue(counter[0] >= 2, "Task should have run at least twice");
+ }
+
+ @Test
+ void timeoutNegativeShouldNotExpireImmediately() {
+ long timeout = -1;
+ long checkInterval = 100;
+
+ var budgetBuilder = Budgets.iterationTimeBudget()
+ .withInterval(Duration.ofMillis(checkInterval));
+ if (timeout > 0) {
+ budgetBuilder.withMaxDuration(Duration.ofMillis(timeout));
+ } else {
+ budgetBuilder.withUnlimitedDuration();
+ }
+ BlockingTask task = Tasks.foregroundTask()
+ .withBudget(budgetBuilder.build())
+ .withName("ftp-acquire-exclusive-read-lock")
+ .build();
+
+ int[] counter = { 0 };
+ boolean result = task.run(null, () -> {
+ counter[0]++;
+ return counter[0] >= 2;
+ });
+
+ assertTrue(result, "Task with timeout=-1 should have acquired the
lock");
+ }
+
+ @Test
+ void timeoutPositiveShouldExpireWhenExceeded() {
+ long timeout = 200;
+ long checkInterval = 100;
+
+ var budgetBuilder = Budgets.iterationTimeBudget()
+ .withInterval(Duration.ofMillis(checkInterval));
+ if (timeout > 0) {
+ budgetBuilder.withMaxDuration(Duration.ofMillis(timeout));
+ } else {
+ budgetBuilder.withUnlimitedDuration();
+ }
+ BlockingTask task = Tasks.foregroundTask()
+ .withBudget(budgetBuilder.build())
+ .withName("ftp-acquire-exclusive-read-lock")
+ .build();
+
+ boolean result = task.run(null, () -> false);
+
+ assertFalse(result, "Task with positive timeout should have timed
out");
+ }
+}
diff --git
a/components/camel-smb/src/main/java/org/apache/camel/component/smb/strategy/SmbChangedExclusiveReadLockStrategy.java
b/components/camel-smb/src/main/java/org/apache/camel/component/smb/strategy/SmbChangedExclusiveReadLockStrategy.java
index a1666b263ef5..161436bafd16 100644
---
a/components/camel-smb/src/main/java/org/apache/camel/component/smb/strategy/SmbChangedExclusiveReadLockStrategy.java
+++
b/components/camel-smb/src/main/java/org/apache/camel/component/smb/strategy/SmbChangedExclusiveReadLockStrategy.java
@@ -56,11 +56,15 @@ public class SmbChangedExclusiveReadLockStrategy
LOG.trace("Waiting for exclusive read lock to file: {}", file);
+ var budgetBuilder = Budgets.iterationTimeBudget()
+ .withInterval(Duration.ofMillis(checkInterval));
+ if (timeout > 0) {
+ budgetBuilder.withMaxDuration(Duration.ofMillis(timeout));
+ } else {
+ budgetBuilder.withUnlimitedDuration();
+ }
BlockingTask task = Tasks.foregroundTask()
- .withBudget(Budgets.iterationTimeBudget()
- .withMaxDuration(Duration.ofMillis(timeout))
- .withInterval(Duration.ofMillis(checkInterval))
- .build())
+ .withBudget(budgetBuilder.build())
.withName("smb-acquire-exclusive-read-lock")
.build();
