diff --git a/raft/src/main/java/org/apache/kafka/raft/LeaderState.java b/raft/src/main/java/org/apache/kafka/raft/LeaderState.java index 0f0f728f7c04f..d939aacbb1436 100644 --- a/raft/src/main/java/org/apache/kafka/raft/LeaderState.java +++ b/raft/src/main/java/org/apache/kafka/raft/LeaderState.java @@ -16,6 +16,9 @@ */ package org.apache.kafka.raft; +import org.apache.kafka.common.utils.LogContext; +import org.slf4j.Logger; + import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; @@ -33,22 +36,25 @@ * they acknowledge the leader. */ public class LeaderState implements EpochState { + static final long OBSERVER_SESSION_TIMEOUT_MS = 300_000L; + private final int localId; private final int epoch; private final long epochStartOffset; private Optional highWatermark; - private final Map voterReplicaStates = new HashMap<>(); - private final Map observerReplicaStates = new HashMap<>(); + private final Map voterStates = new HashMap<>(); + private final Map observerStates = new HashMap<>(); private final Set grantingVoters = new HashSet<>(); - private static final long OBSERVER_SESSION_TIMEOUT_MS = 300_000L; + private final Logger log; protected LeaderState( int localId, int epoch, long epochStartOffset, Set voters, - Set grantingVoters + Set grantingVoters, + LogContext logContext ) { this.localId = localId; this.epoch = epoch; @@ -57,9 +63,10 @@ protected LeaderState( for (int voterId : voters) { boolean hasAcknowledgedLeader = voterId == localId; - this.voterReplicaStates.put(voterId, new VoterState(voterId, hasAcknowledgedLeader)); + this.voterStates.put(voterId, new ReplicaState(voterId, hasAcknowledgedLeader)); } this.grantingVoters.addAll(grantingVoters); + this.log = logContext.logger(LeaderState.class); } @Override @@ -69,7 +76,7 @@ public Optional highWatermark() { @Override public ElectionState election() { - return ElectionState.withElectedLeader(epoch, localId, voterReplicaStates.keySet()); + return ElectionState.withElectedLeader(epoch, localId, voterStates.keySet()); } @Override @@ -78,7 +85,7 @@ public int epoch() { } public Set followers() { - return voterReplicaStates.keySet().stream().filter(id -> id != localId).collect(Collectors.toSet()); + return voterStates.keySet().stream().filter(id -> id != localId).collect(Collectors.toSet()); } public Set grantingVoters() { @@ -91,7 +98,7 @@ public int localId() { public Set nonAcknowledgingVoters() { Set nonAcknowledging = new HashSet<>(); - for (VoterState state : voterReplicaStates.values()) { + for (ReplicaState state : voterStates.values()) { if (!state.hasAcknowledgedLeader) nonAcknowledging.add(state.nodeId); } @@ -100,9 +107,9 @@ public Set nonAcknowledgingVoters() { private boolean updateHighWatermark() { // Find the largest offset which is replicated to a majority of replicas (the leader counts) - List followersByDescendingFetchOffset = followersByDescendingFetchOffset(); + List followersByDescendingFetchOffset = followersByDescendingFetchOffset(); - int indexOfHw = voterReplicaStates.size() / 2; + int indexOfHw = voterStates.size() / 2; Optional highWatermarkUpdateOpt = followersByDescendingFetchOffset.get(indexOfHw).endOffset; if (highWatermarkUpdateOpt.isPresent()) { @@ -110,16 +117,22 @@ private boolean updateHighWatermark() { // leader. In order to avoid exposing a non-monotonically increasing value, we have // to wait for followers to catch up to the start of the leader's epoch. LogOffsetMetadata highWatermarkUpdateMetadata = highWatermarkUpdateOpt.get(); - long highWatermarkUpdate = highWatermarkUpdateMetadata.offset; + long highWatermarkUpdateOffset = highWatermarkUpdateMetadata.offset; - if (highWatermarkUpdate >= epochStartOffset) { + if (highWatermarkUpdateOffset >= epochStartOffset) { if (highWatermark.isPresent()) { LogOffsetMetadata currentHighWatermarkMetadata = highWatermark.get(); - if (highWatermarkUpdate > currentHighWatermarkMetadata.offset - || (highWatermarkUpdate == currentHighWatermarkMetadata.offset && + if (highWatermarkUpdateOffset > currentHighWatermarkMetadata.offset + || (highWatermarkUpdateOffset == currentHighWatermarkMetadata.offset && !highWatermarkUpdateMetadata.metadata.equals(currentHighWatermarkMetadata.metadata))) { highWatermark = highWatermarkUpdateOpt; return true; + } else if (highWatermarkUpdateOffset < currentHighWatermarkMetadata.offset) { + log.error("The latest computed high watermark {} is smaller than the current " + + "value {}, which suggests that one of the voters has lost committed data. " + + "Full voter replication state: {}", highWatermarkUpdateOffset, + currentHighWatermarkMetadata.offset, voterStates.values()); + return false; } else { return false; } @@ -170,8 +183,8 @@ public List nonLeaderVotersByDescendingFetchOffset() { .collect(Collectors.toList()); } - private List followersByDescendingFetchOffset() { - return new ArrayList<>(this.voterReplicaStates.values()).stream() + private List followersByDescendingFetchOffset() { + return new ArrayList<>(this.voterStates.values()).stream() .sorted() .collect(Collectors.toList()); } @@ -179,27 +192,29 @@ private List followersByDescendingFetchOffset() { private boolean updateEndOffset(ReplicaState state, LogOffsetMetadata endOffsetMetadata) { state.endOffset.ifPresent(currentEndOffset -> { - if (currentEndOffset.offset > endOffsetMetadata.offset) - throw new IllegalArgumentException("Non-monotonic update to end offset for nodeId " + state.nodeId); + if (currentEndOffset.offset > endOffsetMetadata.offset) { + if (state.nodeId == localId) { + throw new IllegalStateException("Detected non-monotonic update of local " + + "end offset: " + currentEndOffset.offset + " -> " + endOffsetMetadata.offset); + } else { + log.warn("Detected non-monotonic update of fetch offset from nodeId {}: {} -> {}", + state.nodeId, currentEndOffset.offset, endOffsetMetadata.offset); + } + } }); state.endOffset = Optional.of(endOffsetMetadata); - - if (isVoter(state.nodeId)) { - ((VoterState) state).hasAcknowledgedLeader = true; - addAcknowledgementFrom(state.nodeId); - return updateHighWatermark(); - } - return false; + state.hasAcknowledgedLeader = true; + return isVoter(state.nodeId) && updateHighWatermark(); } public void addAcknowledgementFrom(int remoteNodeId) { - VoterState voterState = ensureValidVoter(remoteNodeId); + ReplicaState voterState = ensureValidVoter(remoteNodeId); voterState.hasAcknowledgedLeader = true; } - private VoterState ensureValidVoter(int remoteNodeId) { - VoterState state = voterReplicaStates.get(remoteNodeId); + private ReplicaState ensureValidVoter(int remoteNodeId) { + ReplicaState state = voterStates.get(remoteNodeId); if (state == null) throw new IllegalArgumentException("Unexpected acknowledgement from non-voter " + remoteNodeId); return state; @@ -209,22 +224,22 @@ public long epochStartOffset() { return epochStartOffset; } - ReplicaState getReplicaState(int remoteNodeId) { - ReplicaState state = voterReplicaStates.get(remoteNodeId); + private ReplicaState getReplicaState(int remoteNodeId) { + ReplicaState state = voterStates.get(remoteNodeId); if (state == null) { - observerReplicaStates.putIfAbsent(remoteNodeId, new ReplicaState(remoteNodeId)); - return observerReplicaStates.get(remoteNodeId); + observerStates.putIfAbsent(remoteNodeId, new ReplicaState(remoteNodeId, false)); + return observerStates.get(remoteNodeId); } return state; } Map getVoterEndOffsets() { - return getReplicaEndOffsets(voterReplicaStates); + return getReplicaEndOffsets(voterStates); } Map getObserverStates(final long currentTimeMs) { clearInactiveObservers(currentTimeMs); - return getReplicaEndOffsets(observerReplicaStates); + return getReplicaEndOffsets(observerStates); } private static Map getReplicaEndOffsets( @@ -237,25 +252,27 @@ private static Map getReplicaEndOffsets( } private void clearInactiveObservers(final long currentTimeMs) { - observerReplicaStates.entrySet().removeIf( + observerStates.entrySet().removeIf( integerReplicaStateEntry -> currentTimeMs - integerReplicaStateEntry.getValue().lastFetchTimestamp.orElse(-1) >= OBSERVER_SESSION_TIMEOUT_MS); } private boolean isVoter(int remoteNodeId) { - return voterReplicaStates.containsKey(remoteNodeId); + return voterStates.containsKey(remoteNodeId); } private static class ReplicaState implements Comparable { final int nodeId; Optional endOffset; OptionalLong lastFetchTimestamp; + boolean hasAcknowledgedLeader; - public ReplicaState(int nodeId) { + public ReplicaState(int nodeId, boolean hasAcknowledgedLeader) { this.nodeId = nodeId; this.endOffset = Optional.empty(); this.lastFetchTimestamp = OptionalLong.empty(); + this.hasAcknowledgedLeader = hasAcknowledgedLeader; } void updateFetchTimestamp(long currentFetchTimeMs) { @@ -275,15 +292,15 @@ else if (!that.endOffset.isPresent()) else return Long.compare(that.endOffset.get().offset, this.endOffset.get().offset); } - } - private static class VoterState extends ReplicaState { - boolean hasAcknowledgedLeader; - - public VoterState(int nodeId, - boolean hasAcknowledgedLeader) { - super(nodeId); - this.hasAcknowledgedLeader = hasAcknowledgedLeader; + @Override + public String toString() { + return "ReplicaState(" + + "nodeId=" + nodeId + + ", endOffset=" + endOffset + + ", lastFetchTimestamp=" + lastFetchTimestamp + + ", hasAcknowledgedLeader=" + hasAcknowledgedLeader + + ')'; } } diff --git a/raft/src/main/java/org/apache/kafka/raft/QuorumState.java b/raft/src/main/java/org/apache/kafka/raft/QuorumState.java index 70c8ed0bda7f7..6216d0f5847e2 100644 --- a/raft/src/main/java/org/apache/kafka/raft/QuorumState.java +++ b/raft/src/main/java/org/apache/kafka/raft/QuorumState.java @@ -81,6 +81,7 @@ public class QuorumState { private final Random random; private final int electionTimeoutMs; private final int fetchTimeoutMs; + private final LogContext logContext; private volatile EpochState state; @@ -100,6 +101,7 @@ public QuorumState(OptionalInt localId, this.time = time; this.log = logContext.logger(QuorumState.class); this.random = random; + this.logContext = logContext; } public void initialize(OffsetAndEpoch logEndOffsetAndEpoch) throws IOException, IllegalStateException { @@ -437,7 +439,8 @@ public void transitionToLeader(long epochStartOffset) throws IOException { epoch(), epochStartOffset, voters, - candidateState.grantingVoters() + candidateState.grantingVoters(), + logContext )); } diff --git a/raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java b/raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java index 8ec91a7032266..cdadf6e00123a 100644 --- a/raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java +++ b/raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java @@ -16,13 +16,18 @@ */ package org.apache.kafka.raft; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.MockTime; import org.junit.jupiter.api.Test; import java.util.Arrays; import java.util.Collections; import java.util.Objects; import java.util.Optional; +import java.util.Set; +import static java.util.Collections.emptySet; +import static java.util.Collections.singleton; import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; import static org.apache.kafka.common.utils.Utils.mkSet; @@ -34,47 +39,62 @@ public class LeaderStateTest { private final int localId = 0; private final int epoch = 5; + private final LogContext logContext = new LogContext(); + + private LeaderState newLeaderState( + Set voters, + long epochStartOffset + ) { + return new LeaderState( + localId, + epoch, + epochStartOffset, + voters, + voters, + logContext + ); + } @Test public void testFollowerAcknowledgement() { int node1 = 1; int node2 = 2; - LeaderState state = new LeaderState(localId, epoch, 0L, mkSet(localId, node1, node2), Collections.emptySet()); + LeaderState state = newLeaderState(mkSet(localId, node1, node2), 0L); assertEquals(mkSet(node1, node2), state.nonAcknowledgingVoters()); state.addAcknowledgementFrom(node1); - assertEquals(Collections.singleton(node2), state.nonAcknowledgingVoters()); + assertEquals(singleton(node2), state.nonAcknowledgingVoters()); state.addAcknowledgementFrom(node2); - assertEquals(Collections.emptySet(), state.nonAcknowledgingVoters()); + assertEquals(emptySet(), state.nonAcknowledgingVoters()); } @Test public void testNonFollowerAcknowledgement() { int nonVoterId = 1; - LeaderState state = new LeaderState(localId, epoch, 0L, Collections.singleton(localId), Collections.emptySet()); + LeaderState state = newLeaderState(singleton(localId), 0L); assertThrows(IllegalArgumentException.class, () -> state.addAcknowledgementFrom(nonVoterId)); } @Test public void testUpdateHighWatermarkQuorumSizeOne() { - LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId), Collections.emptySet()); + LeaderState state = newLeaderState(singleton(localId), 15L); assertEquals(Optional.empty(), state.highWatermark()); assertTrue(state.updateLocalState(0, new LogOffsetMetadata(15L))); assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark()); } @Test - public void testNonMonotonicEndOffsetUpdate() { - LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId), Collections.emptySet()); + public void testNonMonotonicLocalEndOffsetUpdate() { + LeaderState state = newLeaderState(singleton(localId), 15L); assertEquals(Optional.empty(), state.highWatermark()); assertTrue(state.updateLocalState(0, new LogOffsetMetadata(15L))); assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark()); - assertThrows(IllegalArgumentException.class, + assertThrows(IllegalStateException.class, () -> state.updateLocalState(0, new LogOffsetMetadata(14L))); } @Test public void testIdempotentEndOffsetUpdate() { - LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId), Collections.emptySet()); + LeaderState state = newLeaderState(singleton(localId), 15L); assertEquals(Optional.empty(), state.highWatermark()); assertTrue(state.updateLocalState(0, new LogOffsetMetadata(15L))); assertFalse(state.updateLocalState(0, new LogOffsetMetadata(15L))); @@ -83,7 +103,7 @@ public void testIdempotentEndOffsetUpdate() { @Test public void testUpdateHighWatermarkMetadata() { - LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId), Collections.emptySet()); + LeaderState state = newLeaderState(singleton(localId), 15L); assertEquals(Optional.empty(), state.highWatermark()); LogOffsetMetadata initialHw = new LogOffsetMetadata(15L, Optional.of(new MockOffsetMetadata("foo"))); @@ -98,11 +118,11 @@ public void testUpdateHighWatermarkMetadata() { @Test public void testUpdateHighWatermarkQuorumSizeTwo() { int otherNodeId = 1; - LeaderState state = new LeaderState(localId, epoch, 10L, mkSet(localId, otherNodeId), Collections.emptySet()); + LeaderState state = newLeaderState(mkSet(localId, otherNodeId), 10L); assertFalse(state.updateLocalState(0, new LogOffsetMetadata(15L))); assertEquals(Optional.empty(), state.highWatermark()); assertTrue(state.updateReplicaState(otherNodeId, 0, new LogOffsetMetadata(10L))); - assertEquals(Collections.emptySet(), state.nonAcknowledgingVoters()); + assertEquals(emptySet(), state.nonAcknowledgingVoters()); assertEquals(Optional.of(new LogOffsetMetadata(10L)), state.highWatermark()); assertTrue(state.updateReplicaState(otherNodeId, 0, new LogOffsetMetadata(15L))); assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark()); @@ -111,7 +131,7 @@ public void testUpdateHighWatermarkQuorumSizeTwo() { @Test public void testHighWatermarkUnknownUntilStartOfLeaderEpoch() { int otherNodeId = 1; - LeaderState state = new LeaderState(localId, epoch, 15L, mkSet(localId, otherNodeId), Collections.emptySet()); + LeaderState state = newLeaderState(mkSet(localId, otherNodeId), 15L); assertFalse(state.updateLocalState(0, new LogOffsetMetadata(20L))); assertEquals(Optional.empty(), state.highWatermark()); assertFalse(state.updateReplicaState(otherNodeId, 0, new LogOffsetMetadata(10L))); @@ -124,11 +144,11 @@ public void testHighWatermarkUnknownUntilStartOfLeaderEpoch() { public void testUpdateHighWatermarkQuorumSizeThree() { int node1 = 1; int node2 = 2; - LeaderState state = new LeaderState(localId, epoch, 10L, mkSet(localId, node1, node2), Collections.emptySet()); + LeaderState state = newLeaderState(mkSet(localId, node1, node2), 10L); assertFalse(state.updateLocalState(0, new LogOffsetMetadata(15L))); assertEquals(Optional.empty(), state.highWatermark()); assertTrue(state.updateReplicaState(node1, 0, new LogOffsetMetadata(10L))); - assertEquals(Collections.singleton(node2), state.nonAcknowledgingVoters()); + assertEquals(singleton(node2), state.nonAcknowledgingVoters()); assertEquals(Optional.of(new LogOffsetMetadata(10L)), state.highWatermark()); assertTrue(state.updateReplicaState(node2, 0, new LogOffsetMetadata(15L))); assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark()); @@ -140,6 +160,22 @@ public void testUpdateHighWatermarkQuorumSizeThree() { assertEquals(Optional.of(new LogOffsetMetadata(20L)), state.highWatermark()); } + @Test + public void testNonMonotonicHighWatermarkUpdate() { + MockTime time = new MockTime(); + int node1 = 1; + LeaderState state = newLeaderState(mkSet(localId, node1), 0L); + state.updateLocalState(time.milliseconds(), new LogOffsetMetadata(10L)); + state.updateReplicaState(node1, time.milliseconds(), new LogOffsetMetadata(10L)); + assertEquals(Optional.of(new LogOffsetMetadata(10L)), state.highWatermark()); + + // Follower crashes and disk is lost. It fetches an earlier offset to rebuild state. + // The leader will report an error in the logs, but will not let the high watermark rewind + assertFalse(state.updateReplicaState(node1, time.milliseconds(), new LogOffsetMetadata(5L))); + assertEquals(5L, state.getVoterEndOffsets().get(node1)); + assertEquals(Optional.of(new LogOffsetMetadata(10L)), state.highWatermark()); + } + @Test public void testGetNonLeaderFollowersByFetchOffsetDescending() { int node1 = 1; @@ -173,7 +209,7 @@ private LeaderState setUpLeaderAndFollowers(int follower1, int follower2, long leaderStartOffset, long leaderEndOffset) { - LeaderState state = new LeaderState(localId, epoch, leaderStartOffset, mkSet(localId, follower1, follower2), Collections.emptySet()); + LeaderState state = newLeaderState(mkSet(localId, follower1, follower2), leaderStartOffset); state.updateLocalState(0, new LogOffsetMetadata(leaderEndOffset)); assertEquals(Optional.empty(), state.highWatermark()); state.updateReplicaState(follower1, 0, new LogOffsetMetadata(leaderStartOffset)); @@ -184,26 +220,40 @@ private LeaderState setUpLeaderAndFollowers(int follower1, @Test public void testGetObserverStatesWithObserver() { int observerId = 10; - long endOffset = 10L; + long epochStartOffset = 10L; - LeaderState state = new LeaderState(localId, epoch, endOffset, mkSet(localId), Collections.emptySet()); + LeaderState state = newLeaderState(mkSet(localId), epochStartOffset); long timestamp = 20L; - assertFalse(state.updateReplicaState(observerId, timestamp, new LogOffsetMetadata(endOffset))); + assertFalse(state.updateReplicaState(observerId, timestamp, new LogOffsetMetadata(epochStartOffset))); - assertEquals(Collections.singletonMap(observerId, endOffset), state.getObserverStates(timestamp)); + assertEquals(Collections.singletonMap(observerId, epochStartOffset), state.getObserverStates(timestamp)); } @Test public void testNoOpForNegativeRemoteNodeId() { int observerId = -1; - long endOffset = 10L; + long epochStartOffset = 10L; - LeaderState state = new LeaderState(localId, epoch, endOffset, mkSet(localId), Collections.emptySet()); - assertFalse(state.updateReplicaState(observerId, 0, new LogOffsetMetadata(endOffset))); + LeaderState state = newLeaderState(mkSet(localId), epochStartOffset); + assertFalse(state.updateReplicaState(observerId, 0, new LogOffsetMetadata(epochStartOffset))); assertEquals(Collections.emptyMap(), state.getObserverStates(10)); } + @Test + public void testObserverStateExpiration() { + MockTime time = new MockTime(); + int observerId = 10; + long epochStartOffset = 10L; + LeaderState state = newLeaderState(mkSet(localId), epochStartOffset); + + state.updateReplicaState(observerId, time.milliseconds(), new LogOffsetMetadata(epochStartOffset)); + assertEquals(singleton(observerId), state.getObserverStates(time.milliseconds()).keySet()); + + time.sleep(LeaderState.OBSERVER_SESSION_TIMEOUT_MS); + assertEquals(emptySet(), state.getObserverStates(time.milliseconds()).keySet()); + } + private static class MockOffsetMetadata implements OffsetMetadata { private final String value; diff --git a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java index 00d6e0aec0ede..223c2694c767d 100644 --- a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java +++ b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java @@ -304,6 +304,45 @@ void canMakeProgressAfterBackToBackLeaderFailures( scheduler.runUntil(() -> cluster.anyReachedHighWatermark(targetHighWatermark)); } + @Property(tries = 100) + void canRecoverFromSingleNodeCommittedDataLoss( + @ForAll Random random, + @ForAll @IntRange(min = 3, max = 5) int numVoters, + @ForAll @IntRange(min = 0, max = 2) int numObservers + ) { + // We run this test without the `MonotonicEpoch` and `MajorityReachedHighWatermark` + // invariants since the loss of committed data on one node can violate them. + + Cluster cluster = new Cluster(numVoters, numObservers, random); + EventScheduler scheduler = new EventScheduler(cluster.random, cluster.time); + scheduler.addInvariant(new MonotonicHighWatermark(cluster)); + scheduler.addInvariant(new SingleLeader(cluster)); + scheduler.addValidation(new ConsistentCommittedData(cluster)); + + MessageRouter router = new MessageRouter(cluster); + + cluster.startAll(); + schedulePolling(scheduler, cluster, 3, 5); + scheduler.schedule(router::deliverAll, 0, 2, 5); + scheduler.schedule(new SequentialAppendAction(cluster), 0, 2, 3); + scheduler.runUntil(() -> cluster.anyReachedHighWatermark(10)); + + RaftNode node = cluster.randomRunning().orElseThrow(() -> + new AssertionError("Failed to find running node") + ); + + // Kill a random node and drop all of its persistent state. The Raft + // protocol guarantees should still ensure we lose no committed data + // as long as a new leader is elected before the failed node is restarted. + cluster.killAndDeletePersistentState(node.nodeId); + scheduler.runUntil(() -> !cluster.hasLeader(node.nodeId) && cluster.hasConsistentLeader()); + + // Now restart the failed node and ensure that it recovers. + long highWatermarkBeforeRestart = cluster.maxHighWatermarkReached(); + cluster.start(node.nodeId); + scheduler.runUntil(() -> cluster.allReachedHighWatermark(highWatermarkBeforeRestart + 10)); + } + private EventScheduler schedulerWithDefaultInvariants(Cluster cluster) { EventScheduler scheduler = new EventScheduler(cluster.random, cluster.time); scheduler.addInvariant(new MonotonicHighWatermark(cluster)); @@ -489,10 +528,6 @@ int majoritySize() { return voters.size() / 2 + 1; } - Set voters() { - return voters; - } - OptionalLong leaderHighWatermark() { Optional leaderWithMaxEpoch = running.values().stream().filter(node -> node.client.quorum().isLeader()) .max((node1, node2) -> Integer.compare(node2.client.quorum().epoch(), node1.client.quorum().epoch())); @@ -533,6 +568,11 @@ boolean allReachedHighWatermark(long offset) { .allMatch(node -> node.highWatermark() > offset); } + boolean hasLeader(int nodeId) { + OptionalInt latestLeader = latestLeader(); + return latestLeader.isPresent() && latestLeader.getAsInt() == nodeId; + } + OptionalInt latestLeader() { OptionalInt latestLeader = OptionalInt.empty(); int latestEpoch = 0; @@ -599,11 +639,12 @@ void ifRunning(int nodeId, Consumer action) { nodeIfRunning(nodeId).ifPresent(action); } - void forRandomRunning(Consumer action) { + Optional randomRunning() { List nodes = new ArrayList<>(running.values()); - if (!nodes.isEmpty()) { - RaftNode randomNode = nodes.get(random.nextInt(nodes.size())); - action.accept(randomNode); + if (nodes.isEmpty()) { + return Optional.empty(); + } else { + return Optional.of(nodes.get(random.nextInt(nodes.size()))); } } @@ -627,6 +668,11 @@ void startAll() { } } + void killAndDeletePersistentState(int nodeId) { + kill(nodeId); + nodes.put(nodeId, new PersistentState()); + } + private static RaftConfig.AddressSpec nodeAddress(int id) { return new RaftConfig.InetAddressSpec(new InetSocketAddress("localhost", 9990 + id)); } @@ -830,8 +876,13 @@ public void verify() { Integer nodeId = nodeStateEntry.getKey(); PersistentState state = nodeStateEntry.getValue(); Integer oldEpoch = nodeEpochs.get(nodeId); - Integer newEpoch = state.store.readElectionState().epoch; + ElectionState electionState = state.store.readElectionState(); + if (electionState == null) { + continue; + } + + Integer newEpoch = electionState.epoch; if (oldEpoch > newEpoch) { fail("Non-monotonic update of epoch detected on node " + nodeId + ": " + oldEpoch + " -> " + newEpoch); @@ -880,7 +931,7 @@ public void verify() { PersistentState state = nodeEntry.getValue(); ElectionState electionState = state.store.readElectionState(); - if (electionState.epoch >= epoch && electionState.hasLeader()) { + if (electionState != null && electionState.epoch >= epoch && electionState.hasLeader()) { if (epoch == electionState.epoch && leaderId.isPresent()) { assertEquals(leaderId.getAsInt(), electionState.leaderId()); } else { @@ -889,7 +940,6 @@ public void verify() { } } } - } }