CAMEL-22784 - Improve FileLockClusterService resilience to long blocking network based file I/O

Author: jamesnetherton

components/camel-file/pom.xml

@@ -53,6 +53,11 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-core-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
 
     </dependencies>
 </project>

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

@@ -68,4 +68,13 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hashCode(id);
     }
+
+    @Override
+    public String toString() {
+        return "FileLockClusterLeaderInfo{" +
+               "id='" + id + '\'' +
+               ", heartbeatUpdateIntervalMilliseconds=" + heartbeatUpdateIntervalMilliseconds +
+               ", heartbeatMilliseconds=" + heartbeatMilliseconds +
+               '}';
+    }
 }

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

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.file.cluster;
 
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
@@ -32,13 +33,20 @@ public class FileLockClusterService extends AbstractCamelClusterService<FileLock
     private TimeUnit acquireLockIntervalUnit;
     private ScheduledExecutorService executor;
     private int heartbeatTimeoutMultiplier;
+    private int clusterDataTaskMaxAttempts;
+    private long clusterDataTaskTimeout;
+    private TimeUnit clusterDataTaskTimeoutUnit;
+    private ExecutorService clusterDataTaskExecutor;
 
     public FileLockClusterService() {
         this.acquireLockDelay = 1;
         this.acquireLockDelayUnit = TimeUnit.SECONDS;
         this.acquireLockInterval = 10;
         this.acquireLockIntervalUnit = TimeUnit.SECONDS;
         this.heartbeatTimeoutMultiplier = 5;
+        this.clusterDataTaskMaxAttempts = 5;
+        this.clusterDataTaskTimeout = 10;
+        this.clusterDataTaskTimeoutUnit = TimeUnit.SECONDS;
     }
 
     @Override
@@ -120,13 +128,74 @@ public void setAcquireLockIntervalUnit(TimeUnit acquireLockIntervalUnit) {
      * <p>
      */
     public void setHeartbeatTimeoutMultiplier(int heartbeatTimeoutMultiplier) {
+        if (heartbeatTimeoutMultiplier <= 0) {
+            throw new IllegalArgumentException("HeartbeatTimeoutMultiplier must be greater than 0");
+        }
         this.heartbeatTimeoutMultiplier = heartbeatTimeoutMultiplier;
     }
 
     public int getHeartbeatTimeoutMultiplier() {
         return heartbeatTimeoutMultiplier;
     }
 
+    /**
+     * Sets how many times a cluster data task will run, counting both the first execution and subsequent retries in
+     * case of failure or timeout. The default is 5 attempts.
+     * <p>
+     * This can be useful when the cluster data root is on network based file storage, where I/O operations may
+     * occasionally block for long or unpredictable periods.
+     */
+    public void setClusterDataTaskMaxAttempts(int clusterDataTaskMaxAttempts) {
+        if (clusterDataTaskMaxAttempts <= 0) {
+            throw new IllegalArgumentException("clusterDataTaskMaxRetries must be greater than 0");
+        }
+        this.clusterDataTaskMaxAttempts = clusterDataTaskMaxAttempts;
+    }
+
+    public int getClusterDataTaskMaxAttempts() {
+        return clusterDataTaskMaxAttempts;
+    }
+
+    /**
+     * Sets the timeout for a cluster data task (reading or writing cluster data). The default is 10 seconds.
+     * <p>
+     * Timeouts are useful when the cluster data root is on network storage, where I/O operations may occasionally block
+     * for long or unpredictable periods.
+     */
+    public void setClusterDataTaskTimeout(long clusterDataTaskTimeout) {
+        if (clusterDataTaskTimeout <= 0) {
+            throw new IllegalArgumentException("clusterDataTaskMaxRetries must be greater than 0");
+        }
+        this.clusterDataTaskTimeout = clusterDataTaskTimeout;
+    }
+
+    public long getClusterDataTaskTimeout() {
+        return clusterDataTaskTimeout;
+    }
+
+    /**
+     * The time unit for the clusterDataTaskTimeoutUnit, default to TimeUnit.SECONDS.
+     */
+    public void setClusterDataTaskTimeoutUnit(TimeUnit clusterDataTaskTimeoutUnit) {
+        this.clusterDataTaskTimeoutUnit = clusterDataTaskTimeoutUnit;
+    }
+
+    public TimeUnit getClusterDataTaskTimeoutUnit() {
+        return clusterDataTaskTimeoutUnit;
+    }
+
+    /**
+     * Sets the timeout for a cluster data task (reading or writing cluster data). The default is 10 seconds.
+     * <p>
+     * Timeouts are useful when the cluster data root is on network storage, where I/O operations may occasionally block
+     * for long or unpredictable periods.
+     * <p>
+     */
+    public void setClusterDataTaskTimeout(long clusterDataTaskTimeout, TimeUnit clusterDataTaskTimeoutUnit) {
+        setClusterDataTaskTimeout(clusterDataTaskTimeout);
+        setClusterDataTaskTimeoutUnit(clusterDataTaskTimeoutUnit);
+    }
+
     @Override
     protected void doStop() throws Exception {
         super.doStop();
@@ -142,6 +211,14 @@ protected void doStop() throws Exception {
 
             executor = null;
         }
+
+        if (clusterDataTaskExecutor != null) {
+            if (context != null) {
+                context.getExecutorServiceManager().shutdown(clusterDataTaskExecutor);
+            } else {
+                clusterDataTaskExecutor.shutdown();
+            }
+        }
     }
 
     ScheduledExecutorService getExecutor() {
@@ -161,4 +238,19 @@ ScheduledExecutorService getExecutor() {
             internalLock.unlock();
         }
     }
+
+    ExecutorService getClusterDataTaskExecutor() {
+        Lock internalLock = getInternalLock();
+        internalLock.lock();
+        try {
+            if (clusterDataTaskExecutor == null) {
+                final CamelContext context = ObjectHelper.notNull(getCamelContext(), "CamelContext");
+                clusterDataTaskExecutor = context.getExecutorServiceManager().newFixedThreadPool(this,
+                        "FileLockClusterDataTask-" + getId(), 5);
+            }
+            return clusterDataTaskExecutor;
+        } finally {
+            internalLock.unlock();
+        }
+    }
 }

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

