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 97bbffa0a89 KAFKA-20742: Start offset for newly initialized share
partitions of known topics should be 0 (#22714)
97bbffa0a89 is described below
commit 97bbffa0a89ebc3d88eb05d7e189ea1cf9798f26
Author: Sanskar Jhajharia <[email protected]>
AuthorDate: Thu Jul 2 02:52:55 2026 +0530
KAFKA-20742: Start offset for newly initialized share partitions of known
topics should be 0 (#22714)
### Summary
When a share group heartbeat detects subscribed-but-uninitialized
topic-partitions, the group coordinator builds an
`InitializeShareGroupStateParameters` request. Previously it sent
`startOffset = -1` (`PartitionFactory.UNINITIALIZED_START_OFFSET`) for
*every* partition.
`-1` is a sentinel meaning "not yet decided":
`SharePartition.startOffsetDuringInitialization()` resolves the real
offset from the group's `share.auto.offset.reset` strategy (default
`LATEST`). This is correct for a brand-new topic subscription, but
incorrect when **new partitions are added to a topic the group already
knows about**. Resolving those new partitions via `LATEST` can skip
records produced before initialization completes, causing data loss.
This change makes newly initialized partitions of an **already-known
topic** (present in the group's `initializedTopics` or
`initializingTopics`) start at offset `0`, guaranteeing no records are
missed. Partitions of a topic seen for the first time keep `-1` so that
`share.auto.offset.reset` still applies to fresh subscriptions.
### Change
In `GroupMetadataManager`:
- `maybeCreateInitializeShareGroupStateRequest` computes the set of
already-known topics (`initializedTopics ∪ initializingTopics`) from the
current persisted metadata. This is read before the heartbeat's records
are replayed, so a brand-new topic is correctly absent.
- `buildInitializeShareGroupStateRequest` selects the start offset per
topic: `0` for an already-known topic, `-1` for a first-sighting topic.
### Testing
- Added `testShareGroupHeartbeatInitializeMixedNewAndExpandedTopics`: a
single heartbeat that both adds a brand-new topic and expands an
already-initialized topic produces one request where the new topic uses
`-1` and the expanded topic's new partitions use `0` — verifying the
decision is per-topic, not per-request.
Reviewers: Sushant Mahajan <[email protected]>, Apoorv Mittal
<[email protected]>
---
.../kafka/clients/consumer/ShareConsumerTest.java | 60 ++++++++++
.../coordinator/group/GroupMetadataManager.java | 40 +++++--
.../group/GroupMetadataManagerTest.java | 129 ++++++++++++++++++++-
3 files changed, 220 insertions(+), 9 deletions(-)
diff --git
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java
index a9f45bf73b1..d4865ab6cf8 100644
---
a/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java
+++
b/clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.clients.consumer;
import kafka.server.KafkaBroker;
import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.clients.admin.RecordsToDelete;
import org.apache.kafka.clients.admin.SharePartitionOffsetInfo;
import org.apache.kafka.clients.producer.Producer;
@@ -2033,4 +2034,63 @@ public class ShareConsumerTest extends
ShareConsumerTestBase {
throw new RuntimeException(e);
}
}
+
+ @ClusterTest
+ public void testAddedPartitionsOfKnownTopicStartAtOffsetZero() throws
InterruptedException, ExecutionException {
+ String groupId = "expand-group";
+ String topic = "expand-topic";
+ createTopic(topic, 1, 1);
+
+ alterShareAutoOffsetReset(groupId, "latest");
+
+ try (Producer<byte[], byte[]> producer = createProducer();
+ Admin admin = createAdminClient();
+ ShareConsumer<byte[], byte[]> shareConsumer =
createShareConsumer(groupId)) {
+
+ // Produce 10 records to partition 0 before anyone subscribes.
+ for (int i = 0; i < 10; i++) {
+ producer.send(new ProducerRecord<>(topic, 0, null, ("p0-key-"
+ i).getBytes(), ("p0-value-" + i).getBytes()));
+ }
+ producer.flush();
+
+ // The consumer joins and polls: with "latest" it starts partition
0 at the log end, so it receives
+ // none of the pre-existing records. This also makes the group
aware of (initializes) the topic.
+ shareConsumer.subscribe(Set.of(topic));
+ int received = 0;
+ long deadline = System.currentTimeMillis() + 10000L;
+ while (System.currentTimeMillis() < deadline) {
+ received +=
shareConsumer.poll(Duration.ofMillis(1000)).count();
+ }
+ assertEquals(0, received, "latest reset should skip records
produced before the topic was known");
+
+ // Expand the topic and produce 10 records to the brand-new
partition 1, each with a distinct,
+ // known value so we can assert all of them are delivered (none
skipped by the latest reset).
+ admin.createPartitions(Map.of(topic,
NewPartitions.increaseTo(2))).all().get();
+ cluster.waitTopicCreation(topic, 2);
+
+ List<String> expectedValues = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ String value = "p1-value-" + i;
+ expectedValues.add(value);
+ producer.send(new ProducerRecord<>(topic, 1, null, ("p1-key-"
+ i).getBytes(), value.getBytes()));
+ }
+ producer.flush();
+
+ // Partition 1 belongs to an already-known topic, so it must be
initialized at offset 0 and deliver
+ // every record produced above — even though it was created under
a "latest" reset. Share consumers
+ // deliver at-least-once, so records may be redelivered; track
distinct values in a set so
+ // duplicates don't cause spurious failures, then assert the count
of distinct records.
+ Set<String> receivedValues = new HashSet<>();
+ waitForCondition(() -> {
+ ConsumerRecords<byte[], byte[]> records =
shareConsumer.poll(Duration.ofMillis(1000));
+ for (ConsumerRecord<byte[], byte[]> record : records) {
+ if (record.partition() == 1) {
+ receivedValues.add(new String(record.value()));
+ }
+ }
+ return receivedValues.size() >= expectedValues.size();
+ }, 30000L, 500L, () -> "did not receive all records from the newly
added partition");
+ assertEquals(expectedValues.size(), receivedValues.size());
+ }
+ }
}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index 238e637e05c..97a49040f56 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -3133,18 +3133,42 @@ public class GroupMetadataManager {
}
addInitializingTopicsRecords(groupId, records,
topicPartitionChangeMap);
- return Optional.of(buildInitializeShareGroupStateRequest(groupId,
groupEpoch, topicPartitionChangeMap));
+
+ // Topics already known to the group (already initialized or already
initializing) whose
+ // newly added partitions are now being initialized must start at
offset 0 to avoid losing
+ // records produced before initialization completes. Brand-new topics
(seen for the first
+ // time, present in neither map) use -1 so that the group's
share.auto.offset.reset strategy
+ // applies. This is read before the records above are replayed, so a
brand-new topic is
+ // correctly absent here.
+ Set<Uuid> alreadyKnownTopics = null;
+ if (shareGroupStatePartitionMetadata.containsKey(groupId)) {
+ ShareGroupStatePartitionMetadataInfo info =
shareGroupStatePartitionMetadata.get(groupId);
+ alreadyKnownTopics = new HashSet<>();
+ alreadyKnownTopics.addAll(info.initializedTopics().keySet());
+ alreadyKnownTopics.addAll(info.initializingTopics().keySet());
+ }
+
+ return Optional.of(buildInitializeShareGroupStateRequest(groupId,
groupEpoch, topicPartitionChangeMap, alreadyKnownTopics != null ?
alreadyKnownTopics : Set.of()));
}
- private InitializeShareGroupStateParameters
buildInitializeShareGroupStateRequest(String groupId, int groupEpoch, Map<Uuid,
InitMapValue> topicPartitions) {
+ private InitializeShareGroupStateParameters
buildInitializeShareGroupStateRequest(
+ String groupId,
+ int groupEpoch,
+ Map<Uuid, InitMapValue> topicPartitions,
+ Set<Uuid> alreadyKnownTopics
+ ) {
return new
InitializeShareGroupStateParameters.Builder().setGroupTopicPartitionData(
new GroupTopicPartitionData<>(groupId,
topicPartitions.entrySet().stream()
- .map(entry -> new TopicData<>(
- entry.getKey(),
- entry.getValue().partitions().stream()
- .map(partitionId ->
PartitionFactory.newPartitionStateData(partitionId, groupEpoch, -1))
- .toList())
- ).toList()
+ .map(entry -> {
+ // New partitions of an already-known topic start at 0;
brand-new topics use -1
+ // so that the group's share.auto.offset.reset strategy
applies.
+ long startOffset =
alreadyKnownTopics.contains(entry.getKey()) ? 0L :
PartitionFactory.UNINITIALIZED_START_OFFSET;
+ return new TopicData<>(
+ entry.getKey(),
+ entry.getValue().partitions().stream()
+ .map(partitionId ->
PartitionFactory.newPartitionStateData(partitionId, groupEpoch, startOffset))
+ .toList());
+ }).toList()
)).build();
}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index ef816ee2ac6..9f348600f4c 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -27247,12 +27247,15 @@ public class GroupMetadataManagerTest {
assertTrue(actual.isPresent());
assertRecordEquals(expected, actual.get());
+ // t1 is already initialized, so its newly added partitions (2, 3)
must start at offset 0
+ // to avoid losing records produced before initialization completes.
verifyShareGroupHeartbeatInitializeRequest(
result.response().getValue(),
Map.of(
t1Uuid,
Set.of(2, 3)
),
+ Map.of(t1Uuid, 0L),
groupId,
3,
true
@@ -27262,6 +27265,114 @@ public class GroupMetadataManagerTest {
verify(context.metrics,
times(2)).record(SHARE_GROUP_REBALANCES_SENSOR_NAME);
}
+ @Test
+ public void testShareGroupHeartbeatInitializeMixedNewAndExpandedTopics() {
+ // A single heartbeat that both adds a brand-new topic and expands an
already-initialized
+ // topic must produce one initialize request where the new topic uses
the uninitialized (-1)
+ // start offset (so share.auto.offset.reset applies) while the
expanded topic's new
+ // partitions use a fixed start offset of 0 (to avoid losing records).
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of()));
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withShareGroupAssignor(assignor)
+ .build();
+
+ Uuid t1Uuid = Uuid.randomUuid();
+ String t1Name = "t1";
+ Uuid t2Uuid = Uuid.randomUuid();
+ String t2Name = "t2";
+ Uuid t3Uuid = Uuid.randomUuid();
+ String t3Name = "t3";
+ CoordinatorMetadataImage image = new MetadataImageBuilder()
+ .addTopic(t1Uuid, t1Name, 2)
+ .addTopic(t2Uuid, t2Name, 2)
+ .buildCoordinatorMetadataImage();
+
+ String groupId = "share-group";
+
+
context.groupMetadataManager.onMetadataUpdate(mock(CoordinatorMetadataDelta.class),
image);
+
+ Uuid memberId = Uuid.randomUuid();
+ CoordinatorResult<Map.Entry<ShareGroupHeartbeatResponseData,
Optional<InitializeShareGroupStateParameters>>, CoordinatorRecord> result =
context.shareGroupHeartbeat(
+ new ShareGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId.toString())
+ .setMemberEpoch(0)
+ .setSubscribedTopicNames(List.of(t1Name, t2Name)));
+
+ // Both topics are brand-new, so both initialize at the uninitialized
(-1) start offset.
+ verifyShareGroupHeartbeatInitializeRequest(
+ result.response().getValue(),
+ Map.of(
+ t1Uuid, Set.of(0, 1),
+ t2Uuid, Set.of(0, 1)
+ ),
+ groupId,
+ 2,
+ true
+ );
+
+ context.groupMetadataManager.replay(
+ new ShareGroupStatePartitionMetadataKey()
+ .setGroupId(groupId),
+ new ShareGroupStatePartitionMetadataValue()
+ .setInitializingTopics(List.of())
+ .setInitializedTopics(List.of(
+ new
ShareGroupStatePartitionMetadataValue.TopicPartitionsInfo()
+ .setTopicId(t1Uuid)
+ .setTopicName(t1Name)
+ .setPartitions(List.of(0, 1)),
+ new
ShareGroupStatePartitionMetadataValue.TopicPartitionsInfo()
+ .setTopicId(t2Uuid)
+ .setTopicName(t2Name)
+ .setPartitions(List.of(0, 1))
+ ))
+ .setDeletingTopics(List.of())
+ );
+
+ // t1 expands from 2 to 4 partitions and a brand-new topic t3 (3
partitions) is added.
+ image = new MetadataImageBuilder()
+ .addTopic(t1Uuid, t1Name, 4)
+ .addTopic(t2Uuid, t2Name, 2)
+ .addTopic(t3Uuid, t3Name, 3)
+ .buildCoordinatorMetadataImage();
+
+
context.groupMetadataManager.onMetadataUpdate(mock(CoordinatorMetadataDelta.class),
image);
+
+ assignor.prepareGroupAssignment(new GroupAssignment(
+ Map.of(
+ memberId.toString(),
+ new MemberAssignmentImpl(
+ Map.of(
+ t1Uuid, Set.of(0, 1),
+ t2Uuid, Set.of(0, 1)
+ )
+ )
+ )
+ ));
+
+ result = context.shareGroupHeartbeat(
+ new ShareGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId.toString())
+ .setMemberEpoch(2)
+ .setSubscribedTopicNames(List.of(t1Name, t2Name, t3Name)));
+
+ // Single request: t1's new partitions {2,3} (already-initialized
topic) start at 0, while
+ // brand-new topic t3's partitions {0,1,2} use the uninitialized (-1)
start offset.
+ verifyShareGroupHeartbeatInitializeRequest(
+ result.response().getValue(),
+ Map.of(
+ t1Uuid, Set.of(2, 3),
+ t3Uuid, Set.of(0, 1, 2)
+ ),
+ Map.of(t1Uuid, 0L),
+ groupId,
+ 3,
+ true
+ );
+ }
+
@Test
public void
testShareGroupHeartbeatDoesNotBumpGroupEpochDuringAssignmentDelay() {
Uuid t1Uuid = Uuid.randomUuid();
@@ -27441,9 +27552,12 @@ public class GroupMetadataManagerTest {
))
);
+ // t1 was already in the initializing state, so retrying its
initialization uses a fixed
+ // start offset of 0 rather than re-deferring to the
share.auto.offset.reset strategy.
verifyShareGroupHeartbeatInitializeRequest(
result.response().getValue(),
Map.of(t1Uuid, Set.of(0, 1)),
+ Map.of(t1Uuid, 0L),
groupId,
3,
true
@@ -29119,6 +29233,18 @@ public class GroupMetadataManagerTest {
String groupId,
int stateEpoch,
boolean shouldExist
+ ) {
+ // Brand-new topics are expected to use the uninitialized (-1) start
offset.
+ verifyShareGroupHeartbeatInitializeRequest(initRequest,
expectedTopicPartitionsMap, Map.of(), groupId, stateEpoch, shouldExist);
+ }
+
+ private void verifyShareGroupHeartbeatInitializeRequest(
+ Optional<InitializeShareGroupStateParameters> initRequest,
+ Map<Uuid, Set<Integer>> expectedTopicPartitionsMap,
+ Map<Uuid, Long> expectedStartOffsetByTopic,
+ String groupId,
+ int stateEpoch,
+ boolean shouldExist
) {
if (shouldExist) {
assertTrue(initRequest.isPresent());
@@ -29126,10 +29252,11 @@ public class GroupMetadataManagerTest {
assertEquals(groupId, request.groupTopicPartitionData().groupId());
Map<Uuid, Set<Integer>> actualTopicPartitionsMap = new HashMap<>();
for (TopicData<PartitionStateData> topicData :
request.groupTopicPartitionData().topicsData()) {
+ long expectedStartOffset =
expectedStartOffsetByTopic.getOrDefault(topicData.topicId(), -1L);
actualTopicPartitionsMap.computeIfAbsent(topicData.topicId(),
k -> new HashSet<>())
.addAll(topicData.partitions().stream().map(partitionData
-> {
assertEquals(stateEpoch, partitionData.stateEpoch());
- assertEquals(-1, partitionData.startOffset());
+ assertEquals(expectedStartOffset,
partitionData.startOffset());
return partitionData.partition();
}).toList());
}
