KAFKA-16933: New consumer unsubscribe close commit fixes #16272
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -158,7 +158,12 @@ public class AsyncKafkaConsumerTest { | |
| public void resetAll() { | ||
| backgroundEventQueue.clear(); | ||
| if (consumer != null) { | ||
| try { | ||
| consumer.close(Duration.ZERO); | ||
| } catch (Exception e) { | ||
| // best effort to clean up after each test, but may throw (ex. if callbacks where | ||
| // throwing errors) | ||
|
There was a problem hiding this comment. I did this before but was asked to not do it, see https://github.com/apache/kafka/pull/15613/files/9fb917e4b1e60f238183c92d1ad3bc2565a7e1ea#r1559907295. That's why I added a "clean-up close" in the tests where close fails, with an expected exception (search for "clean-up" in this file). I'd also be fine with your (and my original approach) to have a best-effort clean up and ignore exceptions here. But then, let's remove the "clean-up close" code in the other tests. Any consistent approach is fine with me here. There was a problem hiding this comment. Oh I see, good point. Agree with going with a consistent approach here, but could I bring it in a follow-up PR right after this? (it wouldn't need to get into 3.8). The trick is that now the close may throw for any of the tests that throw on the callback, so I would prefer to play safe, make sure the close does not throw here, and I'll review all the tests that we expect could rightfully throw on close now (Also the other clean-up logic has some Timeout related checks so better not removing blindly, could have some value it seems) |
||
| } | ||
| } | ||
| consumer = null; | ||
| Mockito.framework().clearInlineMocks(); | ||
|
|
@@ -832,8 +837,8 @@ public void testPartitionRevocationOnClose() { | |
|
|
||
| @Test | ||
| public void testFailedPartitionRevocationOnClose() { | ||
| // If rebalance listener failed to execute during close, we still send the leave group, | ||
| // and proceed with closing the consumer. | ||
| ConsumerRebalanceListener listener = mock(ConsumerRebalanceListener.class); | ||
| SubscriptionState subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE); | ||
| consumer = newConsumer( | ||
|
|
@@ -848,7 +853,7 @@ public void testFailedPartitionRevocationOnClose() { | |
| subscriptions.assignFromSubscribed(singleton(tp)); | ||
| doThrow(new KafkaException()).when(listener).onPartitionsRevoked(eq(singleton(tp))); | ||
| assertThrows(KafkaException.class, () -> consumer.close(Duration.ZERO)); | ||
| verify(applicationEventHandler).addAndGet(any(LeaveOnCloseEvent.class)); | ||
| verify(listener).onPartitionsRevoked(eq(singleton(tp))); | ||
| assertEquals(emptySet(), subscriptions.assignedPartitions()); | ||
| } | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
So the point here seems to be, if
maybeRevokePartitionsfails, likely due to rebalancelistener, still leave the group.It looks like a good change to me. I'm just surprised to see that the legacy consumer does not seem to do this? If
onPrepareLeaveinAbstractCoordinatorfails, we won't reachmaybeLeaveGroup.So is this a bug also in the legacy consumer?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You got it right, in this case the legacy consumer won't send a leave group either. We discussed it with @dajac and it seemed right to ensure the new consumer leaves on close (even if the callbacks fail). No changes on the legacy for now though (would require more thought).