@@ -0,0 +1,75 @@
+/*
+ * 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.cluster;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Executes cluster data read / write tasks asynchronously, with timeouts to guard against potential unpredictable
+ * blocking I/O periods.
+ */
+class FileLockClusterTaskExecutor {
+    private static final Logger LOGGER = LoggerFactory.getLogger(FileLockClusterTaskExecutor.class);
+    private final FileLockClusterService service;
+
+    FileLockClusterTaskExecutor(FileLockClusterService service) {
+        Objects.requireNonNull(service, "FileLockClusterService cannot be null");
+        this.service = service;
+    }
+
+    /**
+     * If the cluster data root is network based, like an NFS mount, avoid potential long blocking I/O to fail fast and
+     * reliably reason about the cluster state.
+     *
+     * @param task Supplier representing a task to run
+     */
+    <T> T run(Supplier<T> task) throws ExecutionException, TimeoutException {
+        Objects.requireNonNull(task, "Task cannot be null");
+
+        int maxAttempts = service.getClusterDataTaskMaxAttempts();
+        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+            LOGGER.debug("Running cluster task attempt {} of {}", attempt, maxAttempts);
+
+            CompletableFuture<T> future = CompletableFuture.supplyAsync(task, service.getClusterDataTaskExecutor());
+            try {
+                return future.get(service.getClusterDataTaskTimeout(), service.getClusterDataTaskTimeoutUnit());
+            } catch (InterruptedException e) {
+                LOGGER.trace("Cluster task interrupted on attempt {} of {}", attempt, maxAttempts);
+                future.cancel(true);
+                Thread.currentThread().interrupt();
+                return null;
+            } catch (ExecutionException | TimeoutException e) {
+                LOGGER.debug("Cluster task encountered an exception on attempt {} of {}", attempt, maxAttempts, e);
+                future.cancel(true);
+                if (attempt == maxAttempts) {
+                    LOGGER.debug("Cluster task retry limit ({}) reached", maxAttempts, e);
+                    throw e;
+                }
+            } finally {
+                LOGGER.debug("Cluster task attempt {} ended", attempt);
+            }
+        }
+        return null;
+    }
+}

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

@@ -62,7 +62,7 @@ static void writeClusterLeaderInfo(
             FileChannel channel,
             FileLockClusterLeaderInfo clusterLeaderInfo,
             boolean forceMetaData)
-            throws IOException {
+            throws Exception {
 
         Objects.requireNonNull(channel, "channel cannot be null");
         Objects.requireNonNull(clusterLeaderInfo, "clusterLeaderInfo cannot be null");
@@ -100,7 +100,7 @@ static void writeClusterLeaderInfo(
      *                        inconsistent state
      * @throws IOException    If reading the lock file failed
      */
-    static FileLockClusterLeaderInfo readClusterLeaderInfo(Path leaderDataPath) throws IOException {
+    static FileLockClusterLeaderInfo readClusterLeaderInfo(Path leaderDataPath) throws Exception {
         try {
             byte[] bytes = Files.readAllBytes(leaderDataPath);
 
@@ -119,7 +119,7 @@ static FileLockClusterLeaderInfo readClusterLeaderInfo(Path leaderDataPath) thro
             long lastHeartbeat = buf.getLong();
 
             return new FileLockClusterLeaderInfo(uuidStr, intervalMillis, lastHeartbeat);
-        } catch (NoSuchFileException e) {
+        } catch (FileNotFoundException | NoSuchFileException e) {
             // Handle NoSuchFileException to give the ClusterView a chance to recreate the leadership data
             return null;
         }