Skip to content

KAFKA-12440: ClusterId validation for Vote, BeginQuorum, EndQuorum and FetchSnapshot #10289

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 3 commits into from
Mar 19, 2021
Merged

KAFKA-12440: ClusterId validation for Vote, BeginQuorum, EndQuorum and FetchSnapshot #10289

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
58612e2
KAFKA-12440: ClusterId validation for Vote, BeginQourum, EndQuorum an…
Mar 9, 2021
ce8a99d
fix
Mar 15, 2021
ef799af
improvement
Mar 19, 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
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@
files="(Utils|Topic|KafkaLZ4BlockOutputStream|AclData|JoinGroupRequest).java"/>

<suppress checks="CyclomaticComplexity"
files="(ConsumerCoordinator|Fetcher|Sender|KafkaProducer|BufferPool|ConfigDef|RecordAccumulator|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer).java"/>
files="(ConsumerCoordinator|Fetcher|Sender|KafkaProducer|BufferPool|ConfigDef|RecordAccumulator|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|KafkaRaftClient).java"/>

@jsancio jsancio Mar 10, 2021

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.

It is sad that we have to add KafkaRaftClient to this list. Do you know what exactly pushed this over the threshold? This would allow us to look into ways to re-organize the code so that it is not so complex.

@dengziming dengziming Mar 11, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

handleVoteRequest() method has too many if condition.

