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 all commits
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
107 changes: 62 additions & 45 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 @@ -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;
Expand All @@ -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<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 static final long OBSERVER_SESSION_TIMEOUT_MS = 300_000L;
private final Logger log;

protected LeaderState(
int localId,
int epoch,
long epochStartOffset,
Set<Integer> voters,
Set<Integer> grantingVoters
Set<Integer> grantingVoters,
LogContext logContext

@abbccdda abbccdda Mar 20, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Just for my own education, when is it preferable to use upper class log context vs creating own log context?

@hachikuji hachikuji Mar 22, 2021

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.

The log context is useful because it carries with it a logging prefix which can be used to distinguish log messages. For example, in a streams application, the fact that we have multiple producers can make debugging difficult. Or in the context of integration/system/simulation testing, we often get logs from multiple nodes mixed together. With a common prefix, it is easy to grep messages for a particular instance so long as the LogContext is carried through to all the dependencies. Sometimes it is a little annoying to add the extra parameter, but it is worthwhile for improved debugging whenever the parent object already has a log context.

) {
this.localId = localId;
this.epoch = epoch;
Expand All @@ -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
Expand All @@ -69,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 @@ -78,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 @@ -91,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 @@ -100,26 +107,32 @@ 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()) {
// When a leader is first elected, it cannot know the high watermark of the previous
// 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;
}
Expand Down Expand Up @@ -170,36 +183,38 @@ 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());
}

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 {}: {} -> {}",

@abbccdda abbccdda Mar 21, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

I wonder whether the current approach is too loose. Maybe this is already done, but do we want to inform failed replica to cleanup or truncate in the FetchResponse?

@hachikuji hachikuji Mar 22, 2021

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.

The situation we are trying to handle is when a follower loses its disk. Basically the damage is already done by the time we receive the Fetch and the only thing we can do is let the follower try to catch back up. The problem with the old logic is that it prevented this even in situations which would not violate guarantees. I am planning to file a follow-up jira to think of some ways to handle disk loss situations more generally. We would like to at least detect the situation and see if we can prevent it from causing too much damage.

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;
Expand All @@ -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<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 @@ -237,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 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 @@ -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 +
')';
}
}

Expand Down
5 changes: 4 additions & 1 deletion raft/src/main/java/org/apache/kafka/raft/QuorumState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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 {
Expand Down Expand Up @@ -437,7 +439,8 @@ public void transitionToLeader(long epochStartOffset) throws IOException {
epoch(),
epochStartOffset,
voters,
candidateState.grantingVoters()
candidateState.grantingVoters(),
logContext
));
}

Expand Down
Loading