This is an automated email from the ASF dual-hosted git repository.
apoorvmittal10 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new cde0d5687d2 KAFKA-20743: Restrict sentinel leader and state epoch in
share coord. (#22685)
cde0d5687d2 is described below
commit cde0d5687d29e7b0bced4f5b57f234b2dd63f8a0
Author: Sushant Mahajan <[email protected]>
AuthorDate: Fri Jun 26 21:10:34 2026 +0530
KAFKA-20743: Restrict sentinel leader and state epoch in share coord.
(#22685)
* currently in ShareCoordinatorShard the writeState RPC allows negative
leader and state epoch values, readState allows negative
leaderEpoch and initState allows negative stateEpoch.
* though these epoch's could never be injected by the client and are
server controlled, still we need to harden them.
* the initState call always has the correct stateEpoch (viz groupEpoch)
by virtue of
being called from the group coordinator (both in initializing and
altering offsets).
* readState is called from
SharePartition and will always be preceded by initializeState call,
which sets the stateEpoch (non negative). Furthermore, SharePartition
also creates new leaderEpoch per leadership change.
* writeState will always be preceded by readState which will have
correctly set the leaderEpoch by virtue of SharePartition. And since
initializeState correctly sets stateEpoch - writes should contain valid
values for both the epochs.
* in light of this ShareCoordinatorShard has been tightened.
* new tests have been added to verify the change.
Reviewers: Andrew Schofield <[email protected]>, Apoorv Mittal
<[email protected]>
---
.../coordinator/share/ShareCoordinatorShard.java | 43 +++++--
.../share/ShareCoordinatorShardTest.java | 123 ++++++++++++++++++++-
2 files changed, 151 insertions(+), 15 deletions(-)
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 705ba35a35f..1f5296eeb9c 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
@@ -89,6 +89,8 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
public static final Exception NULL_TOPIC_ID = new Exception("The topic id
cannot be null.");
public static final Exception NEGATIVE_PARTITION_ID = new Exception("The
partition id cannot be a negative number.");
+ public static final Exception NEGATIVE_LEADER_EPOCH = new Exception("The
leader epoch cannot be a negative number.");
+ public static final Exception NEGATIVE_STATE_EPOCH = new Exception("The
state epoch cannot be a negative number.");
public static final Exception WRITE_UNINITIALIZED_SHARE_PARTITION = new
Exception("Write operation on uninitialized share partition not allowed.");
public static final Exception READ_UNINITIALIZED_SHARE_PARTITION = new
Exception("Read operation on uninitialized share partition not allowed.");
@@ -386,8 +388,7 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
);
// Optimization in case leaderEpoch update is not required.
- if (leaderEpoch == -1 ||
- (leaderEpochMap.get(key) != null && leaderEpochMap.get(key) ==
leaderEpoch)) {
+ if (leaderEpochMap.get(key) != null && leaderEpochMap.get(key) ==
leaderEpoch) {
return new CoordinatorResult<>(List.of(), responseData);
}
@@ -613,7 +614,8 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
/**
* Remove share partitions corresponding to the input topic ids, if
present.
- * @param deletedTopicIds The topic ids which have been deleted
+ *
+ * @param deletedTopicIds The topic ids which have been deleted
* @return A result containing relevant coordinator records and void
response
*/
public CoordinatorResult<Void, CoordinatorRecord>
maybeCleanupShareState(Set<Uuid> deletedTopicIds) {
@@ -654,10 +656,7 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
int updatesPerSnapshotLimit =
config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
ShareGroupOffset currentState = shareStateMap.get(key); // This method
assumes containsKey is true.
- int newLeaderEpoch = currentState.leaderEpoch();
- if (updateLeaderEpoch) {
- newLeaderEpoch = partitionData.leaderEpoch() != -1 ?
partitionData.leaderEpoch() : newLeaderEpoch;
- }
+ int newLeaderEpoch = updateLeaderEpoch ? partitionData.leaderEpoch() :
currentState.leaderEpoch();
if (snapshotUpdateCount.getOrDefault(key, 0) >=
updatesPerSnapshotLimit) {
// shareStateMap will have the entry as containsKey is true
@@ -747,6 +746,8 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
Uuid topicId = topicData.topicId();
int partitionId = partitionData.partition();
+ int leaderEpoch = partitionData.leaderEpoch();
+ int stateEpoch = partitionData.stateEpoch();
if (topicId == null) {
return
Optional.of(getWriteErrorCoordinatorResult(Errors.INVALID_REQUEST,
NULL_TOPIC_ID, null, partitionId));
@@ -756,17 +757,25 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
return
Optional.of(getWriteErrorCoordinatorResult(Errors.INVALID_REQUEST,
NEGATIVE_PARTITION_ID, topicId, partitionId));
}
+ if (leaderEpoch < 0) {
+ return
Optional.of(getWriteErrorCoordinatorResult(Errors.INVALID_REQUEST,
NEGATIVE_LEADER_EPOCH, topicId, partitionId));
+ }
+
+ if (stateEpoch < 0) {
+ return
Optional.of(getWriteErrorCoordinatorResult(Errors.INVALID_REQUEST,
NEGATIVE_STATE_EPOCH, topicId, partitionId));
+ }
+
SharePartitionKey mapKey = SharePartitionKey.getInstance(groupId,
topicId, partitionId);
if (!shareStateMap.containsKey(mapKey)) {
return
Optional.of(getWriteErrorCoordinatorResult(Errors.INVALID_REQUEST,
WRITE_UNINITIALIZED_SHARE_PARTITION, topicId, partitionId));
}
- if (partitionData.leaderEpoch() != -1 &&
leaderEpochMap.containsKey(mapKey) && leaderEpochMap.get(mapKey) >
partitionData.leaderEpoch()) {
+ if (leaderEpochMap.containsKey(mapKey) && leaderEpochMap.get(mapKey) >
partitionData.leaderEpoch()) {
log.error("Write request leader epoch is smaller than last
recorded current: {}, requested: {}.", leaderEpochMap.get(mapKey),
partitionData.leaderEpoch());
return
Optional.of(getWriteErrorCoordinatorResult(Errors.FENCED_LEADER_EPOCH, null,
topicId, partitionId));
}
- if (partitionData.stateEpoch() != -1 &&
stateEpochMap.containsKey(mapKey) && stateEpochMap.get(mapKey) >
partitionData.stateEpoch()) {
+ if (stateEpochMap.containsKey(mapKey) && stateEpochMap.get(mapKey) >
partitionData.stateEpoch()) {
log.info("Write request state epoch is smaller than last recorded
current: {}, requested: {}.", stateEpochMap.get(mapKey),
partitionData.stateEpoch());
return
Optional.of(getWriteErrorCoordinatorResult(Errors.FENCED_STATE_EPOCH, null,
topicId, partitionId));
}
@@ -789,6 +798,7 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
String groupId = request.groupId();
ReadShareGroupStateRequestData.ReadStateData topicData =
request.topics().get(0);
ReadShareGroupStateRequestData.PartitionData partitionData =
topicData.partitions().get(0);
+ int leaderEpoch = partitionData.leaderEpoch();
Uuid topicId = topicData.topicId();
int partitionId = partitionData.partition();
@@ -805,6 +815,12 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
topicId, partitionId, Errors.INVALID_REQUEST,
NEGATIVE_PARTITION_ID.getMessage()));
}
+ if (leaderEpoch < 0) {
+ log.error("Request leader epoch is negative.");
+ return Optional.of(ReadShareGroupStateResponse.toErrorResponseData(
+ topicId, partitionId, Errors.INVALID_REQUEST,
NEGATIVE_LEADER_EPOCH.getMessage()));
+ }
+
SharePartitionKey mapKey = SharePartitionKey.getInstance(groupId,
topicId, partitionId);
if (!shareStateMap.containsKey(mapKey)) {
@@ -813,7 +829,7 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
topicId, partitionId, Errors.INVALID_REQUEST,
READ_UNINITIALIZED_SHARE_PARTITION.getMessage()));
}
- if (partitionData.leaderEpoch() != -1 &&
leaderEpochMap.containsKey(mapKey) && leaderEpochMap.get(mapKey) >
partitionData.leaderEpoch()) {
+ if (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()));
}
@@ -910,6 +926,7 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
Uuid topicId = topicData.topicId();
int partitionId = partitionData.partition();
+ int stateEpoch = partitionData.stateEpoch();
if (topicId == null) {
return
Optional.of(getInitializeErrorCoordinatorResult(Errors.INVALID_REQUEST,
NULL_TOPIC_ID, null, partitionId));
@@ -919,8 +936,12 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
return
Optional.of(getInitializeErrorCoordinatorResult(Errors.INVALID_REQUEST,
NEGATIVE_PARTITION_ID, topicId, partitionId));
}
+ if (stateEpoch < 0) {
+ return
Optional.of(getInitializeErrorCoordinatorResult(Errors.INVALID_REQUEST,
NEGATIVE_STATE_EPOCH, topicId, partitionId));
+ }
+
SharePartitionKey key =
SharePartitionKey.getInstance(request.groupId(), topicId, partitionId);
- if (partitionData.stateEpoch() != -1 && stateEpochMap.containsKey(key)
&& stateEpochMap.get(key) > partitionData.stateEpoch()) {
+ if (stateEpochMap.containsKey(key) && stateEpochMap.get(key) >
partitionData.stateEpoch()) {
log.info("Initialize request state epoch is smaller than last
recorded current: {}, requested: {}.", stateEpochMap.get(key),
partitionData.stateEpoch());
return
Optional.of(getInitializeErrorCoordinatorResult(Errors.FENCED_STATE_EPOCH,
Errors.FENCED_STATE_EPOCH.exception(), topicId, partitionId));
}
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 7e44c9b4a6a..23558216dc5 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
@@ -411,7 +411,7 @@ class ShareCoordinatorShardTest {
}
@Test
- public void testWriteStateInvalidRequestData() {
+ public void testWriteStateInvalidPartitionRequestData() {
int partition = -1;
WriteShareGroupStateRequestData request = new
WriteShareGroupStateRequestData()
@@ -443,6 +443,72 @@ class ShareCoordinatorShardTest {
assertNull(shard.getLeaderMapValue(SHARE_PARTITION_KEY));
}
+ @Test
+ public void testWriteStateInvalidLeaderEpochRequestData() {
+ int leaderEpoch = -1;
+
+ WriteShareGroupStateRequestData request = new
WriteShareGroupStateRequestData()
+ .setGroupId(GROUP_ID)
+ .setTopics(List.of(new
WriteShareGroupStateRequestData.WriteStateData()
+ .setTopicId(TOPIC_ID)
+ .setPartitions(List.of(new
WriteShareGroupStateRequestData.PartitionData()
+ .setPartition(0)
+ .setStartOffset(0)
+ .setDeliveryCompleteCount(11)
+ .setStateEpoch(0)
+ .setLeaderEpoch(leaderEpoch)
+ .setStateBatches(List.of(new
WriteShareGroupStateRequestData.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0)))))));
+
+ CoordinatorResult<WriteShareGroupStateResponseData, CoordinatorRecord>
result = shard.writeState(request);
+
+ WriteShareGroupStateResponseData expectedData =
WriteShareGroupStateResponse.toErrorResponseData(
+ TOPIC_ID, 0, Errors.INVALID_REQUEST,
ShareCoordinatorShard.NEGATIVE_LEADER_EPOCH.getMessage());
+ List<CoordinatorRecord> expectedRecords = List.of();
+
+ assertEquals(expectedData, result.response());
+ assertEquals(expectedRecords, result.records());
+
+ assertNull(shard.getShareStateMapValue(SHARE_PARTITION_KEY));
+ assertNull(shard.getLeaderMapValue(SHARE_PARTITION_KEY));
+ }
+
+ @Test
+ public void testWriteStateInvalidStateEpochRequestData() {
+ int stateEpoch = -1;
+
+ WriteShareGroupStateRequestData request = new
WriteShareGroupStateRequestData()
+ .setGroupId(GROUP_ID)
+ .setTopics(List.of(new
WriteShareGroupStateRequestData.WriteStateData()
+ .setTopicId(TOPIC_ID)
+ .setPartitions(List.of(new
WriteShareGroupStateRequestData.PartitionData()
+ .setPartition(0)
+ .setStartOffset(0)
+ .setDeliveryCompleteCount(11)
+ .setStateEpoch(stateEpoch)
+ .setLeaderEpoch(0)
+ .setStateBatches(List.of(new
WriteShareGroupStateRequestData.StateBatch()
+ .setFirstOffset(0)
+ .setLastOffset(10)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0)))))));
+
+ CoordinatorResult<WriteShareGroupStateResponseData, CoordinatorRecord>
result = shard.writeState(request);
+
+ WriteShareGroupStateResponseData expectedData =
WriteShareGroupStateResponse.toErrorResponseData(
+ TOPIC_ID, 0, Errors.INVALID_REQUEST,
ShareCoordinatorShard.NEGATIVE_STATE_EPOCH.getMessage());
+ List<CoordinatorRecord> expectedRecords = List.of();
+
+ assertEquals(expectedData, result.response());
+ assertEquals(expectedRecords, result.records());
+
+ assertNull(shard.getShareStateMapValue(SHARE_PARTITION_KEY));
+ assertNull(shard.getLeaderMapValue(SHARE_PARTITION_KEY));
+ }
+
@Test
public void testWriteNullMetadataImage() {
initSharePartition(shard, SHARE_PARTITION_KEY);
@@ -1000,12 +1066,12 @@ class ShareCoordinatorShardTest {
.setTopicId(TOPIC_ID)
.setPartitions(List.of(new
ReadShareGroupStateRequestData.PartitionData()
.setPartition(PARTITION)
- .setLeaderEpoch(-1)
+ .setLeaderEpoch(2)
))));
CoordinatorResult<ReadShareGroupStateResponseData, CoordinatorRecord>
result2 = shard.readStateAndMaybeUpdateLeaderEpoch(request2);
- assertTrue(result2.records().isEmpty()); // Leader epoch -1 - no
update.
+ assertTrue(result2.records().isEmpty()); // Leader epoch same - no
update.
assertEquals(Errors.NONE.code(),
result2.response().results().get(0).partitions().get(0).errorCode());
assertEquals(2, shard.getLeaderMapValue(SHARE_PARTITION_KEY));
@@ -1015,7 +1081,7 @@ class ShareCoordinatorShardTest {
.setTopicId(TOPIC_ID)
.setPartitions(List.of(new
ReadShareGroupStateRequestData.PartitionData()
.setPartition(PARTITION)
- .setLeaderEpoch(-1)
+ .setLeaderEpoch(2)
))));
CoordinatorResult<ReadShareGroupStateResponseData, CoordinatorRecord>
result3 = shard.readStateAndMaybeUpdateLeaderEpoch(request3);
@@ -1026,6 +1092,29 @@ class ShareCoordinatorShardTest {
verify(shard.getMetricsShard()).record(ShareCoordinatorMetrics.SHARE_COORDINATOR_WRITE_SENSOR_NAME);
}
+ @Test
+ public void testReadStateNegativeLeaderEpochDisallowed() {
+ ReadShareGroupStateRequestData request1 = new
ReadShareGroupStateRequestData()
+ .setGroupId(GROUP_ID)
+ .setTopics(List.of(new
ReadShareGroupStateRequestData.ReadStateData()
+ .setTopicId(TOPIC_ID)
+ .setPartitions(List.of(new
ReadShareGroupStateRequestData.PartitionData()
+ .setPartition(PARTITION)
+ .setLeaderEpoch(-1)
+ ))));
+
+ CoordinatorResult<ReadShareGroupStateResponseData, CoordinatorRecord>
result = shard.readStateAndMaybeUpdateLeaderEpoch(request1);
+ short errCode =
result.response().results().get(0).partitions().get(0).errorCode();
+ String errMessage =
result.response().results().get(0).partitions().get(0).errorMessage();
+
+ assertEquals(Errors.INVALID_REQUEST.code(), errCode);
+ assertEquals(ShareCoordinatorShard.NEGATIVE_LEADER_EPOCH.getMessage(),
errMessage);
+ assertTrue(result.records().isEmpty()); // Record not generated.
+
+ assertNull(shard.getLeaderMapValue(SHARE_PARTITION_KEY));
+ verify(shard.getMetricsShard(),
times(0)).record(ShareCoordinatorMetrics.SHARE_COORDINATOR_WRITE_SENSOR_NAME);
+ }
+
@Test
public void testDeleteStateSuccess() {
DeleteShareGroupStateRequestData request = new
DeleteShareGroupStateRequestData()
@@ -1294,6 +1383,32 @@ class ShareCoordinatorShardTest {
assertEquals(expectedRecords, result.records());
}
+ @Test
+ public void testInitializeStateInvalidStateEpochRequestData() {
+ // invalid stateEpoch
+ int stateEpoch = -1;
+
+ InitializeShareGroupStateRequestData request = new
InitializeShareGroupStateRequestData()
+ .setGroupId(GROUP_ID)
+ .setTopics(List.of(new
InitializeShareGroupStateRequestData.InitializeStateData()
+ .setTopicId(TOPIC_ID)
+ .setPartitions(List.of(new
InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(0)
+ .setStateEpoch(stateEpoch)
+ .setStartOffset(1)
+ ))
+ ));
+
+ CoordinatorResult<InitializeShareGroupStateResponseData,
CoordinatorRecord> result = shard.initializeState(request);
+
+ InitializeShareGroupStateResponseData expectedData =
InitializeShareGroupStateResponse.toErrorResponseData(
+ TOPIC_ID, 0, Errors.INVALID_REQUEST,
ShareCoordinatorShard.NEGATIVE_STATE_EPOCH.getMessage());
+ List<CoordinatorRecord> expectedRecords = List.of();
+
+ assertEquals(expectedData, result.response());
+ assertEquals(expectedRecords, result.records());
+ }
+
@Test
public void testInitializeNullMetadataImage() {
shard.onMetadataUpdate(null, null);
