From c102b7790128c20ea855c5c8d4ef49fa2d87f3f3 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 12 Mar 2021 14:08:42 -0800 Subject: [PATCH 1/6] KAFKA-12181; Loosen end offset validation of remote replicas --- .../org/apache/kafka/raft/LeaderState.java | 68 ++++++++++--- .../org/apache/kafka/raft/QuorumState.java | 5 +- .../apache/kafka/raft/LeaderStateTest.java | 96 ++++++++++++++----- 3 files changed, 134 insertions(+), 35 deletions(-) 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..a8a15dafce524 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,6 +36,8 @@ * 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; @@ -41,14 +46,15 @@ public class LeaderState implements EpochState { private final Map voterReplicaStates = new HashMap<>(); private final Map observerReplicaStates = 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; @@ -60,6 +66,7 @@ protected LeaderState( this.voterReplicaStates.put(voterId, new VoterState(voterId, hasAcknowledgedLeader)); } this.grantingVoters.addAll(grantingVoters); + this.log = logContext.logger(LeaderState.class); } @Override @@ -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, voterReplicaStates.values()); + return false; } else { return false; } @@ -179,8 +192,15 @@ 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); @@ -209,10 +229,10 @@ public long epochStartOffset() { return epochStartOffset; } - ReplicaState getReplicaState(int remoteNodeId) { + private ReplicaState getReplicaState(int remoteNodeId) { ReplicaState state = voterReplicaStates.get(remoteNodeId); if (state == null) { - observerReplicaStates.putIfAbsent(remoteNodeId, new ReplicaState(remoteNodeId)); + observerReplicaStates.putIfAbsent(remoteNodeId, new ObserverState(remoteNodeId)); return observerReplicaStates.get(remoteNodeId); } return state; @@ -247,7 +267,7 @@ private boolean isVoter(int remoteNodeId) { return voterReplicaStates.containsKey(remoteNodeId); } - private static class ReplicaState implements Comparable { + private static abstract class ReplicaState implements Comparable { final int nodeId; Optional endOffset; OptionalLong lastFetchTimestamp; @@ -277,6 +297,22 @@ else if (!that.endOffset.isPresent()) } } + private static class ObserverState extends ReplicaState { + + public ObserverState(int nodeId) { + super(nodeId); + } + + @Override + public String toString() { + return "Observer(" + + "nodeId=" + nodeId + + ", endOffset=" + endOffset + + ", lastFetchTimestamp=" + lastFetchTimestamp + + ')'; + } + } + private static class VoterState extends ReplicaState { boolean hasAcknowledgedLeader; @@ -285,6 +321,16 @@ public VoterState(int nodeId, super(nodeId); this.hasAcknowledgedLeader = hasAcknowledgedLeader; } + + @Override + public String toString() { + return "Voter(" + + "nodeId=" + nodeId + + ", endOffset=" + endOffset + + ", lastFetchTimestamp=" + lastFetchTimestamp + + ", hasAcknowledgedLeader=" + hasAcknowledgedLeader + + ')'; + } } @Override 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..5c9fdf5409e66 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. + state.updateReplicaState(node1, time.milliseconds(), new LogOffsetMetadata(5L)); + + // The leader will report an error in the logs, but will not let the high watermark rewind + 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; From 7d87ce6c04920dee0e9481ca3873101a5d863a25 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 12 Mar 2021 18:32:54 -0800 Subject: [PATCH 2/6] Review comments and add simulation test --- .../org/apache/kafka/raft/LeaderState.java | 93 ++++++------------- .../apache/kafka/raft/LeaderStateTest.java | 4 +- .../kafka/raft/RaftEventSimulationTest.java | 79 ++++++++++++++-- 3 files changed, 98 insertions(+), 78 deletions(-) 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 a8a15dafce524..81c47d604897d 100644 --- a/raft/src/main/java/org/apache/kafka/raft/LeaderState.java +++ b/raft/src/main/java/org/apache/kafka/raft/LeaderState.java @@ -43,8 +43,8 @@ public class LeaderState implements EpochState { 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 final Logger log; @@ -63,7 +63,7 @@ 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); @@ -76,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 @@ -85,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() { @@ -98,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); } @@ -107,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()) { @@ -131,7 +131,7 @@ private boolean updateHighWatermark() { 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, voterReplicaStates.values()); + currentHighWatermarkMetadata.offset, voterStates.values()); return false; } else { return false; @@ -183,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()); } @@ -204,22 +204,17 @@ private boolean updateEndOffset(ReplicaState state, }); 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; @@ -230,21 +225,21 @@ public long epochStartOffset() { } private ReplicaState getReplicaState(int remoteNodeId) { - ReplicaState state = voterReplicaStates.get(remoteNodeId); + ReplicaState state = voterStates.get(remoteNodeId); if (state == null) { - observerReplicaStates.putIfAbsent(remoteNodeId, new ObserverState(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( @@ -257,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 abstract class ReplicaState implements Comparable { + 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) { @@ -297,42 +294,6 @@ else if (!that.endOffset.isPresent()) } } - private static class ObserverState extends ReplicaState { - - public ObserverState(int nodeId) { - super(nodeId); - } - - @Override - public String toString() { - return "Observer(" + - "nodeId=" + nodeId + - ", endOffset=" + endOffset + - ", lastFetchTimestamp=" + lastFetchTimestamp + - ')'; - } - } - - private static class VoterState extends ReplicaState { - boolean hasAcknowledgedLeader; - - public VoterState(int nodeId, - boolean hasAcknowledgedLeader) { - super(nodeId); - this.hasAcknowledgedLeader = hasAcknowledgedLeader; - } - - @Override - public String toString() { - return "Voter(" + - "nodeId=" + nodeId + - ", endOffset=" + endOffset + - ", lastFetchTimestamp=" + lastFetchTimestamp + - ", hasAcknowledgedLeader=" + hasAcknowledgedLeader + - ')'; - } - } - @Override public String toString() { return "Leader(" + 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 5c9fdf5409e66..cdadf6e00123a 100644 --- a/raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java +++ b/raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java @@ -170,9 +170,9 @@ public void testNonMonotonicHighWatermarkUpdate() { assertEquals(Optional.of(new LogOffsetMetadata(10L)), state.highWatermark()); // Follower crashes and disk is lost. It fetches an earlier offset to rebuild state. - state.updateReplicaState(node1, time.milliseconds(), new LogOffsetMetadata(5L)); - // 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()); } 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 28a6618d25bb8..b3608f09ff00c 100644 --- a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java +++ b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java @@ -401,6 +401,51 @@ private void checkBackToBackLeaderFailures(QuorumConfig config) { } } + @Test + public void checkSingleNodeCommittedDataLossQuorumSizeThree() { + checkSingleNodeCommittedDataLoss(new QuorumConfig(3, 0)); + } + + private void checkSingleNodeCommittedDataLoss(QuorumConfig config) { + assertTrue(config.numVoters > 2, + "This test requires the cluster to be able to recover from one failed node"); + + for (int seed = 0; seed < 100; seed++) { + // 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(config, seed); + 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.kill(node.nodeId); + cluster.deletePersistentState(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)); @@ -590,10 +635,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())); @@ -635,6 +676,11 @@ boolean allReachedHighWatermark(long offset) { .allMatch(node -> node.client.quorum().highWatermark().map(hw -> hw.offset).orElse(0L) > offset); } + boolean hasLeader(int nodeId) { + OptionalInt latestLeader = latestLeader(); + return latestLeader.isPresent() && latestLeader.getAsInt() == nodeId; + } + OptionalInt latestLeader() { OptionalInt latestLeader = OptionalInt.empty(); int latestEpoch = 0; @@ -701,11 +747,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()))); } } @@ -729,6 +776,10 @@ void startAll() { } } + void deletePersistentState(int nodeId) { + nodes.put(nodeId, new PersistentState()); + } + private static RaftConfig.AddressSpec nodeAddress(int id) { return new RaftConfig.InetAddressSpec(new InetSocketAddress("localhost", 9990 + id)); } @@ -921,8 +972,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); @@ -971,6 +1027,10 @@ public void verify() { PersistentState state = nodeEntry.getValue(); ElectionState electionState = state.store.readElectionState(); + if (electionState == null) { + continue; + } + if (electionState.epoch >= epoch && electionState.hasLeader()) { if (epoch == electionState.epoch && leaderId.isPresent()) { assertEquals(leaderId.getAsInt(), electionState.leaderId()); @@ -980,7 +1040,6 @@ public void verify() { } } } - } } From 39ee5e7d55fea2c5015dc0f59e79497dba1330b3 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 12 Mar 2021 20:59:36 -0800 Subject: [PATCH 3/6] Add toString for ReplicaState --- .../main/java/org/apache/kafka/raft/LeaderState.java | 10 ++++++++++ 1 file changed, 10 insertions(+) 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 81c47d604897d..d939aacbb1436 100644 --- a/raft/src/main/java/org/apache/kafka/raft/LeaderState.java +++ b/raft/src/main/java/org/apache/kafka/raft/LeaderState.java @@ -292,6 +292,16 @@ else if (!that.endOffset.isPresent()) else return Long.compare(that.endOffset.get().offset, this.endOffset.get().offset); } + + @Override + public String toString() { + return "ReplicaState(" + + "nodeId=" + nodeId + + ", endOffset=" + endOffset + + ", lastFetchTimestamp=" + lastFetchTimestamp + + ", hasAcknowledgedLeader=" + hasAcknowledgedLeader + + ')'; + } } @Override From a6b12615935cafe87f9463e9b4a5696c5f7e1db8 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Wed, 17 Mar 2021 12:40:34 -0700 Subject: [PATCH 4/6] Create `killAndDeletePersistentState` --- raft/.jqwik-database | Bin 0 -> 1354 bytes .../kafka/raft/RaftEventSimulationTest.java | 6 +++--- 2 files changed, 3 insertions(+), 3 deletions(-) create mode 100644 raft/.jqwik-database diff --git a/raft/.jqwik-database b/raft/.jqwik-database new file mode 100644 index 0000000000000000000000000000000000000000..a29e3b43f06f9298fe234fef14d62cf89fd44811 GIT binary patch literal 1354 zcmchXO^(wr6o9P}Y>*fbGY1&KW)Raf&X1FnRS|~}< z;sOY+z!q^B&MX0YDbSF|{Ru3fT^TOg zeXh;RK6#)(X2ooXm>Ke9{a%0^w}(dA57`~NmE`jZd)fNQoO~cV?@MAwX~=$y#a@yG z5p#V;6Gp?3GDALJu8BJh{-}MK3edV!sIdbjr0>v~LOumYL2jIZosJhM&5+M30frZ{ zPS2hyRZ*fct4<2?1g3bdim^hid-(}W3dPN;rw3@`!Ii+nIya}z%|@(qIy&IeJRj+r zxEYpr&u3u}yP>7_Jm04*@nS>1Z6EhoU}{cr*+a5;(&#rA)mGnKd?={_@hi$h|?lN z{gw-X+2*u+Eh~3e+u_kLVUbHC>&eIM!#0}tq}9RxaT?V(g|yB1dK)}9+S`WQy&dv5 D6yVXh literal 0 HcmV?d00001 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 b3608f09ff00c..84026aa25af55 100644 --- a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java +++ b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java @@ -435,8 +435,7 @@ private void checkSingleNodeCommittedDataLoss(QuorumConfig config) { // 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.kill(node.nodeId); - cluster.deletePersistentState(node.nodeId); + cluster.killAndDeletePersistentState(node.nodeId); scheduler.runUntil(() -> !cluster.hasLeader(node.nodeId) && cluster.hasConsistentLeader()); // Now restart the failed node and ensure that it recovers. @@ -776,7 +775,8 @@ void startAll() { } } - void deletePersistentState(int nodeId) { + void killAndDeletePersistentState(int nodeId) { + kill(nodeId); nodes.put(nodeId, new PersistentState()); } From bc8cc9c8b990b8e49ad25e2c1e0a921e10793be9 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Wed, 17 Mar 2021 21:32:51 -0700 Subject: [PATCH 5/6] Refactor new simulation test as a property test --- raft/.jqwik-database | Bin 1354 -> 0 bytes .../kafka/raft/RaftEventSimulationTest.java | 67 ++++++++---------- 2 files changed, 31 insertions(+), 36 deletions(-) delete mode 100644 raft/.jqwik-database diff --git a/raft/.jqwik-database b/raft/.jqwik-database deleted file mode 100644 index a29e3b43f06f9298fe234fef14d62cf89fd44811..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 1354 zcmchXO^(wr6o9P}Y>*fbGY1&KW)Raf&X1FnRS|~}< z;sOY+z!q^B&MX0YDbSF|{Ru3fT^TOg zeXh;RK6#)(X2ooXm>Ke9{a%0^w}(dA57`~NmE`jZd)fNQoO~cV?@MAwX~=$y#a@yG z5p#V;6Gp?3GDALJu8BJh{-}MK3edV!sIdbjr0>v~LOumYL2jIZosJhM&5+M30frZ{ zPS2hyRZ*fct4<2?1g3bdim^hid-(}W3dPN;rw3@`!Ii+nIya}z%|@(qIy&IeJRj+r zxEYpr&u3u}yP>7_Jm04*@nS>1Z6EhoU}{cr*+a5;(&#rA)mGnKd?={_@hi$h|?lN z{gw-X+2*u+Eh~3e+u_kLVUbHC>&eIM!#0}tq}9RxaT?V(g|yB1dK)}9+S`WQy&dv5 D6yVXh 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 b26d9929a2c8b..53ff2d2d830fa 100644 --- a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java +++ b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java @@ -304,48 +304,43 @@ void canMakeProgressAfterBackToBackLeaderFailures( scheduler.runUntil(() -> cluster.anyReachedHighWatermark(targetHighWatermark)); } - @Test - public void checkSingleNodeCommittedDataLossQuorumSizeThree() { - checkSingleNodeCommittedDataLoss(new QuorumConfig(3, 0)); - } - - private void checkSingleNodeCommittedDataLoss(QuorumConfig config) { - assertTrue(config.numVoters > 2, - "This test requires the cluster to be able to recover from one failed node"); - - for (int seed = 0; seed < 100; seed++) { - // We run this test without the `MonotonicEpoch` and `MajorityReachedHighWatermark` - // invariants since the loss of committed data on one node can violate them. + @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(config, seed); - EventScheduler scheduler = new EventScheduler(cluster.random, cluster.time); - scheduler.addInvariant(new MonotonicHighWatermark(cluster)); - scheduler.addInvariant(new SingleLeader(cluster)); - scheduler.addValidation(new ConsistentCommittedData(cluster)); + 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); + 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)); + 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") - ); + 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()); + // 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)); - } + // 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) { From 4fc7bf96b2722317ac418a1d6ff0dedbc01a9ea8 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Mon, 22 Mar 2021 11:47:09 -0700 Subject: [PATCH 6/6] Consolidate check in `SingleLeader` invariant --- .../java/org/apache/kafka/raft/RaftEventSimulationTest.java | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) 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 53ff2d2d830fa..223c2694c767d 100644 --- a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java +++ b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java @@ -931,11 +931,7 @@ public void verify() { PersistentState state = nodeEntry.getValue(); ElectionState electionState = state.store.readElectionState(); - if (electionState == null) { - continue; - } - - 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 {