You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by or...@apache.org on 2021/10/29 11:11:23 UTC
[camel] 04/04: CAMEL-17121: convert the ftp component to use the
repeatable tasks
This is an automated email from the ASF dual-hosted git repository.
orpiske pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit ae7928cab7fac63fa9ff61011cf0249a8943c5c4
Author: Otavio Rodolfo Piske <op...@redhat.com>
AuthorDate: Tue Oct 26 11:49:36 2021 +0200
CAMEL-17121: convert the ftp component to use the repeatable tasks
---
.../camel/component/file/remote/FtpOperations.java | 128 +++++++++----------
.../component/file/remote/SftpOperations.java | 104 ++++++++--------
.../remote/strategy/ExclusiveReadLockCheck.java | 136 +++++++++++++++++++++
.../FtpChangedExclusiveReadLockStrategy.java | 121 +++---------------
4 files changed, 275 insertions(+), 214 deletions(-)
diff --git a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java
index a9f96a5..c8bf472 100644
--- a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java
+++ b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java
@@ -23,6 +23,7 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.time.Duration;
import java.util.Iterator;
import org.apache.camel.Exchange;
@@ -33,6 +34,9 @@ import org.apache.camel.component.file.GenericFileEndpoint;
import org.apache.camel.component.file.GenericFileExist;
import org.apache.camel.component.file.GenericFileOperationFailedException;
import org.apache.camel.support.ObjectHelper;
+import org.apache.camel.support.task.BlockingTask;
+import org.apache.camel.support.task.Tasks;
+import org.apache.camel.support.task.budget.Budgets;
import org.apache.camel.util.FileUtil;
import org.apache.camel.util.IOHelper;
import org.apache.camel.util.StopWatch;
@@ -56,6 +60,15 @@ public class FtpOperations implements RemoteFileOperations<FTPFile> {
protected FtpEndpoint<FTPFile> endpoint;
protected FtpClientActivityListener clientActivityListener;
+ private static class TaskPayload {
+ final RemoteFileConfiguration configuration;
+ private Exception exception;
+
+ public TaskPayload(RemoteFileConfiguration configuration) {
+ this.configuration = configuration;
+ }
+ }
+
public FtpOperations(FTPClient client, FTPClientConfig clientConfig) {
this.client = client;
this.clientConfig = clientConfig;
@@ -100,7 +113,6 @@ public class FtpOperations implements RemoteFileOperations<FTPFile> {
log.trace("Connecting using FTPClient: {}", client);
String host = configuration.getHost();
- int port = configuration.getPort();
String username = configuration.getUsername();
String account = ((FtpConfiguration) configuration).getAccount();
@@ -118,71 +130,32 @@ public class FtpOperations implements RemoteFileOperations<FTPFile> {
client.getConnectTimeout());
}
- boolean connected = false;
- int attempt = 0;
+ BlockingTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(Budgets.atLeastOnce(endpoint.getMaximumReconnectAttempts()))
+ .withInterval(Duration.ofMillis(endpoint.getReconnectDelay()))
+ .build())
+ .build();
- while (!connected) {
- try {
- if (log.isTraceEnabled() && attempt > 0) {
- log.trace("Reconnect attempt #{} connecting to {}", attempt, configuration.remoteServerInformation());
- }
- clientActivityListener.onConnecting(host);
- client.connect(host, port);
- // must check reply code if we are connected
- int reply = client.getReplyCode();
-
- if (FTPReply.isPositiveCompletion(reply)) {
- // yes we could connect
- connected = true;
- } else {
- // throw an exception to force the retry logic in the catch
- // exception block
- throw new GenericFileOperationFailedException(
- client.getReplyCode(), client.getReplyString(), "Server refused connection");
- }
- } catch (Exception e) {
- if (client.isConnected()) {
- log.trace("Disconnecting due to exception during connect");
- try {
- client.disconnect(); // ensures socket is closed
- } catch (IOException ignore) {
- log.trace("Ignore exception during disconnect: {}", ignore.getMessage());
- }
- }
- // check if we are interrupted so we can break out
- if (Thread.currentThread().isInterrupted()) {
- throw new GenericFileOperationFailedException(
- "Interrupted during connecting", new InterruptedException("Interrupted during connecting"));
- }
+ TaskPayload payload = new TaskPayload(configuration);
- GenericFileOperationFailedException failed;
- if (e instanceof GenericFileOperationFailedException) {
- failed = (GenericFileOperationFailedException) e;
- } else {
- failed = new GenericFileOperationFailedException(
- client.getReplyCode(), client.getReplyString(), e.getMessage(), e);
- }
+ if (!task.run(this::tryConnect, payload)) {
+ if (exchange != null) {
+ exchange.getIn().setHeader(FtpConstants.FTP_REPLY_CODE, client.getReplyCode());
+ exchange.getIn().setHeader(FtpConstants.FTP_REPLY_STRING, client.getReplyString());
+ }
- log.trace("Cannot connect due: {}", failed.getMessage());
- attempt++;
- if (attempt > endpoint.getMaximumReconnectAttempts()) {
- throw failed;
- }
- if (endpoint.getReconnectDelay() > 0) {
- try {
- Thread.sleep(endpoint.getReconnectDelay());
- } catch (InterruptedException ie) {
- // we could potentially also be interrupted during sleep
- Thread.currentThread().interrupt();
- throw new GenericFileOperationFailedException("Interrupted during sleeping", ie);
- }
- }
- } finally {
- if (exchange != null) {
- // store client reply information after the operation
- exchange.getIn().setHeader(FtpConstants.FTP_REPLY_CODE, client.getReplyCode());
- exchange.getIn().setHeader(FtpConstants.FTP_REPLY_STRING, client.getReplyString());
+ if (payload.exception != null) {
+ if (payload.exception instanceof GenericFileOperationFailedException) {
+ throw (GenericFileOperationFailedException) payload.exception;
+ } else {
+ throw new GenericFileOperationFailedException(
+ client.getReplyCode(), client.getReplyString(), payload.exception.getMessage(), payload.exception);
}
+ } else {
+ throw new GenericFileOperationFailedException(
+ client.getReplyCode(), client.getReplyString(),
+ "Server refused connection");
}
}
@@ -270,6 +243,37 @@ public class FtpOperations implements RemoteFileOperations<FTPFile> {
return true;
}
+ private boolean tryConnect(TaskPayload payload) {
+ final RemoteFileConfiguration configuration = payload.configuration;
+ final String host = configuration.getHost();
+ final int port = configuration.getPort();
+
+ try {
+ log.trace("Reconnect attempt to {}", configuration.remoteServerInformation());
+
+ clientActivityListener.onConnecting(host);
+ client.connect(host, port);
+
+ // must check reply code if we are connected
+ int reply = client.getReplyCode();
+
+ return FTPReply.isPositiveCompletion(reply);
+ } catch (Exception e) {
+
+ payload.exception = e;
+
+ if (client.isConnected()) {
+ log.trace("Disconnecting due to exception during connect");
+ try {
+ client.disconnect(); // ensures socket is closed
+ } catch (IOException ignore) {
+ log.trace("Ignore exception during disconnect: {}", ignore.getMessage());
+ }
+ }
+ }
+ return false;
+ }
+
@Override
public boolean isConnected() throws GenericFileOperationFailedException {
return client.isConnected();
diff --git a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpOperations.java b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpOperations.java
index b966c9a..71453ef 100644
--- a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpOperations.java
+++ b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/SftpOperations.java
@@ -28,6 +28,7 @@ import java.net.InetAddress;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.security.KeyPair;
+import java.time.Duration;
import java.util.Base64;
import java.util.Hashtable;
import java.util.List;
@@ -53,6 +54,9 @@ import org.apache.camel.component.file.GenericFileEndpoint;
import org.apache.camel.component.file.GenericFileExist;
import org.apache.camel.component.file.GenericFileOperationFailedException;
import org.apache.camel.support.ResourceHelper;
+import org.apache.camel.support.task.BlockingTask;
+import org.apache.camel.support.task.Tasks;
+import org.apache.camel.support.task.budget.Budgets;
import org.apache.camel.util.FileUtil;
import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
@@ -78,6 +82,15 @@ public class SftpOperations implements RemoteFileOperations<SftpRemoteFile> {
private ChannelSftp channel;
private Session session;
+ private static class TaskPayload {
+ final RemoteFileConfiguration configuration;
+ private Exception exception;
+
+ public TaskPayload(RemoteFileConfiguration configuration) {
+ this.configuration = configuration;
+ }
+ }
+
public SftpOperations() {
}
@@ -109,70 +122,63 @@ public class SftpOperations implements RemoteFileOperations<SftpRemoteFile> {
return true;
}
- boolean connected = false;
- int attempt = 0;
+ BlockingTask task = Tasks
+ .foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(endpoint.getMaximumReconnectAttempts())
+ .withInterval(Duration.ofMillis(endpoint.getReconnectDelay()))
+ .build())
+ .build();
- while (!connected) {
- try {
- if (LOG.isTraceEnabled() && attempt > 0) {
- LOG.trace("Reconnect attempt #{} connecting to + {}", attempt, configuration.remoteServerInformation());
- }
+ TaskPayload payload = new TaskPayload(configuration);
- if (channel == null || !channel.isConnected()) {
- if (session == null || !session.isConnected()) {
- LOG.trace("Session isn't connected, trying to recreate and connect.");
- session = createSession(configuration);
- if (endpoint.getConfiguration().getConnectTimeout() > 0) {
- LOG.trace("Connecting use connectTimeout: {} ...", endpoint.getConfiguration().getConnectTimeout());
- session.connect(endpoint.getConfiguration().getConnectTimeout());
- } else {
- LOG.trace("Connecting ...");
- session.connect();
- }
- }
+ if (!task.run(this::tryConnect, payload)) {
+ throw new GenericFileOperationFailedException(
+ "Cannot connect to " + configuration.remoteServerInformation(),
+ payload.exception);
+ }
+
+ configureBulkRequests();
+
+ return true;
+ }
- LOG.trace("Channel isn't connected, trying to recreate and connect.");
- channel = (ChannelSftp) session.openChannel("sftp");
+ private boolean tryConnect(TaskPayload payload) {
+ LOG.trace("Reconnect attempt to {}", payload.configuration.remoteServerInformation());
+
+ try {
+ if (channel == null || !channel.isConnected()) {
+ if (session == null || !session.isConnected()) {
+ LOG.trace("Session isn't connected, trying to recreate and connect.");
+
+ session = createSession(payload.configuration);
if (endpoint.getConfiguration().getConnectTimeout() > 0) {
LOG.trace("Connecting use connectTimeout: {} ...", endpoint.getConfiguration().getConnectTimeout());
- channel.connect(endpoint.getConfiguration().getConnectTimeout());
+ session.connect(endpoint.getConfiguration().getConnectTimeout());
} else {
LOG.trace("Connecting ...");
- channel.connect();
+ session.connect();
}
- LOG.debug("Connected to {}", configuration.remoteServerInformation());
}
- // yes we could connect
- connected = true;
- } catch (Exception e) {
- // check if we are interrupted so we can break out
- if (Thread.currentThread().isInterrupted()) {
- throw new GenericFileOperationFailedException(
- "Interrupted during connecting", new InterruptedException("Interrupted during connecting"));
- }
+ LOG.trace("Channel isn't connected, trying to recreate and connect.");
+ channel = (ChannelSftp) session.openChannel("sftp");
- GenericFileOperationFailedException failed = new GenericFileOperationFailedException(
- "Cannot connect to " + configuration.remoteServerInformation(), e);
- LOG.trace("Cannot connect due: {}", failed.getMessage());
- attempt++;
- if (attempt > endpoint.getMaximumReconnectAttempts()) {
- throw failed;
- }
- if (endpoint.getReconnectDelay() > 0) {
- try {
- Thread.sleep(endpoint.getReconnectDelay());
- } catch (InterruptedException ie) {
- // we could potentially also be interrupted during sleep
- Thread.currentThread().interrupt();
- throw new GenericFileOperationFailedException("Interrupted during sleeping", ie);
- }
+ if (endpoint.getConfiguration().getConnectTimeout() > 0) {
+ LOG.trace("Connecting use connectTimeout: {} ...", endpoint.getConfiguration().getConnectTimeout());
+ channel.connect(endpoint.getConfiguration().getConnectTimeout());
+ } else {
+ LOG.trace("Connecting ...");
+ channel.connect();
}
+ LOG.debug("Connected to {}", payload.configuration.remoteServerInformation());
}
- }
+ } catch (JSchException e) {
+ payload.exception = e;
- configureBulkRequests();
+ return false;
+ }
return true;
}
diff --git a/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/ExclusiveReadLockCheck.java b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/ExclusiveReadLockCheck.java
new file mode 100644
index 0000000..3e98b27
--- /dev/null
+++ b/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/strategy/ExclusiveReadLockCheck.java
@@ -0,0 +1,136 @@
+/*
+ * 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.util.Date;
+
+import org.apache.camel.component.file.GenericFile;
+import org.apache.camel.component.file.GenericFileOperations;
+import org.apache.camel.util.StopWatch;
+import org.apache.commons.net.ftp.FTPFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExclusiveReadLockCheck {
+ private static final Logger LOG = LoggerFactory.getLogger(ExclusiveReadLockCheck.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 ExclusiveReadLockCheck(boolean fastExistsCheck, long minAge, long minLength) {
+ this.fastExistsCheck = fastExistsCheck;
+ this.startTime = new Date().getTime();
+ this.minAge = minAge;
+ this.minLength = minLength;
+ this.watch = new StopWatch();
+
+ this.lastModified = Long.MIN_VALUE;
+ this.length = Long.MIN_VALUE;
+ }
+
+ public boolean tryAcquireExclusiveReadLock(GenericFileOperations<FTPFile> operations, GenericFile<FTPFile> file) {
+ long newLastModified = 0;
+ long newLength = 0;
+
+ FTPFile[] files = getFtpFiles(operations, file);
+
+ LOG.trace("List files {} found {} files", file.getAbsoluteFilePath(), files.length);
+ for (FTPFile f : files) {
+ boolean match;
+ if (fastExistsCheck) {
+ // uses the absolute file path as well
+ match = f.getName().equals(file.getAbsoluteFilePath()) || f.getName().equals(file.getFileNameOnly());
+ } else {
+ match = f.getName().equals(file.getFileNameOnly());
+ }
+ if (match) {
+ newLength = f.getSize();
+ if (f.getTimestamp() != null) {
+ newLastModified = f.getTimestamp().getTimeInMillis();
+ }
+ }
+ }
+
+ 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 FTPFile[] getFtpFiles(GenericFileOperations<FTPFile> operations, GenericFile<FTPFile> file) {
+ FTPFile[] 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
+ files = getFtpFilesFast(operations, file);
+ } else {
+ files = getFtpFilesByFilter(operations, file);
+ }
+ return files;
+ }
+
+ private FTPFile[] getFtpFilesByFilter(GenericFileOperations<FTPFile> operations, GenericFile<FTPFile> file) {
+ // fast option not enabled, so list the directory and filter the
+ // file name
+ 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.");
+ 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 FTPFile[] getFtpFilesFast(GenericFileOperations<FTPFile> operations, GenericFile<FTPFile> file) {
+ 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");
+ 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/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 e2fc306..54130f5 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
@@ -16,7 +16,7 @@
*/
package org.apache.camel.component.file.remote.strategy;
-import java.util.Date;
+import java.time.Duration;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
@@ -25,7 +25,9 @@ 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.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.apache.commons.net.ftp.FTPFile;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,113 +52,26 @@ public class FtpChangedExclusiveReadLockStrategy implements GenericFileExclusive
public boolean acquireExclusiveReadLock(
GenericFileOperations<FTPFile> operations, GenericFile<FTPFile> 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 = new Date().getTime();
-
- 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;
-
- FTPFile[] 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 {
- // fast option not enabled, so list the directory and filter the
- // file name
- 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 (FTPFile f : files) {
- boolean match;
- if (fastExistsCheck) {
- // uses the absolute file path as well
- match = f.getName().equals(file.getAbsoluteFilePath()) || f.getName().equals(file.getFileNameOnly());
- } else {
- match = f.getName().equals(file.getFileNameOnly());
- }
- if (match) {
- newLength = f.getSize();
- if (f.getTimestamp() != null) {
- newLastModified = f.getTimestamp().getTimeInMillis();
- }
- }
- }
-
- 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;
- }
- }
- }
+ BlockingTask task = Tasks.foregroundTask()
+ .withBudget(Budgets.iterationTimeBudget()
+ .withMaxDuration(Duration.ofMillis(timeout))
+ .withInterval(Duration.ofMillis(checkInterval))
+ .build())
+ .withName("ftp-acquire-exclusive-read-lock")
+ .build();
- return exclusive;
- }
+ ExclusiveReadLockCheck exclusiveReadLockCheck = new ExclusiveReadLockCheck(fastExistsCheck, minAge, minLength);
+
+ if (!task.run(() -> 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) {
- LOG.debug("Sleep interrupted while waiting for exclusive read lock, so breaking out");
- return true;
}
+
+ return true;
}
@Override