This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch 10.1.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/10.1.x by this push:
new 7a8335428a Review "Run TcpFailureDetector liveness checks outside the
... monitor"
7a8335428a is described below
commit 7a8335428a9ad7aa63fc0341cf988614b044d95b
Author: Mark Thomas <[email protected]>
AuthorDate: Wed Oct 21 17:46:05 2026 +0100
Review "Run TcpFailureDetector liveness checks outside the ... monitor"
Correctly handle events received during the liveness probe
Co-authored by: Claude / Sonnet 5.5
Co-authored by: CoPilot / GPT-5.6 Sol
---
.../group/interceptors/TcpFailureDetector.java | 240 ++++++++++++----
.../TestTcpFailureDetectorLivenessChecks.java | 304 +++++++++++++++++++++
2 files changed, 487 insertions(+), 57 deletions(-)
diff --git
a/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java
b/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java
index 161b482539..6a8a48c6a0 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java
@@ -22,9 +22,11 @@ import java.net.InetSocketAddress;
import java.net.NoRouteToHostException;
import java.net.Socket;
import java.net.SocketTimeoutException;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
import org.apache.catalina.tribes.Channel;
import org.apache.catalina.tribes.ChannelException;
@@ -102,6 +104,24 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
*/
protected Membership membership = null;
+ /**
+ * The sequence number of the last membership event of each member, used
to detect that a member event
+ * was processed while a liveness check of the same member was running.
Only populated while at least
+ * one liveness check is running. Guarded by {@link #membership}.
+ */
+ private final HashMap<Member,Long> lastMemberEvent = new HashMap<>();
+
+ /**
+ * The last membership event sequence number issued. Guarded by {@link
#membership}.
+ */
+ private long memberEventSequence = 0;
+
+ /**
+ * The number of liveness checks that have started and not yet been
applied. Guarded by
+ * {@link #membership}.
+ */
+ private int checksInFlight = 0;
+
/**
* Map of members suspected of being dead and their detection time.
*/
@@ -160,24 +180,28 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
}
boolean notify = false;
boolean verify = false;
+ long stamp = 0;
synchronized (membership) {
+ recordMemberEvent(member);
if (removeSuspects.containsKey(member)) {
// previously marked suspect, system below picked up the
member again
removeSuspects.remove(member);
} else if (membership.getMember(member) == null) {
verify = true;
+ stamp = beginCheck(member);
}
}
if (verify) {
// if we add it here, then add it upwards too
// check to see if it is alive, outside the membership monitor so
// that slow TCP connects do not block other membership operations
- boolean alive = memberAlive(member);
+ boolean alive = memberAliveForCheck(member, stamp);
synchronized (membership) {
// the state may have changed while the member was being
checked
- if (!removeSuspects.containsKey(member) &&
membership.getMember(member) == null) {
+ if (endCheck(member, stamp) && membership.getMember(member) ==
null) {
if (alive) {
membership.memberAlive(member);
+ recordMemberEvent(member);
addSuspects.remove(member);
notify = true;
} else if (member instanceof StaticMember) {
@@ -197,8 +221,10 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
setupMembership();
}
boolean shutdown = Arrays.equals(member.getCommand(),
Member.SHUTDOWN_PAYLOAD);
+ long stamp = 0;
if (shutdown) {
synchronized (membership) {
+ recordMemberEvent(member);
if (!membership.contains(member)) {
return;
}
@@ -212,7 +238,11 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
} else {
boolean verify;
synchronized (membership) {
+ recordMemberEvent(member);
verify = membership.contains(member);
+ if (verify) {
+ stamp = beginCheck(member);
+ }
}
if (!verify) {
if (log.isInfoEnabled()) {
@@ -228,17 +258,19 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
// check to see if the member really is gone, outside the
membership
// monitor so that slow TCP connects do not block other membership
// operations
- boolean alive = memberAlive(member);
+ boolean alive = memberAliveForCheck(member, stamp);
synchronized (membership) {
// the state may have changed while the member was being
checked
- if (membership.contains(member)) {
+ if (endCheck(member, stamp) && membership.contains(member)) {
applied = true;
if (alive) {
// add the member as suspect
removeSuspects.put(member,
Long.valueOf(System.currentTimeMillis()));
+ recordMemberEvent(member);
} else {
// not correct, we need to maintain the map
membership.removeMember(member);
+ recordMemberEvent(member);
removeSuspects.remove(member);
if (member instanceof StaticMember) {
addSuspects.put(member,
Long.valueOf(System.currentTimeMillis()));
@@ -321,18 +353,28 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
Member[] members = super.getMembers();
for (int i = 0; members != null && i < members.length; i++) {
Member member = members[i];
- boolean alive = memberAlive(member);
+ long stamp;
+ synchronized (membership) {
+ stamp = beginCheck(member);
+ }
+ boolean alive = memberAliveForCheck(member, stamp);
boolean added = false;
boolean disappeared = false;
synchronized (membership) {
+ if (!endCheck(member, stamp)) {
+ // a member event for this member was processed meanwhile
and wins
+ continue;
+ }
if (alive) {
if (membership.memberAlive(member)) {
added = true;
+ recordMemberEvent(member);
}
addSuspects.remove(member);
} else {
if (membership.getMember(member) != null) {
membership.removeMember(member);
+ recordMemberEvent(member);
removeSuspects.remove(member);
if (member instanceof StaticMember) {
addSuspects.put(member,
Long.valueOf(System.currentTimeMillis()));
@@ -358,38 +400,53 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
protected void performBasicCheck() {
// update all alive times
Member[] members = super.getMembers();
- ArrayList<Member> toCheck = new ArrayList<>();
- for (int i = 0; members != null && i < members.length; i++) {
- synchronized (membership) {
- if (addSuspects.containsKey(members[i]) &&
membership.getMember(members[i]) == null) {
- // avoid temporary adding member.
- continue;
+ // Members unknown to the tracker and the stamps of their running
checks. They are only added to the
+ // tracker once the check has found them alive, so an unverified
member is never visible and a stale
+ // result can simply be discarded.
+ Map<Member,Long> toCheck = new LinkedHashMap<>();
+ try {
+ for (int i = 0; members != null && i < members.length; i++) {
+ synchronized (membership) {
+ if (membership.getMember(members[i]) != null) {
+ // known member, just update its alive time
+ membership.memberAlive(members[i]);
+ } else if (!addSuspects.containsKey(members[i])) {
+ // we don't have this one in our membership, check to
see if the member is alive
+ // (members in addSuspects are not added temporarily,
see below)
+ toCheck.put(members[i],
Long.valueOf(beginCheck(members[i])));
+ }
}
- if (membership.memberAlive(members[i])) {
- // we don't have this one in our membership, check to see
if the member is alive
- toCheck.add(members[i]);
- } // end if
- }
- } // for
+ } // for
- // The liveness checks run outside the monitor, so a concurrent
memberDisappeared can
- // remove a member from the tracker between the two passes. The apply
below then skips
- // that member and the concurrent disappearance is the one that is
reported.
- for (Member member : toCheck) {
- boolean alive = memberAlive(member);
- boolean notify = false;
- synchronized (membership) {
- if (membership.getMember(member) != null) {
- if (alive) {
+ // The liveness checks run outside the monitor. A member event or
a change made by another check
+ // for the same member while it runs supersedes the result, which
is then discarded.
+ for (Iterator<Map.Entry<Member,Long>> it =
toCheck.entrySet().iterator(); it.hasNext();) {
+ Map.Entry<Member,Long> entry = it.next();
+ it.remove();
+ Member member = entry.getKey();
+ long stamp = entry.getValue().longValue();
+ boolean alive = memberAliveForCheck(member, stamp);
+ boolean notify = false;
+ synchronized (membership) {
+ if (endCheck(member, stamp) && alive &&
membership.getMember(member) == null) {
+ membership.memberAlive(member);
+ recordMemberEvent(member);
notify = true;
- } else {
- membership.removeMember(member);
- } // end if
+ }
+ }
+ if (notify) {
+
log.warn(sm.getString("tcpFailureDetector.performBasicCheck.memberAdded",
member));
+ super.memberAdded(member);
}
}
- if (notify) {
-
log.warn(sm.getString("tcpFailureDetector.performBasicCheck.memberAdded",
member));
- super.memberAdded(member);
+ } finally {
+ if (!toCheck.isEmpty()) {
+ // an exception interrupted the checks, release the ones that
did not run
+ synchronized (membership) {
+ for (Map.Entry<Member,Long> entry : toCheck.entrySet()) {
+ endCheck(entry.getKey(), entry.getValue().longValue());
+ }
+ }
}
}
@@ -401,27 +458,29 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
}
for (Member m : keys) {
boolean verify;
+ long stamp = 0;
synchronized (membership) {
verify = membership.getMember(m) != null;
+ if (verify) {
+ stamp = beginCheck(m);
+ }
}
boolean disappeared = false;
if (verify) {
- boolean alive = memberAlive(m);
- if (!alive) {
- synchronized (membership) {
- // the state may have changed while the member was
being checked
- if (membership.getMember(m) != null) {
- membership.removeMember(m);
- if (m instanceof StaticMember) {
- addSuspects.put(m,
Long.valueOf(System.currentTimeMillis()));
- }
- removeSuspects.remove(m);
- disappeared = true;
- }
- }
- } else {
- synchronized (membership) {
+ boolean alive = memberAliveForCheck(m, stamp);
+ synchronized (membership) {
+ // the state may have changed while the member was being
checked
+ boolean current = endCheck(m, stamp);
+ if (current && alive) {
expireRemoveSuspect(m);
+ } else if (current && membership.getMember(m) != null) {
+ membership.removeMember(m);
+ recordMemberEvent(m);
+ if (m instanceof StaticMember) {
+ addSuspects.put(m,
Long.valueOf(System.currentTimeMillis()));
+ }
+ removeSuspects.remove(m);
+ disappeared = true;
}
}
} else {
@@ -444,22 +503,25 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
}
for (Member m : keys) {
boolean verify;
+ long stamp = 0;
synchronized (membership) {
verify = membership.getMember(m) == null;
+ if (verify) {
+ stamp = beginCheck(m);
+ }
}
if (!verify) {
continue;
}
- boolean alive = memberAlive(m);
+ boolean alive = memberAliveForCheck(m, stamp);
boolean notify = false;
- if (alive) {
- synchronized (membership) {
- // the state may have changed while the member was being
checked
- if (membership.getMember(m) == null) {
- membership.memberAlive(m);
- addSuspects.remove(m);
- notify = true;
- }
+ synchronized (membership) {
+ // the state may have changed while the member was being
checked
+ if (endCheck(m, stamp) && alive && membership.getMember(m) ==
null) {
+ membership.memberAlive(m);
+ recordMemberEvent(m);
+ addSuspects.remove(m);
+ notify = true;
}
}
if (notify) {
@@ -471,6 +533,70 @@ public class TcpFailureDetector extends
ChannelInterceptorBase implements TcpFai
}
}
+ /**
+ * Records that a member event was processed or a change was made to the
membership state of the member. A
+ * liveness check of the same member that started earlier is then stale
and must not be applied, since the
+ * later event or change wins. Events of other members do not affect it.
Must be called while holding the
+ * membership monitor.
+ *
+ * @param member the member the event is for
+ */
+ private void recordMemberEvent(Member member) {
+ if (checksInFlight > 0) {
+ lastMemberEvent.put(member, Long.valueOf(++memberEventSequence));
+ }
+ }
+
+ /**
+ * Registers the start of a liveness check of the member. Must be called
while holding the membership
+ * monitor and followed by a call to {@link #endCheck(Member, long)} when
the result is applied.
+ *
+ * @param member the member to be checked
+ * @return the stamp to pass to {@link #endCheck(Member, long)}
+ */
+ private long beginCheck(Member member) {
+ checksInFlight++;
+ Long last = lastMemberEvent.get(member);
+ return last == null ? 0 : last.longValue();
+ }
+
+ /**
+ * Registers the end of a liveness check of the member and reports if its
result is still current. Must
+ * be called while holding the membership monitor.
+ *
+ * @param member the member that was checked
+ * @param stamp the stamp returned by {@link #beginCheck(Member)}
+ * @return {@code true} if no member event for the member was processed
since the check started
+ */
+ private boolean endCheck(Member member, long stamp) {
+ Long last = lastMemberEvent.get(member);
+ boolean current = (last == null ? 0 : last.longValue()) == stamp;
+ if (--checksInFlight == 0) {
+ // no running check can compare against the recorded events any
more
+ lastMemberEvent.clear();
+ }
+ return current;
+ }
+
+ /**
+ * Performs a registered liveness check and releases its registration if
the check fails before its result can be
+ * applied.
+ *
+ * @param member the member to check
+ * @param stamp the stamp returned by {@link #beginCheck(Member)}
+ * @return {@code true} if the member is alive
+ */
+ private boolean memberAliveForCheck(Member member, long stamp) {
+ try {
+ return memberAlive(member);
+ } catch (RuntimeException | Error e) {
+ synchronized (membership) {
+ endCheck(member, stamp);
+ }
+ throw e;
+ }
+ }
+
/**
* Removes the member from the remove suspects map if it has been suspect
for longer than the
* configured timeout. Must be called while holding the membership monitor.
diff --git
a/test/org/apache/catalina/tribes/group/interceptors/TestTcpFailureDetectorLivenessChecks.java
b/test/org/apache/catalina/tribes/group/interceptors/TestTcpFailureDetectorLivenessChecks.java
index 676435c472..03ca67b2e9 100644
---
a/test/org/apache/catalina/tribes/group/interceptors/TestTcpFailureDetectorLivenessChecks.java
+++
b/test/org/apache/catalina/tribes/group/interceptors/TestTcpFailureDetectorLivenessChecks.java
@@ -19,6 +19,10 @@ package org.apache.catalina.tribes.group.interceptors;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import org.junit.Assert;
import org.junit.Test;
@@ -71,6 +75,45 @@ public class TestTcpFailureDetectorLivenessChecks {
}
}
+ private static final int GATED_PORT = 4000;
+
+ private static class GatedAddDetector extends TcpFailureDetector {
+
+ private final CountDownLatch checkStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseCheck = new CountDownLatch(1);
+ private volatile boolean gatedResult = true;
+ // Result of the gated thread's checks after the first one, which was
held
+ private volatile boolean laterGatedResult = true;
+ private final AtomicInteger gatedChecks = new AtomicInteger();
+ // When set, only checks run by this thread are gated and all others
return otherResult
+ private volatile Thread gatedThread;
+ private volatile boolean otherResult = true;
+
+ @Override
+ protected boolean memberAlive(Member mbr) {
+ if (gatedThread != null) {
+ if (Thread.currentThread() != gatedThread) {
+ return otherResult;
+ }
+ } else if (mbr.getPort() != GATED_PORT) {
+ return true;
+ }
+ if (gatedChecks.incrementAndGet() > 1 && gatedThread != null) {
+ return laterGatedResult;
+ }
+ checkStarted.countDown();
+ try {
+ if (!releaseCheck.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting to
release liveness check");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ return gatedResult;
+ }
+ }
+
// Downstream stand-in: provides the channel membership view the detector
checks
private static class FakeMembershipChain extends ChannelInterceptorBase {
@@ -201,4 +244,265 @@ public class TestTcpFailureDetectorLivenessChecks {
detector.assertNoCheckRanUnderMonitor();
}
+
+ @Test
+ public void testShutdownInvalidatesInFlightAddCheck() throws Exception {
+ FakeMembershipChain chain = new FakeMembershipChain();
+ RecordingInterceptor recorder = new RecordingInterceptor();
+ GatedAddDetector detector = new GatedAddDetector();
+ detector.setChannel(new GroupChannel());
+ detector.setNext(chain);
+ detector.setPrevious(recorder);
+ chain.localMember = new StaticMember("localhost", 5000, -1);
+ StaticMember member = new StaticMember("localhost", 4000, -1);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ Thread addThread = new Thread(() -> {
+ try {
+ detector.memberAdded(member);
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ addThread.start();
+ Assert.assertTrue("Liveness check did not start",
+ detector.checkStarted.await(10, TimeUnit.SECONDS));
+
+ member.setCommand(Member.SHUTDOWN_PAYLOAD);
+ detector.memberDisappeared(member);
+ detector.releaseCheck.countDown();
+ addThread.join(10000);
+
+ Assert.assertFalse("Member add thread did not finish",
addThread.isAlive());
+ Assert.assertNull("Member add failed unexpectedly", failure.get());
+ Assert.assertNull("A stale successful check must not resurrect a
member after shutdown",
+ detector.getMember(member));
+ Assert.assertTrue("A stale add must not be forwarded",
recorder.added.isEmpty());
+ }
+
+ @Test
+ public void testEventOfOtherMemberDoesNotInvalidateAddCheck() throws
Exception {
+ FakeMembershipChain chain = new FakeMembershipChain();
+ RecordingInterceptor recorder = new RecordingInterceptor();
+ GatedAddDetector detector = new GatedAddDetector();
+ detector.setChannel(new GroupChannel());
+ detector.setNext(chain);
+ detector.setPrevious(recorder);
+ chain.localMember = new StaticMember("localhost", 5000, -1);
+ StaticMember gated = new StaticMember("localhost", GATED_PORT, -1);
+ StaticMember other = new StaticMember("localhost", 4001, -1);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ Thread addThread = new Thread(() -> {
+ try {
+ detector.memberAdded(gated);
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ addThread.start();
+ Assert.assertTrue("Liveness check did not start",
+ detector.checkStarted.await(10, TimeUnit.SECONDS));
+
+ // Events of other members, including a disappearance, are processed
meanwhile
+ detector.memberAdded(other);
+ detector.memberDisappeared(other);
+ detector.releaseCheck.countDown();
+ addThread.join(10000);
+
+ Assert.assertFalse("Member add thread did not finish",
addThread.isAlive());
+ Assert.assertNull("Member add failed unexpectedly", failure.get());
+ Assert.assertNotNull("An event of another member must not drop the
add", detector.getMember(gated));
+ Assert.assertTrue("The add must be forwarded",
recorder.added.contains(gated));
+ }
+
+ @Test
+ public void testLaterEventOfSameMemberInvalidatesAddCheck() throws
Exception {
+ FakeMembershipChain chain = new FakeMembershipChain();
+ RecordingInterceptor recorder = new RecordingInterceptor();
+ GatedAddDetector detector = new GatedAddDetector();
+ detector.setChannel(new GroupChannel());
+ detector.setNext(chain);
+ detector.setPrevious(recorder);
+ chain.localMember = new StaticMember("localhost", 5000, -1);
+ StaticMember member = new StaticMember("localhost", GATED_PORT, -1);
+
+ Thread addThread = new Thread(() -> detector.memberAdded(member));
+ addThread.start();
+ Assert.assertTrue("Liveness check did not start",
+ detector.checkStarted.await(10, TimeUnit.SECONDS));
+
+ // A non-shutdown disappearance of a member not yet in the table is
ignored but still
+ // supersedes the running add check
+ detector.memberDisappeared(member);
+ detector.releaseCheck.countDown();
+ addThread.join(10000);
+
+ Assert.assertFalse("Member add thread did not finish",
addThread.isAlive());
+ Assert.assertNull("A later event for the member supersedes the running
check",
+ detector.getMember(member));
+ Assert.assertTrue("A superseded add must not be forwarded",
recorder.added.isEmpty());
+ }
+
+ /*
+ * Runs a basic check whose probe of the member is held, while the same
member is reported by the
+ * membership service, which runs its own (ungated) check with the given
result. Then the held probe is
+ * released with its own result.
+ */
+ private void runBasicCheckWithConcurrentAdd(boolean probeAlive, boolean
eventAlive, GatedAddDetector detector,
+ RecordingInterceptor recorder, FakeMembershipChain chain,
StaticMember member) throws Exception {
+ detector.setChannel(new GroupChannel());
+ detector.setNext(chain);
+ detector.setPrevious(recorder);
+ chain.localMember = new StaticMember("localhost", 5000, -1);
+ chain.members = new Member[] { member };
+ detector.gatedResult = probeAlive;
+ // the same heartbeat re-checks add suspects after the held probe,
which must agree with the event
+ detector.laterGatedResult = eventAlive;
+ detector.otherResult = eventAlive;
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ Thread checkThread = new Thread(() -> {
+ try {
+ detector.checkMembers(false);
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ detector.gatedThread = checkThread;
+ checkThread.start();
+ Assert.assertTrue("Liveness check did not start",
+ detector.checkStarted.await(10, TimeUnit.SECONDS));
+
+ // The basic check is verifying the member, which must not be visible
until verified
+ Assert.assertNull("An unverified member must not be in the table",
detector.getMember(member));
+ detector.memberAdded(member);
+ detector.releaseCheck.countDown();
+ checkThread.join(10000);
+
+ Assert.assertFalse("Check thread did not finish",
checkThread.isAlive());
+ Assert.assertNull("Check failed unexpectedly", failure.get());
+ }
+
+ @Test
+ public void testBasicCheckAddsVerifiedMemberOnlyAfterProbe() throws
Exception {
+ FakeMembershipChain chain = new FakeMembershipChain();
+ RecordingInterceptor recorder = new RecordingInterceptor();
+ GatedAddDetector detector = new GatedAddDetector();
+ detector.setChannel(new GroupChannel());
+ detector.setNext(chain);
+ detector.setPrevious(recorder);
+ chain.localMember = new StaticMember("localhost", 5000, -1);
+ StaticMember member = new StaticMember("localhost", GATED_PORT, -1);
+ chain.members = new Member[] { member };
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ Thread checkThread = new Thread(() -> {
+ try {
+ detector.checkMembers(false);
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ detector.gatedThread = checkThread;
+ checkThread.start();
+ Assert.assertTrue("Liveness check did not start",
+ detector.checkStarted.await(10, TimeUnit.SECONDS));
+ Assert.assertNull("An unverified member must not be in the table",
detector.getMember(member));
+ Assert.assertTrue("An unverified member must not be reported",
recorder.added.isEmpty());
+
+ detector.releaseCheck.countDown();
+ checkThread.join(10000);
+
+ Assert.assertFalse("Check thread did not finish",
checkThread.isAlive());
+ Assert.assertNull("Check failed unexpectedly", failure.get());
+ Assert.assertNotNull("A verified member must be added",
detector.getMember(member));
+ Assert.assertEquals("A verified member must be reported once", 1,
recorder.added.size());
+ }
+
+ @Test
+ public void testBasicCheckDeadProbeDoesNotOverrideLaterAdd() throws
Exception {
+ FakeMembershipChain chain = new FakeMembershipChain();
+ RecordingInterceptor recorder = new RecordingInterceptor();
+ GatedAddDetector detector = new GatedAddDetector();
+ StaticMember member = new StaticMember("localhost", GATED_PORT, -1);
+
+ runBasicCheckWithConcurrentAdd(false, true, detector, recorder, chain,
member);
+
+ Assert.assertNotNull("The later add must win over the older dead
probe", detector.getMember(member));
+ Assert.assertEquals("The member must be reported exactly once", 1,
recorder.added.size());
+ }
+
+ @Test
+ public void testBasicCheckLiveProbeDoesNotOverrideLaterAdd() throws
Exception {
+ FakeMembershipChain chain = new FakeMembershipChain();
+ RecordingInterceptor recorder = new RecordingInterceptor();
+ GatedAddDetector detector = new GatedAddDetector();
+ StaticMember member = new StaticMember("localhost", GATED_PORT, -1);
+
+ runBasicCheckWithConcurrentAdd(true, true, detector, recorder, chain,
member);
+
+ Assert.assertNotNull("The member must stay in the table",
detector.getMember(member));
+ Assert.assertEquals("The member must be reported exactly once, by the
later add", 1,
+ recorder.added.size());
+ }
+
+ @Test
+ public void testBasicCheckLiveProbeDoesNotOverrideLaterFailedAdd() throws
Exception {
+ FakeMembershipChain chain = new FakeMembershipChain();
+ RecordingInterceptor recorder = new RecordingInterceptor();
+ GatedAddDetector detector = new GatedAddDetector();
+ StaticMember member = new StaticMember("localhost", GATED_PORT, -1);
+
+ runBasicCheckWithConcurrentAdd(true, false, detector, recorder, chain,
member);
+
+ Assert.assertNull("The later failed add must win over the older live
probe", detector.getMember(member));
+ Assert.assertTrue("A superseded probe must not be reported",
recorder.added.isEmpty());
+ Assert.assertTrue("The failed static member must be an add suspect",
+ detector.addSuspects.containsKey(member));
+ }
+
+ @Test
+ public void testHeartbeatChangesInvalidateInFlightAddCheck() throws
Exception {
+ FakeMembershipChain chain = new FakeMembershipChain();
+ RecordingInterceptor recorder = new RecordingInterceptor();
+ GatedAddDetector detector = new GatedAddDetector();
+ detector.setChannel(new GroupChannel());
+ detector.setNext(chain);
+ detector.setPrevious(recorder);
+ chain.localMember = new StaticMember("localhost", 5000, -1);
+ StaticMember member = new StaticMember("localhost", GATED_PORT, -1);
+ chain.members = new Member[] { member };
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ Thread addThread = new Thread(() -> {
+ try {
+ detector.memberAdded(member);
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ detector.gatedThread = addThread;
+ addThread.start();
+ Assert.assertTrue("Liveness check did not start",
+ detector.checkStarted.await(10, TimeUnit.SECONDS));
+
+ // Forced heartbeat checks add the member and then find it dead again,
while the
+ // check started by the member event is still running
+ detector.otherResult = true;
+ detector.checkMembers(true);
+ Assert.assertNotNull("The forced check must add the member",
detector.getMember(member));
+ detector.otherResult = false;
+ detector.checkMembers(true);
+ Assert.assertNull("The forced check must remove the member",
detector.getMember(member));
+
+ detector.releaseCheck.countDown();
+ addThread.join(10000);
+
+ Assert.assertFalse("Member add thread did not finish",
addThread.isAlive());
+ Assert.assertNull("Member add failed unexpectedly", failure.get());
+ Assert.assertNull("A stale live result must not resurrect a member the
heartbeat found dead",
+ detector.getMember(member));
+ Assert.assertEquals("Only the forced check may have reported the
member", 1, recorder.added.size());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
