You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2024/01/26 18:12:49 UTC
[PR] KAFKA-15832: Trigger client reconciliation based on manager poll [kafka]
lucasbru opened a new pull request, #15275:
URL: https://github.com/apache/kafka/pull/15275
Currently, the reconciliation logic on the client is triggered when a new target assignment is received and resolved, or when new unresolved target assignments are discovered in metadata.
This change improves this by triggering the reconciliation logic on each poll iteration, to reconcile whatever is ready to be reconciled. This would require changes to support poll on the MembershipManager, and integrate it with the current polling logic in the background thread. Receiving a new target assignment from the broker, or resolving new topic names via a metadata update could only ensure that the #assignmentReadyToReconcile is properly updated (currently done), but wouldn't trigger the #reconcile() logic, leaving that to the #poll() operation.
Stacked PR, do not review the first commit.
Draft - still working out some details and some tests missing.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #15275:
URL: https://github.com/apache/kafka/pull/15275
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1485886563
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1356,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING) {
+ maybeReconcile();
Review Comment:
Yes, this is also my understanding.
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1485888892
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -929,6 +938,9 @@ private void revokeAndAssign(SortedSet<TopicIdPartition> assignedTopicIdPartitio
String reason = interruptedReconciliationErrorMessage();
log.error("Interrupting reconciliation after partitions assigned callback " +
"completed. " + reason);
+ if (memberHasRejoined) {
+ attemptReconciliation = true;
Review Comment:
The point was that we avoid the reconciliation-on-every-poll when metadata is missing by using `attemptReconciliation`, and I think we had to enable it here as well. However, since we decided that reconciliation-on-every-poll is fine for a short timeframe where metadata is incomplete, this does not matter anymore, I removed it.
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1485885246
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -1521,6 +1660,18 @@ private ConsumerRebalanceListenerInvoker consumerRebalanceListenerInvoker() {
);
}
+ private SortedSet<TopicPartition> topicPartitions(Map<Uuid, SortedSet<Integer>> topicIdMap, Map<Uuid, String> topicIdNames) {
Review Comment:
Done
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15275:
URL: https://github.com/apache/kafka/pull/15275#issuecomment-1939663048
Hey folks -- looks like this broke DescribeConsumerGroupTest. Can we take a look? cc: @lucasbru
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #15275:
URL: https://github.com/apache/kafka/pull/15275#issuecomment-1921371376
@lianetm Could you please have a look?
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1486257781
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1356,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING) {
+ maybeReconcile();
Review Comment:
Just for the record you are both right, us requesting metadata updates from the reconciliation, is only setting a flag in the metadata object indicating that it needs a full update, but effectively a single metadata request is going to be issued when that flag is on, on the next network client poll. Here btw, we just have a related jira for a potential future improvement, regarding partial metadata updates when reconciling (https://issues.apache.org/jira/browse/KAFKA-15847), since right now we request it for all topics, even if we only need the ones involved in the 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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1478881882
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1402,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING && attemptReconciliation) {
+ reconcile();
Review Comment:
Regarding the reconcile on poll, agree with @dajac that we were moving away from the multiple triggers looking for a more unified approach, that would reconcile not based on different events, but rather on a regular check of the target assignment, that could be affected by all those events. So with this, we have a `reconcile()` as the single place to determine if we should reconcile ([this initial block](https://github.com/apache/kafka/blob/2158ab7b66e0b57bdaf7872511e48ff076160320/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java#L787-L822)),
1. no reconciliation in progress
2. target assignment different from the current assignment
3. resolved target different from the current assignment
In that sense, calling that reconcile regularly on every poll is probably all we want, and aligns with what @dajac described, as it will run the checks above and proceed. But the `attemptReconciliation` introduces something extra that maybe we can simplify a bit.
If I get it right, all we want with that attempt logic is to avoid going through the unresolved assignment and requesting metadata if we know we were waiting for it already and it hasn't been received, right? In that sense, it looks to me as an internal improvement to how we resolve metadata in the `[findResolvableAssignmentAndTriggerMetadataUpdate](https://github.com/apache/kafka/blob/2158ab7b66e0b57bdaf7872511e48ff076160320/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java#L1012)`, and not as another condition to reconcile. The `findResolvableAssignmentAndTriggerMetadataUpdate` could consider if we are awaiting metadata for a target assignment before checking the target and requesting it again. It would be no-op if we are awaiting metadata for the target, and the same common reconcile checks (check # 3) would ensure that a reconciliation is not triggered if resolved target didn't change.
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1484072806
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1356,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING) {
+ maybeReconcile();
Review Comment:
I have one remaining question for my understanding. If there is unresolved topic ids, is it going to request a metadata update (in `findResolvableAssignmentAndTriggerMetadataUpdate`) every time that `poll` is called? I believe that it won't be the case because the `Metadata` class will request a full update only once. Is my understanding correct?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -1521,6 +1660,18 @@ private ConsumerRebalanceListenerInvoker consumerRebalanceListenerInvoker() {
);
}
+ private SortedSet<TopicPartition> topicPartitions(Map<Uuid, SortedSet<Integer>> topicIdMap, Map<Uuid, String> topicIdNames) {
Review Comment:
nit: Could it be static?
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1485890253
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1402,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING && attemptReconciliation) {
+ reconcile();
Review Comment:
Good points. Let's keep it simple for now and remove `attemptReconciliation` and the metadata update listener, and just reconcile on every poll until the metadata is complete. We can always bring back the extra complexity if this causes problems.
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1477976583
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1402,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING && attemptReconciliation) {
+ reconcile();
Review Comment:
Got it. But if we completely move away from triggers (so remove `attemptReconciliation` boolean in this change) and re-attempt reconciliation in every poll, we'd stay in a reconciliation loop as long as topic metadata is missing. Do we know that missing target IDs are always resolved in a relatively short amount of time? Otherwise, we'd need a back-off mechanism.
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1474503361
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1402,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING && attemptReconciliation) {
+ reconcile();
Review Comment:
I think doing this in `poll` isn't strictly necessary. We could call `reconcile` in 3 places (on metadata update, on new assignment, after an ongoing reconciliation finished), and wouldn't have to rely on `poll` to attempt the reconciliation. That would be mostly the same change, but simpler. But since the ticket was written that way, for now I implemented it inside poll.
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1477951854
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1402,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING && attemptReconciliation) {
+ reconcile();
Review Comment:
For the context, I think that the idea was to move away from the edge reconciliation triggers to a model where we would check if the current assignment matches the target assignment in `poll` and reconcile if it does not and there is no on-going reconciliation. The benefits of this is that it works in all cases even if we, for instance, miss a trigger.
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1478861404
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1402,16 @@ public void registerStateListener(MemberStateListener listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING && attemptReconciliation) {
+ reconcile();
Review Comment:
We stay RECONCILING as long as topic metadata is missing as you mentioned, OR, the broker removes the assignment. And that's what makes this assumption sensible I would say. We know the missing targets will be resolved in a timely manner, or removed from the target.
--
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-15832: Trigger client reconciliation based on manager poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1478904746
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -929,6 +938,9 @@ private void revokeAndAssign(SortedSet<TopicIdPartition> assignedTopicIdPartitio
String reason = interruptedReconciliationErrorMessage();
log.error("Interrupting reconciliation after partitions assigned callback " +
"completed. " + reason);
+ if (memberHasRejoined) {
+ attemptReconciliation = true;
Review Comment:
I guess this is trying to trigger a reconciliation if a member rejoins with the same assignment it was stuck reconciling? I would say we don't need to involve this new condition to reconcile here, and could simply let the `reconcile()` decide if it needs to reconcile when called on poll. We do need to fix the underlying problem, which is that we are discarding the reconciliation when a member rejoins with the same assignment ([KAFKA-16185](https://issues.apache.org/jira/browse/KAFKA-16185) to come in another PR). Once we got that right, I expect the reconciliation of the target received after rejoining would happen on poll if the target changed, or be reused when reconciliation completes, if the target is the same.
--
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