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/09/27 09:41:06 UTC
[PR] KAFKA-21170: Validate Streams task assignor output before accepting target assignments. [kafka]
chickenchickenlove opened a new pull request, #23597:
URL: https://github.com/apache/kafka/pull/23597
Fixes KAFKA-21170.
A faulty custom Streams task assignor can return conflicting assignments that prevent reconciliation from completing.
Validate task IDs, active owner uniqueness, and task placement within each process before accepting the target assignment. Reject invalid output through the existing TaskAssignorException failure path, preserving the previous assignment.
--
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-21170: Validate Streams task assignor output before accepting target assignments. [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #23597:
URL: https://github.com/apache/kafka/pull/23597#issuecomment-6030248440
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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-21170: Validate Streams task assignor output before accepting target assignments. [kafka]
Posted by "mjsax (via GitHub) via jira" <ji...@kafka.apache.org>.
mjsax commented on code in PR #23597:
URL: https://github.com/apache/kafka/pull/23597#discussion_r4498537954
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
@@ -289,6 +293,72 @@ public TargetAssignmentResult build() throws TaskAssignorException {
return new TargetAssignmentResult(records, newTargetAssignment);
}
+ private void validateAssignment(Map<String, org.apache.kafka.coordinator.group.streams.TasksTuple> assignment) {
+ // The builder produces empty assignments when the topology is not ready.
+ if (!topology.isReady()) {
+ return;
+ }
+
+ final Map<String, ConfiguredSubtopology> subtopologies = topology.subtopologies().orElseThrow();
+ final Map<TaskId, String> activeOwners = new HashMap<>();
+ final Map<String, Map<TaskId, String>> processOwners = new HashMap<>();
+
+ assignment.forEach((memberId, memberAssignment) -> {
+ final String processId = members.get(memberId).processId();
+ final Map<TaskId, String> ownersOnProcess = processOwners.computeIfAbsent(processId, ignored -> new HashMap<>());
+
+ for (TaskRole role : List.of(TaskRole.ACTIVE, TaskRole.STANDBY)) {
+ final Map<String, Set<Integer>> tasks = role == TaskRole.ACTIVE
+ ? memberAssignment.activeTasks() : memberAssignment.standbyTasks();
+
+ tasks.forEach((subtopologyId, partitions) -> {
+ // Both active and standby tasks must reference a subtopology in the configured topology.
+ ConfiguredSubtopology subtopology = subtopologies.get(subtopologyId);
+ if (subtopology == null) {
+ throw new TaskAssignorException(String.format(
Review Comment:
Not sure if throwing this exception is right? Won't it kill the GC/broker?
The fall back should maybe be, to stay on the previous assignment? -- At least this is what we do elsewhere IIRC, even if I am not sure if it's the best strategy? Assume a member left the group: for this case, we would get offline partitions/tasks which are not assigned.
\cc @lucasbru @dajac
--
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-21170: Validate Streams task assignor output before accepting target assignments. [kafka]
Posted by "chickenchickenlove (via GitHub) via jira" <ji...@kafka.apache.org>.
chickenchickenlove commented on code in PR #23597:
URL: https://github.com/apache/kafka/pull/23597#discussion_r4499478241
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
@@ -289,6 +293,72 @@ public TargetAssignmentResult build() throws TaskAssignorException {
return new TargetAssignmentResult(records, newTargetAssignment);
}
+ private void validateAssignment(Map<String, org.apache.kafka.coordinator.group.streams.TasksTuple> assignment) {
+ // The builder produces empty assignments when the topology is not ready.
+ if (!topology.isReady()) {
+ return;
+ }
+
+ final Map<String, ConfiguredSubtopology> subtopologies = topology.subtopologies().orElseThrow();
+ final Map<TaskId, String> activeOwners = new HashMap<>();
+ final Map<String, Map<TaskId, String>> processOwners = new HashMap<>();
+
+ assignment.forEach((memberId, memberAssignment) -> {
+ final String processId = members.get(memberId).processId();
+ final Map<TaskId, String> ownersOnProcess = processOwners.computeIfAbsent(processId, ignored -> new HashMap<>());
+
+ for (TaskRole role : List.of(TaskRole.ACTIVE, TaskRole.STANDBY)) {
+ final Map<String, Set<Integer>> tasks = role == TaskRole.ACTIVE
+ ? memberAssignment.activeTasks() : memberAssignment.standbyTasks();
+
+ tasks.forEach((subtopologyId, partitions) -> {
+ // Both active and standby tasks must reference a subtopology in the configured topology.
+ ConfiguredSubtopology subtopology = subtopologies.get(subtopologyId);
+ if (subtopology == null) {
+ throw new TaskAssignorException(String.format(
Review Comment:
@mjsax Thanks for taking the time to look into this!
When a new assignment is computed while processing a heartbeat, `TaskAssignorException` is caught by the existing exception handler in `GroupMetadataManager` and converted to an `UnknownServerException`, which is ultimately returned as an `UNKNOWN_SERVER_ERROR` heartbeat response. Therefore, this exception handling path does not terminate the coordinator or broker!
By the way, after checking the client-side handling, I found that the current Streams client treats this response as a fatal error and transitions the member to the `FATAL` state. Therefore, preserving the previous target assignment on the server does not provide a fallback that allows the client to continue operating with its previous assignment. To support the fallback you suggested, I think we would also need to revisit this error handling.
Also, even if we change the behavior to retain the previous assignment, I don’t think that would allow the group to keep processing tasks in every case.
1. An existing member’s heartbeat triggers a new assignment (For example, new computed hash)
- If the previous assignment is still valid, we could retain it temporarily. However, as you mentioned, tasks belonging to members that have already left need to be reassigned. If the assignor keeps returning invalid results while existing members gradually leave or are replaced, more tasks could remain unprocessed, potentially leaving the group unable to process any tasks.
2. A new member joins the group(also leave group?)
- This member has no previous assignment. If we allow it to join while retaining the existing group assignment, it may receive no tasks for as long as the assignor continues to fail. In particular, if all existing members are replaced during a rolling restart, the group could still have members but no longer process any tasks. We should also consider the initial assignment of a new group, where there is no previous assignment to fall back to.
Keeping the previous assignment could help with temporary failures, but it would not fully address recovery when the assignor persistently returns invalid results. Would it make sense to first define the failure behavior separately for cases where we can retain the previous assignment and cases that need different handling, such as the initial assignment?
Also, please refer to this comments!
(https://github.com/apache/kafka/pull/23597#issuecomment-6700401404)
Please let me know your opinion!
cc. @lucasbru @dajac
--
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-21170: Validate Streams task assignor output before accepting target assignments. [kafka]
Posted by "chickenchickenlove (via GitHub) via jira" <ji...@kafka.apache.org>.
chickenchickenlove commented on PR #23597:
URL: https://github.com/apache/kafka/pull/23597#issuecomment-6700401404
@mjsax Thanks for your comments!
Sure, I’d be happy to look into the consumer group case as well and add JMH benchmarks to measure the validation overhead for both consumer and Streams groups.
I’d also appreciate your thoughts on whether some checks could be performed at broker startup. For example, we could validate the loaded assignors against the requirements defined in the relevant KIPs.
Of course, startup validation alone cannot guarantee valid assignments for every subsequent group state, so I don’t think it could fully replace runtime validation. Still, could we reduce the repeated validation cost by distinguishing between checks that can be performed at startup and those that must run for each computed assignment?
cc. @dajac
--
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-21170: Validate Streams task assignor output before accepting target assignments. [kafka]
Posted by "mjsax (via GitHub) via jira" <ji...@kafka.apache.org>.
mjsax commented on PR #23597:
URL: https://github.com/apache/kafka/pull/23597#issuecomment-6697913934
@chickenchickenlove -- I was talking to @dajac about this, and we are thinking if we should do a similar thing for "consumer" groups. But there is also a performance concern. -- Can you maybe also look into the "consumer" case, and also work out some JMH benchmark for both, so we can see what the load we are adding?
--
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-21170: Validate Streams task assignor output before accepting target assignments. [kafka]
Posted by "chickenchickenlove (via GitHub) via jira" <ji...@kafka.apache.org>.
chickenchickenlove commented on code in PR #23597:
URL: https://github.com/apache/kafka/pull/23597#discussion_r4499478241
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
@@ -289,6 +293,72 @@ public TargetAssignmentResult build() throws TaskAssignorException {
return new TargetAssignmentResult(records, newTargetAssignment);
}
+ private void validateAssignment(Map<String, org.apache.kafka.coordinator.group.streams.TasksTuple> assignment) {
+ // The builder produces empty assignments when the topology is not ready.
+ if (!topology.isReady()) {
+ return;
+ }
+
+ final Map<String, ConfiguredSubtopology> subtopologies = topology.subtopologies().orElseThrow();
+ final Map<TaskId, String> activeOwners = new HashMap<>();
+ final Map<String, Map<TaskId, String>> processOwners = new HashMap<>();
+
+ assignment.forEach((memberId, memberAssignment) -> {
+ final String processId = members.get(memberId).processId();
+ final Map<TaskId, String> ownersOnProcess = processOwners.computeIfAbsent(processId, ignored -> new HashMap<>());
+
+ for (TaskRole role : List.of(TaskRole.ACTIVE, TaskRole.STANDBY)) {
+ final Map<String, Set<Integer>> tasks = role == TaskRole.ACTIVE
+ ? memberAssignment.activeTasks() : memberAssignment.standbyTasks();
+
+ tasks.forEach((subtopologyId, partitions) -> {
+ // Both active and standby tasks must reference a subtopology in the configured topology.
+ ConfiguredSubtopology subtopology = subtopologies.get(subtopologyId);
+ if (subtopology == null) {
+ throw new TaskAssignorException(String.format(
Review Comment:
@mjsax Thanks for taking the time to look into this!
When a new assignment is computed while processing a heartbeat, `TaskAssignorException` is caught by the existing exception handler in `GroupMetadataManager` and converted to an `UnknownServerException`, which is ultimately returned as an `UNKNOWN_SERVER_ERROR` heartbeat response. Therefore, this exception handling path does not terminate the coordinator or broker!
By the way, after checking the client-side handling, I found that the current Streams client treats this response as a fatal error and transitions the member to the `FATAL` state. Therefore, preserving the previous target assignment on the server does not provide a fallback that allows the client to continue operating with its previous assignment. To support the fallback you suggested, I think we would also need to revisit this error handling.
Also, even if we change the behavior to retain the previous assignment, I don’t think that would allow the group to keep processing tasks in every case.
1. An existing member’s heartbeat triggers a new assignment (For example, new computed hash)
- If the previous assignment is still valid, we could retain it temporarily. However, as you mentioned, tasks belonging to members that have already left need to be reassigned. If the assignor keeps returning invalid results while existing members gradually leave or are replaced, more tasks could remain unprocessed, potentially leaving the group unable to process any tasks.
2. A new member joins the group(also leave group?)
- This member has no previous assignment. If we allow it to join while retaining the existing group assignment, it may receive no tasks for as long as the assignor continues to fail. In particular, if all existing members are replaced during a rolling restart, the group could still have members but no longer process any tasks. We should also consider the initial assignment of a new group, where there is no previous assignment to fall back to.
Keeping the previous assignment could help with temporary failures, but it would not fully address recovery when the assignor persistently returns invalid results. Would it make sense to first define the failure behavior separately for cases where we can retain the previous assignment and cases that need different handling, such as the initial assignment?
--
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