Skip to content

KAFKA-16530: Fix high-watermark calculation to not assume the leader is in the voter set #16079

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
showuon merged 10 commits into from
Jun 6, 2024
Merged

KAFKA-16530: Fix high-watermark calculation to not assume the leader is in the voter set #16079

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
372e02c
additional tests in LeaderStateTest to enforce monotonically increasi…
ahuang98 May 24, 2024
e7e067b
Merge branch 'trunk' into KAFKA-16530
ahuang98 May 29, 2024
c3a00c0
voter set updated from partitionstate
ahuang98 May 29, 2024
93fd95d
more comments
ahuang98 Jun 3, 2024
66e27d7
refactoring tests
ahuang98 Jun 3, 2024
06bf0f8
Merge branch 'trunk' into KAFKA-16530
ahuang98 Jun 3, 2024
9acdb02
test comments
ahuang98 Jun 3, 2024
d33f923
more test improvements
ahuang98 Jun 4, 2024
4eb7961
more tests
ahuang98 Jun 4, 2024
5d73624
comments
ahuang98 Jun 5, 2024
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
27 changes: 22 additions & 5 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 @@ -243,8 +243,9 @@ private boolean maybeUpdateHighWatermark() {
);
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. " +
log.warn("The latest computed high watermark {} is smaller than the current " +
"value {}, which should only happen when voter set membership changes. If the voter " +
"set has not changed this suggests that one of the voters has lost committed data. " +

@jsancio jsancio May 27, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In general we should avoid having error or warn log messages. In most cases we want to throw an exception if there is an error.

In this case, this state is expected but infrequent so log.info seems appropriate.

"Full voter replication state: {}", highWatermarkUpdateOffset,
currentHighWatermarkMetadata.offset, voterStates.values());
return false;
Expand Down Expand Up @@ -341,9 +342,16 @@ public boolean updateReplicaState(
state.nodeId, currentEndOffset.offset, fetchOffsetMetadata.offset);
}
});

Optional<LogOffsetMetadata> leaderEndOffsetOpt =
voterStates.get(localId).endOffset;
Optional<LogOffsetMetadata> leaderEndOffsetOpt;

@jsancio jsancio Jun 3, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you move this right before the if statement and mark it as final it makes it more obvious that this variable can take two different values.

if (voterStates.containsKey(localId)) {
leaderEndOffsetOpt = voterStates.get(localId).endOffset;

@jsancio jsancio May 27, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's avoid the pattern if (map.containsKey(key)) { T value = map.get(key); ... } This is the same as:

T value = map.get(key);
if (value != null) {
    ...
}

// Or
Optional.OfNullable(map.get(key)).ifPresent(value -> ...);

} else if (observerStates.containsKey(localId)) {
// The leader is not guaranteed to be in the voter set when in the process of being removed from the quorum.
log.info("Updating end offset for leader {} which is also an observer.", localId);
leaderEndOffsetOpt = observerStates.get(localId).endOffset;
} else {
throw new IllegalStateException("Leader state not found for localId " + localId);
}

state.updateFollowerState(
currentTimeMs,
Expand Down Expand Up @@ -445,6 +453,15 @@ private boolean isVoter(int remoteNodeId) {
return voterStates.containsKey(remoteNodeId);
}

// for testing purposes
boolean removeVoter(int nodeId) {
if (voterStates.containsKey(nodeId)) {
voterStates.remove(nodeId);
return true;
}
return false;
}

@jsancio jsancio May 27, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should remove this method. We should test and use the same functionality that will be used by KafkaRaftClient. I don't think this is how voter changes are going to get communicated to the replica state (LeaderState).

I am thinking that we should replace the Set<Integer> voters parameter in the constructor with Supplier<VoterSet> latestVoterSet. Every time the partitionState gets updated, we should compare the current LeaderState against the latest voter set and update the LeaderState accordingly.


private static class ReplicaState implements Comparable<ReplicaState> {
final int nodeId;
Optional<LogOffsetMetadata> endOffset;
Expand Down
54 changes: 54 additions & 0 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 @@ -280,6 +280,60 @@ public void testUpdateHighWatermarkQuorumSizeThree() {
assertEquals(Optional.of(new LogOffsetMetadata(20L)), state.highWatermark());
}

@Test
public void testUpdateHighWatermarkRemovingFollowerFromVoterStates() {
Comment thread
jsancio marked this conversation as resolved.
int node1 = 1;
int node2 = 2;
LeaderState<?> state = newLeaderState(mkSet(localId, node1, node2), 10L);
assertFalse(state.updateLocalState(new LogOffsetMetadata(15L)));
assertTrue(state.updateReplicaState(node1, 0, new LogOffsetMetadata(15L)));
assertFalse(state.updateReplicaState(node2, 0, new LogOffsetMetadata(10L)));
assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());

// removing node1 should not decrement HW to 10L
assertTrue(state.removeVoter(node1));
assertFalse(state.updateLocalState(new LogOffsetMetadata(17L)));
assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());

// HW cannot change until after node2 catches up to last HW
assertFalse(state.updateReplicaState(node2, 0, new LogOffsetMetadata(14L)));

@showuon showuon Jun 4, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can also increase leader's LEO to 16 and show that it doesn't increase the HWM.

@ahuang98 ahuang98 Jun 4, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

leader's LEO at this point is 17 (line 330L), however I can bump it up one further to demonstrate the HW doesn't change

assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());
assertFalse(state.updateReplicaState(node2, 0, new LogOffsetMetadata(15L)));
assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());

// HW should update to 16L
assertTrue(state.updateReplicaState(node2, 0, new LogOffsetMetadata(16L)));
assertEquals(Optional.of(new LogOffsetMetadata(16L)), state.highWatermark());
}

@Test
public void testUpdateHighWatermarkQuorumRemovingLeaderFromVoterStates() {
int node1 = 1;
int node2 = 2;
LeaderState<?> state = newLeaderState(mkSet(localId, node1, node2), 10L);
assertFalse(state.updateLocalState(new LogOffsetMetadata(15L)));
assertTrue(state.updateReplicaState(node1, 0, new LogOffsetMetadata(15L)));
assertFalse(state.updateReplicaState(node2, 0, new LogOffsetMetadata(10L)));
assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());

// removing leader should not decrement HW to 10L
assertTrue(state.removeVoter(localId));
assertFalse(state.updateLocalState(new LogOffsetMetadata(17L)));
assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());

// HW cannot change until node2 catches up to last HW

@jsancio jsancio Jun 3, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Before checking this, let's increase node1's LEO to 16 and show that it doesn't increase the HWM.

assertFalse(state.updateReplicaState(node2, 0, new LogOffsetMetadata(14L)));
assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());
assertFalse(state.updateReplicaState(node2, 0, new LogOffsetMetadata(15L)));
assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());

// HW will not update to 16L until majority of remaining voterSet (node1, node2) are at least 16L
assertFalse(state.updateReplicaState(node2, 0, new LogOffsetMetadata(16L)));
assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());
assertTrue(state.updateReplicaState(node1, 0, new LogOffsetMetadata(16L)));
assertEquals(Optional.of(new LogOffsetMetadata(16L)), state.highWatermark());
}

@Test
public void testNonMonotonicHighWatermarkUpdate() {
MockTime time = new MockTime();
Expand Down