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 b77567478f0 MINOR: Honor higher start offset value in share write
state. (#23074)
b77567478f0 is described below
commit b77567478f0d51d135e3d1e33419d9c7cb5fee69
Author: Sushant Mahajan <[email protected]>
AuthorDate: Tue Aug 4 20:20:25 2026 +0530
MINOR: Honor higher start offset value in share write state. (#23074)
In share coordinator, there was no restriction on the start offset value
causing out of order requests to regress the value. This might result in
redelivery of some records when the SharePartition leadership changes.
This PR remedies the issue by always taking the max of current view and
incoming write state request's start offset value and uses the new value
to combine and prune the state batches. A new test has been added to
verify the behavior.
Reviewers: Apoorv Mittal <[email protected]>, Andrew Schofield
<[email protected]>
---
.../coordinator/share/ShareCoordinatorShard.java | 18 ++--
.../share/ShareCoordinatorShardTest.java | 113 ++++++++++++++++++++-
2 files changed, 118 insertions(+), 13 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 1f5296eeb9c..2acf749cccb 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
@@ -311,6 +311,9 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
* This method as called by the ShareCoordinatorService will be provided
with
* the request data which covers only a single key i.e.
group1:topic1:partition1. The implementation
* below was done keeping this in mind.
+ * <p>
+ * If the value of startOffset is lower than that of the current state,
the current state
+ * version will be used.
*
* @param request - WriteShareGroupStateRequestData for a single key
* @return CoordinatorResult(records, response)
@@ -655,13 +658,10 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
long timestamp = time.milliseconds();
int updatesPerSnapshotLimit =
config.shareCoordinatorSnapshotUpdateRecordsPerSnapshot();
ShareGroupOffset currentState = shareStateMap.get(key); // This method
assumes containsKey is true.
-
int newLeaderEpoch = updateLeaderEpoch ? partitionData.leaderEpoch() :
currentState.leaderEpoch();
+ long newStartOffset = Math.max(partitionData.startOffset(),
currentState.startOffset());
if (snapshotUpdateCount.getOrDefault(key, 0) >=
updatesPerSnapshotLimit) {
- // shareStateMap will have the entry as containsKey is true
- long newStartOffset = partitionData.startOffset() == -1 ?
currentState.startOffset() : partitionData.startOffset();
-
// Since the number of update records for this share part key
exceeds snapshotUpdateRecordsPerSnapshot
// or state epoch has incremented, we should be creating a share
snapshot record.
// The incoming partition data could have overlapping state
batches, we must merge them.
@@ -685,10 +685,10 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
key.groupId(), key.topicId(), partitionData.partition(),
new ShareGroupOffset.Builder()
.setSnapshotEpoch(currentState.snapshotEpoch()) // Use
same snapshotEpoch as last share snapshot.
- .setStartOffset(partitionData.startOffset())
+ .setStartOffset(newStartOffset) // Prevents start offset
from regressing.
.setDeliveryCompleteCount(partitionData.deliveryCompleteCount())
.setLeaderEpoch(newLeaderEpoch)
- .setStateBatches(mergeBatches(List.of(), partitionData))
+ .setStateBatches(mergeBatches(List.of(), partitionData,
newStartOffset))
.build());
}
}
@@ -719,12 +719,6 @@ public class ShareCoordinatorShard implements
CoordinatorShard<CoordinatorRecord
);
}
- private List<PersisterStateBatch> mergeBatches(
- List<PersisterStateBatch> soFar,
- WriteShareGroupStateRequestData.PartitionData partitionData) {
- return mergeBatches(soFar, partitionData, partitionData.startOffset());
- }
-
private List<PersisterStateBatch> mergeBatches(
List<PersisterStateBatch> soFar,
WriteShareGroupStateRequestData.PartitionData partitionData,
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 23558216dc5..95ba59dc38d 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
@@ -163,6 +163,7 @@ class ShareCoordinatorShardTest {
shard.replay(0L, 0L, (short) 0, result.records().get(0));
+ int readStateEpoch =
result.response().results().get(0).partitions().get(0).stateEpoch();
WriteShareGroupStateRequestData request = new
WriteShareGroupStateRequestData()
.setGroupId(GROUP_ID)
.setTopics(List.of(new
WriteShareGroupStateRequestData.WriteStateData()
@@ -171,7 +172,7 @@ class ShareCoordinatorShardTest {
.setPartition(PARTITION)
.setStartOffset(0)
.setDeliveryCompleteCount(0)
- .setStateEpoch(0)
+ .setStateEpoch(readStateEpoch)
.setLeaderEpoch(leaderEpoch)
.setStateBatches(List.of(new
WriteShareGroupStateRequestData.StateBatch()
.setFirstOffset(0)
@@ -608,6 +609,115 @@ class ShareCoordinatorShardTest {
assertEquals(1, shard.getStateEpochMapValue(SHARE_PARTITION_KEY));
}
+ /**
+ * The test validates that if a previous write state request with higher
start offset
+ * had been applied to the state machine, then a subsequent write request
+ * on the same share partition key with lower start offset will have the
start offset value
+ * superseded by the previous higher value.
+ * <p>
+ * For example,
+ * req1: {
+ * startOffset: 5,
+ * stateEpoch: 5,
+ * leaderEpoch: 3,
+ * stateBatches: [(11, 20, 1, ARCHIVED)]
+ * }
+ * <p>
+ * req2: {
+ * startOffset: 2,
+ * stateEpoch: 5,
+ * leaderEpoch: 3,
+ * stateBatches: [(0, 20, 1, AVAILABLE)]
+ * }
+ * <p>
+ * Then, final state would be
+ * result: {
+ * startOffset: 5,
+ * stateEpoch: 5,
+ * leaderEpoch: 3,
+ * stateBatches: [(5, 10, 1, AVAILABLE), (11, 20, 1, ARCHIVED)]
+ * }
+ *
+ */
+ @Test
+ public void testWriteStateDiscardsStartOffsetRegression() {
+ initSharePartition(shard, SHARE_PARTITION_KEY, 5);
+ writeAndReplayRecord(shard, 3);
+ int request1StartOffset = 5;
+ int request2StartOffset = 2;
+
+ WriteShareGroupStateRequestData request1 = new
WriteShareGroupStateRequestData()
+ .setGroupId(GROUP_ID)
+ .setTopics(List.of(new
WriteShareGroupStateRequestData.WriteStateData()
+ .setTopicId(TOPIC_ID)
+ .setPartitions(List.of(new
WriteShareGroupStateRequestData.PartitionData()
+ .setPartition(PARTITION)
+ .setStartOffset(request1StartOffset)
+ .setDeliveryCompleteCount(10)
+ .setStateEpoch(5)
+ .setLeaderEpoch(3)
+ .setStateBatches(List.of(new
WriteShareGroupStateRequestData.StateBatch()
+ .setFirstOffset(11)
+ .setLastOffset(20)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 4)))))));
+
+ WriteShareGroupStateRequestData request2 = new
WriteShareGroupStateRequestData()
+ .setGroupId(GROUP_ID)
+ .setTopics(List.of(new
WriteShareGroupStateRequestData.WriteStateData()
+ .setTopicId(TOPIC_ID)
+ .setPartitions(List.of(new
WriteShareGroupStateRequestData.PartitionData()
+ .setPartition(PARTITION)
+ .setStartOffset(request2StartOffset) // Lower start
offset in second request
+ .setDeliveryCompleteCount(10)
+ .setStateEpoch(5)
+ .setLeaderEpoch(3)
+ .setStateBatches(List.of(new
WriteShareGroupStateRequestData.StateBatch()
+ .setFirstOffset(0) // First offset lower than in
request1
+ .setLastOffset(20)
+ .setDeliveryCount((short) 1)
+ .setDeliveryState((byte) 0)))))));
+
+ CoordinatorResult<WriteShareGroupStateResponseData, CoordinatorRecord>
result = shard.writeState(request1);
+ shard.replay(0L, 0L, (short) 0, result.records().get(0));
+
+ WriteShareGroupStateResponseData expectedData =
WriteShareGroupStateResponse.toResponseData(TOPIC_ID, PARTITION);
+ List<CoordinatorRecord> expectedRecords =
List.of(ShareCoordinatorRecordHelpers.newShareUpdateRecord(
+ GROUP_ID, TOPIC_ID, PARTITION,
ShareGroupOffset.fromRequest(request1.topics().get(0).partitions().get(0),
TIME.milliseconds())
+ ));
+
+ assertEquals(expectedData, result.response());
+ assertEquals(expectedRecords, result.records());
+
+ // Verify request2
+ CoordinatorResult<WriteShareGroupStateResponseData, CoordinatorRecord>
result2 = shard.writeState(request2);
+ shard.replay(0L, 0L, (short) 0, result2.records().get(0));
+
+ WriteShareGroupStateResponseData expectedData2 =
WriteShareGroupStateResponse.toResponseData(TOPIC_ID, PARTITION);
+ List<CoordinatorRecord> expectedRecords2 =
List.of(ShareCoordinatorRecordHelpers.newShareUpdateRecord(
+ GROUP_ID, TOPIC_ID, PARTITION,
ShareGroupOffset.fromRequest(request2.topics().get(0).partitions().get(0),
TIME.milliseconds()).builderSupplier()
+ .setStartOffset(request1StartOffset)
+ .setStateBatches(List.of(new
PersisterStateBatch(request1StartOffset, 20, (byte) 0, (short) 1)))
+ .build()
+ ));
+
+ assertEquals(expectedData2, result2.response());
+ assertEquals(expectedRecords2, result2.records());
+
+ ShareGroupOffset expectedFinalValue = new ShareGroupOffset.Builder()
+ .setStartOffset(request1StartOffset)
+ .setStateEpoch(5)
+ .setLeaderEpoch(3)
+ .setDeliveryCompleteCount(10)
+ .setStateBatches(List.of(
+ new PersisterStateBatch(5, 10, (byte) 0, (short) 1),
+ new PersisterStateBatch(11, 20, (byte) 4, (short) 1)
+ ))
+ .build();
+
+ assertEquals(expectedFinalValue,
shard.getShareStateMapValue(SHARE_PARTITION_KEY));
+ }
+
@Test
public void testReadFailsOnUninitializedPartition() {
ReadShareGroupStateRequestData request = new
ReadShareGroupStateRequestData()
@@ -1950,6 +2060,7 @@ class ShareCoordinatorShardTest {
private void initSharePartition(ShareCoordinatorShard shard,
SharePartitionKey key) {
initSharePartition(shard, key, 0);
}
+
private void initSharePartition(ShareCoordinatorShard shard,
SharePartitionKey key, int stateEpoch) {
shard.replay(0L, 0L, (short) 0, CoordinatorRecord.record(
new ShareSnapshotKey()
