Skip to content

KAFKA-13940; Return NOT_LEADER_OR_FOLLOWER if DescribeQuorum sent to non-leader #12517

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 2 commits into from
Aug 17, 2022
Merged

KAFKA-13940; Return NOT_LEADER_OR_FOLLOWER if DescribeQuorum sent to non-leader #12517

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fc63e2e
KAFKA-13940; Return NOT_LEADER_OR_FOLLOWER if DescribeQuorum sent to …
hachikuji Aug 15, 2022
4172e55
Set partition index explicitly in DescribeQuorumResponse
hachikuji Aug 16, 2022
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
16 changes: 15 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/DescribeQuorumResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@
* - {@link Errors#BROKER_NOT_AVAILABLE}
*
* Partition level errors:
* - {@link Errors#INVALID_REQUEST}
* - {@link Errors#NOT_LEADER_OR_FOLLOWER}
* - {@link Errors#UNKNOWN_TOPIC_OR_PARTITION}
*/
public class DescribeQuorumResponse extends AbstractResponse {
Expand Down Expand Up @@ -72,6 +72,19 @@ public int throttleTimeMs() {
return DEFAULT_THROTTLE_TIME;
}

public static DescribeQuorumResponseData singletonErrorResponse(
TopicPartition topicPartition,
Errors error
) {
return new DescribeQuorumResponseData()
.setTopics(Collections.singletonList(new DescribeQuorumResponseData.TopicData()
.setTopicName(topicPartition.topic())
.setPartitions(Collections.singletonList(new DescribeQuorumResponseData.PartitionData()
.setPartitionIndex(topicPartition.partition())
.setErrorCode(error.code())))));
Comment thread
hachikuji marked this conversation as resolved.
}


public static DescribeQuorumResponseData singletonResponse(TopicPartition topicPartition,
int leaderId,
int leaderEpoch,
Expand All @@ -82,6 +95,7 @@ public static DescribeQuorumResponseData singletonResponse(TopicPartition topicP
.setTopics(Collections.singletonList(new DescribeQuorumResponseData.TopicData()
.setTopicName(topicPartition.topic())
.setPartitions(Collections.singletonList(new DescribeQuorumResponseData.PartitionData()
.setPartitionIndex(topicPartition.partition())
.setErrorCode(Errors.NONE.code())
.setLeaderId(leaderId)
.setLeaderEpoch(leaderEpoch)
Expand Down
23 changes: 23 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5192,6 +5192,29 @@ public void testDescribeMetadataQuorumSuccess() throws Exception {
}
}

@Test
public void testDescribeMetadataQuorumRetriableError() throws Exception {
try (final AdminClientUnitTestEnv env = mockClientEnv()) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create(ApiKeys.DESCRIBE_QUORUM.id,
ApiKeys.DESCRIBE_QUORUM.oldestVersion(),
ApiKeys.DESCRIBE_QUORUM.latestVersion()));

// First request fails with a NOT_LEADER_OR_FOLLOWER error (which is retriable)
env.kafkaClient().prepareResponse(
body -> body instanceof DescribeQuorumRequest,
prepareDescribeQuorumResponse(Errors.NONE, Errors.NOT_LEADER_OR_FOLLOWER, false, false, false, false, false));

// The second request succeeds
env.kafkaClient().prepareResponse(
body -> body instanceof DescribeQuorumRequest,
prepareDescribeQuorumResponse(Errors.NONE, Errors.NONE, false, false, false, false, false));

KafkaFuture<QuorumInfo> future = env.adminClient().describeMetadataQuorum().quorumInfo();
QuorumInfo quorumInfo = future.get();
assertEquals(defaultQuorumInfo(false), quorumInfo);
}
}

@Test
public void testDescribeMetadataQuorumFailure() {
try (final AdminClientUnitTestEnv env = mockClientEnv()) {
Expand Down
5 changes: 4 additions & 1 deletion 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 @@ -1171,7 +1171,10 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
}

if (!quorum.isLeader()) {
return DescribeQuorumRequest.getTopLevelErrorResponse(Errors.INVALID_REQUEST);
return DescribeQuorumResponse.singletonErrorResponse(

@dengziming dengziming Aug 17, 2022

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.

We have documentation comments for handleFetchRequest and other handleXXXRequest methods, how about adding documentation comments for handleDescribeQuorumRequest.

@hachikuji hachikuji Aug 17, 2022

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.

Hmm, this part at least seems self-explanatory. Do you see something else that deserves more explanation? Would you mind submitting a follow-up PR if you do? I am happy to review.

log.topicPartition(),
Errors.NOT_LEADER_OR_FOLLOWER
);
}

LeaderState<T> leaderState = quorum.leaderStateOrThrow();
Expand Down
29 changes: 29 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 @@ -20,6 +20,7 @@
import org.apache.kafka.common.errors.RecordBatchTooLargeException;
import org.apache.kafka.common.memory.MemoryPool;
import org.apache.kafka.common.message.BeginQuorumEpochResponseData;
import org.apache.kafka.common.message.DescribeQuorumResponseData;
import org.apache.kafka.common.message.DescribeQuorumResponseData.ReplicaState;
import org.apache.kafka.common.message.EndQuorumEpochResponseData;
import org.apache.kafka.common.message.FetchResponseData;
Expand Down Expand Up @@ -1946,6 +1947,34 @@ public void testEndQuorumEpochSentBasedOnFetchOffset() throws Exception {
);
}

@Test
public void testDescribeQuorumNonLeader() throws Exception {
int localId = 0;
int voter2 = localId + 1;
int voter3 = localId + 2;
int epoch = 2;
Set<Integer> voters = Utils.mkSet(localId, voter2, voter3);

RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
.withUnknownLeader(epoch)
.build();

context.deliverRequest(DescribeQuorumRequest.singletonRequest(context.metadataPartition));
context.pollUntilResponse();

DescribeQuorumResponseData responseData = context.collectDescribeQuorumResponse();
assertEquals(Errors.NONE, Errors.forCode(responseData.errorCode()));

assertEquals(1, responseData.topics().size());
DescribeQuorumResponseData.TopicData topicData = responseData.topics().get(0);
assertEquals(context.metadataPartition.topic(), topicData.topicName());

assertEquals(1, topicData.partitions().size());
DescribeQuorumResponseData.PartitionData partitionData = topicData.partitions().get(0);
assertEquals(context.metadataPartition.partition(), partitionData.partitionIndex());
assertEquals(Errors.NOT_LEADER_OR_FOLLOWER, Errors.forCode(partitionData.errorCode()));
}

@Test
public void testDescribeQuorum() throws Exception {
int localId = 0;
Expand Down
27 changes: 14 additions & 13 deletions raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
*/
package org.apache.kafka.raft;

import java.util.function.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.memory.MemoryPool;
Expand Down Expand Up @@ -57,8 +56,8 @@
import org.apache.kafka.raft.internals.BatchBuilder;
import org.apache.kafka.raft.internals.StringSerde;
import org.apache.kafka.server.common.serialization.RecordSerde;
import org.apache.kafka.snapshot.SnapshotReader;
import org.apache.kafka.snapshot.RawSnapshotWriter;
import org.apache.kafka.snapshot.SnapshotReader;
import org.apache.kafka.test.TestCondition;
import org.apache.kafka.test.TestUtils;

Expand All @@ -76,6 +75,7 @@
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import static org.apache.kafka.raft.RaftUtil.hasValidTopicPartition;
Expand Down Expand Up @@ -128,7 +128,7 @@ public static final class Builder {
private final QuorumStateStore quorumStateStore = new MockQuorumStateStore();
private final MockableRandom random = new MockableRandom(1L);
private final LogContext logContext = new LogContext();
private final MockLog log = new MockLog(METADATA_PARTITION, Uuid.METADATA_TOPIC_ID, logContext);
private final MockLog log = new MockLog(METADATA_PARTITION, Uuid.METADATA_TOPIC_ID, logContext);
private final Set<Integer> voters;
private final OptionalInt localId;

Expand Down Expand Up @@ -440,31 +440,32 @@ void assertResignedLeader(int epoch, int leaderId) throws IOException {
assertEquals(ElectionState.withElectedLeader(epoch, leaderId, voters), quorumStateStore.readElectionState());
}

int assertSentDescribeQuorumResponse(
int leaderId,
int leaderEpoch,
long highWatermark,
List<ReplicaState> voterStates,
List<ReplicaState> observerStates
) {
DescribeQuorumResponseData collectDescribeQuorumResponse() {
List<RaftResponse.Outbound> sentMessages = drainSentResponses(ApiKeys.DESCRIBE_QUORUM);
assertEquals(1, sentMessages.size());
RaftResponse.Outbound raftMessage = sentMessages.get(0);
assertTrue(
raftMessage.data() instanceof DescribeQuorumResponseData,
"Unexpected request type " + raftMessage.data());
DescribeQuorumResponseData response = (DescribeQuorumResponseData) raftMessage.data();
return (DescribeQuorumResponseData) raftMessage.data();
}

void assertSentDescribeQuorumResponse(
int leaderId,
int leaderEpoch,
long highWatermark,
List<ReplicaState> voterStates,
List<ReplicaState> observerStates
) {
DescribeQuorumResponseData response = collectDescribeQuorumResponse();
DescribeQuorumResponseData expectedResponse = DescribeQuorumResponse.singletonResponse(
metadataPartition,
leaderId,
leaderEpoch,
highWatermark,
voterStates,
observerStates);

assertEquals(expectedResponse, response);
return raftMessage.correlationId();
}

int assertSentVoteRequest(int epoch, int lastEpoch, long lastEpochOffset, int numVoteReceivers) {
Expand Down