lianetm commented on code in PR #16449:
URL: https://github.com/apache/kafka/pull/16449#discussion_r1667035370
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -213,14 +217,14 @@ private void process(final SubscriptionChangeEvent
ignored) {
* the group is sent out.
*/
private void process(final UnsubscribeEvent event) {
- if (!requestManagers.heartbeatRequestManager.isPresent()) {
- KafkaException error = new KafkaException("Group membership
manager not present when processing an unsubscribe event");
- event.future().completeExceptionally(error);
- return;
+ event.future().whenComplete((__, ___) -> subscriptions.unsubscribe());
+ if (requestManagers.heartbeatRequestManager.isPresent()) {
+ MembershipManager membershipManager =
requestManagers.heartbeatRequestManager.get().membershipManager();
+ CompletableFuture<Void> future = membershipManager.leaveGroup();
+ future.whenComplete(complete(event.future()));
+ } else {
+ event.future().complete(null);
Review Comment:
Agree, and we should clarify it here, and in the UnsubscribeEvent itself I
would say. I checked the comments on both, and seems ok if we keep them like
they are (all holds true), and just add a final sentence to describe that if no
group ID has been defined by the consumer, the UnsubscribeEvent will still
ensure that all topics and assigned partitions are cleared.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
