Skip to content

KAFKA-12181; Loosen raft fetch offset validation of remote replicas #10309

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 7 commits into from
Mar 22, 2021
Merged

KAFKA-12181; Loosen raft fetch offset validation of remote replicas #10309

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c102b77
KAFKA-12181; Loosen end offset validation of remote replicas
hachikuji Mar 12, 2021
7d87ce6
Review comments and add simulation test
hachikuji Mar 13, 2021
39ee5e7
Add toString for ReplicaState
hachikuji Mar 13, 2021
a6b1261
Create `killAndDeletePersistentState`
hachikuji Mar 17, 2021
778d76d
Merge remote-tracking branch 'upstream/trunk' into KAFKA-12181
hachikuji Mar 18, 2021
bc8cc9c
Refactor new simulation test as a property test
hachikuji Mar 18, 2021
4fc7bf9
Consolidate check in `SingleLeader` invariant
hachikuji Mar 22, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 27 additions & 66 deletions raft/src/main/java/org/apache/kafka/raft/LeaderState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,8 @@ public class LeaderState implements EpochState {
private final long epochStartOffset;

private Optional<LogOffsetMetadata> highWatermark;
private final Map<Integer, VoterState> voterReplicaStates = new HashMap<>();
private final Map<Integer, ReplicaState> observerReplicaStates = new HashMap<>();
private final Map<Integer, ReplicaState> voterStates = new HashMap<>();
private final Map<Integer, ReplicaState> observerStates = new HashMap<>();
private final Set<Integer> grantingVoters = new HashSet<>();
private final Logger log;

Expand All @@ -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);
Expand All @@ -76,7 +76,7 @@ public Optional<LogOffsetMetadata> highWatermark() {

@Override
public ElectionState election() {
return ElectionState.withElectedLeader(epoch, localId, voterReplicaStates.keySet());
return ElectionState.withElectedLeader(epoch, localId, voterStates.keySet());
}

@Override
Expand All @@ -85,7 +85,7 @@ public int epoch() {
}

public Set<Integer> 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<Integer> grantingVoters() {
Expand All @@ -98,7 +98,7 @@ public int localId() {

public Set<Integer> nonAcknowledgingVoters() {
Set<Integer> nonAcknowledging = new HashSet<>();
for (VoterState state : voterReplicaStates.values()) {
for (ReplicaState state : voterStates.values()) {
if (!state.hasAcknowledgedLeader)
nonAcknowledging.add(state.nodeId);
}
Expand All @@ -107,9 +107,9 @@ public Set<Integer> nonAcknowledgingVoters() {

private boolean updateHighWatermark() {
// Find the largest offset which is replicated to a majority of replicas (the leader counts)
List<VoterState> followersByDescendingFetchOffset = followersByDescendingFetchOffset();
List<ReplicaState> followersByDescendingFetchOffset = followersByDescendingFetchOffset();

int indexOfHw = voterReplicaStates.size() / 2;
int indexOfHw = voterStates.size() / 2;
Optional<LogOffsetMetadata> highWatermarkUpdateOpt = followersByDescendingFetchOffset.get(indexOfHw).endOffset;

if (highWatermarkUpdateOpt.isPresent()) {
Expand All @@ -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;
Expand Down Expand Up @@ -183,8 +183,8 @@ public List<Integer> nonLeaderVotersByDescendingFetchOffset() {
.collect(Collectors.toList());
}

private List<VoterState> followersByDescendingFetchOffset() {
return new ArrayList<>(this.voterReplicaStates.values()).stream()
private List<ReplicaState> followersByDescendingFetchOffset() {
return new ArrayList<>(this.voterStates.values()).stream()
.sorted()
.collect(Collectors.toList());
}
Expand All @@ -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;
Expand All @@ -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<Integer, Long> getVoterEndOffsets() {
return getReplicaEndOffsets(voterReplicaStates);
return getReplicaEndOffsets(voterStates);
}

Map<Integer, Long> getObserverStates(final long currentTimeMs) {
clearInactiveObservers(currentTimeMs);
return getReplicaEndOffsets(observerReplicaStates);
return getReplicaEndOffsets(observerStates);
}

private static <R extends ReplicaState> Map<Integer, Long> getReplicaEndOffsets(
Expand All @@ -257,25 +252,27 @@ private static <R extends ReplicaState> Map<Integer, Long> 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<ReplicaState> {
private static class ReplicaState implements Comparable<ReplicaState> {
final int nodeId;
Optional<LogOffsetMetadata> 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) {
Expand All @@ -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(" +
Expand Down
4 changes: 2 additions & 2 deletions raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}

Expand Down
79 changes: 69 additions & 10 deletions raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Comment thread
hachikuji marked this conversation as resolved.
Outdated
scheduler.runUntil(() -> !cluster.hasLeader(node.nodeId) && cluster.hasConsistentLeader());
Comment thread
hachikuji marked this conversation as resolved.
Outdated

// 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));
Expand Down Expand Up @@ -590,10 +635,6 @@ int majoritySize() {
return voters.size() / 2 + 1;
}

Set<Integer> voters() {
return voters;
}

OptionalLong leaderHighWatermark() {
Optional<RaftNode> leaderWithMaxEpoch = running.values().stream().filter(node -> node.client.quorum().isLeader())
.max((node1, node2) -> Integer.compare(node2.client.quorum().epoch(), node1.client.quorum().epoch()));
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -701,11 +747,12 @@ void ifRunning(int nodeId, Consumer<RaftNode> action) {
nodeIfRunning(nodeId).ifPresent(action);
}

void forRandomRunning(Consumer<RaftNode> action) {
Optional<RaftNode> randomRunning() {
List<RaftNode> 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())));
}
}

Expand All @@ -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));
}
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -971,6 +1027,10 @@ public void verify() {
PersistentState state = nodeEntry.getValue();
ElectionState electionState = state.store.readElectionState();

if (electionState == null) {
Comment thread
hachikuji marked this conversation as resolved.
Outdated
continue;
}

if (electionState.epoch >= epoch && electionState.hasLeader()) {
if (epoch == electionState.epoch && leaderId.isPresent()) {
assertEquals(leaderId.getAsInt(), electionState.leaderId());
Expand All @@ -980,7 +1040,6 @@ public void verify() {
}
}
}

}
}

Expand Down