You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sc...@apache.org on 2026/06/16 09:33:21 UTC
(kafka) branch 4.3 updated: KAFKA-20694: Fix share read leader epoch sentinel (#22581)
This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new e61da6429c1 KAFKA-20694: Fix share read leader epoch sentinel (#22581)
e61da6429c1 is described below
commit e61da6429c1509df00de86f5990d240d31143101
Author: Shekhar Prasad Rajak <57...@users.noreply.github.com>
AuthorDate: Tue Jun 16 14:39:56 2026 +0530
KAFKA-20694: Fix share read leader epoch sentinel (#22581)
https://issues.apache.org/jira/browse/KAFKA-20694
Fixes ReadShareGroupStateRequest handling for leaderEpoch = -1. This
value is used as a sentinel for “do not update leader epoch”, so share
coordinator read validation should not fence it as stale. The read path
now matches write-state validation behavior.
Reviewers: Sushant Mahajan <sm...@confluent.io>, Andrew Schofield
<as...@confluent.io>
---
.../org/apache/kafka/coordinator/share/ShareCoordinatorShard.java | 2 +-
.../org/apache/kafka/coordinator/share/ShareCoordinatorShardTest.java | 4 ++++
2 files changed, 5 insertions(+), 1 deletion(-)
diff --git a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java
index 9eb4ea3f0da..07f7ce9d825 100644
--- a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java
+++ b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java
@@ -812,7 +812,7 @@ public class ShareCoordinatorShard implements CoordinatorShard<CoordinatorRecord
topicId, partitionId, Errors.INVALID_REQUEST, READ_UNINITIALIZED_SHARE_PARTITION.getMessage()));
}
- if (leaderEpochMap.containsKey(mapKey) && leaderEpochMap.get(mapKey) > partitionData.leaderEpoch()) {
+ if (partitionData.leaderEpoch() != -1 && leaderEpochMap.containsKey(mapKey) && leaderEpochMap.get(mapKey) > partitionData.leaderEpoch()) {
log.error("Read request leader epoch is smaller than last recorded current: {}, requested: {}.", leaderEpochMap.get(mapKey), partitionData.leaderEpoch());
return Optional.of(ReadShareGroupStateResponse.toErrorResponseData(topicId, partitionId, Errors.FENCED_LEADER_EPOCH, Errors.FENCED_LEADER_EPOCH.message()));
}
diff --git a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorShardTest.java b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorShardTest.java
index ecdbb991554..4787e34eb0e 100644
--- a/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorShardTest.java
+++ b/share-coordinator/src/test/java/org/apache/kafka/coordinator/share/ShareCoordinatorShardTest.java
@@ -1006,6 +1006,8 @@ class ShareCoordinatorShardTest {
CoordinatorResult<ReadShareGroupStateResponseData, CoordinatorRecord> result2 = shard.readStateAndMaybeUpdateLeaderEpoch(request2);
assertTrue(result2.records().isEmpty()); // Leader epoch -1 - no update.
+ assertEquals(Errors.NONE.code(), result2.response().results().get(0).partitions().get(0).errorCode());
+ assertEquals(2, shard.getLeaderMapValue(SHARE_PARTITION_KEY));
ReadShareGroupStateRequestData request3 = new ReadShareGroupStateRequestData()
.setGroupId(GROUP_ID)
@@ -1019,6 +1021,8 @@ class ShareCoordinatorShardTest {
CoordinatorResult<ReadShareGroupStateResponseData, CoordinatorRecord> result3 = shard.readStateAndMaybeUpdateLeaderEpoch(request3);
assertTrue(result3.records().isEmpty()); // Same leader epoch - no update.
+ assertEquals(Errors.NONE.code(), result3.response().results().get(0).partitions().get(0).errorCode());
+ assertEquals(2, shard.getLeaderMapValue(SHARE_PARTITION_KEY));
verify(shard.getMetricsShard()).record(ShareCoordinatorMetrics.SHARE_COORDINATOR_WRITE_SENSOR_NAME);
}