@hachikuji hachikuji Mar 18, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I agree it is unfortunate. There are probably ways we can improve this. For example, this logic smells a little bit:

        if (quorum.isLeader()) {
            logger.debug("Rejecting vote request {} with epoch {} since we are already leader on that epoch",
                    request, candidateEpoch);
            voteGranted = false;
        } else if (quorum.isCandidate()) {
            logger.debug("Rejecting vote request {} with epoch {} since we are already candidate on that epoch",
                    request, candidateEpoch);
            voteGranted = false;
        } else if (quorum.isResigned()) {
            logger.debug("Rejecting vote request {} with epoch {} since we have resigned as candidate/leader in this epoch",
                request, candidateEpoch);
            voteGranted = false;
        } else if (quorum.isFollower()) {
            FollowerState state = quorum.followerStateOrThrow();
            logger.debug("Rejecting vote request {} with epoch {} since we already have a leader {} on that epoch",
                request, candidateEpoch, state.leaderId());
            voteGranted = false;

It might be possible to push this logic into EpochState or at least to use make use of the name() method in the logging. @dengziming would you be interested in following up on this separately?

@dengziming dengziming Mar 20, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Thank you, I will take some time to improve this.


<suppress checks="JavaNCSS"
files="(AbstractRequest|AbstractResponse|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest|KafkaAdminClientTest).java"/>
Expand Down
2 changes: 2 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/requests/FetchSnapshotRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ public FetchSnapshotRequestData data() {
* @return the created fetch snapshot request data
*/
public static FetchSnapshotRequestData singleton(
String clusterId,
TopicPartition topicPartition,
UnaryOperator<FetchSnapshotRequestData.PartitionSnapshot> operator
) {
Expand All @@ -68,6 +69,7 @@ public static FetchSnapshotRequestData singleton(
);

return new FetchSnapshotRequestData()
.setClusterId(clusterId)
.setTopics(
Collections.singletonList(
new FetchSnapshotRequestData.TopicSnapshot()
Expand Down
51 changes: 47 additions & 4 deletions raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -549,6 +549,8 @@ private VoteResponseData buildVoteResponse(Errors partitionLevelError, boolean v
/**
* Handle a Vote request. This API may return the following errors:
*
* - {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
* but different from this node
* - {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
* - {@link Errors#FENCED_LEADER_EPOCH} if the epoch is smaller than this node's epoch
* - {@link Errors#INCONSISTENT_VOTER_SET} if the request suggests inconsistent voter membership (e.g.
Expand All @@ -560,6 +562,10 @@ private VoteResponseData handleVoteRequest(
) throws IOException {
VoteRequestData request = (VoteRequestData) requestMetadata.data;

if (!hasValidClusterId(request.clusterId())) {
return new VoteResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code());
}

if (!hasValidTopicPartition(request, log.topicPartition())) {
// Until we support multi-raft, we treat topic partition mismatches as invalid requests
return new VoteResponseData().setErrorCode(Errors.INVALID_REQUEST.code());
Expand Down Expand Up @@ -720,6 +726,8 @@ private BeginQuorumEpochResponseData buildBeginQuorumEpochResponse(Errors partit
/**
* Handle a BeginEpoch request. This API may return the following errors:
*
* - {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
* but different from this node
* - {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
* - {@link Errors#INCONSISTENT_VOTER_SET} if the request suggests inconsistent voter membership (e.g.
* if this node or the sender is not one of the current known voters)
Expand All @@ -731,6 +739,10 @@ private BeginQuorumEpochResponseData handleBeginQuorumEpochRequest(
) throws IOException {
BeginQuorumEpochRequestData request = (BeginQuorumEpochRequestData) requestMetadata.data;

if (!hasValidClusterId(request.clusterId())) {
return new BeginQuorumEpochResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code());
}

if (!hasValidTopicPartition(request, log.topicPartition())) {
// Until we support multi-raft, we treat topic partition mismatches as invalid requests
return new BeginQuorumEpochResponseData().setErrorCode(Errors.INVALID_REQUEST.code());
Expand Down Expand Up @@ -803,6 +815,8 @@ private EndQuorumEpochResponseData buildEndQuorumEpochResponse(Errors partitionL
/**
* Handle an EndEpoch request. This API may return the following errors:
*
* - {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
* but different from this node
* - {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
* - {@link Errors#INCONSISTENT_VOTER_SET} if the request suggests inconsistent voter membership (e.g.
* if this node or the sender is not one of the current known voters)
Expand All @@ -814,6 +828,10 @@ private EndQuorumEpochResponseData handleEndQuorumEpochRequest(
) throws IOException {
EndQuorumEpochRequestData request = (EndQuorumEpochRequestData) requestMetadata.data;

if (!hasValidClusterId(request.clusterId())) {
return new EndQuorumEpochResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code());
}

if (!hasValidTopicPartition(request, log.topicPartition())) {
// Until we support multi-raft, we treat topic partition mismatches as invalid requests
return new EndQuorumEpochResponseData().setErrorCode(Errors.INVALID_REQUEST.code());
Expand Down Expand Up @@ -939,12 +957,12 @@ private FetchResponseData buildEmptyFetchResponse(
);
}

private boolean hasValidClusterId(FetchRequestData request) {
private boolean hasValidClusterId(String requestClusterId) {
// We don't enforce the cluster id if it is not provided.
if (request.clusterId() == null) {
if (requestClusterId == null) {
return true;
}
return clusterId.equals(request.clusterId());
return clusterId.equals(requestClusterId);
}

/**
Expand All @@ -955,6 +973,8 @@ private boolean hasValidClusterId(FetchRequestData request) {
*
* This API may return the following errors:
*
* - {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
* but different from this node
* - {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
* - {@link Errors#FENCED_LEADER_EPOCH} if the epoch is smaller than this node's epoch
* - {@link Errors#INVALID_REQUEST} if the request epoch is larger than the leader's current epoch
Expand All @@ -966,7 +986,7 @@ private CompletableFuture<FetchResponseData> handleFetchRequest(
) {
FetchRequestData request = (FetchRequestData) requestMetadata.data;

if (!hasValidClusterId(request)) {
if (!hasValidClusterId(request.clusterId())) {
return completedFuture(new FetchResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code()));
}

Expand Down Expand Up @@ -1200,11 +1220,30 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}

/**
* Handle a FetchSnapshot request, similar to the Fetch request but we use {@link UnalignedRecords}
* in response because the records are not necessarily offset-aligned.
*
* This API may return the following errors:
*
* - {@link Errors#INCONSISTENT_CLUSTER_ID} if the cluster id is presented in request
* but different from this node
* - {@link Errors#BROKER_NOT_AVAILABLE} if this node is currently shutting down
* - {@link Errors#FENCED_LEADER_EPOCH} if the epoch is smaller than this node's epoch
* - {@link Errors#INVALID_REQUEST} if the request epoch is larger than the leader's current epoch
* or if either the fetch offset or the last fetched epoch is invalid
* - {@link Errors#SNAPSHOT_NOT_FOUND} if the request snapshot id does not exists
* - {@link Errors#POSITION_OUT_OF_RANGE} if the request snapshot offset out of range
*/
private FetchSnapshotResponseData handleFetchSnapshotRequest(
RaftRequest.Inbound requestMetadata
) throws IOException {
FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;

if (!hasValidClusterId(data.clusterId())) {
return new FetchSnapshotResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code());
}

if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
return FetchSnapshotResponse.withTopLevelError(Errors.INVALID_REQUEST);
}
Expand Down Expand Up @@ -1728,6 +1767,7 @@ private EndQuorumEpochRequestData buildEndQuorumEpochRequest(
) {
return EndQuorumEpochRequest.singletonRequest(
log.topicPartition(),
clusterId,
quorum.epoch(),
quorum.localIdOrThrow(),
state.preferredSuccessors()
Expand All @@ -1752,6 +1792,7 @@ private long maybeSendRequests(
private BeginQuorumEpochRequestData buildBeginQuorumEpochRequest() {
return BeginQuorumEpochRequest.singletonRequest(
log.topicPartition(),
clusterId,
quorum.epoch(),
quorum.localIdOrThrow()
);
Expand All @@ -1761,6 +1802,7 @@ private VoteRequestData buildVoteRequest() {
OffsetAndEpoch endOffset = endOffset();
return VoteRequest.singletonRequest(
log.topicPartition(),
clusterId,
quorum.epoch(),
quorum.localIdOrThrow(),
endOffset.epoch,
Expand Down Expand Up @@ -1801,6 +1843,7 @@ private FetchSnapshotRequestData buildFetchSnapshotRequest(OffsetAndEpoch snapsh
.setEndOffset(snapshotId.offset);

FetchSnapshotRequestData request = FetchSnapshotRequest.singleton(
clusterId,
log.topicPartition(),
snapshotPartition -> {
return snapshotPartition
Expand Down
78 changes: 78 additions & 0 deletions raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1269,7 +1269,84 @@ public void testFetchSnapshotResponseToNotFollower() throws Exception {
context.assertVotedCandidate(epoch + 1, localId);
}

@Test
public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
int localId = 0;
int otherNodeId = 1;
int epoch = 5;
Set<Integer> voters = Utils.mkSet(localId, otherNodeId);

RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);

// null cluster id is accepted
context.deliverRequest(
fetchSnapshotRequest(
context.clusterId.toString(),
context.metadataPartition,
epoch,
new OffsetAndEpoch(0, 0),
Integer.MAX_VALUE,
0
)
);
context.pollUntilResponse();
context.assertSentFetchSnapshotResponse(context.metadataPartition);

// null cluster id is accepted
context.deliverRequest(
fetchSnapshotRequest(
null,
context.metadataPartition,
epoch,
new OffsetAndEpoch(0, 0),
Integer.MAX_VALUE,
0
)
);
context.pollUntilResponse();
context.assertSentFetchSnapshotResponse(context.metadataPartition);

// empty cluster id is rejected
context.deliverRequest(
fetchSnapshotRequest(
"",
context.metadataPartition,
epoch,
new OffsetAndEpoch(0, 0),
Integer.MAX_VALUE,
0
)
);
context.pollUntilResponse();
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);

// invalid cluster id is rejected
context.deliverRequest(
fetchSnapshotRequest(
"invalid-uuid",
context.metadataPartition,
epoch,
new OffsetAndEpoch(0, 0),
Integer.MAX_VALUE,
0
)
);
context.pollUntilResponse();
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}

private static FetchSnapshotRequestData fetchSnapshotRequest(
TopicPartition topicPartition,
int epoch,
OffsetAndEpoch offsetAndEpoch,
int maxBytes,
long position
) {
return fetchSnapshotRequest(null, topicPartition, epoch, offsetAndEpoch, maxBytes, position);
}

private static FetchSnapshotRequestData fetchSnapshotRequest(
String clusterId,
TopicPartition topicPartition,
int epoch,
OffsetAndEpoch offsetAndEpoch,
Expand All @@ -1281,6 +1358,7 @@ private static FetchSnapshotRequestData fetchSnapshotRequest(
.setEpoch(offsetAndEpoch.epoch);

FetchSnapshotRequestData request = FetchSnapshotRequest.singleton(
clusterId,
topicPartition,
snapshotPartition -> {
return snapshotPartition
Expand Down
96 changes: 96 additions & 0 deletions raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1107,6 +1107,12 @@ public void testFetchRequestClusterIdValidation() throws Exception {

RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);

// valid cluster id is accepted
context.deliverRequest(context.fetchRequest(
epoch, context.clusterId.toString(), otherNodeId, -5L, 0, 0));
context.pollUntilResponse();
context.assertSentFetchPartitionResponse(Errors.INVALID_REQUEST, epoch, OptionalInt.of(localId));

// null cluster id is accepted
context.deliverRequest(context.fetchRequest(
epoch, null, otherNodeId, -5L, 0, 0));
Expand All @@ -1126,6 +1132,96 @@ public void testFetchRequestClusterIdValidation() throws Exception {
context.assertSentFetchPartitionResponse(Errors.INCONSISTENT_CLUSTER_ID);
}

@Test
public void testVoteRequestClusterIdValidation() throws Exception {
int localId = 0;
int otherNodeId = 1;
int epoch = 5;
Set<Integer> voters = Utils.mkSet(localId, otherNodeId);

RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);

// valid cluster id is accepted
context.deliverRequest(context.voteRequest(epoch, localId, 0, 0));
context.pollUntilResponse();
context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(localId), false);

// null cluster id is accepted
context.deliverRequest(context.voteRequest(epoch, localId, 0, 0));
context.pollUntilResponse();
context.assertSentVoteResponse(Errors.NONE, epoch, OptionalInt.of(localId), false);

// empty cluster id is rejected
context.deliverRequest(context.voteRequest("", epoch, localId, 0, 0));
context.pollUntilResponse();
context.assertSentVoteResponse(Errors.INCONSISTENT_CLUSTER_ID);

// invalid cluster id is rejected
context.deliverRequest(context.voteRequest("invalid-uuid", epoch, localId, 0, 0));
context.pollUntilResponse();
context.assertSentVoteResponse(Errors.INCONSISTENT_CLUSTER_ID);
}

@Test
public void testBeginQuorumEpochRequestClusterIdValidation() throws Exception {
int localId = 0;
int otherNodeId = 1;
int epoch = 5;
Set<Integer> voters = Utils.mkSet(localId, otherNodeId);

RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);

// valid cluster id is accepted
context.deliverRequest(context.beginEpochRequest(context.clusterId.toString(), epoch, localId));
context.pollUntilResponse();
context.assertSentBeginQuorumEpochResponse(Errors.NONE, epoch, OptionalInt.of(localId));

// null cluster id is accepted
context.deliverRequest(context.beginEpochRequest(epoch, localId));
context.pollUntilResponse();
context.assertSentBeginQuorumEpochResponse(Errors.NONE, epoch, OptionalInt.of(localId));

// empty cluster id is rejected
context.deliverRequest(context.beginEpochRequest("", epoch, localId));
context.pollUntilResponse();
context.assertSentBeginQuorumEpochResponse(Errors.INCONSISTENT_CLUSTER_ID);

// invalid cluster id is rejected
context.deliverRequest(context.beginEpochRequest("invalid-uuid", epoch, localId));
context.pollUntilResponse();
context.assertSentBeginQuorumEpochResponse(Errors.INCONSISTENT_CLUSTER_ID);
}

@Test
public void testEndQuorumEpochRequestClusterIdValidation() throws Exception {
int localId = 0;
int otherNodeId = 1;
int epoch = 5;
Set<Integer> voters = Utils.mkSet(localId, otherNodeId);

RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);

// valid cluster id is accepted
context.deliverRequest(context.endEpochRequest(context.clusterId.toString(), epoch, localId, Collections.singletonList(otherNodeId)));
context.pollUntilResponse();
context.assertSentEndQuorumEpochResponse(Errors.NONE, epoch, OptionalInt.of(localId));

// null cluster id is accepted
context.deliverRequest(context.endEpochRequest(epoch, localId, Collections.singletonList(otherNodeId)));
context.pollUntilResponse();
context.assertSentEndQuorumEpochResponse(Errors.NONE, epoch, OptionalInt.of(localId));

// empty cluster id is rejected
context.deliverRequest(context.endEpochRequest("", epoch, localId, Collections.singletonList(otherNodeId)));
context.pollUntilResponse();
context.assertSentEndQuorumEpochResponse(Errors.INCONSISTENT_CLUSTER_ID);

// invalid cluster id is rejected
context.deliverRequest(context.endEpochRequest("invalid-uuid", epoch, localId, Collections.singletonList(otherNodeId)));
context.pollUntilResponse();
context.assertSentEndQuorumEpochResponse(Errors.INCONSISTENT_CLUSTER_ID);
}

@Test
public void testVoterOnlyRequestValidation() throws Exception {
int localId = 0;
Expand Down
Loading