This is an automated email from the ASF dual-hosted git repository.
jamesnetherton pushed a commit to branch camel-4.14.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.14.x by this push:
new 4d371879e924 CAMEL-22784: Remove acquireLeadershipBackoff option from
FileLockClusterService
4d371879e924 is described below
commit 4d371879e924c4d621c3d36594e1494437eb343a
Author: James Netherton <[email protected]>
AuthorDate: Fri Dec 19 08:44:25 2025 +0000
CAMEL-22784: Remove acquireLeadershipBackoff option from
FileLockClusterService
---
.../file/cluster/FileLockClusterService.java | 29 --------
.../file/cluster/FileLockClusterView.java | 51 ++------------
.../FileLockClusterServiceBasicFailoverTest.java | 78 ----------------------
.../cluster/FileLockClusterServiceTestBase.java | 10 ---
.../user-manual/modules/ROOT/pages/clustering.adoc | 2 -
5 files changed, 4 insertions(+), 166 deletions(-)
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/cluster/FileLockClusterService.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/cluster/FileLockClusterService.java
index 5b53ce16b7c7..e21568925f0a 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/cluster/FileLockClusterService.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/cluster/FileLockClusterService.java
@@ -32,8 +32,6 @@ public class FileLockClusterService extends
AbstractCamelClusterService<FileLock
private TimeUnit acquireLockIntervalUnit;
private ScheduledExecutorService executor;
private int heartbeatTimeoutMultiplier;
- private long acquireLeadershipBackoff;
- private TimeUnit acquireLeadershipBackoffIntervalUnit;
public FileLockClusterService() {
this.acquireLockDelay = 1;
@@ -41,8 +39,6 @@ public class FileLockClusterService extends
AbstractCamelClusterService<FileLock
this.acquireLockInterval = 10;
this.acquireLockIntervalUnit = TimeUnit.SECONDS;
this.heartbeatTimeoutMultiplier = 5;
- this.acquireLeadershipBackoff = 0;
- this.acquireLeadershipBackoffIntervalUnit = TimeUnit.SECONDS;
}
@Override
@@ -131,31 +127,6 @@ public class FileLockClusterService extends
AbstractCamelClusterService<FileLock
return heartbeatTimeoutMultiplier;
}
- /**
- * The time to wait before a cluster member broadcasts acquisition of
gaining cluster leadership. A value of 0 (the
- * default) disables the backoff wait period. This can be useful to
introduce a delay to ensure that the cluster
- * leader has fully surrendered its leadership. A sensible value is
acquireLockDelay + 5 seconds. Or whatever is
- * approprite for your environment and requirements.
- */
- public void setAcquireLeadershipBackoff(long acquireLeadershipBackoff) {
- this.acquireLeadershipBackoff = acquireLeadershipBackoff;
- }
-
- public long getAcquireLeadershipBackoff() {
- return acquireLeadershipBackoff;
- }
-
- /**
- * The time unit for the acquireLeadershipBackoff, default to
TimeUnit.SECONDS.
- */
- public void setAcquireLeadershipBackoffIntervalUnit(TimeUnit
acquireLeadershipBackoffIntervalUnit) {
- this.acquireLeadershipBackoffIntervalUnit =
acquireLeadershipBackoffIntervalUnit;
- }
-
- public TimeUnit getAcquireLeadershipBackoffIntervalUnit() {
- return acquireLeadershipBackoffIntervalUnit;
- }
-
@Override
protected void doStop() throws Exception {
super.doStop();
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/cluster/FileLockClusterView.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/cluster/FileLockClusterView.java
index 5ce7c554ebc7..8156f00cfaff 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/cluster/FileLockClusterView.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/cluster/FileLockClusterView.java
@@ -32,7 +32,6 @@ import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
@@ -52,14 +51,12 @@ public class FileLockClusterView extends
AbstractCamelClusterView {
private final Path leaderLockPath;
private final Path leaderDataPath;
private final AtomicReference<FileLockClusterLeaderInfo>
clusterLeaderInfoRef = new AtomicReference<>();
- private final AtomicBoolean firstAcquireLockAttemptCompleted = new
AtomicBoolean();
private RandomAccessFile leaderLockFile;
private RandomAccessFile leaderDataFile;
private FileLock lock;
private ScheduledFuture<?> task;
private int heartbeatTimeoutMultiplier;
private long acquireLockIntervalMilliseconds;
- private long acquireLeadershipBackoffMilliseconds;
FileLockClusterView(FileLockClusterService cluster, String namespace) {
super(cluster, namespace);
@@ -126,14 +123,6 @@ public class FileLockClusterView extends
AbstractCamelClusterView {
throw new IllegalArgumentException("HeartbeatTimeoutMultiplier
must be greater than 0");
}
- long acquireLeadershipBackoff = service.getAcquireLeadershipBackoff();
- if (acquireLeadershipBackoff < 0) {
- throw new IllegalArgumentException("acquireLeadershipBackoff must
not be a negative value");
- }
- acquireLeadershipBackoffMilliseconds = TimeUnit.MILLISECONDS.convert(
- acquireLeadershipBackoff,
- service.getAcquireLeadershipBackoffIntervalUnit());
-
ScheduledExecutorService executor = service.getExecutor();
task = executor.scheduleWithFixedDelay(this::tryLock,
TimeUnit.MILLISECONDS.convert(service.getAcquireLockDelay(),
service.getAcquireLockDelayUnit()),
@@ -159,7 +148,6 @@ public class FileLockClusterView extends
AbstractCamelClusterView {
closeInternal();
localMember.setStatus(ClusterMemberStatus.STOPPED);
clusterLeaderInfoRef.set(null);
- firstAcquireLockAttemptCompleted.set(false);
}
private void closeInternal() {
@@ -262,33 +250,16 @@ public class FileLockClusterView extends
AbstractCamelClusterView {
if (lock != null) {
LOGGER.info("Lock on file {} acquired (lock={},
cluster-member-id={})", leaderLockPath, lock,
localMember.getUuid());
-
- boolean interrupted = false;
- if (firstAcquireLockAttemptCompleted.get()) {
- // If this is not the initial attempt to claim
leadership, give some time for the existing leader to fully relinquish
leadership
- interrupted = applyAcquireLeadershipBackoff();
- }
-
- if (interrupted) {
- // applyAcquireLeadershipBackoff sleep was
interrupted, likely because we are being shutdown
- lock = null;
- } else {
- localMember.setStatus(ClusterMemberStatus.LEADER);
- clusterLeaderInfoRef.set(null);
- firstAcquireLockAttemptCompleted.set(false);
- fireLeadershipChangedEvent(localMember);
- writeClusterLeaderInfo(true);
- }
+ localMember.setStatus(ClusterMemberStatus.LEADER);
+ clusterLeaderInfoRef.set(null);
+ fireLeadershipChangedEvent(localMember);
+ writeClusterLeaderInfo(true);
} else {
LOGGER.debug("Lock on file {} not acquired",
leaderLockPath);
}
} else {
LOGGER.debug("Existing cluster leader is valid. Retrying
leadership acquisition on next interval");
}
-
- if
(localMember.getStatus().equals(ClusterMemberStatus.FOLLOWER)) {
- firstAcquireLockAttemptCompleted.set(true);
- }
} catch (OverlappingFileLockException e) {
reason = new IOException(e);
} catch (Exception e) {
@@ -345,20 +316,6 @@ public class FileLockClusterView extends
AbstractCamelClusterView {
return false;
}
- boolean applyAcquireLeadershipBackoff() {
- boolean interrupted = false;
- if (acquireLeadershipBackoffMilliseconds > 0) {
- try {
- LOGGER.debug("Waiting {} milliseconds before claiming
leadership", acquireLeadershipBackoffMilliseconds);
- Thread.sleep(acquireLeadershipBackoffMilliseconds);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- interrupted = true;
- }
- }
- return interrupted;
- }
-
private final class ClusterMember implements CamelClusterMember {
private final AtomicReference<ClusterMemberStatus> status = new
AtomicReference<>(ClusterMemberStatus.STOPPED);
private final String uuid = UUID.randomUUID().toString();
diff --git
a/components/camel-master/src/test/java/org/apache/camel/component/file/cluster/FileLockClusterServiceBasicFailoverTest.java
b/components/camel-master/src/test/java/org/apache/camel/component/file/cluster/FileLockClusterServiceBasicFailoverTest.java
index a6b3e1c784d1..72b49bc1b486 100644
---
a/components/camel-master/src/test/java/org/apache/camel/component/file/cluster/FileLockClusterServiceBasicFailoverTest.java
+++
b/components/camel-master/src/test/java/org/apache/camel/component/file/cluster/FileLockClusterServiceBasicFailoverTest.java
@@ -167,71 +167,6 @@ class FileLockClusterServiceBasicFailoverTest extends
FileLockClusterServiceTest
assertEquals(0, Files.size(dataFile));
}
- @Test
- void clusterFailoverWithBackoffWhenLeaderCamelContextStopped() throws
Exception {
- CamelContext clusterLeader = createCamelContext();
-
- ClusterConfig followerConfig = new ClusterConfig();
- followerConfig.setAcquireLockDelay(2);
- followerConfig.setAcquireLeadershipBackoff(5);
- CamelContext clusterFollower = createCamelContext(followerConfig);
-
- try {
- MockEndpoint mockEndpointClustered =
clusterLeader.getEndpoint("mock:result", MockEndpoint.class);
- mockEndpointClustered.expectedMessageCount(5);
-
- clusterLeader.start();
- clusterFollower.start();
-
- mockEndpointClustered.assertIsSatisfied();
-
- AtomicReference<String> leaderId = new AtomicReference<>();
- Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(()
-> {
- assertTrue(Files.exists(lockFile));
- assertTrue(Files.exists(dataFile));
- assertTrue(getClusterMember(clusterLeader).isLeader());
-
- FileLockClusterLeaderInfo clusterLeaderInfo =
FileLockClusterUtils.readClusterLeaderInfo(dataFile);
- assertNotNull(clusterLeaderInfo);
-
- assertNotNull(clusterLeaderInfo.getId());
- assertDoesNotThrow(() ->
UUID.fromString(clusterLeaderInfo.getId()));
- leaderId.set(clusterLeaderInfo.getId());
- });
-
- // Wait enough time for the follower to have run its lock
acquisition scheduled task
- Thread.sleep(followerConfig.getStartupDelayWithOffsetMillis());
-
- // The follower should not have produced any messages
- MockEndpoint mockEndpointFollower =
clusterFollower.getEndpoint("mock:result", MockEndpoint.class);
- assertTrue(mockEndpointFollower.getExchanges().isEmpty());
-
- // Stop the cluster leader
- clusterLeader.stop();
-
- // Backoff is configured so the follower should not claim
leadership within most of that period
-
Awaitility.await().during(Duration.ofMillis(4900)).untilAsserted(() -> {
- assertFalse(getClusterMember(clusterFollower).isLeader());
- });
-
- // Verify the follower was elected as the new cluster leader
- Awaitility.await().atMost(Duration.ofSeconds(10)).untilAsserted(()
-> {
- FileLockClusterLeaderInfo updatedClusterLeaderInfo =
FileLockClusterUtils.readClusterLeaderInfo(dataFile);
- assertNotNull(updatedClusterLeaderInfo);
-
- String newLeaderId = updatedClusterLeaderInfo.getId();
- assertNotNull(newLeaderId);
- assertDoesNotThrow(() -> UUID.fromString(newLeaderId));
- assertNotEquals(leaderId.get(), newLeaderId);
- assertEquals(5, mockEndpointFollower.getExchanges().size());
- });
- } finally {
- clusterFollower.stop();
- }
-
- assertEquals(0, Files.size(dataFile));
- }
-
@Test
void singleClusterMemberRecoversLeadershipIfUUIDRemovedFromLockFile()
throws Exception {
try (CamelContext clusterLeader = createCamelContext()) {
@@ -310,17 +245,4 @@ class FileLockClusterServiceBasicFailoverTest extends
FileLockClusterServiceTest
});
assertIsInstanceOf(IllegalArgumentException.class,
exception.getCause());
}
-
- @Test
- void negativeAcquireLeadershipBackoffThrowsException() throws Exception {
- ClusterConfig config = new ClusterConfig();
- config.setAcquireLeadershipBackoff(-1);
-
- Exception exception = assertThrows(Exception.class, () -> {
- try (CamelContext camelContext = createCamelContext(config)) {
- camelContext.start();
- }
- });
- assertIsInstanceOf(IllegalArgumentException.class,
exception.getCause());
- }
}
diff --git
a/components/camel-master/src/test/java/org/apache/camel/component/file/cluster/FileLockClusterServiceTestBase.java
b/components/camel-master/src/test/java/org/apache/camel/component/file/cluster/FileLockClusterServiceTestBase.java
index ee846f168467..72290aaacbbb 100644
---
a/components/camel-master/src/test/java/org/apache/camel/component/file/cluster/FileLockClusterServiceTestBase.java
+++
b/components/camel-master/src/test/java/org/apache/camel/component/file/cluster/FileLockClusterServiceTestBase.java
@@ -67,7 +67,6 @@ abstract class FileLockClusterServiceTestBase {
service.setAcquireLockInterval(1);
service.setRoot(clusterDir.toString());
service.setHeartbeatTimeoutMultiplier(config.getHeartbeatTimeoutMultiplier());
-
service.setAcquireLeadershipBackoff(config.getAcquireLeadershipBackoff());
return service;
}
@@ -84,7 +83,6 @@ abstract class FileLockClusterServiceTestBase {
private long acquireLockDelay = 1;
private long timerRepeatCount = 5;
private int heartbeatTimeoutMultiplier = 5;
- private long acquireLeadershipBackoff = 0;
long getAcquireLockDelay() {
return acquireLockDelay;
@@ -113,13 +111,5 @@ abstract class FileLockClusterServiceTestBase {
public void setHeartbeatTimeoutMultiplier(int
heartbeatTimeoutMultiplier) {
this.heartbeatTimeoutMultiplier = heartbeatTimeoutMultiplier;
}
-
- public long getAcquireLeadershipBackoff() {
- return acquireLeadershipBackoff;
- }
-
- public void setAcquireLeadershipBackoff(long acquireLeadershipBackoff)
{
- this.acquireLeadershipBackoff = acquireLeadershipBackoff;
- }
}
}
diff --git a/docs/user-manual/modules/ROOT/pages/clustering.adoc
b/docs/user-manual/modules/ROOT/pages/clustering.adoc
index 52ee7aaff7c8..669a56d13216 100644
--- a/docs/user-manual/modules/ROOT/pages/clustering.adoc
+++ b/docs/user-manual/modules/ROOT/pages/clustering.adoc
@@ -65,8 +65,6 @@ Configuration options:
| acquireLockDelayUnit | The time unit for acquireLockDelay | SECONDS |
TimeUnit
| acquireLockInterval | The time to wait between attempts to try to acquire
the cluster lock | 10 | long
| acquireLockIntervalUnit | The time unit for acquireLockInterval | SECONDS |
TimeUnit
-| acquireLeadershipBackoff | The time to wait before a cluster member
broadcasts acquisition of gaining cluster leadership. A value of 0 (the
default) disables the backoff wait period. This can be useful to introduce a
delay to ensure that the cluster leader has fully surrendered its leadership. A
sensible value is acquireLockDelay + 5 seconds. Or whatever is appropriate for
your environment and requirements | 0 | long
-| acquireLeadershipBackoffIntervalUnit | The time unit for
acquireLeadershipBackoff | SECONDS | TimeUnit
| heartbeatTimeoutMultiplier | Multiplier applied to the cluster leader
acquireLockInterval to determine how long followers should wait before
considering the leader "stale". For example, if the leader updates its
heartbeat every 2 seconds and the heartbeatTimeoutMultiplier is 3, followers
will tolerate up to code 2s * 3 = 6s of silence before declaring the leader
unavailable | 5 | int
| rootPath | The file cluster root directory path | | String
|===
