You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chickenchickenlove (via GitHub)" <gi...@apache.org> on 2026/03/31 10:16:51 UTC
[PR] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
chickenchickenlove opened a new pull request, #21909:
URL: https://github.com/apache/kafka/pull/21909
One possible way to support server-side regex in the `StreamsGroup` Protocol is for the server to send topic IDs, and for Kafka Streams to resolve those topic IDs back to topic names on the client side before using them. The `ConsumerGroup`
Protocol already has a similar model, where the broker sends topic IDs and the client only consumes topics whose names it can resolve. This POC explores a similar approach for the `StreamsGroup` Protocol.
- In the current implementation, broken method signatures in JavaDoc and test code have not yet been fixed.
- The current implementation does not include regex source topic support itself. However, it appears that this could be completed by having the broker resolve regex source topics and send them to the client, while the client relaxes the regex-related constraints accordingly.
## Layering
This implementation can be divided into four layers.
### (1) Broker Assignment Layer
This layer produces the authoritative assignment and includes resolved topic IDs by subtopology in the response. The source of truth in this layer is the full task snapshot.
### (2) Membership / Client Control Plane Layer
StreamsMembershipManager manages the broker assignment in this layer. It operates on the full task snapshot and is responsible for owned-task reporting to the broker as well as callback sequencing.
### (3) Materialization / Fetch Layer
This layer resolves the topic IDs received from the broker into topic names using client metadata, and builds the actual fetchable TaskId -> TopicPartition plan. The source of truth in this layer is the materialized subset, and active tasks in subtopologies with unresolved source topics are excluded here.
### (4) Streams Runtime Layer
Callbacks and task control-plane behavior continue to use the full task snapshot. In contrast, the TopicPartitions used for actual runtime assignment and revocation are based on the materialized subset.
In this POC, if even one source topic ID in a given subtopology cannot be resolved to a topic name, the active tasks for that subtopology:
1. are not registered in SubscriptionState as TopicPartitions, and
2. are not included in the Streams runtime active assignment.
Once topic id -> topic name resolution becomes possible, they are included again in the materialization plan during the next maybeReconcile(), and are then reflected in fetch and runtime.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3073623162
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
The purpose of this record was to make `StreamsGroup` use topic IDs. However, my implementation intent was a bit different from the direction used in `ConsumerGroup`.
1. Similar to what `ConsumerGroup` already does, I wanted `StreamsGroup` to use topic IDs for both source topics and source repartition topics. That means topic IDs would be used not only for regex cases, but also for non-regex cases.
2. However, the current `StreamsGroupTargetAssignmentMemberValue` record does not contain topic-partition information. I thought that if we `changed StreamsGroupTargetAssignmentMemberValue` to include topic partitions, the scope of the change would become much larger. At the moment, it only contains partition information.
3. To minimize the scope of change, I chose to keep `StreamsGroupTargetAssignmentMemberValue` as it is, and instead persist the resolved topic IDs in a separate `StreamsGroupTargetAssignmentResolvedTopicIdsValue` record. Here, `StreamsGroupTargetAssignmentResolvedTopicIdsValue` persists not only regex-related topics, but all source topics and source repartition topics per subtopology.
When the task assignor assigns tasks, it uses only partition and topology information, not topic information. Based on that, I separated the task-assignment-related record from the resolved-topic-ID-related record, and structured it so that they are combined only before sending the response. In other words, they are loosely coupled as two separate records that are assembled just before the response is returned.
If we instead added topic-partition information to `StreamsGroupTargetAssignmentMemberValue`, as `ConsumerGroup` does, it could allow for a more tightly coupled design. I would need to actually try implementing it to be sure, but I agree that it could turn out to be a better direction.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3093745128
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -1062,9 +1093,14 @@ public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
* - Another reconciliation is already in progress.
*/
private void maybeReconcile() {
- if (targetAssignmentReconciled()) {
+ MaterializationPlan targetMaterializationPlan = buildMaterializationPlan(targetAssignment);
+ final SortedSet<TopicPartition> targetFetchablePartitions = targetMaterializationPlan.fetchablePartitions();
+ final SortedSet<TopicPartition> currentFetchablePartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
+ currentFetchablePartitions.addAll(subscriptionState.assignedPartitions());
+
+ if (targetAssignmentReconciled() && currentFetchablePartitions.equals(targetFetchablePartitions)) {
Review Comment:
@lucasbru
Thanks for taking the time to look at this!
As you mentioned before, it has become quite complex... 🥲
Yes, I agree with your point.. I will focus on the implementation of static membership.
And when KAFKA-20171 becomes a higher priority, I would like to revisit this work and focus on it together with you at that point. Until then, I’ll continue studying it whenever I have time and keep thinking about whether there might be a better approach!
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3072823008
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
But it seems we only need this in the regular expressions case. Why do we need to introduce in this commit?
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3073533179
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
@lucasbru
I think I understand your point.
In `ConsumerGroup`, `ConsumerGroupTargetAssignmentMemberValue` and `ConsumerGroupRegularExpressionValue` are stored separately. I understood your question as asking: if we borrow that approach for `StreamsGroup`, we would only need to add `StreamsGroupRegularExpressionValue`—so is there really a need to introduce `StreamsGroupTargetAssignmentResolvedTopicIdsValue` right now in the commit that makes `StreamsGroup` use topic IDs?
My current implementation intent is different from the direction you mentioned. (I will explain this separately below.)
To make it work that way, I think `StreamsGroupTargetAssignmentMemberValue` would need to contain the topic-partition information. Then, once the regex feature is implemented, we would need to move in the direction of additionally recording `StreamsGroupRegularExpressionValue`.
First, I would like to confirm whether I have correctly understood the intent of your question.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3074445951
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
I'm mostly asking questions about your design, I'm not saying it's not correct. We just need to think about these things, and I haven't fully thought about it. I think you are right, we want to persist the topic IDs either way, whether we are using regular expressions or not.
I don't think there can be "replay timing issues" with two separate records. However, we have to consider any order -- so if we separate the two records, they may appear in any order on the consumer changelog topic. We always need to be able to load them.
IIUC, in consumer groups, we have two independent processes. One is for resolving regular expressions -- resulting in ConsumerGroupRegularExpressionValue and one is computing the target assignment, which is using ConsumerGroupTargetAssignmentMemberValue. So they serve different purposes and may be out of sync. For example, if we have already resolved a regular expression, but not computed a new target assignment. So if we want to follow the same model as consumer groups, we need topic IDs in StreamsGroupTargetAssignmentMemberValue, or an third record StreamsGroupTargetAssignmentResolvedTopicIdsValue.
ConsumerGroupRegularExpressionValue doesn't actually seem to store topic IDs at all. It stores topic names.
It seems the metadata hash would be sufficient if we always produce a new target assignment, since we'll anyways re-resolve all topic names to topic IDs and get a new target assignment. But I agree if we continue using a previous assignment and the metadata image has changed, there may be no way to recover the topic IDs.
So all in all, I agree that we need to store the topic IDs in StreamsGroupTargetAssignmentMemberValue or StreamsGroupTargetAssignmentResolvedTopicIdsValue. But it seems we have to handle the absence of the second anyways for compatibility reasons, so I see no advantage of putting it in a separate record.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3073658098
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
It is possible that I have not fully understood your intent, so I’d like to make sure we are aligned.
- Is the introduction of topic IDs in StreamsGroup intended only for regex patterns, or more generally for the source topics and source repartition topics of active tasks?
- If it is the latter, are you envisioning a direction where `StreamsGroupTargetAssignmentMemberValue` would include topic-partition information, similar to what `ConsumerGroupTargetAssignmentMemberValue does`?
IMHO, Including topic-partition information in `StreamsGroupTargetAssignmentMemberValue` has the advantage of avoiding the need to consider replay timing issues, such as the case where the target assignment has been replayed but the resolved topic IDs have not yet been replayed. However, it seems that we would then need to address the record compatibility issue.
I just wanted to confirm that I am understanding your point correctly 🙇♂️
Please let me know your thoughts!
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3092180563
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -1062,9 +1093,14 @@ public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
* - Another reconciliation is already in progress.
*/
private void maybeReconcile() {
- if (targetAssignmentReconciled()) {
+ MaterializationPlan targetMaterializationPlan = buildMaterializationPlan(targetAssignment);
+ final SortedSet<TopicPartition> targetFetchablePartitions = targetMaterializationPlan.fetchablePartitions();
+ final SortedSet<TopicPartition> currentFetchablePartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
+ currentFetchablePartitions.addAll(subscriptionState.assignedPartitions());
+
+ if (targetAssignmentReconciled() && currentFetchablePartitions.equals(targetFetchablePartitions)) {
Review Comment:
@chickenchickenlove - thanks for your deep investigation into this topic. This is indeed becoming pretty complex. Right now, I'm not sure I will be able to shepard this change in the coming weeks and also the static membership changes, due to other obligations that are prioritized right now. I would suggest we focus on the static membership change for now. This POC can stay and we have learned something from it, but I am not sure now is the right time to pull all the changes through.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #21909: [POC] KAFKA-20171: Kafka Streams use topic ids.
URL: https://github.com/apache/kafka/pull/21909
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3045846973
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
We actually persist it too for the consumer protocol: https://github.com/apache/kafka/blob/trunk/group-coordinator/src/main/resources/common/message/ConsumerGroupRegularExpressionValue.json.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3073658098
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
It is possible that I have not fully understood your intent, so I’d like to make sure we are aligned.
- Is the introduction of topic IDs in StreamsGroup intended only for regex patterns, or more generally for the source topics and source repartition topics of active tasks?
- If it is the latter, are you envisioning a direction where StreamsGroupTargetAssignmentMemberValue would include topic-partition information, similar to what ConsumerGroupTargetAssignmentMemberValue does?
I just wanted to confirm that I am understanding your point correctly 🙇♂️
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3073533179
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
@lucasbru
I think I understand your point.
In ConsumerGroup, ConsumerGroupTargetAssignmentMemberValue and ConsumerGroupRegularExpressionValue are stored separately. I understood your question as asking: if we borrow that approach for StreamsGroup, we would only need to add StreamsGroupRegularExpressionValue—so is there really a need to introduce StreamsGroupTargetAssignmentResolvedTopicIdsValue right now in the commit that makes StreamsGroup use topic IDs?
My current implementation intent is different from the direction you mentioned. (I will explain this separately below.)
To make it work that way, I think StreamsGroupTargetAssignmentMemberValue would need to contain the topic-partition information. Then, once the regex feature is implemented, we would need to move in the direction of additionally recording StreamsGroupRegularExpressionValue.
First, I would like to confirm whether I have correctly understood the intent of your question.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3044899410
##########
clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json:
##########
@@ -63,6 +63,9 @@
"about": "Assigned standby tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Assigned warm-up tasks for this client. Null if unchanged since last heartbeat." },
+
+ { "name": "ResolvedTopicIdsBySubtopology", "type": "[]ResolvedTopicIds", "versions": "1+", "nullableVersions": "1+", "default": "null",
Review Comment:
Could this be moved into "TaskIDs"? Then we don't have top repeat all the subtopologies.
We could consider having a separate task IDs structure with topic IDs, that we only use for active tasks.
We should put in the name or description that these are all source topics (including repartition source topics) of the streams group.
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
We already persist a hash of the metadata. Why do we need this extra record?
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3080256810
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
Thanks for the careful explanation. 🙇♂️
It helped me revisit the design.
I still see one benefit of keeping `ResolvedTopicIds` separate -- if the task assignment stays the same and only the resolved topic-id snapshot changes, it minimizes the number of records that need to be rewritten.
However, I now think the downside is more important. If assignment and resolved topic IDs are stored separately, then after `TargetAssignmentBuilder.build(...)` the coordinator may not be able to precisely reconstruct which resolved topic-id snapshot a given member has already observed. Reasoning only from the assignment epoch does not seem sufficient, because a member may already have received the new assignment payload and resolved topic IDs while still carrying the previous member epoch due to pending revocations.
As I understand it, `StreamsGroup` does not always compute a new target assignment. In some cases it reuses the previous one via `fromLastTargetAssignment(...)`. So I agree that resolved topic IDs must be persisted somewhere.
From that perspective, I currently think persisting them in `StreamsGroupTargetAssignmentMemberValue` is the cleaner direction.
1. We need to handle compatibility whether resolved topic IDs are split out or embedded, so separating them does not seem to provide a clear compatibility advantage by itself.
2. Putting them in `StreamsGroupTargetAssignmentMemberValue` makes the persisted state more cohesive and easier to understand, and also aligns better with the convention used by `ConsumerGroup`.
So at this point, I think including topic IDs in `StreamsGroupTargetAssignmentMemberValue` is the better direction. It also seems natural for `StreamsGroupHeartbeatResponse` to carry them as part of the assignment payload.
I also now think `StreamsGroupHeartbeatRequest` should carry them back to the coordinator as part of the member’s reported current assignment state. The broker remains authoritative for target assignment, but the request is still the channel through which the member reports what it currently believes it owns during reconciliation. If topic IDs are treated as part of that applied assignment state, then they should be represented there as well, so that the coordinator can reason about member state consistently in both persisted state and heartbeat exchange!
What do you think?
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3045712864
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
IMHO, persisting only the metadata hash is insufficient.
The metadata hash is useful as a compact change detector, but it is not a lossless representation of the resolved topic IDs per subtopology. In other words, once we only retain the hash, we cannot reconstruct the exact resolved topic ID snapshot that corresponded to a given assignment epoch.
That matters in failover/replay scenarios. After a coordinator failover, the new coordinator can replay the log and recover the metadata hash, but it cannot derive the previously returned resolved topic IDs from that hash alone. For the client-side reconciliation flow, what we need is not just evidence that the metadata changed, but the exact broker-authoritative resolved topic ID snapshot associated with the target assignment.
So my view is that the resolved topic IDs need to be persisted as assignment-scoped state, rather than inferred later from the metadata hash.
What do you think? Please share your thoughts 🙇♂️
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3083094388
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -1062,9 +1093,14 @@ public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
* - Another reconciliation is already in progress.
*/
private void maybeReconcile() {
- if (targetAssignmentReconciled()) {
+ MaterializationPlan targetMaterializationPlan = buildMaterializationPlan(targetAssignment);
+ final SortedSet<TopicPartition> targetFetchablePartitions = targetMaterializationPlan.fetchablePartitions();
+ final SortedSet<TopicPartition> currentFetchablePartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
+ currentFetchablePartitions.addAll(subscriptionState.assignedPartitions());
+
+ if (targetAssignmentReconciled() && currentFetchablePartitions.equals(targetFetchablePartitions)) {
Review Comment:
@lucasbru
One additional area I would like to discuss more closely is the assignment behavior.
As you mentioned earlier, in the Consumer case, if the client cannot resolve the topic IDs received from the coordinator, it effectively keeps that portion of the assignment pending until those unresolved topic IDs can be resolved. I implemented a similar convention on the Streams client side, and I wanted to ask for your thoughts on that approach.
The current implementation behaves as follows:
1. Even if some topic IDs included in the assignment received from the coordinator are unresolved, the client still performs the assignment-related processing. This part remains the same as before.
2. However, if any of the topics required by a given task are unresolved, that task is not moved to the running state, and instead waits until metadata resolution catches up. On the client side, this is checked whenever `maybeReconcile()` is invoked, and once the unresolved topics become resolvable, the task can then be moved to running. For example, on one client, tasks `0_0` and `0_1` may remain not running, while tasks `1_0` and `1_1` are already running.
This behavior was originally implemented when `ResolvedTopicIds` were still carried in a separate record.
Now that we seem to be converging on the direction where the client-side assignment itself should include the topic IDs, I do think that change is necessary. What I am less sure about is whether the mechanism described above also needs to change as a result.
At the moment, my inclination is that adding topic IDs into the assignment representation is necessary, but that this unresolved-topic handling model may still remain valid.
What do you think?
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21909:
URL: https://github.com/apache/kafka/pull/21909#issuecomment-6183883398
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3045677404
##########
clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json:
##########
@@ -63,6 +63,9 @@
"about": "Assigned standby tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Assigned warm-up tasks for this client. Null if unchanged since last heartbeat." },
+
+ { "name": "ResolvedTopicIdsBySubtopology", "type": "[]ResolvedTopicIds", "versions": "1+", "nullableVersions": "1+", "default": "null",
Review Comment:
Thanks for your comments!
That is a very valuable suggestion, I think that makes sense. 😄
My initial reason for keeping them separate was to avoid coupling the assignor itself to resolved topic IDs, since task assignment is still fundamentally based on task IDs. But I agree we could keep that separation internally and still expose a more cohesive response model.
Concretely, after the assignor finishes computing the task assignment, we could build an active-task-specific response structure that additionally carries the resolved source topic IDs needed by the client. IMHO, That would seem to give us a few advantages:
1. No change to the persisted target assignment records, since the assigned tasks themselves are already logged independently. (Compatiability)
2. A simpler server/client response model, since the client would no longer need to join parallel collections by subtopology ID.
3. A clearer expression of intent, because resolved topic IDs are only used to materialize active task inputs on the client side.
I’ll put together another draft PR for this alternative so we can compare it directly with the current approach.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3080544555
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
I agree that StreamsGroupTargetAssignmentMemberValue is probably the place.
Wouldn't that mean that we also need to add the topic IDs to the CurrentAssignment 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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3083096513
##########
group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentResolvedTopicIdsValue.json:
##########
@@ -0,0 +1,36 @@
+// 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.
+
+{
+ "apiKey": 24,
+ "type": "coordinator-value",
+ "name": "StreamsGroupTargetAssignmentResolvedTopicIdsValue",
Review Comment:
Yes, I agree!
I’ll address that in the PoC PR and follow up again once I have it ready.
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21909:
URL: https://github.com/apache/kafka/pull/21909#discussion_r3069425701
##########
clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json:
##########
@@ -63,6 +63,9 @@
"about": "Assigned standby tasks for this client. Null if unchanged since last heartbeat." },
{ "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Assigned warm-up tasks for this client. Null if unchanged since last heartbeat." },
+
+ { "name": "ResolvedTopicIdsBySubtopology", "type": "[]ResolvedTopicIds", "versions": "1+", "nullableVersions": "1+", "default": "null",
Review Comment:
@lucasbru
I also prepared a separate draft PR for the version that includes resolved topic IDs within the active task structure based on your comments.
When you get a chance, please take a look. 🙇♂️
https://github.com/apache/kafka/pull/22030
--
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] [POC] KAFKA-20171: Kafka Streams use topic ids. [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21909:
URL: https://github.com/apache/kafka/pull/21909#issuecomment-4988207199
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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