KAFKA-18185: remove internal.leave.group.on.close config #19400
Conversation
github-actions
bot
added
triage
frankvicky
changed the title
frankvicky
changed the title
| if (leaveGroup) { | ||
| processStreamThread(streamThreadLeaveConsumerGroup(timeoutMs)); | ||
| } |
There was a problem hiding this comment.
As previous discussion, we should only close the consumer in StreamThread#completeShutdown
| } | ||
| } | ||
|
|
||
| private void completeShutdown(final boolean cleanRun, final boolean leaveGroup) { |
There was a problem hiding this comment.
We could leverage the leaveGroupRequested and don't need a new argument.
| @@ -1823,14 +1826,7 @@ private void completeShutdown(final boolean cleanRun, final boolean leaveGroup) | |||
| log.error("Failed to close changelog reader due to the following error:", e); | |||
| } | |||
| try { | |||
| if (leaveGroupRequested.get()) { | |||
| mainConsumer.unsubscribe(); | |||
There was a problem hiding this comment.
Consumer# unsubscribe() will make consumers leave the group.
I believe this is a workaround, and now we could rely onthe new close API
There was a problem hiding this comment.
@frankvicky thanks for this patch. a couple of comments are left.
| final Optional<String> groupInstanceID = streamThread.groupInstanceID(); | ||
| streamThread.requestLeaveGroupDuringShutdown(); | ||
| streamThread.shutdown(); | ||
| streamThread.shutdown(true); | ||
| if (!streamThread.getName().equals(Thread.currentThread().getName())) { |
There was a problem hiding this comment.
As you are updating this file, please replace !streamThread.getName().equals(Thread.currentThread().getName()) with callingThreadIsNotCurrentStreamThread.
| } catch (final Throwable e) { | ||
| log.error("Failed to close consumer due to the following error:", e); | ||
| } | ||
| try { | ||
| restoreConsumer.close(); | ||
| restoreConsumer.close(CloseOptions.groupMembershipOperation(REMAIN_IN_GROUP)); |
There was a problem hiding this comment.
The restore consumer does not use a consumer group, so it should be fine to call restoreConsumer.close();, right?
| log.info("Informed to shut down"); | ||
| final State oldState = setState(State.PENDING_SHUTDOWN); | ||
| if (leaveGroup) { | ||
| requestLeaveGroupDuringShutdown(); |
There was a problem hiding this comment.
requestLeaveGroupDuringShutdown is not used by other classes now, so we can remove requestLeaveGroupDuringShutdown by calling leaveGroupRequested.set(true);
| @@ -490,7 +487,7 @@ private void replaceStreamThread(final Throwable throwable) { | |||
| closeToError(); | |||
| } | |||
| final StreamThread deadThread = (StreamThread) Thread.currentThread(); | |||
| deadThread.shutdown(); | |||
| deadThread.shutdown(true); | |||
There was a problem hiding this comment.
It seems the origin behavior does not want to leave group, right? otherwise, it will trigger the extra rebalance.
| + " for the following reason: ", | ||
| exception.getCause() | ||
| ); | ||
| } |
There was a problem hiding this comment.
It seems the behavior of KafkaStreams#removeStreamThread has changed. Without an explicit removal, the consumer is kicked off from the group after stream thread completes its run function. This means users CANNOT assume KafkaStreams#removeStreamThread will remove the member from the group.
JIRA: KAFKA-18185
This is a follow-up of #17614 The patch is to remove the
internal.leave.group.on.closeconfig.