Skip to content

Commit 843f8fe

Browse files
sstremlerbbejeck
sstremler
authored and
bbejeck
committed
KAFKA-20398: Fix memory leak by removing StreamsThreadMetricsDelegatingReporter from metrics registry on thread shutdown (#21973)
When a stream thread in Kafka Streams crashes, `StreamsUncaughtExceptionHandler` can be configured to replace the crashed thread with a new one. Each time a new stream thread is created, it adds a new `StreamsThreadMetricsDelegatingReporter` to it. However, when the old thread shuts down, its reporter is never removed from the registry implying "unbounded" memory growth. This bug was introduced via #17021 This PR ensures that the existing `StreamsThreadMetricsDelegatingReporter` is removed when a thread is shutting down. Reviewers: Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bbejeck@apache.org> Signed-off-by: Szabolcs Stremler <stremler.sz@gmail.com>
1 parent 15793d8 commit 843f8fe

2 files changed

Lines changed: 65 additions & 10 deletions

File tree

‎streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -384,6 +384,7 @@ public boolean isStartingRunningOrPartitionAssigned() {
384384
private final WindowedSum punctuateLatencyWindowedSum = new WindowedSum();
385385
private final WindowedSum runOnceLatencyWindowedSum = new WindowedSum();
386386
private final MetricConfig metricsConfig;
387+
private final StreamsThreadMetricsDelegatingReporter metricsReporter;
387388

388389
private boolean latencyWindowsInitialized = false;
389390

@@ -514,8 +515,8 @@ public static StreamThread create(final TopologyMetadata topologyMetadata,
514515
taskManager.setMainConsumer(mainConsumerSetup.mainConsumer);
515516
referenceContainer.mainConsumer = mainConsumerSetup.mainConsumer;
516517

517-
final StreamsThreadMetricsDelegatingReporter reporter = new StreamsThreadMetricsDelegatingReporter(mainConsumerSetup.mainConsumer, threadId, Optional.of(stateUpdaterId));
518-
streamsMetrics.metricsRegistry().addReporter(reporter);
518+
final StreamsThreadMetricsDelegatingReporter metricsReporter = new StreamsThreadMetricsDelegatingReporter(mainConsumerSetup.mainConsumer, threadId, Optional.of(stateUpdaterId));
519+
streamsMetrics.metricsRegistry().addReporter(metricsReporter);
519520

520521
final StreamThread streamThread = new StreamThread(
521522
time,
@@ -539,7 +540,8 @@ public static StreamThread create(final TopologyMetadata topologyMetadata,
539540
streamsUncaughtExceptionHandler,
540541
cache::resize,
541542
mainConsumerSetup.streamsRebalanceData,
542-
streamsMetadataState
543+
streamsMetadataState,
544+
metricsReporter
543545
);
544546

545547
return streamThread.updateThreadMetadata(adminClientId(clientId));
@@ -786,7 +788,8 @@ public StreamThread(final Time time,
786788
final BiConsumer<Throwable, Boolean> streamsUncaughtExceptionHandler,
787789
final java.util.function.Consumer<Long> cacheResizer,
788790
final Optional<StreamsRebalanceData> streamsRebalanceData,
789-
final StreamsMetadataState streamsMetadataState
791+
final StreamsMetadataState streamsMetadataState,
792+
final StreamsThreadMetricsDelegatingReporter metricsReporter
790793
) {
791794
super(threadId);
792795
this.stateLock = new Object();
@@ -809,6 +812,7 @@ public StreamThread(final Time time,
809812
this.streamsUncaughtExceptionHandler = streamsUncaughtExceptionHandler;
810813
this.cacheResizer = cacheResizer;
811814
this.metricsConfig = streamsMetrics.metricsRegistry().config();
815+
this.metricsReporter = metricsReporter;
812816

813817
// The following sensors are created here but their references are not stored in this object, since within
814818
// this object they are not recorded. The sensors are created here so that the stream threads starts with all
@@ -1905,6 +1909,7 @@ private void completeShutdown(final boolean cleanRun) {
19051909
}
19061910
streamsMetrics.removeAllThreadLevelSensors(getName());
19071911
streamsMetrics.removeAllThreadLevelMetrics(getName());
1912+
streamsMetrics.metricsRegistry().removeReporter(metricsReporter);
19081913

19091914
setState(State.DEAD);
19101915

@@ -2102,6 +2107,10 @@ Optional<StreamsRebalanceData> streamsRebalanceData() {
21022107
return streamsRebalanceData;
21032108
}
21042109

2110+
StreamsMetricsImpl streamsMetrics() {
2111+
return streamsMetrics;
2112+
}
2113+
21052114
/**
21062115
* Initialize both WindowedSum instances at exactly the same timestamp so
21072116
* their windows are aligned from the very beginning.

‎streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java‎

Lines changed: 52 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
import org.apache.kafka.common.metrics.Measurable;
5353
import org.apache.kafka.common.metrics.Metrics;
5454
import org.apache.kafka.common.metrics.MetricsContext;
55+
import org.apache.kafka.common.metrics.MetricsReporter;
5556
import org.apache.kafka.common.record.TimestampType;
5657
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse;
5758
import org.apache.kafka.common.serialization.Serdes;
@@ -71,6 +72,7 @@
7172
import org.apache.kafka.streams.errors.TaskCorruptedException;
7273
import org.apache.kafka.streams.errors.TaskMigratedException;
7374
import org.apache.kafka.streams.errors.TopologyException;
75+
import org.apache.kafka.streams.internals.metrics.StreamsThreadMetricsDelegatingReporter;
7476
import org.apache.kafka.streams.kstream.Consumed;
7577
import org.apache.kafka.streams.kstream.Materialized;
7678
import org.apache.kafka.streams.kstream.internals.ConsumedInternal;
@@ -1449,6 +1451,7 @@ public void shouldNotReturnDataAfterTaskMigrated(final boolean processingThreads
14491451
HANDLER,
14501452
null,
14511453
Optional.empty(),
1454+
null,
14521455
null
14531456
).updateThreadMetadata(adminClientId(CLIENT_ID));
14541457

@@ -2474,6 +2477,7 @@ public void shouldCatchHandleCorruptionOnTaskCorruptedExceptionPath(final boolea
24742477
HANDLER,
24752478
null,
24762479
Optional.empty(),
2480+
null,
24772481
null
24782482
) {
24792483
@Override
@@ -2535,6 +2539,7 @@ public void shouldCatchTimeoutExceptionFromHandleCorruptionAndInvokeExceptionHan
25352539
HANDLER,
25362540
null,
25372541
Optional.empty(),
2542+
null,
25382543
null
25392544
) {
25402545
@Override
@@ -2604,6 +2609,7 @@ public void shouldCatchTaskMigratedExceptionOnOnTaskCorruptedExceptionPath(final
26042609
HANDLER,
26052610
null,
26062611
Optional.empty(),
2612+
null,
26072613
null
26082614
) {
26092615
@Override
@@ -2670,6 +2676,7 @@ public void shouldEnforceRebalanceWhenTaskCorruptedExceptionIsThrownForAnActiveT
26702676
HANDLER,
26712677
null,
26722678
Optional.empty(),
2679+
null,
26732680
null
26742681
) {
26752682
@Override
@@ -2733,6 +2740,7 @@ public void shouldNotEnforceRebalanceWhenTaskCorruptedExceptionIsThrownForAnInac
27332740
HANDLER,
27342741
null,
27352742
Optional.empty(),
2743+
null,
27362744
null
27372745
) {
27382746
@Override
@@ -2965,6 +2973,7 @@ public void shouldConstructAdminMetrics(final boolean processingThreadsEnabled)
29652973
HANDLER,
29662974
null,
29672975
Optional.empty(),
2976+
null,
29682977
null
29692978
);
29702979
final MetricName testMetricName = new MetricName("test_metric", "", "", new HashMap<>());
@@ -3024,6 +3033,7 @@ public void runAndVerifyFailedStreamThreadRecording(final boolean shouldFail, fi
30243033
(e, b) -> { },
30253034
null,
30263035
Optional.empty(),
3036+
null,
30273037
null
30283038
) {
30293039
@Override
@@ -3611,7 +3621,8 @@ public void testStreamsProtocolRunOnceWithoutProcessingThreads() {
36113621
HANDLER,
36123622
null,
36133623
Optional.of(streamsRebalanceData),
3614-
streamsMetadataState
3624+
streamsMetadataState,
3625+
null
36153626
).updateThreadMetadata(adminClientId(CLIENT_ID));
36163627

36173628
thread.setState(State.STARTING);
@@ -3672,7 +3683,8 @@ public void testStreamsProtocolRunOnceWithoutProcessingThreadsMissingSourceTopic
36723683
HANDLER,
36733684
null,
36743685
Optional.of(streamsRebalanceData),
3675-
streamsMetadataState
3686+
streamsMetadataState,
3687+
null
36763688
).updateThreadMetadata(adminClientId(CLIENT_ID));
36773689

36783690
thread.setState(State.STARTING);
@@ -3742,7 +3754,8 @@ public void testStreamsProtocolIncorrectlyPartitionedTopics() {
37423754
HANDLER,
37433755
null,
37443756
Optional.of(streamsRebalanceData),
3745-
streamsMetadataState
3757+
streamsMetadataState,
3758+
null
37463759
).updateThreadMetadata(adminClientId(CLIENT_ID));
37473760

37483761
thread.setState(State.STARTING);
@@ -3803,7 +3816,8 @@ public void testStreamsProtocolRunOnceWithProcessingThreads() {
38033816
HANDLER,
38043817
null,
38053818
Optional.of(streamsRebalanceData),
3806-
streamsMetadataState
3819+
streamsMetadataState,
3820+
null
38073821
).updateThreadMetadata(adminClientId(CLIENT_ID));
38083822

38093823
thread.setState(State.STARTING);
@@ -3864,7 +3878,8 @@ public void testStreamsProtocolRunOnceWithProcessingThreadsMissingSourceTopic()
38643878
HANDLER,
38653879
null,
38663880
Optional.of(streamsRebalanceData),
3867-
streamsMetadataState
3881+
streamsMetadataState,
3882+
null
38683883
).updateThreadMetadata(adminClientId(CLIENT_ID));
38693884

38703885
thread.setState(State.STARTING);
@@ -3934,7 +3949,8 @@ public void testStreamsProtocolMissingSourceTopicRecovery() {
39343949
HANDLER,
39353950
null,
39363951
Optional.of(streamsRebalanceData),
3937-
streamsMetadataState
3952+
streamsMetadataState,
3953+
null
39383954
).updateThreadMetadata(adminClientId(CLIENT_ID));
39393955

39403956
thread.setState(State.STARTING);
@@ -3998,6 +4014,34 @@ t2p1, new PartitionInfo(t2p1.topic(), t2p1.partition(), null, new Node[0], new N
39984014
);
39994015
}
40004016

4017+
@Test
4018+
public void shouldRemoveMetricsDelegatingReporterOnShutdown() throws InterruptedException {
4019+
thread = createStreamThread(CLIENT_ID, false);
4020+
4021+
final List<MetricsReporter> reportersAfterCreate = thread.streamsMetrics().metricsRegistry().reporters();
4022+
assertThat(
4023+
reportersAfterCreate.stream()
4024+
.filter(r -> r instanceof StreamsThreadMetricsDelegatingReporter)
4025+
.count(),
4026+
equalTo(1L)
4027+
);
4028+
4029+
thread.shutdown(CloseOptions.GroupMembershipOperation.LEAVE_GROUP);
4030+
TestUtils.waitForCondition(
4031+
() -> thread.state() == StreamThread.State.DEAD,
4032+
10 * 1000,
4033+
"Thread never shut down."
4034+
);
4035+
4036+
final List<MetricsReporter> reportersAfterShutdown = thread.streamsMetrics().metricsRegistry().reporters();
4037+
assertThat(
4038+
reportersAfterShutdown.stream()
4039+
.filter(r -> r instanceof StreamsThreadMetricsDelegatingReporter)
4040+
.count(),
4041+
equalTo(0L)
4042+
);
4043+
}
4044+
40014045
private StreamThread setUpThread(final Properties streamsConfigProps) {
40024046
final StreamsConfig config = new StreamsConfig(streamsConfigProps);
40034047
final ConsumerGroupMetadata consumerGroupMetadata = Mockito.mock(ConsumerGroupMetadata.class);
@@ -4030,6 +4074,7 @@ private StreamThread setUpThread(final Properties streamsConfigProps) {
40304074
null,
40314075
null,
40324076
Optional.empty(),
4077+
null,
40334078
null
40344079
);
40354080
}
@@ -4131,6 +4176,7 @@ private StreamThread buildStreamThread(final Consumer<byte[], byte[]> consumer,
41314176
HANDLER,
41324177
null,
41334178
Optional.empty(),
4179+
null,
41344180
null
41354181
);
41364182
}

0 commit comments

Comments
 (0)