MINOR: log warning when topology override for cache size is non-zero by ableegoldman · Pull Request #11959 · apache/kafka

ableegoldman · 2022-03-29T06:28:47Z

Since the topology-level cache size config only controls whether we disable the caching layer entirely for that topology, setting it to anything other than 0 has no effect. The actual cache memory is still just split evenly between the threads, and shared by all topologies.

It's possible we'll want to change this in the future, but for now we should make sure to log a warning so that users who do try to set this override to some nonzero value are made aware that it doesn't work like this.

Also includes some minor refactoring plus an off-by-one fix from #11796

ableegoldman · 2022-03-29T06:29:51Z

Hey @vamossagar12 can you take a look at this? Also cc @guozhangwang @wcarlson5

vamossagar12 · 2022-03-29T07:24:25Z

@ableegoldman , I did a quick pass.. Looks good to me..

showuon

Overall LGTM. Left some minor comments. Thank you.

showuon · 2022-03-29T07:36:04Z

-                resizeThreadCacheAndBufferMemory(numLiveThreads + 1);
+                resizeThreadCacheAndBufferMemory(numLiveThreads);
                log.info("Adding StreamThread-{}, there are now {} threads with cache size/max buffer size values as {} per thread.",
-                        threadIdx, numLiveThreads + 1, getThreadCacheAndBufferMemoryString());
+                        threadIdx, numLiveThreads, getThreadCacheAndBufferMemoryString());


I think we should add tests for it, to make sure after adding a thread, the cache size and buffer memory is set as what we expected.

@showuon , there are some tests for this here like : https://github.com/vamossagar12/kafka/blob/84c75ce44167e5e1bf8748a7ff9ba95309cd3d7e/streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java#L373. which were added as part of original PR. Do you think they are enough?

That test is to test thread removal, right? We should have the similar tests for thread addition. Otherwise, we should be able to catch this issue via these tests.

ok.. yeah there are 2 more tests for threadReplacement which seem to be printing out an extra thread count. But I agree, a test for thread addition would be good..

vamossagar12 · 2022-03-29T11:26:42Z

-                    CACHE_MAX_BYTES_BUFFERING_CONFIG);
-        } else if (isTopologyOverride(STATESTORE_CACHE_MAX_BYTES_CONFIG, topologyOverrides)) {
-            cacheSize = getLong(STATESTORE_CACHE_MAX_BYTES_CONFIG);
+        final boolean stateStoreCacheMaxBytesOverridden = isTopologyOverride(STATESTORE_CACHE_MAX_BYTES_CONFIG, topologyOverrides);


@ableegoldman , another thing is my initial logic was similar to what you have below (without the boolean variables stateStoreCacheMaxBytesOverridden and cacheMaxBytesBufferingOverridden but I had to change it to the above one as I was getting =>

[2022-03-29T06:53:02.757Z] [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-11959/streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:127:5: NPath Complexity is 960 (max allowed is 500). [NPathComplexity]

Looks like this PR suffers from the same problem.

Ah :/

I think this is a case where it's reasonable to suppress this checkstyle exception, as the NPath complexity is sort of "artificially" complex, ie the code is easy to follow it just has many branches because it goes through each config

Ok.. yeah we can suppress it but after reading the error, I also felt it has too many branches. Anyways, it's fine I guess.

ableegoldman · 2022-03-29T20:49:48Z

@showuon addressed your comment and added test coverage for the cache & buffer size after adding a thread, plz give a +1 if all looks good

wcarlson5

I don't have anything else to add tbh

showuon

Thanks for the update and the tests. LGTM!

ableegoldman · 2022-03-30T04:40:10Z


                for (final String log : appender.getMessages()) {
-                    // after we replace the thread there should be two remaining threads with 5 bytes each
-                    if (log.endsWith("Adding StreamThread-3, there are now 3 threads with cache size/max buffer size values as 3/178956970 per thread.")) {


Apparently there was already a test for the cache being sized correctly after a thread replacement/addition, I'm guessing this was updated with the incorrect values in the PR that introduced the off by one bug -- probably should have been a red flag if the value of the cache size changed here, not to mention the comment above which explicitly mentions it should be 5 bytes per thread 🙂

yeah the comment is off.

No actually I meant that the comment was correct -- the test was just verifying incorrect results (after the thread replacement there should be 2 threads with 5MB of cache, as it says). But no worries

ableegoldman · 2022-03-30T04:43:51Z

-    }
-
-    @Test
-    public void shouldResizeMaxBufferAfterThreadReplacement() throws InterruptedException {


This test seems to be identical to the one above except that it uses the default cache size and a custom input buffer size -- however both tests are still validating both sizes in the log message so this doesn't need to be a separate test, we can just cover both in the original test.

(The total time to run streams tests has been getting a bit out of hand lately, so we should have a concrete reason to split something out into a second test like this, especially the more "heavy" integration tests)

ableegoldman · 2022-03-30T23:18:35Z

All test failures are in Connect, so unrelated. Going to merge

ableegoldman · 2022-03-30T23:24:50Z

Merged to trunk

showuon · 2022-03-31T02:41:00Z

Thanks for fixing the tests, @ableegoldman !

…on-zero (apache#11959)" This reverts commit 1317f3f.

some refactoring, fix, and warning

1245fec

ableegoldman requested review from guozhangwang and showuon March 29, 2022 06:28

ableegoldman mentioned this pull request Mar 29, 2022

KAFKA-13152: Replace "buffered.records.per.partition" with "input.buffer.max.bytes" #11796

Merged

deprecation warning

1ac63e0

showuon reviewed Mar 29, 2022

View reviewed changes

vamossagar12 reviewed Mar 29, 2022

View reviewed changes

ableegoldman added 2 commits March 29, 2022 12:56

checkstyle

75a5cc1

add test

958a709

wcarlson5 approved these changes Mar 29, 2022

View reviewed changes

showuon approved these changes Mar 29, 2022

View reviewed changes

ableegoldman added 2 commits March 29, 2022 17:48

fix NPE

4dd9de9

fix test

0d2ac29

ableegoldman commented Mar 30, 2022

View reviewed changes

ableegoldman added 2 commits March 29, 2022 21:44

use variable

116cdcb

fix num threads in log

6ee5d08

ableegoldman merged commit 1317f3f into apache:trunk Mar 30, 2022

mjsax added a commit to mjsax/kafka that referenced this pull request Jul 6, 2022

Revert "MINOR: log warning when topology override for cache size is n…

b3d6696

…on-zero (apache#11959)" This reverts commit 1317f3f.

Conversation

ableegoldman commented Mar 29, 2022

Uh oh!

ableegoldman commented Mar 29, 2022

Uh oh!

vamossagar12 commented Mar 29, 2022

Uh oh!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ableegoldman commented Mar 29, 2022

Uh oh!

wcarlson5 left a comment

Choose a reason for hiding this comment

Uh oh!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ableegoldman commented Mar 30, 2022

Uh oh!

ableegoldman commented Mar 30, 2022

Uh oh!

showuon commented Mar 31, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants