You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2025/05/24 05:46:42 UTC
[PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group (wip) [kafka]
FrankYang0529 opened a new pull request, #19802:
URL: https://github.com/apache/kafka/pull/19802
* Use metadata hash to replace subscription metadata.
* Remove `StreamsGroupPartitionMetadataKey` and `StreamsGroupPartitionMetadataValue`.
* Check whether `configuredTopology` is empty. If it's, call `InternalTopicManager.configureTopics` and set the result to the group.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #19802:
URL: https://github.com/apache/kafka/pull/19802
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group (wip) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2106715246
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -582,54 +592,44 @@ public Set<String> currentWarmupTaskProcessIds(
}
/**
- * @return An immutable map of partition metadata for each topic that are inputs for this streams group.
+ * @return The metadata hash.
*/
- public Map<String, TopicMetadata> partitionMetadata() {
- return Collections.unmodifiableMap(partitionMetadata);
+ public long metadataHash() {
+ return metadataHash.get();
}
/**
- * Updates the partition metadata. This replaces the previous one.
+ * Updates the metadata hash.
*
- * @param partitionMetadata The new partition metadata.
+ * @param metadataHash The new metadata hash.
*/
- public void setPartitionMetadata(
- Map<String, TopicMetadata> partitionMetadata
- ) {
- this.partitionMetadata.clear();
- this.partitionMetadata.putAll(partitionMetadata);
- maybeUpdateConfiguredTopology();
- maybeUpdateGroupState();
Review Comment:
ConfiguredTopology can be derived completely from the records for `PartitionMetadata` and `Topology`. So we just need to cache it, and can recreate it upon the first heartbeat after failover. We could consider persisting it, but that would just mean storing somewhat duplicate data. `state` is also derived from the other records.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2115184350
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -582,54 +592,44 @@ public Set<String> currentWarmupTaskProcessIds(
}
/**
- * @return An immutable map of partition metadata for each topic that are inputs for this streams group.
+ * @return The metadata hash.
*/
- public Map<String, TopicMetadata> partitionMetadata() {
- return Collections.unmodifiableMap(partitionMetadata);
+ public long metadataHash() {
+ return metadataHash.get();
}
/**
- * Updates the partition metadata. This replaces the previous one.
+ * Updates the metadata hash.
*
- * @param partitionMetadata The new partition metadata.
+ * @param metadataHash The new metadata hash.
*/
- public void setPartitionMetadata(
- Map<String, TopicMetadata> partitionMetadata
- ) {
- this.partitionMetadata.clear();
- this.partitionMetadata.putAll(partitionMetadata);
- maybeUpdateConfiguredTopology();
- maybeUpdateGroupState();
Review Comment:
@lucasbru, I add `metadataHash` to `ConfiguredTopology` and keep `TimelineLong` in `StreamsGroup`. Could you review again when you have time? Thank you.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group (wip) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2107522589
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -582,54 +592,44 @@ public Set<String> currentWarmupTaskProcessIds(
}
/**
- * @return An immutable map of partition metadata for each topic that are inputs for this streams group.
+ * @return The metadata hash.
*/
- public Map<String, TopicMetadata> partitionMetadata() {
- return Collections.unmodifiableMap(partitionMetadata);
+ public long metadataHash() {
+ return metadataHash.get();
}
/**
- * Updates the partition metadata. This replaces the previous one.
+ * Updates the metadata hash.
*
- * @param partitionMetadata The new partition metadata.
+ * @param metadataHash The new metadata hash.
*/
- public void setPartitionMetadata(
- Map<String, TopicMetadata> partitionMetadata
- ) {
- this.partitionMetadata.clear();
- this.partitionMetadata.putAll(partitionMetadata);
- maybeUpdateConfiguredTopology();
- maybeUpdateGroupState();
Review Comment:
We may not be able to set metadata hash in `ConfiguredTopology`. When replaying `StreamsGroupMetadataValue`, we need to set `metadataHash` back. If we add it to `ConfiguredTopology`, that means there will have a non-empty `ConfiguredTopology` in the group and the `metadataHash` may be the same as latest computed value. Then we don't have a value to compare whether need to compute a new `ConfiguredTopology`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group (wip) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2107633695
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -582,54 +592,44 @@ public Set<String> currentWarmupTaskProcessIds(
}
/**
- * @return An immutable map of partition metadata for each topic that are inputs for this streams group.
+ * @return The metadata hash.
*/
- public Map<String, TopicMetadata> partitionMetadata() {
- return Collections.unmodifiableMap(partitionMetadata);
+ public long metadataHash() {
+ return metadataHash.get();
}
/**
- * Updates the partition metadata. This replaces the previous one.
+ * Updates the metadata hash.
*
- * @param partitionMetadata The new partition metadata.
+ * @param metadataHash The new metadata hash.
*/
- public void setPartitionMetadata(
- Map<String, TopicMetadata> partitionMetadata
- ) {
- this.partitionMetadata.clear();
- this.partitionMetadata.putAll(partitionMetadata);
- maybeUpdateConfiguredTopology();
- maybeUpdateGroupState();
Review Comment:
Ah, I did not mean as a replacement for the `TimelineLong` inside, `StreamsGroup`. We need an additional hash to be stored inside `ConfiguredTopology`, to remember for which state of the topics it was calculated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #19802:
URL: https://github.com/apache/kafka/pull/19802#issuecomment-2933708725
https://confluent-open-source-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/trunk/2025-06-02--001.c311e62d-9e3f-4594-b4e6-3fac3fa04638--1748907426--FrankYang0529--KAFKA-17747-5--1cc843467c/report.html
System test results look good, the failing tests are a problem already fixed on trunk.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2121800290
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java:
##########
@@ -54,7 +54,7 @@ public void testConstructorWithNullInternalTopicsToBeCreated() {
assertThrows(NullPointerException.class,
() -> new ConfiguredTopology(
0,
- Optional.of(new TreeMap<>()),
+ 0, Optional.of(new TreeMap<>()),
Review Comment:
```suggestion
0,
Optional.of(new TreeMap<>()),
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java:
##########
@@ -78,7 +78,7 @@ public void testConstructorWithInvalidTopologyEpoch() {
assertThrows(IllegalArgumentException.class,
() -> new ConfiguredTopology(
-1,
- Optional.of(new TreeMap<>()),
+ 0, Optional.of(new TreeMap<>()),
Review Comment:
```suggestion
0,
Optional.of(new TreeMap<>()),
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -16707,13 +16732,18 @@ public void testStreamsUpdatingMemberMetadataTriggersNewTargetAssignment() {
TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
.withTargetAssignmentEpoch(10)
.withTopology(StreamsTopology.fromHeartbeatRequest(topology))
- .withPartitionMetadata(Map.of(
- fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, 6),
- barTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(barTopicId, barTopicName, 3)
- ))
+ .withMetadataHash(groupMetadataHash)
)
.build();
+ context.groupMetadataManager.getStreamsGroupOrThrow(groupId)
Review Comment:
Same as above.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -17158,13 +17214,17 @@ public void testStreamsReconciliationProcess() {
TaskAssignmentTestUtil.mkTasks(subtopology1, 3, 4, 5),
TaskAssignmentTestUtil.mkTasks(subtopology2, 2)))
.withTargetAssignmentEpoch(10)
- .withPartitionMetadata(Map.of(
- fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, 6),
- barTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(barTopicId, barTopicName, 3)
- ))
+ .withMetadataHash(groupMetadataHash)
)
.build();
+ context.groupMetadataManager.getStreamsGroupOrThrow(groupId)
Review Comment:
Same as above.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java:
##########
@@ -42,7 +42,7 @@ public void testConstructorWithNullSubtopologies() {
assertThrows(NullPointerException.class,
() -> new ConfiguredTopology(
0,
- null,
+ 0, null,
Review Comment:
```suggestion
0,
null,
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -4434,6 +4435,15 @@ public void testUpdateStreamsGroupSizeCounter() {
.build()))
.build();
+ for (int i = 1; i < 4; i++) {
Review Comment:
It's not so nice to call `InternalTopicManager` from within this test. Could we turn this into explicit `.setConfiguredTopology` calls above? That would seem more idiomatic.
EDIT: I see that you do that a couple of times below. I guess we can keep it this way if we don't find an easy way to define those topologies statically. If this will take too much time, we could define a little follow-up ticket to get rid of the `InternalTopicManager` calls in here. Definitely get rid of the loop here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -16526,12 +16539,18 @@ public void testStreamsGroupMemberRequestingShutdownApplication() {
TaskAssignmentTestUtil.mkTasks(subtopology1, 3, 4, 5)))
.withTargetAssignmentEpoch(10)
.withTopology(StreamsTopology.fromHeartbeatRequest(topology))
- .withPartitionMetadata(Map.of(
- fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, 6)
- ))
+ .withMetadataHash(groupMetadataHash)
)
.build();
+ context.groupMetadataManager.getStreamsGroupOrThrow(groupId)
Review Comment:
Same here. If possible, let's make the ConfiguredTopology explicit here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java:
##########
@@ -90,7 +90,7 @@ public void testNoExceptionButNoSubtopologies() {
final IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
() -> new ConfiguredTopology(
1,
- Optional.empty(),
+ 0, Optional.empty(),
Review Comment:
```suggestion
0,
Optional.empty(),
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -17593,16 +17663,16 @@ public void testStreamsStreamsGroupStates() {
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupTopologyRecord(groupId, topology));
context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupMemberRecord(groupId, streamsGroupMemberBuilderWithDefaults(memberId1)
.build()));
- context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord(groupId, 11, 0));
+ context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupEpochRecord(groupId, 11, groupMetadataHash));
assertEquals(StreamsGroupState.NOT_READY, context.streamsGroupState(groupId));
- context.replay(StreamsCoordinatorRecordHelpers.newStreamsGroupPartitionMetadataRecord(groupId,
- Map.of(
- fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, 6),
- barTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(barTopicId, barTopicName, 3)
- )
- ));
+ context.groupMetadataManager.getStreamsGroupOrThrow(groupId)
Review Comment:
Same as above.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -16807,13 +16848,18 @@ public void testStreamsUpdatingPartitionMetadataTriggersNewTargetAssignment() {
TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3, 4, 5)))
.withTargetAssignmentEpoch(10)
.withTopology(StreamsTopology.fromHeartbeatRequest(topology))
- .withPartitionMetadata(Map.of(
- fooTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(fooTopicId, fooTopicName, 6),
- barTopicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(barTopicId, barTopicName, 3)
- ))
+ .withMetadataHash(oldGroupMetadataHash)
)
.build();
+ context.groupMetadataManager.getStreamsGroupOrThrow(groupId)
Review Comment:
Same as above.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java:
##########
@@ -66,7 +66,7 @@ public void testConstructorWithNullTopicConfigurationException() {
assertThrows(NullPointerException.class,
() -> new ConfiguredTopology(
0,
- Optional.empty(),
+ 0, Optional.empty(),
Review Comment:
```suggestion
0,
Optional.empty(),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group (wip) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2106731186
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -582,54 +592,44 @@ public Set<String> currentWarmupTaskProcessIds(
}
/**
- * @return An immutable map of partition metadata for each topic that are inputs for this streams group.
+ * @return The metadata hash.
*/
- public Map<String, TopicMetadata> partitionMetadata() {
- return Collections.unmodifiableMap(partitionMetadata);
+ public long metadataHash() {
+ return metadataHash.get();
}
/**
- * Updates the partition metadata. This replaces the previous one.
+ * Updates the metadata hash.
*
- * @param partitionMetadata The new partition metadata.
+ * @param metadataHash The new metadata hash.
*/
- public void setPartitionMetadata(
- Map<String, TopicMetadata> partitionMetadata
- ) {
- this.partitionMetadata.clear();
- this.partitionMetadata.putAll(partitionMetadata);
- maybeUpdateConfiguredTopology();
- maybeUpdateGroupState();
Review Comment:
Thanks for the reply. I will keep working on this PR and change `streamsGroupHeartbeat` to rebuild these two fields if they are empty.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group (wip) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2105717603
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -582,54 +592,44 @@ public Set<String> currentWarmupTaskProcessIds(
}
/**
- * @return An immutable map of partition metadata for each topic that are inputs for this streams group.
+ * @return The metadata hash.
*/
- public Map<String, TopicMetadata> partitionMetadata() {
- return Collections.unmodifiableMap(partitionMetadata);
+ public long metadataHash() {
+ return metadataHash.get();
}
/**
- * Updates the partition metadata. This replaces the previous one.
+ * Updates the metadata hash.
*
- * @param partitionMetadata The new partition metadata.
+ * @param metadataHash The new metadata hash.
*/
- public void setPartitionMetadata(
- Map<String, TopicMetadata> partitionMetadata
- ) {
- this.partitionMetadata.clear();
- this.partitionMetadata.putAll(partitionMetadata);
- maybeUpdateConfiguredTopology();
- maybeUpdateGroupState();
Review Comment:
To implement [KIP-1101](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1101%3A+Trigger+rebalance+on+rack+topology+changes#KIP1101:Triggerrebalanceonracktopologychanges-StreamsGroupPartitionMetadataKey/StreamsGroupPartitionMetadataValue/ShareGroupPartitionMetadataKey/ShareGroupPartitionMetadataValue) in stream group, we have to remove `StreamsGroupPartitionMetadataKey` and `StreamsGroupPartitionMetadataValue`. This can save disk storage and the group coordinator still can trigger a rebalance when metadata hash is changed.
In stream group, it tries to calculate configured topology and update group state when `partitionMetadata` or `topology` data is updated. I wonder why it doesn't save `configuredTopology` and `state` to records, so it can recover when replaying records?
cc @lucasbru @bbejeck @chia7712
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group (wip) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2106982543
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -582,54 +592,44 @@ public Set<String> currentWarmupTaskProcessIds(
}
/**
- * @return An immutable map of partition metadata for each topic that are inputs for this streams group.
+ * @return The metadata hash.
*/
- public Map<String, TopicMetadata> partitionMetadata() {
- return Collections.unmodifiableMap(partitionMetadata);
+ public long metadataHash() {
+ return metadataHash.get();
}
/**
- * Updates the partition metadata. This replaces the previous one.
+ * Updates the metadata hash.
*
- * @param partitionMetadata The new partition metadata.
+ * @param metadataHash The new metadata hash.
*/
- public void setPartitionMetadata(
- Map<String, TopicMetadata> partitionMetadata
- ) {
- this.partitionMetadata.clear();
- this.partitionMetadata.putAll(partitionMetadata);
- maybeUpdateConfiguredTopology();
- maybeUpdateGroupState();
Review Comment:
Could you instead store the group metadata hash along with / inside ConfiguredTopology, and rebuild configuredTopology not only when it is empty, but also when the group metadata hash does not match the current group metadata hash?
Also, that would mean, when we call `setTopology` or `setPartitionMetadata` inside `StreamsGroup`, we will not call `maybeUpdateConfiguredTopology`, since it will be only be created in one place - from the `streamsGroupHeartbeat`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2122885582
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -4434,6 +4435,15 @@ public void testUpdateStreamsGroupSizeCounter() {
.build()))
.build();
+ for (int i = 1; i < 4; i++) {
Review Comment:
I moved `setConfiguredTopology` to `GroupMetadataManagerTestContext`. If a stream group has topology, the context calls `InternalTopicManager.configureTopics` and set the result to the group.
Only `testStreamsStreamsGroupStates` does `InternalTopicManager.configureTopics` in `GroupMetadataManagerTest.java`. It tried to replay records to rebuild the group. Since we don't have subscription metadata now, it has to call `InternalTopicManager.configureTopics` and set the result to configured topology.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org