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
68 changes: 57 additions & 11 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,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;
Expand All @@ -41,14 +46,15 @@ public class LeaderState implements EpochState {
private final Map<Integer, VoterState> voterReplicaStates = new HashMap<>();
private final Map<Integer, ReplicaState> observerReplicaStates = 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 @@ -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
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -179,8 +192,15 @@ private List<VoterState> 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 {}: {} -> {}",

@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);
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -247,7 +267,7 @@ private boolean isVoter(int remoteNodeId) {
return voterReplicaStates.containsKey(remoteNodeId);
}

private static class ReplicaState implements Comparable<ReplicaState> {
private static abstract class ReplicaState implements Comparable<ReplicaState> {
Comment thread
hachikuji marked this conversation as resolved.
Outdated
final int nodeId;
Optional<LogOffsetMetadata> endOffset;
OptionalLong lastFetchTimestamp;
Expand Down Expand Up @@ -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;

Expand All @@ -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
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
96 changes: 73 additions & 23 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 @@ -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;
Expand All @@ -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<Integer> 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)));
Expand All @@ -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")));
Expand All @@ -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());
Expand All @@ -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)));
Expand All @@ -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());
Expand All @@ -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));
Comment thread
hachikuji marked this conversation as resolved.
Outdated

// 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;
Expand Down Expand Up @@ -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));
Expand All @@ -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;

@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.

So this offset was named wrong previously?

@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.

I'm not sure I'd call it wrong. The epoch start offset is initialized as the current log end offset. But I thought it was better to choose a more explicit name.


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;

Expand Down