mjsax commented on code in PR #22917:
URL: https://github.com/apache/kafka/pull/22917#discussion_r3639922791
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1113,20 +1112,22 @@ public void maybeSendShutdown() {
if (assignmentErrorCode.get() ==
AssignorError.SHUTDOWN_REQUESTED.code()) {
final long now = time.milliseconds();
final long lastLogged = lastShutdownWarningTimestamp.get();
- if (now - lastLogged >= 10_000L) {
- if (lastShutdownWarningTimestamp.compareAndSet(lastLogged,
now)) {
- log.warn("Detected that shutdown was requested. " +
- "All clients in this app will now begin to
shutdown");
+ // The run loop calls this method on every iteration while a
shutdown is pending, and while
+ // the shutdown rebalance is in progress the thread polls
non-blocking, so throttle both the
+ // warning and the rebalance enforcement to at most once per 10s
to avoid flooding the logs.
+ // The first detection still fires immediately (lastLogged == 0),
so shutdown is not delayed.
+ if (now - lastLogged >= 10_000L &&
lastShutdownWarningTimestamp.compareAndSet(lastLogged, now)) {
+ log.warn("Detected that shutdown was requested. " +
+ "All clients in this app will now begin to shutdown");
+ // Under the classic protocol the shutdown request is
propagated to the rest of the group
+ // by the assignor during a rebalance, so we need to enforce
one. Under the Streams group
+ // protocol (KIP-1071) the request is propagated through the
group heartbeat (see
+ // sendShutdownRequest), and enforceRebalance is not supported
by the consumer (it would
+ // only log a warning), so we skip it.
Review Comment:
I know this is an existing comment, but it's also very verbose. I think we
can say this in a single sentence, instead of a whole paragraph.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
