You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ma...@apache.org on 2026/08/21 07:27:29 UTC
(kafka) branch 4.2 updated: MINOR: Fix fetch snapshot RPC topic partition check (#23185) (#23217)
This is an automated email from the ASF dual-hosted git repository.
omkreddy pushed a commit to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new 1356af4e8a1 MINOR: Fix fetch snapshot RPC topic partition check (#23185) (#23217)
1356af4e8a1 is described below
commit 1356af4e8a12d031a203bd827268bb397ada5334
Author: Kevin Wu <kw...@confluent.io>
AuthorDate: Fri Aug 21 00:27:22 2026 -0700
MINOR: Fix fetch snapshot RPC topic partition check (#23185) (#23217)
Cherry-pick of #23185 to the `4.2` branch.
This PR fixes the check in `FetchSnapshot` RPC handling to be more like
the existing `hasValidTopicPartition` check used by `Fetch` handlers.
Original PR: https://github.com/apache/kafka/pull/23185
---
.../org/apache/kafka/raft/KafkaRaftClient.java | 4 +-
.../main/java/org/apache/kafka/raft/RaftUtil.java | 8 ++++
.../kafka/raft/KafkaRaftClientSnapshotTest.java | 44 ++++++++++++++++++++++
.../java/org/apache/kafka/raft/RaftUtilTest.java | 41 ++++++++++++++++++++
4 files changed, 95 insertions(+), 2 deletions(-)
diff --git a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
index d073819f248..8347f3670b2 100644
--- a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
+++ b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
@@ -1922,7 +1922,7 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
return new FetchSnapshotResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code());
}
- if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ if (!hasValidTopicPartition(data)) {
return FetchSnapshotResponse.withTopLevelError(Errors.INVALID_REQUEST);
}
@@ -2078,7 +2078,7 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
return handleTopLevelError(topLevelError, responseMetadata);
}
- if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ if (!hasValidTopicPartition(data)) {
return false;
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/RaftUtil.java b/raft/src/main/java/org/apache/kafka/raft/RaftUtil.java
index d3a4f760968..eb9de4e0c93 100644
--- a/raft/src/main/java/org/apache/kafka/raft/RaftUtil.java
+++ b/raft/src/main/java/org/apache/kafka/raft/RaftUtil.java
@@ -715,6 +715,14 @@ public class RaftUtil {
data.responses().get(0).partitions().get(0).partitionIndex() == topicPartition.partition();
}
+ static boolean hasValidTopicPartition(FetchSnapshotRequestData data) {
+ return data.topics().size() == 1 && data.topics().get(0).partitions().size() == 1;
+ }
+
+ static boolean hasValidTopicPartition(FetchSnapshotResponseData data) {
+ return data.topics().size() == 1 && data.topics().get(0).partitions().size() == 1;
+ }
+
static boolean hasValidTopicPartition(VoteResponseData data, TopicPartition topicPartition) {
return data.topics().size() == 1 &&
data.topics().get(0).topicName().equals(topicPartition.topic()) &&
diff --git a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
index fd696458b80..90bc5803705 100644
--- a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
@@ -51,6 +51,7 @@ import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Stream;
+import static org.apache.kafka.raft.RaftClientTestContext.RaftProtocol.KIP_1186_PROTOCOL;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -764,6 +765,49 @@ public final class KafkaRaftClientSnapshotTest {
assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.forCode(response.errorCode()));
}
+ @Test
+ public void testFetchSnapshotRequestWithoutTopics() throws Exception {
+ int localId = randomReplicaId();
+ Set<Integer> voters = Set.of(localId, localId + 1);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withUnknownLeader(3)
+ .withRaftProtocol(KIP_1186_PROTOCOL)
+ .build();
+
+ context.unattachedToLeader();
+ context.deliverRequest(new FetchSnapshotRequestData());
+ context.pollUntilResponse();
+ context.assertSentFetchSnapshotResponse(Errors.INVALID_REQUEST);
+ }
+
+ @Test
+ public void testFetchSnapshotRequestWithoutPartitions() throws Exception {
+ int localId = randomReplicaId();
+ Set<Integer> voters = Set.of(localId, localId + 1);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withUnknownLeader(3)
+ .withRaftProtocol(KIP_1186_PROTOCOL)
+ .build();
+
+ context.unattachedToLeader();
+
+ // FetchSnapshot is invalid for a topic that doesn't contain any partition.
+ context.deliverRequest(
+ new FetchSnapshotRequestData()
+ .setTopics(
+ List.of(
+ new FetchSnapshotRequestData.TopicSnapshot()
+ .setName(context.metadataPartition.topic())
+ )
+ )
+ );
+
+ context.pollUntilResponse();
+ context.assertSentFetchSnapshotResponse(Errors.INVALID_REQUEST);
+ }
+
@ParameterizedTest
@ValueSource(booleans = { false, true })
public void testFetchSnapshotRequestAsLeader(boolean withKip853Rpc) throws Exception {
diff --git a/raft/src/test/java/org/apache/kafka/raft/RaftUtilTest.java b/raft/src/test/java/org/apache/kafka/raft/RaftUtilTest.java
index 5aacab01df6..28261f66b54 100644
--- a/raft/src/test/java/org/apache/kafka/raft/RaftUtilTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/RaftUtilTest.java
@@ -72,8 +72,10 @@ import java.util.Map;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class RaftUtilTest {
@@ -631,6 +633,45 @@ public class RaftUtilTest {
assertEquals(expectedJson, json.toString());
}
+ @Test
+ public void testHasValidTopicPartitionFetchSnapshotRequest() {
+ FetchSnapshotRequestData data = RaftUtil.singletonFetchSnapshotRequest(
+ clusterId,
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ topicPartition,
+ 1,
+ new OffsetAndEpoch(10, 1),
+ 1000,
+ 10
+ );
+ assertTrue(RaftUtil.hasValidTopicPartition(data));
+
+ data.topics().get(0).setPartitions(List.of());
+ assertFalse(RaftUtil.hasValidTopicPartition(data));
+
+ data.setTopics(List.of());
+ assertFalse(RaftUtil.hasValidTopicPartition(data));
+ }
+
+ @Test
+ public void testHasValidTopicPartitionFetchSnapshotResponse() {
+ FetchSnapshotResponseData data = RaftUtil.singletonFetchSnapshotResponse(
+ listenerName,
+ (short) 0,
+ topicPartition,
+ 1,
+ Endpoints.fromInetSocketAddresses(Map.of(listenerName, address)),
+ responsePartitionSnapshot -> responsePartitionSnapshot
+ );
+ assertTrue(RaftUtil.hasValidTopicPartition(data));
+
+ data.topics().get(0).setPartitions(List.of());
+ assertFalse(RaftUtil.hasValidTopicPartition(data));
+
+ data.setTopics(List.of());
+ assertFalse(RaftUtil.hasValidTopicPartition(data));
+ }
+
@Test
public void testAddRaftVoterRequestAckWhenCommittedNotIgnorable() {
AddRaftVoterRequestData request = new AddRaftVoterRequestData()