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/08/21 13:25:49 UTC
(kafka) branch trunk updated: KAFKA-20972: Change initializeState to take a single key [1/N] (#23224)
This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield 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 8f9338ba4b0 KAFKA-20972: Change initializeState to take a single key [1/N] (#23224)
8f9338ba4b0 is described below
commit 8f9338ba4b0cfaf3c58a3cb40fe30d9a69dbde17
Author: Sushant Mahajan <sm...@confluent.io>
AuthorDate: Fri Aug 21 18:55:43 2026 +0530
KAFKA-20972: Change initializeState to take a single key [1/N] (#23224)
Instead of accepting the full InitializeShareGroupStateRequestData,
initializeState (and maybeGetInitializeStateError) now take the groupId,
topicId, and PartitionData directly, since the shard only ever operates
on a single group:topic:partition key. Updated ShareCoordinatorService's
caller and ShareCoordinatorShardTest accordingly, simplifying tests to
build only the PartitionData instead of wrapping it in a full request
object.
Other RPCs will be updated separately.
Reviewers: Andrew Schofield <as...@confluent.io>
---------
Co-authored-by: Claude Sonnet 5 <no...@anthropic.com>
---
.../coordinator/share/ShareCoordinatorService.java | 8 +-
.../coordinator/share/ShareCoordinatorShard.java | 34 +++--
.../share/ShareCoordinatorShardTest.java | 140 +++++++--------------
3 files changed, 59 insertions(+), 123 deletions(-)
diff --git a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
index 33c30af8c42..71356d72599 100644
--- a/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
+++ b/share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
@@ -964,16 +964,10 @@ public class ShareCoordinatorService implements ShareCoordinator {
for (InitializeShareGroupStateRequestData.PartitionData partitionData : topicData.partitions()) {
SharePartitionKey coordinatorKey = SharePartitionKey.getInstance(request.groupId(), topicId, partitionData.partition());
- InitializeShareGroupStateRequestData requestForCurrentPartition = new InitializeShareGroupStateRequestData()
- .setGroupId(groupId)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(topicId)
- .setPartitions(List.of(partitionData))));
-
CompletableFuture<InitializeShareGroupStateResponseData> initializeFuture = runtime.scheduleWriteOperation(
"initialize-share-group-state",
topicPartitionFor(coordinatorKey),
- coordinator -> coordinator.initializeState(requestForCurrentPartition)
+ coordinator -> coordinator.initializeState(groupId, topicId, partitionData)
).exceptionally(initializeException ->
handleOperationException(
"initialize-share-group-state",
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 1da43572065..6b5a64be81e 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
@@ -539,29 +539,29 @@ public class ShareCoordinatorShard implements CoordinatorShard<CoordinatorRecord
}
/**
- * This method writes a share snapshot records corresponding to the requested topic partitions.
+ * This method writes a share snapshot records corresponding to the requested share partition.
* <p>
* This method as called by the ShareCoordinatorService will be provided with
- * the request data which covers only key i.e. group1:topic1:partition1. The implementation
- * below was done keeping this in mind.
+ * the request data for a single share partition (group, topicId and partition data).
*
- * @param request - InitializeShareGroupStateRequestData for a single key
+ * @param groupId - String representing the group id.
+ * @param topicId - Uuid representing the topic id.
+ * @param partitionData - InitializeShareGroupStateRequestData.PartitionData representing partition request data.
* @return CoordinatorResult(records, response)
*/
public CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> initializeState(
- InitializeShareGroupStateRequestData request
+ String groupId,
+ Uuid topicId,
+ InitializeShareGroupStateRequestData.PartitionData partitionData
) {
- // Records to write (with both key and value of snapshot type), response to caller
- // only one key will be there in the request by design.
- Optional<CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord>> error = maybeGetInitializeStateError(request);
+ // Records to write (with both key and value of snapshot type), response to caller.
+ Optional<CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord>> error = maybeGetInitializeStateError(groupId, topicId, partitionData);
if (error.isPresent()) {
return error.get();
}
- InitializeShareGroupStateRequestData.InitializeStateData topicData = request.topics().get(0);
- InitializeShareGroupStateRequestData.PartitionData partitionData = topicData.partitions().get(0);
- SharePartitionKey key = SharePartitionKey.getInstance(request.groupId(), topicData.topicId(), partitionData.partition());
+ SharePartitionKey key = SharePartitionKey.getInstance(groupId, topicId, partitionData.partition());
Integer currentStateEpoch = stateEpochMap.get(key);
ShareGroupOffset currentState = shareStateMap.get(key);
@@ -575,7 +575,7 @@ public class ShareCoordinatorShard implements CoordinatorShard<CoordinatorRecord
if (currentStateEpoch != null && currentStateEpoch == partitionData.stateEpoch() &&
currentState != null && currentState.startOffset() == partitionData.startOffset()) {
- log.debug("Duplicate initialize request {}. Treating as no-op.", request);
+ log.debug("Duplicate initialize request {} {} {}. Treating as no-op.", groupId, topicId, partitionData);
return new CoordinatorResult<>(List.of(), responseData);
}
@@ -933,12 +933,10 @@ public class ShareCoordinatorShard implements CoordinatorShard<CoordinatorRecord
}
private Optional<CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord>> maybeGetInitializeStateError(
- InitializeShareGroupStateRequestData request
+ String groupId,
+ Uuid topicId,
+ InitializeShareGroupStateRequestData.PartitionData partitionData
) {
- InitializeShareGroupStateRequestData.InitializeStateData topicData = request.topics().get(0);
- InitializeShareGroupStateRequestData.PartitionData partitionData = topicData.partitions().get(0);
-
- Uuid topicId = topicData.topicId();
int partitionId = partitionData.partition();
int stateEpoch = partitionData.stateEpoch();
@@ -954,7 +952,7 @@ public class ShareCoordinatorShard implements CoordinatorShard<CoordinatorRecord
return Optional.of(getInitializeErrorCoordinatorResult(Errors.INVALID_REQUEST, NEGATIVE_STATE_EPOCH, topicId, partitionId));
}
- SharePartitionKey key = SharePartitionKey.getInstance(request.groupId(), topicId, partitionId);
+ SharePartitionKey key = SharePartitionKey.getInstance(groupId, topicId, partitionId);
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 d70a958087d..d0573d89f35 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
@@ -1717,26 +1717,21 @@ class ShareCoordinatorShardTest {
@Test
public void testInitializeStateSuccess() {
- InitializeShareGroupStateRequestData request = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(PARTITION)
- .setStartOffset(10)
- .setStateEpoch(5)))
- ));
+ InitializeShareGroupStateRequestData.PartitionData partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(PARTITION)
+ .setStartOffset(10)
+ .setStateEpoch(5);
assertNull(shard.getShareStateMapValue(SHARE_PARTITION_KEY));
assertNull(shard.getStateEpochMapValue(SHARE_PARTITION_KEY));
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
result.records().forEach(record -> shard.replay(0L, 0L, (short) 0, record));
InitializeShareGroupStateResponseData expectedData = InitializeShareGroupStateResponse.toResponseData(TOPIC_ID, PARTITION);
List<CoordinatorRecord> expectedRecords = List.of(
ShareCoordinatorRecordHelpers.newShareSnapshotRecord(
- GROUP_ID, TOPIC_ID, PARTITION, ShareGroupOffset.fromRequest(request.topics().get(0).partitions().get(0), TIME.milliseconds())
+ GROUP_ID, TOPIC_ID, PARTITION, ShareGroupOffset.fromRequest(partitionData, TIME.milliseconds())
));
assertEquals(expectedData, result.response());
@@ -1748,18 +1743,13 @@ class ShareCoordinatorShardTest {
@Test
public void testInitializeStateIdempotent() {
- InitializeShareGroupStateRequestData request = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(PARTITION)
- .setStartOffset(10)
- .setStateEpoch(5)))
- ));
+ InitializeShareGroupStateRequestData.PartitionData partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(PARTITION)
+ .setStartOffset(10)
+ .setStateEpoch(5);
// First call initializes the state and writes a record.
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
result.records().forEach(record -> shard.replay(0L, 0L, (short) 0, record));
InitializeShareGroupStateResponseData expectedData = InitializeShareGroupStateResponse.toResponseData(TOPIC_ID, PARTITION);
@@ -1775,7 +1765,7 @@ class ShareCoordinatorShardTest {
// A retry of the exact same request (same stateEpoch and startOffset) should be
// treated as a no-op -- it should return the same successful response but generate
// no new record, and the in-memory state should remain unchanged.
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> retryResult = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> retryResult = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
assertEquals(expectedData, retryResult.response());
assertEquals(List.of(), retryResult.records());
@@ -1786,33 +1776,23 @@ class ShareCoordinatorShardTest {
@Test
public void testInitializeStateNotIdempotentWhenStartOffsetChanges() {
- InitializeShareGroupStateRequestData request = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(PARTITION)
- .setStartOffset(10)
- .setStateEpoch(5)))
- ));
+ InitializeShareGroupStateRequestData.PartitionData partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(PARTITION)
+ .setStartOffset(10)
+ .setStateEpoch(5);
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
result.records().forEach(record -> shard.replay(0L, 0L, (short) 0, record));
assertEquals(1, result.records().size());
// Same stateEpoch but a different startOffset should not be treated as a duplicate
// and must produce a new record.
- InitializeShareGroupStateRequestData sameEpochDifferentOffsetRequest = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(PARTITION)
- .setStartOffset(20)
- .setStateEpoch(5)))
- ));
+ InitializeShareGroupStateRequestData.PartitionData sameEpochDifferentOffsetPartitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(PARTITION)
+ .setStartOffset(20)
+ .setStateEpoch(5);
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> secondResult = shard.initializeState(sameEpochDifferentOffsetRequest);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> secondResult = shard.initializeState(GROUP_ID, TOPIC_ID, sameEpochDifferentOffsetPartitionData);
InitializeShareGroupStateResponseData expectedData = InitializeShareGroupStateResponse.toResponseData(TOPIC_ID, PARTITION);
assertEquals(expectedData, secondResult.response());
@@ -1824,16 +1804,10 @@ class ShareCoordinatorShardTest {
// invalid partition
int partition = -1;
- InitializeShareGroupStateRequestData request = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(partition)
- ))
- ));
+ InitializeShareGroupStateRequestData.PartitionData partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(partition);
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
InitializeShareGroupStateResponseData expectedData = InitializeShareGroupStateResponse.toErrorResponseData(
TOPIC_ID, partition, Errors.INVALID_REQUEST, ShareCoordinatorShard.NEGATIVE_PARTITION_ID.getMessage());
@@ -1852,16 +1826,10 @@ class ShareCoordinatorShardTest {
.build()
));
- request = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(partition)
- ))
- ));
+ partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(partition);
- result = shard.initializeState(request);
+ result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
expectedData = InitializeShareGroupStateResponse.toErrorResponseData(
TOPIC_ID, partition, Errors.FENCED_STATE_EPOCH, Errors.FENCED_STATE_EPOCH.exception().getMessage());
@@ -1876,18 +1844,12 @@ class ShareCoordinatorShardTest {
// 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)
- ))
- ));
+ InitializeShareGroupStateRequestData.PartitionData partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(0)
+ .setStateEpoch(stateEpoch)
+ .setStartOffset(1);
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
InitializeShareGroupStateResponseData expectedData = InitializeShareGroupStateResponse.toErrorResponseData(
TOPIC_ID, 0, Errors.INVALID_REQUEST, ShareCoordinatorShard.NEGATIVE_STATE_EPOCH.getMessage());
@@ -1901,16 +1863,10 @@ class ShareCoordinatorShardTest {
public void testInitializeNullMetadataImage() {
shard.onMetadataUpdate(null, null);
- InitializeShareGroupStateRequestData request = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(0)
- ))
- ));
+ InitializeShareGroupStateRequestData.PartitionData partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(0);
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
InitializeShareGroupStateResponseData expectedData = InitializeShareGroupStateResponse.toErrorResponseData(
TOPIC_ID, 0, Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.UNKNOWN_TOPIC_OR_PARTITION.message());
@@ -1925,20 +1881,14 @@ class ShareCoordinatorShardTest {
MetadataImage image = mock(MetadataImage.class);
shard.onMetadataUpdate(null, new KRaftCoordinatorMetadataImage(image));
- InitializeShareGroupStateRequestData request = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(0)
- ))
- ));
+ InitializeShareGroupStateRequestData.PartitionData partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(0);
// topic id not found in cache
TopicsImage topicsImage = mock(TopicsImage.class);
when(topicsImage.getTopic(eq(TOPIC_ID))).thenReturn(null);
when(image.topics()).thenReturn(topicsImage);
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
InitializeShareGroupStateResponseData expectedData = InitializeShareGroupStateResponse.toErrorResponseData(
TOPIC_ID, 0, Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.UNKNOWN_TOPIC_OR_PARTITION.message());
@@ -1955,14 +1905,8 @@ class ShareCoordinatorShardTest {
when(image.cluster()).thenReturn(mock(ClusterImage.class));
shard.onMetadataUpdate(null, new KRaftCoordinatorMetadataImage(image));
- InitializeShareGroupStateRequestData request = new InitializeShareGroupStateRequestData()
- .setGroupId(GROUP_ID)
- .setTopics(List.of(new InitializeShareGroupStateRequestData.InitializeStateData()
- .setTopicId(TOPIC_ID)
- .setPartitions(List.of(new InitializeShareGroupStateRequestData.PartitionData()
- .setPartition(0)
- ))
- ));
+ InitializeShareGroupStateRequestData.PartitionData partitionData = new InitializeShareGroupStateRequestData.PartitionData()
+ .setPartition(0);
// topic id found in cache
TopicImage topicImage = mock(TopicImage.class);
@@ -1973,7 +1917,7 @@ class ShareCoordinatorShardTest {
// partition id not found
when(topicsImage.getPartition(eq(TOPIC_ID), eq(0))).thenReturn(null);
- CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(request);
+ CoordinatorResult<InitializeShareGroupStateResponseData, CoordinatorRecord> result = shard.initializeState(GROUP_ID, TOPIC_ID, partitionData);
InitializeShareGroupStateResponseData expectedData = InitializeShareGroupStateResponse.toErrorResponseData(
TOPIC_ID, 0, Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.UNKNOWN_TOPIC_OR_PARTITION.message());