This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new a1d9d8a5175 KAFKA-20829: Throttle StreamThread shutdown rebalance
enforcement to avoid log flood (#22917)
a1d9d8a5175 is described below
commit a1d9d8a5175600ec01b9fd07cef038c63c6dcc47
Author: Adam Souquières <[email protected]>
AuthorDate: Sat Jul 25 05:07:47 2026 +0200
KAFKA-20829: Throttle StreamThread shutdown rebalance enforcement to avoid
log flood (#22917)
When a Kafka Streams application requests an app-wide shutdown, every
StreamThread on the classic rebalance protocol floods the log with:
`Request joining group due to: Shutdown requested`
This PR gates the `enforceRebalance` call behind the existing 10s
throttle that already guards WARN logs, so both fire together at most
once per 10s.
Reviewers: Matthias J. Sax <[email protected]>
---
.../streams/processor/internals/StreamThread.java | 20 ++++-----
.../processor/internals/StreamThreadTest.java | 52 ++++++++++++++++++++++
2 files changed, 60 insertions(+), 12 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index dcd853e2a88..8a74ae6df24 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -1115,20 +1115,16 @@ public class StreamThread extends Thread implements
ProcessingThread {
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");
+
+ 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");
+ // The classic protocol propagates the shutdown request via an
enforced rebalance,
+ // whereas the Streams protocol (KIP-1071) uses the group
heartbeat.
+ if (streamsRebalanceData.isEmpty()) {
+ mainConsumer.enforceRebalance("Shutdown requested");
}
}
- // 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.
- if (streamsRebalanceData.isEmpty()) {
- mainConsumer.enforceRebalance("Shutdown requested");
- }
}
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index dd37e70ef02..13d4d3631da 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -4095,6 +4095,58 @@ public class StreamThreadTest {
verify(mainConsumer).enforceRebalance("Shutdown requested");
}
+ @Test
+ public void
shouldThrottleEnforceRebalanceOnRepeatedShutdownRequestsUnderClassicProtocol() {
+ final MockTime shutdownTime = new MockTime(0, 100_000L, 0L);
+ final ConsumerGroupMetadata consumerGroupMetadata =
Mockito.mock(ConsumerGroupMetadata.class);
+
when(consumerGroupMetadata.groupInstanceId()).thenReturn(Optional.empty());
+ when(mainConsumer.groupMetadata()).thenReturn(consumerGroupMetadata);
+ final Properties props = configProps(false, false);
+ final StreamsMetadataState streamsMetadataState = new
StreamsMetadataState(
+ new TopologyMetadata(internalTopologyBuilder, new
StreamsConfig(props)),
+ StreamsMetadataState.UNKNOWN_HOST,
+ new LogContext(String.format("stream-client [%s] ", CLIENT_ID))
+ );
+ final StreamsConfig config = new StreamsConfig(props);
+ thread = new StreamThread(
+ shutdownTime,
+ config,
+ null,
+ mainConsumer,
+ consumer,
+ changelogReader,
+ null,
+ mock(TaskManager.class),
+ null,
+ new StreamsMetricsImpl(metrics, CLIENT_ID, mockTime),
+ new TopologyMetadata(internalTopologyBuilder, config),
+ PROCESS_ID,
+ CLIENT_ID,
+ new LogContext(""),
+ new AtomicInteger(),
+ new AtomicLong(Long.MAX_VALUE),
+ new LinkedList<>(),
+ mock(Runnable.class),
+ HANDLER,
+ null,
+ Optional.empty(),
+ streamsMetadataState,
+ null,
+ -1L
+ ).updateThreadMetadata(adminClientId(CLIENT_ID));
+
+ thread.sendShutdownRequest();
+
+ for (int i = 0; i < 1_000; i++) {
+ thread.maybeSendShutdown();
+ }
+ verify(mainConsumer, times(1)).enforceRebalance("Shutdown requested");
+
+ shutdownTime.sleep(10_000L);
+ thread.maybeSendShutdown();
+ verify(mainConsumer, times(2)).enforceRebalance("Shutdown requested");
+ }
+
@Test
public void
testStreamsProtocolRunOnceWithoutProcessingThreadsMissingSourceTopic() {
final ConsumerGroupMetadata consumerGroupMetadata =
Mockito.mock(ConsumerGroupMetadata.class);
