KAFKA-10500: Thread Cache Resizes #9572

wcarlson5 · 2020-11-06T18:48:33Z

The thread cache can now be resized. This will go towards being able to scale the number of threads

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

wcarlson5 · 2020-11-06T18:50:13Z

@cadonna Part 2

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

cadonna

Thank you for the PR, @wcarlson5 !

Here my feedback.

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

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

streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

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

mjsax

A few nits. Overall LGTM.

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

mjsax · 2020-11-17T18:41:53Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-        }
-        final long cacheSizePerThread = totalCacheSize / (numStreamThreads + (hasGlobalTopology ? 1 : 0));
+        totalCacheSize = config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG);
+        final long cacheSizePerThread = totalCacheSize / (numStreamThreads + ((globalTaskTopology != null) ? 1 : 0));


Why move off using hasGlobalTopology?

It was in a separate method without access to hasGlobalTopology. I supposes if it stays we can move it back

mjsax · 2020-11-17T18:49:40Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -806,6 +803,13 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
        rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, config);
    }

+    private void resizeThreadCache(final int numStreamThreads) {
+        final long cacheSizePreThread = totalCacheSize / (numStreamThreads + ((globalTaskTopology != null) ? 1 : 0));


Seems this duplicates L733. Might be good to extract into a small helper method.

I did have it in a separate method but helper but when removing the totalCacheSize < 0 check @cadonna thought it would be more readable inline

Not sure why the totalCacheSize check is relevant for avoiding code duplication?

I think it was about readability. I might be misremembering though, as it was a conversation we had last week

If this line is duplicated, it should go in a method. When I proposed to move it inline, I was apparently not aware that the same line was used somewhere else.

Moved to a new method. Glad we got that cleared up. LGTM?

LGTM?

If this is a question, should it be LGTY? 😂

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

streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java

mjsax · 2020-11-17T18:55:00Z

streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java

+            final CircularIterator<NamedCache> circularIterator = new CircularIterator<>(caches.values());
+            while (sizeBytes() > maxCacheSizeBytes) {
+                if (!circularIterator.hasNext()) {
+                    log.error("Unable to remove any more entries as all caches are empty");


Could this ever happen? If we the max cache size is smaller than a single entry, would we not evict the entry and the used cache size would always shrink to zero?

If we add a check to make sure the number of threads is positive then probably not. Ill add that check then remove this one

I see. -- I guess the miss-leading fact was, that this check was done inside the while-loop.

Yeah, in retrospect it was not very clear. Hopefully its better this way now

mjsax · 2020-11-17T21:23:54Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+    }
+
+    private void resizeThreadCache(final int numStreamThreads) {
+        if (numStreamThreads < 0) {


Can it be smaller than 0 ? Should the test be <= 0 or < 1 instead?

It can be zero if you have a global thread, but since this is internal the check might not be entirely necessary

Yes, it can be zero, but the check says < 0, so it would always evaluate to false?

And if we have zero threads, we should not resize the cache as we might end up in an infinite loop? But we would only call this method if we "shrink", ie, if the thread count grows, but it can never grow from negative to zero, right?

That is a good point. Maybe what we need to do it put a minimum size of cache to limit how many stream threads an instance can have?

Well, getCacheSizePerThread would eventually return zero (with growing number of threads), what means that every put() into the cache would result in an immediate eviction. So I don't think we need to do anything for this corner case.

that is a good point

streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java

mjsax · 2020-11-18T23:46:12Z

Thanks for the PR @wcarlson5. Merged to trunk.

wcarlson5 commented Nov 6, 2020

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java Outdated Show resolved Hide resolved

wcarlson5 mentioned this pull request Nov 9, 2020

KAFKA-10500: Add thread #9581

Closed

3 tasks

cadonna reviewed Nov 10, 2020

View reviewed changes

cadonna reviewed Nov 11, 2020

View reviewed changes

mjsax reviewed Nov 17, 2020

View reviewed changes

mjsax added the streams label Nov 17, 2020

mjsax reviewed Nov 17, 2020

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java Outdated Show resolved Hide resolved

wcarlson5 added 12 commits November 18, 2020 09:12

ThreadCache Resizes

ecb17f4

fix typo

41c9dca

address comments and evenly clear cache

a2d3b0f

make final

eade1c7

comments

cb8078f

make maxCacheSize volatile and wrapped thread cache

1ed3914

change to consumer

d7a3edb

improved names and checks

53d92c5

couple of comments

c4b78e5

make helper for cache size

2698000

remove this

26fb2a9

remove redundant check

b69b6b2

mjsax merged commit d12fbb7 into apache:trunk Nov 18, 2020

wcarlson5 deleted the thread-cache-resizeable branch November 19, 2020 00:34

mjsax added the kip Requires or implements a KIP label Jan 8, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-10500: Thread Cache Resizes #9572

KAFKA-10500: Thread Cache Resizes #9572

wcarlson5 commented Nov 6, 2020

wcarlson5 commented Nov 6, 2020

cadonna left a comment

mjsax left a comment

mjsax Nov 17, 2020

wcarlson5 Nov 17, 2020

mjsax Nov 17, 2020

wcarlson5 Nov 17, 2020

mjsax Nov 17, 2020

wcarlson5 Nov 17, 2020

cadonna Nov 17, 2020

wcarlson5 Nov 17, 2020

mjsax Nov 17, 2020

mjsax Nov 17, 2020

wcarlson5 Nov 17, 2020

mjsax Nov 17, 2020

wcarlson5 Nov 17, 2020

mjsax Nov 17, 2020

wcarlson5 Nov 17, 2020

mjsax Nov 17, 2020 •

edited

wcarlson5 Nov 17, 2020

mjsax Nov 17, 2020 •

edited

wcarlson5 Nov 17, 2020

mjsax commented Nov 18, 2020

KAFKA-10500: Thread Cache Resizes #9572

KAFKA-10500: Thread Cache Resizes #9572

Conversation

wcarlson5 commented Nov 6, 2020

Committer Checklist (excluded from commit message)

wcarlson5 commented Nov 6, 2020

cadonna left a comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Nov 17, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Nov 17, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Nov 18, 2020

mjsax Nov 17, 2020 •

edited

mjsax Nov 17, 2020 •

edited