You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "frankvicky (via GitHub)" <gi...@apache.org> on 2026/06/03 15:47:28 UTC
[PR] KAFKA-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
frankvicky opened a new pull request, #22464:
URL: https://github.com/apache/kafka/pull/22464
Sub-task of [KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin)
JIRA: KAFKA-20621
--
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-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22464:
URL: https://github.com/apache/kafka/pull/22464#discussion_r3356120307
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2191,7 +2207,7 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
response.setStatus(returnedStatus);
- return new CoordinatorResult<>(records, new StreamsGroupHeartbeatResult(response, internalTopicsToBeCreated));
+ return new CoordinatorResult<>(records, new StreamsGroupHeartbeatResult(response, internalTopicsToBeCreated, group.currentTopologyEpoch()));
Review Comment:
`group.currentTopologyEpoch()` reads the group topology here, but we may have updated the topology and appended a record for it to the group. I think this should use `updatedTopology.topologyEpoch()` instead.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpers.java:
##########
Review Comment:
Not sure if we should have this overload. It seems like we may easily make a mistake here, call this overload and reset our topology epoch metadata. Are we saving a lot of changes by introducing it?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -1223,17 +1224,15 @@ public CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>>
groupsByTopicPartition.forEach((topicPartition, groupList) -> {
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> future =
- runtime.scheduleReadOperation(
- "streams-group-describe",
- topicPartition,
- (coordinator, lastCommittedOffset) -> coordinator.streamsGroupDescribe(groupList, lastCommittedOffset)
- ).exceptionally(exception -> handleOperationException(
- "streams-group-describe",
- groupList,
- exception,
- (error, __) -> StreamsGroupDescribeRequest.getErrorDescribedGroupList(groupList, error),
- log
- ));
+ runtime.scheduleReadOperation("streams-group-describe",
Review Comment:
nit: somehow, this was reformatted
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupDescribeResult.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams;
+
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Bundles the per-group describe response data with the persisted topology epoch the plugin has stored
+ * for each successfully described group (KIP-1331). The stored epoch lets the service layer decide
+ * whether to call the topology description plugin's {@code getTopology} for a group — the plugin is
+ * only consulted when {@code storedTopologyEpoch == currentTopologyEpoch}.
Review Comment:
Minor: `storedTopologyEpoch == currentTopologyEpoch` also holds when both are -1 (brand-new group, nothing pushed by members and nothing stored), where the plugin has nothing to serve. Should the guard also require `storedTopologyEpoch != -1`?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpersTest.java:
##########
@@ -283,6 +283,39 @@ public void testNewStreamsGroupMetadataRecord() {
)));
}
+ @Test
+ public void testNewStreamsGroupMetadataRecordWithTopologyDescriptionEpochs() {
+ // KIP-1331: the 7-arg overload persists storedTopologyEpoch and lastFailedTopologyEpoch.
+ CoordinatorRecord expectedRecord = CoordinatorRecord.record(
+ new StreamsGroupMetadataKey()
+ .setGroupId(GROUP_ID),
+ new ApiMessageAndVersion(
+ new StreamsGroupMetadataValue()
+ .setEpoch(42)
+ .setMetadataHash(43)
+ .setValidatedTopologyEpoch(44)
+ .setLastAssignmentConfigs(List.of())
+ .setStoredTopologyEpoch(7)
+ .setLastFailedTopologyEpoch(5),
+ (short) 0
+ )
+ );
+
+ assertEquals(expectedRecord, StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ GROUP_ID, 42, 43, 44, Map.of(), 7, 5));
+ }
+
+ @Test
+ public void testNewStreamsGroupMetadataRecordFiveArgOverloadDefaultsToMinusOne() {
+ // The 5-arg back-compat overload must persist -1 for both new fields so a routine call site
+ // that hasn't been retrofitted does not accidentally clear a previously-stored epoch.
Review Comment:
The reasoning reads backwards: writing -1 doesn't preserve a previously-stored epoch, it clears it on replay (`setStoredTopologyEpoch(-1)`). The actual invariant is that every production epoch-bump/fence path must use the 7-arg overload — the 5-arg one persisting -1 is the unsafe fallback, not a safe one. Worth rewording so the comment doesn't imply -1 is preserving.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -737,17 +738,21 @@ public List<ShareGroupDescribeResponseData.DescribedGroup> shareGroupDescribe(
* @param groupIds The IDs of the groups to describe.
* @param committedOffset A specified committed offset corresponding to this shard.
*
- * @return A list containing the StreamsGroupDescribeResponseData.DescribedGroup.
- * If a group is not found, the DescribedGroup will contain the error code and message.
+ * @return A {@link StreamsGroupDescribeResult} bundling the described groups with per-group
+ * storedTopologyEpoch (KIP-1331). If a group is not found, its DescribedGroup carries the
+ * error code and message and is omitted from the stored-epoch map.
*/
- public List<StreamsGroupDescribeResponseData.DescribedGroup> streamsGroupDescribe(
+ public StreamsGroupDescribeResult streamsGroupDescribe(
List<String> groupIds,
long committedOffset
) {
final List<StreamsGroupDescribeResponseData.DescribedGroup> describedGroups = new ArrayList<>();
+ final Map<String, Integer> groupIdToStoredTopologyEpochs = new HashMap<>();
groupIds.forEach(groupId -> {
try {
- describedGroups.add(streamsGroup(groupId, committedOffset).asDescribedGroup(committedOffset));
+ StreamsGroup group = streamsGroup(groupId, committedOffset);
+ describedGroups.add(group.asDescribedGroup(committedOffset));
+ groupIdToStoredTopologyEpochs.put(groupId, group.storedTopologyEpoch());
Review Comment:
I think we want a consistent view of the group here, so we need to pass `committedOffset` into `group.storedTopologyEpoch`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -900,16 +901,34 @@ public List<ConsumerGroupDescribeResponseData.DescribedGroup> consumerGroupDescr
*
* @param groupIds The IDs of the groups to describe.
*
- * @return A list containing the StreamsGroupDescribeResponseData.DescribedGroup.
+ * @return A {@link StreamsGroupDescribeResult} containing the described groups and per-group
+ * stored topology epoch (KIP-1331).
*
*/
- public List<StreamsGroupDescribeResponseData.DescribedGroup> streamsGroupDescribe(
+ public StreamsGroupDescribeResult streamsGroupDescribe(
List<String> groupIds,
long committedOffset
) {
return groupMetadataManager.streamsGroupDescribe(groupIds, committedOffset);
}
+ /**
+ * Validates that a streams group exists and that the given member is a current member of it
+ * (KIP-1331). Read-only; runs against the live state machine, so the caller must schedule this
+ * on the coordinator runtime like any other read.
+ *
+ * @param groupId The group ID.
+ * @param memberId The member ID.
+ * @throws GroupIdNotFoundException if the group does not exist.
+ * @throws UnknownMemberIdException if the member is not in the group.
+ */
+ public void validateStreamsGroupMember(
+ String groupId,
+ String memberId
+ ) {
+ groupMetadataManager.validateStreamsGroupMember(groupId, memberId);
Review Comment:
It probably does not make a big difference in this case, but I think this should carry a `committedOffset` to avoid reading potentially uncommitted fencing / leaves.
--
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-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #22464:
URL: https://github.com/apache/kafka/pull/22464#discussion_r3361787382
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -242,6 +254,10 @@ public StreamsGroup(
this.members = new TimelineHashMap<>(snapshotRegistry, 0);
this.staticMembers = new TimelineHashMap<>(snapshotRegistry, 0);
this.validatedTopologyEpoch = new TimelineInteger(snapshotRegistry);
+ this.storedTopologyEpoch = new TimelineInteger(snapshotRegistry);
+ this.storedTopologyEpoch.set(-1);
+ this.lastFailedTopologyEpoch = new TimelineInteger(snapshotRegistry);
+ this.lastFailedTopologyEpoch.set(-1);
Review Comment:
`validatedTopologyEpoch` is left at the `TimelineInteger` default (0), but the StreamsGroup metadata schema and heartbeat logic use `-1` as the sentinel for “no validated topology yet”. This can cause `validatedTopologyEpoch != group.validatedTopologyEpoch()` to evaluate true on a newly created in-memory group (or any group created outside replay), triggering an unnecessary epoch bump / rebalance decision. Initialize `validatedTopologyEpoch` to `-1` in the constructor for consistency with the message default and the new topology-epoch fields.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpersTest.java:
##########
@@ -283,6 +283,39 @@ public void testNewStreamsGroupMetadataRecord() {
)));
}
+ @Test
+ public void testNewStreamsGroupMetadataRecordWithTopologyDescriptionEpochs() {
+ // KIP-1331: the 7-arg overload persists storedTopologyEpoch and lastFailedTopologyEpoch.
+ CoordinatorRecord expectedRecord = CoordinatorRecord.record(
+ new StreamsGroupMetadataKey()
+ .setGroupId(GROUP_ID),
+ new ApiMessageAndVersion(
+ new StreamsGroupMetadataValue()
+ .setEpoch(42)
+ .setMetadataHash(43)
+ .setValidatedTopologyEpoch(44)
+ .setLastAssignmentConfigs(List.of())
+ .setStoredTopologyEpoch(7)
+ .setLastFailedTopologyEpoch(5),
+ (short) 0
+ )
+ );
+
+ assertEquals(expectedRecord, StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(
+ GROUP_ID, 42, 43, 44, Map.of(), 7, 5));
+ }
+
+ @Test
+ public void testNewStreamsGroupMetadataRecordFiveArgOverloadDefaultsToMinusOne() {
+ // The 5-arg back-compat overload must persist -1 for both new fields so a routine call site
+ // that hasn't been retrofitted does not accidentally clear a previously-stored epoch.
Review Comment:
The comment on the 5-arg `newStreamsGroupMetadataRecord` overload is misleading: persisting `-1` for the new fields represents the default/unknown state, and if such a record were written after a non-(-1) value it would overwrite it (as your replay test later in `GroupMetadataManagerTest` demonstrates). Updating this comment will prevent future readers from assuming the overload is safe to use when they intend to preserve stored epochs.
--
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-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #22464:
URL: https://github.com/apache/kafka/pull/22464#discussion_r3371552600
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -737,17 +738,23 @@ public List<ShareGroupDescribeResponseData.DescribedGroup> shareGroupDescribe(
* @param groupIds The IDs of the groups to describe.
* @param committedOffset A specified committed offset corresponding to this shard.
*
- * @return A list containing the StreamsGroupDescribeResponseData.DescribedGroup.
- * If a group is not found, the DescribedGroup will contain the error code and message.
+ * @return A {@link StreamsGroupDescribeResult} bundling the described groups with per-group
+ * storedTopologyEpoch (KIP-1331). If a group is not found, its DescribedGroup carries the
+ * error code and message and is omitted from the stored-epoch map.
*/
- public List<StreamsGroupDescribeResponseData.DescribedGroup> streamsGroupDescribe(
+ public StreamsGroupDescribeResult streamsGroupDescribe(
List<String> groupIds,
long committedOffset
) {
final List<StreamsGroupDescribeResponseData.DescribedGroup> describedGroups = new ArrayList<>();
+ final Map<String, Integer> groupIdToStoredTopologyEpochs = new HashMap<>();
groupIds.forEach(groupId -> {
try {
- describedGroups.add(streamsGroup(groupId, committedOffset).asDescribedGroup(committedOffset));
+ StreamsGroup group = streamsGroup(groupId, committedOffset);
+ describedGroups.add(group.asDescribedGroup(committedOffset));
+ // KIP-1331: read the stored epoch at the same committedOffset so describedGroup and the
Review Comment:
Remove the inline comment
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -646,6 +688,56 @@ public void setValidatedTopologyEpoch(int validatedTopologyEpoch) {
maybeUpdateGroupState();
}
+ /**
+ * @return The topology epoch most recently accepted by the topology description plugin, or -1 if none.
+ */
+ public int storedTopologyEpoch() {
+ return storedTopologyEpoch.get();
+ }
+
+ /**
+ * @param committedOffset A committed offset corresponding to the desired snapshot.
+ * @return The topology epoch most recently accepted by the topology description plugin at the given
Review Comment:
"accepted" is probably not the perfect word here.
I suppose "successfully stored" would describe it better.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2191,7 +2209,10 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
response.setStatus(returnedStatus);
- return new CoordinatorResult<>(records, new StreamsGroupHeartbeatResult(response, internalTopicsToBeCreated));
+ // KIP-1331: use updatedTopology.topologyEpoch() rather than group.currentTopologyEpoch(). On the first
Review Comment:
Remove this inline comment. I don't need to keep a history of every discussion thread in inline comments
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpersTest.java:
##########
@@ -280,7 +280,29 @@ public void testNewStreamsGroupMetadataRecord() {
assertEquals(expectedRecord, StreamsCoordinatorRecordHelpers.newStreamsGroupMetadataRecord(GROUP_ID, 42, 43, 44, Map.of(
"num.standby.replicas", "2"
- )));
+ ), -1, -1));
+ }
+
+ @Test
+ public void testNewStreamsGroupMetadataRecordWithTopologyDescriptionEpochs() {
+ // KIP-1331: the 7-arg overload persists storedTopologyEpoch and lastFailedTopologyEpoch.
Review Comment:
Remove inline comment
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -8015,6 +8048,29 @@ private CoordinatorResult<HeartbeatResponseData, CoordinatorRecord> classicGroup
);
}
+ /**
+ * Validates that a streams group exists and that the given member is a current member of it.
+ * Used by the StreamsGroupTopologyDescriptionUpdate RPC handler (KIP-1331) to enforce the
Review Comment:
THis seems like a generic valdiation function for me, we don't need to mention KIP-1331 here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -900,16 +901,37 @@ public List<ConsumerGroupDescribeResponseData.DescribedGroup> consumerGroupDescr
*
* @param groupIds The IDs of the groups to describe.
*
- * @return A list containing the StreamsGroupDescribeResponseData.DescribedGroup.
+ * @return A {@link StreamsGroupDescribeResult} containing the described groups and per-group
+ * stored topology epoch (KIP-1331).
*
*/
- public List<StreamsGroupDescribeResponseData.DescribedGroup> streamsGroupDescribe(
+ public StreamsGroupDescribeResult streamsGroupDescribe(
List<String> groupIds,
long committedOffset
) {
return groupMetadataManager.streamsGroupDescribe(groupIds, committedOffset);
}
+ /**
+ * Validates that a streams group exists and that the given member is a current member of it
Review Comment:
Seems like a generic method, we don't need to mention KIP-1331 here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4504,7 +4525,17 @@ private <T> CoordinatorResult<T, CoordinatorRecord> streamsGroupFenceMember(
// We bump the group epoch.
int groupEpoch = group.groupEpoch() + 1;
- records.add(newStreamsGroupMetadataRecord(group.groupId(), groupEpoch, group.metadataHash(), group.validatedTopologyEpoch(), group.lastAssignmentConfigs()));
+ // KIP-1331: preserve storedTopologyEpoch/lastFailedTopologyEpoch across a fence. Fencing a single member
Review Comment:
remove inline comment
##########
group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json:
##########
@@ -27,7 +27,11 @@
{ "name": "ValidatedTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": -1, "type": "int32",
"about": "The topology epoch whose topics are validated to be present in a valid configuration in the metadata." },
{ "name": "LastAssignmentConfigs","taggedVersions": "0+", "nullableVersions": "0+","tag": 1, "default": null,
- "type": "[]LastAssignmentConfig", "about": "The last used configuration parameters as key-value pairs." }
+ "type": "[]LastAssignmentConfig", "about": "The last used configuration parameters as key-value pairs." },
+ { "name": "StoredTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 2, "default": -1, "type": "int32",
Review Comment:
Sorry for raising this late: the names `StoredTopologyEpoch` / `LastFailedTopologyEpoch` read ambiguously alongside the existing `topologyEpoch` (the group's current topology epoch). The description doesn't have its own epoch — it inherits the topology's — so the field really tracks *which* topology epoch we have a stored/failed description for. Suggest renaming to `StoredDescriptionTopologyEpoch` / `FailedDescriptionTopologyEpoch` (parallel to the existing `validatedTopologyEpoch`); same applies to the Java fields/accessors/setters and the `storedTopologyEpochs` map in `StreamsGroupDescribeResult`. Should be a straight search/replace.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java:
##########
@@ -26,14 +26,28 @@
/**
* A simple record to hold the result of a StreamsGroupHeartbeat request.
*
- * @param data The data to be returned to the client.
- * @param creatableTopics The internal topics to be created.
+ * @param data The data to be returned to the client.
+ * @param creatableTopics The internal topics to be created.
+ * @param currentTopologyEpoch The topology epoch the group is operating at after this heartbeat, or -1 if the
+ * group has no topology yet. The service layer uses this to decide whether to set
+ * TopologyDescriptionRequired on the response (KIP-1331).
*/
-public record StreamsGroupHeartbeatResult(StreamsGroupHeartbeatResponseData data, Map<String, CreatableTopic> creatableTopics) {
+public record StreamsGroupHeartbeatResult(
+ StreamsGroupHeartbeatResponseData data,
+ Map<String, CreatableTopic> creatableTopics,
+ int currentTopologyEpoch
+) {
public StreamsGroupHeartbeatResult {
Objects.requireNonNull(data);
creatableTopics = Collections.unmodifiableMap(Objects.requireNonNull(creatableTopics));
}
+ public StreamsGroupHeartbeatResult(
Review Comment:
`StreamsGroupHeartbeatResult` is internal coordinator plumbing, not public API — there's no back-compat to preserve. The 2-arg overload has exactly one remaining caller (`GroupCoordinatorServiceTest.java:516`), and `testTwoArgConstructorDefaultsTopologyEpochToMinusOne` only exists to test the overload itself. Updating that one test line to pass `-1` explicitly lets us drop both the overload and the test.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupDescribeResult.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams;
+
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Bundles the per-group describe response data with the persisted topology epoch the plugin has stored
Review Comment:
I think for this comment, I'd keep the main javadoc description indepedent of KIP-1331 -- And put a shorter description of the purpose of storedTopologyEpochs into the parameter description of storedTopologyEpochs, where it makes sense to mention KIP-1331.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2084,7 +2091,18 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
int groupEpoch = group.groupEpoch();
if (bumpGroupEpoch) {
groupEpoch += 1;
- records.add(newStreamsGroupMetadataRecord(groupId, groupEpoch, metadataHash, validatedTopologyEpoch, currentAssignmentConfigs));
+ // KIP-1331: persist storedTopologyEpoch/lastFailedTopologyEpoch unchanged on a routine epoch bump.
Review Comment:
Remove this inline comment. I don't need to keep a history of every discussion thread in inline comments.
--
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-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #22464:
URL: https://github.com/apache/kafka/pull/22464#discussion_r3349976068
##########
group-coordinator/src/main/resources/common/message/StreamsGroupMetadataValue.json:
##########
@@ -27,7 +27,11 @@
{ "name": "ValidatedTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": -1, "type": "int32",
"about": "The topology epoch whose topics are validated to be present in a valid configuration in the metadata." },
{ "name": "LastAssignmentConfigs","taggedVersions": "0+", "nullableVersions": "0+","tag": 1, "default": null,
- "type": "[]LastAssignmentConfig", "about": "The last used configuration parameters as key-value pairs." }
+ "type": "[]LastAssignmentConfig", "about": "The last used configuration parameters as key-value pairs." },
+ { "name": "StoredTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 2, "default": -1, "type": "int32",
+ "about": "The topology epoch whose description is currently stored in the topology description plugin, or -1 if none is stored." },
+ { "name": "LastFailedTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 3, "default": -1, "type": "int32",
+ "about": "The topology epoch whose description push the plugin permanently rejected (signalled by StreamsTopologyDescriptionPermanentFailureException), or -1 if none. Heartbeat-path solicitation is suppressed while this equals the current topology epoch, to avoid hot-looping." }
Review Comment:
Used tags 2 and 3 (not 3 and 4) — Kafka's message generator enforces contiguous tag IDs, and tag 2 was unused.
--
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-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #22464:
URL: https://github.com/apache/kafka/pull/22464#issuecomment-4629367137
btw, I would be useful to have an actual PR description that describes the change conceptually
--
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-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #22464:
URL: https://github.com/apache/kafka/pull/22464#discussion_r3361534982
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupDescribeResult.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams;
+
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Bundles the per-group describe response data with the persisted topology epoch the plugin has stored
+ * for each successfully described group (KIP-1331). The stored epoch lets the service layer decide
+ * whether to call the topology description plugin's {@code getTopology} for a group — the plugin is
+ * only consulted when {@code storedTopologyEpoch == currentTopologyEpoch}.
Review Comment:
Yes, make sense.
--
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-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #22464:
URL: https://github.com/apache/kafka/pull/22464
--
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-20621: Add StoredTopologyEpoch and LastFailedTopologyEpoch to StreamsGroup model and metadata record [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #22464:
URL: https://github.com/apache/kafka/pull/22464#discussion_r3372372121
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupDescribeResult.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams;
+
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Bundles the per-group describe response data with the persisted topology epoch the plugin has stored
Review Comment:
Make sense
--
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