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/26 11:53:13 UTC
[PR] KAFKA-21167: Filter out members at quota before selecting sticky task candidates. [kafka]
chickenchickenlove opened a new pull request, #23591:
URL: https://github.com/apache/kafka/pull/23591
Fixes KAFKA-21167.
`StickyTaskAssignor` could overlook an eligible member with local state after selecting another member already at quota. Check quota eligibility before selecting candidates for active and standby tasks, allowing available members to reuse local state.
--
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-21167: Filter out members at quota before selecting sticky task candidates. [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23591:
URL: https://github.com/apache/kafka/pull/23591#discussion_r4420952426
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -490,14 +500,17 @@ private static void assignStandby(final LocalState localState, final LinkedList<
// prev standby tasks
final ArrayList<Member> prevStandbyMembers = localState.standbyTaskToPrevMember.get(task);
if (prevStandbyMembers != null && !prevStandbyMembers.isEmpty()) {
- final Member prevStandbyMember = findPrevMemberWithLeastLoad(localState, prevStandbyMembers, Optional.of(task));
+ final Member prevStandbyMember = findPrevMemberWithLeastLoad(
+ localState,
+ prevStandbyMembers,
+ Optional.of(task),
+ member -> hasUnfulfilledTaskQuota(localState, localState.processIdToState.get(member.processId), member)
+ );
if (prevStandbyMember != null) {
Review Comment:
@gabriellefu
Sorry for the delay and Thanks for taking a look! Previously, `findPrevMemberWithLeastLoad()` did not check quota -- that check happened in the caller after selecting a candidate. So a null result did not mean that all previous members were at quota.
The subtle case is two members in the same process: the helper only replaces its candidate when both process load and member load are strictly lower. Since their process loads are equal, it can select a member already at quota and overlook another member that still has room. The caller then rejects the selected member and falls back to general load balancing.
--
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-21167: Filter out members at quota before selecting sticky task candidates. [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #23591:
URL: https://github.com/apache/kafka/pull/23591#issuecomment-6030249443
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-21167: Filter out members at quota before selecting sticky task candidates. [kafka]
Posted by "gabriellefu (via GitHub) via jira" <ji...@kafka.apache.org>.
gabriellefu commented on code in PR #23591:
URL: https://github.com/apache/kafka/pull/23591#discussion_r4469989736
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -490,14 +500,17 @@ private static void assignStandby(final LocalState localState, final LinkedList<
// prev standby tasks
final ArrayList<Member> prevStandbyMembers = localState.standbyTaskToPrevMember.get(task);
if (prevStandbyMembers != null && !prevStandbyMembers.isEmpty()) {
- final Member prevStandbyMember = findPrevMemberWithLeastLoad(localState, prevStandbyMembers, Optional.of(task));
+ final Member prevStandbyMember = findPrevMemberWithLeastLoad(
+ localState,
+ prevStandbyMembers,
+ Optional.of(task),
+ member -> hasUnfulfilledTaskQuota(localState, localState.processIdToState.get(member.processId), member)
+ );
if (prevStandbyMember != null) {
Review Comment:
thanks for the explanation, it make sense to me.
--
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-21167: Filter out members at quota before selecting sticky task candidates. [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23591:
URL: https://github.com/apache/kafka/pull/23591#discussion_r4262987583
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -490,14 +500,17 @@ private static void assignStandby(final LocalState localState, final LinkedList<
// prev standby tasks
final ArrayList<Member> prevStandbyMembers = localState.standbyTaskToPrevMember.get(task);
if (prevStandbyMembers != null && !prevStandbyMembers.isEmpty()) {
- final Member prevStandbyMember = findPrevMemberWithLeastLoad(localState, prevStandbyMembers, Optional.of(task));
+ final Member prevStandbyMember = findPrevMemberWithLeastLoad(
+ localState,
+ prevStandbyMembers,
+ Optional.of(task),
+ member -> hasUnfulfilledTaskQuota(localState, localState.processIdToState.get(member.processId), member)
+ );
if (prevStandbyMember != null) {
Review Comment:
Since findPrevMemberWithLeastLoad() already return the least-loaded pre-member, if it's null already meaning all the premember don't have room, why we need to add member -> hasUnfulfilledTaskQuota(localState, localState.processIdToState.get(member.processId), member) here?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java:
##########
@@ -308,13 +309,16 @@ private static void assignActive(final LocalState localState, final LinkedList<T
for (final Iterator<TaskId> it = activeTasks.iterator(); it.hasNext();) {
final TaskId task = it.next();
final ArrayList<Member> prevMembers = localState.standbyTaskToPrevMember.get(task);
- final Member prevMember = findPrevMemberWithLeastLoad(localState, prevMembers, Optional.empty());
+ final Member prevMember = findPrevMemberWithLeastLoad(
+ localState,
+ prevMembers,
+ Optional.empty(),
+ member -> hasUnfulfilledActiveTaskQuota(localState, localState.processIdToState.get(member.processId), member, stateful)
Review Comment:
Since findPrevMemberWithLeastLoad() already return the least-loaded pre-member, if it's null already meaning all the premember don't have room, why we need to add member -> hasUnfulfilledTaskQuota(localState, localState.processIdToState.get(member.processId), member) here?
--
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