This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24096 in repository https://gitbox.apache.org/repos/asf/camel.git
commit fd33cd761ed6a34a0a336f815c90b5e9af13634e Author: Claus Ibsen <[email protected]> AuthorDate: Thu Jul 16 07:49:17 2026 +0200 CAMEL-24096: readLock=changed - fix readLockTimeout=0 regression and convert SFTP to Tasks API readLockTimeout=0 is documented as "forever" but the CAMEL-17121 Tasks API conversion caused TimeBoundedBudget to expire immediately (only -1 means unlimited, not 0). Fix all affected strategies to use withUnlimitedDuration() when timeout <= 0. Also convert SftpChangedExclusiveReadLockStrategy from the old while-loop to the Tasks API for consistency. Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[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();
