KAFKA-13152: Replace "buffered.records.per.partition" with "input.buffer.max.bytes" by vamossagar12 · Pull Request #11796 · apache/kafka

vamossagar12 · 2022-02-22T17:58:35Z

Implements KIP-770.

vamossagar12 · 2022-02-23T17:24:25Z

@ableegoldman , @guozhangwang , @mjsax this is the new PR for KIP-770(follow up from #11424

last 3-4 commits hold the new set of changes. I haven't added the StreamConfigUtils class, was thinking if i can do it on a separate ticket. WDYT? Also, the javadoc is pending at this moment. Will add once we are ok with the changes here.

vamossagar12 · 2022-02-24T03:45:38Z

@guozhangwang , this PR is showing a failure in streams:compileTestJava.. It seems to be working on my local.

guozhangwang · 2022-02-25T18:08:40Z

Hello @vamossagar12 I checked out your branch, and run the streams:compileTestJava on my local machine it it also fails with:

> Task :streams:compileTestJava
/Users/guozhang/Workspace/github/guozhangwang/kafka-work/streams/src/test/java/org/apache/kafka/streams/integration/ErrorHandlingIntegrationTest.java:95: warning: [deprecation] CACHE_MAX_BYTES_BUFFERING_CONFIG in org.apache.kafka.streams.StreamsConfig has been deprecated
                mkEntry(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0),
                                     ^
error: warnings found and -Werror specified
1 error
1 warning

> Task :streams:compileTestJava FAILED

I guess your branch was not rebased on top of trunk and maybe that's why you did not see the failure?

vamossagar12 · 2022-02-27T10:53:07Z

@guozhangwang i had done a rebase before pushing and hadn't notice the issue. Not sure what went wrong there(maybe an oversight from me). Anyways i did another rebase and now i could see the error. Have fixed it and pushed again.

vamossagar12 · 2022-02-28T02:57:09Z

@guozhangwang , now i see failures with raft/zk tests. that seems unrelated to this PR.

guozhangwang

I only reviewed the code snippet around deprecated config logic, only nit comments regarding the log lines.

@ableegoldman could you also take another look?

guozhangwang · 2022-03-01T18:29:05Z

overriding {} to {} here could also be a bit confusing since it could be interpreted as the second override the first by some one. I'd suggest we just say Both deprecated config {} and new config {} are set, hence {} is ignored and the new config {} (value {}) is used.

Yeah makes sense. I have updated it to use topology name along with the rest of log message. I guess that should be fine?

guozhangwang · 2022-03-01T18:30:10Z

Ditto here: we just say "Only deprecated config {} is set and hence it's value {} would be used; we suggest setting the new config {} instead since the deprecated config {} would be removed in the future."

Same as above.

guozhangwang · 2022-03-09T23:50:51Z

Re-triggered jenkins.

guozhangwang · 2022-03-09T23:51:10Z

cc @ableegoldman

guozhangwang · 2022-03-16T17:05:15Z

@vamossagar12 could you resolve the conflicts before I re-trigger jenkins again?

Co-authored-by: A. Sophie Blee-Goldman <ableegoldman@gmail.com>

…ax buffer bytes size and then resizing

…compatibility

…er-bytes-total

…ache size config overrides

vamossagar12 · 2022-03-21T09:07:38Z

@vamossagar12 could you resolve the conflicts before I re-trigger jenkins again?

@guozhangwang done. On my local, only one test failed in streams which is => org.apache.kafka.streams.integration.PurgeRepartitionTopicIntegrationTest#shouldRestoreState

guozhangwang · 2022-03-22T00:16:20Z

Thanks @vamossagar12 . I've merged the PR, and please go ahead and mark the ticket / KIP as for 3.3.0.

vamossagar12 · 2022-03-22T03:58:09Z

Thanks @guozhangwang !

ableegoldman

Thanks for the new PR @vamossagar12 ! Sorry I wasn't able to take a look earlier but I just gave it a quick pass now. I took the liberty of moving the #getTotalCacheSize method to the StreamsConfigUtils class myself since I'm doing a quick warning log PR in that part of the code anyways.

ableegoldman · 2022-03-29T05:31:44Z

-            createAndAddStreamThread(cacheSizePerThread, i);
+            createAndAddStreamThread(0L, 0L, i);
        }
+        // Initially, all Stream Threads are created with 0 cache size and max buffer size and then resized here.


Why do we do it this way? ie rather than just computing the size upfront and creating the threads with that? I find this a bit confusing, mainly because I can't tell if there is a technical reason for doing it this way or it was just a design choice

ableegoldman · 2022-03-29T05:40:02Z

+                // and then resize them later
+                streamThread = createAndAddStreamThread(0L, 0L, threadIdx);
+                final int numLiveThreads = getNumLiveStreamThreads();
+                resizeThreadCacheAndBufferMemory(numLiveThreads + 1);


I think this feedback got lost in the shuffle when we reverted the original PR, but this needs to be fixed -- the + 1 was only necessary in the old code because we resized the cache before adding the new thread/computing the thread count. Now that we first create the new thread, the numLiveThreads count should accurately reflect the number of current threads, so we shouldn't be adding to it anymore.

I'll include a fix for this on the side in a PR I'm doing so no worries 🙂

I read the code change, and I agree with you. We don't need +1 anymore. Nice catch!

Oh yeah I forgot about it :( Thanks for the new PR. I will take a look..

ableegoldman · 2022-03-29T06:30:16Z

FYI @vamossagar12 here is the PR: #11959

vamossagar12 · 2022-03-29T07:22:10Z

Thanks for the new PR @vamossagar12 ! Sorry I wasn't able to take a look earlier but I just gave it a quick pass now. I took the liberty of moving the #getTotalCacheSize method to the StreamsConfigUtils class myself since I'm doing a quick warning log PR in that part of the code anyways.

Thanks for that! I thought I will create a follow up PR for moving to StreamsConfigUtils after another issue that I have been working on. You have done it already !

…11959) 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 a fix for an off-by-one error in #11796 Reviewers: Luke Chen <showuon@gmail.com>, Walker Carlson <wcarlson@confluent.io>, Sagar Rao <sagarmeansocean@gmail.com>

* KAFKA-13152: Replace "buffered.records.per.partition" with "input.buffer.max.bytes" (apache#11796) Implements KIP-770 Reviewers: Guozhang Wang <wangguoz@gmail.com> (cherry picked from commit 0924fd3) * KAFKA-13945: add bytes/records consumed and produced metrics (apache#12235) Implementation of KIP-846: Source/sink node metrics for Consumed/Produced throughput in Streams Adds the following INFO topic-level metrics for the total bytes/records consumed and produced: bytes-consumed-total records-consumed-total bytes-produced-total records-produced-total Reviewers: Kvicii <Karonazaba@gmail.com>, Guozhang Wang <guozhang@apache.org>, Bruno Cadonna <cadonna@apache.org> * Update CODEOWNERS (#721) (cherry picked from commit ad18b43) Co-authored-by: vamossagar12 <sagarmeansocean@gmail.com> Co-authored-by: A. Sophie Blee-Goldman <sophie@confluent.io> Co-authored-by: Chris Johnson <53450846+cjohnson-confluent@users.noreply.github.com>

…nput.buffer.max.bytes" (apache#11796)" This reverts commit 0924fd3.

vamossagar12 force-pushed the kip-770 branch from 8b3780e to b48f7d5 Compare February 23, 2022 17:20

guozhangwang requested a review from ableegoldman February 25, 2022 18:02

vamossagar12 force-pushed the kip-770 branch from b48f7d5 to 7a41243 Compare February 27, 2022 10:51

guozhangwang reviewed Mar 1, 2022

View reviewed changes

mjsax changed the title ~~Kip 770~~ KAFKA-13152: Replace "buffered.records.per.partition" with "input.buffer.max.bytes" Mar 7, 2022

mjsax added kip Requires or implements a KIP streams labels Mar 7, 2022

vamossagar12 force-pushed the kip-770 branch from 7a41243 to 23e1b0b Compare March 9, 2022 05:52

s0r01ed and others added 14 commits March 21, 2022 13:47

KAFKA-13152: Initial commit

cb6c741

Adding more tests

e30fb0c

Adding test cases and fixing check style issues

92aa5e1

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

118716a

Co-authored-by: A. Sophie Blee-Goldman <ableegoldman@gmail.com>

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

b91cb1a

Co-authored-by: A. Sophie Blee-Goldman <ableegoldman@gmail.com>

KAFKA-13152: Initial commit

bf938dd

Adding more tests

22d5606

Adding test cases and fixing check style issues

b0ecac8

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

fc9e7dc

Co-authored-by: A. Sophie Blee-Goldman <ableegoldman@gmail.com>

Fixing test cases and correcting checkstyle issues

fb2298f

Adding missed files due to rebase

f0cf258

Rebasing with trunk

3658cec

Further Review comments

3c4dd20

Adding logic to first create new StreamThread with 0 cache size and m…

9099d0e

…ax buffer bytes size and then resizing

s0r01ed added 11 commits March 21, 2022 13:54

Trying to remove deprecated config

0915844

Adding suppresswarnings for usage of deprecated configs for backward …

4da8e90

…compatibility

Addressing spotBugsMain errors and renaming total-bytes to input-buff…

d6d8838

…er-bytes-total

Fixing failing test cases plus corrected logic in ToplogyConfig for c…

5b77d9b

…ache size config overrides

Bug fixes + changes to add cache-size metric

666ab76

Adding total cache size sensor

5c88525

Rebasing

50adf47

Another rebase

bfc7be8

Review comments around logging adressed

60c1cbc

Removed deprecated config

5987c3c

Reducing npath complexity of constructor

84c75ce

vamossagar12 force-pushed the kip-770 branch from 60b3dc9 to 84c75ce Compare March 21, 2022 09:06

guozhangwang merged commit 0924fd3 into apache:trunk Mar 22, 2022

ableegoldman reviewed Mar 29, 2022

View reviewed changes

ableegoldman mentioned this pull request Mar 29, 2022

MINOR: log warning when topology override for cache size is non-zero #11959

Merged

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

Revert "KAFKA-13152: Replace "buffered.records.per.partition" with "i…

28ad14e

…nput.buffer.max.bytes" (apache#11796)" This reverts commit 0924fd3.

Conversation

vamossagar12 commented Feb 22, 2022 • edited by mjsax Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

vamossagar12 commented Feb 23, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

vamossagar12 commented Feb 24, 2022

Uh oh!

guozhangwang commented Feb 25, 2022

Uh oh!

vamossagar12 commented Feb 27, 2022

Uh oh!

vamossagar12 commented Feb 28, 2022

Uh oh!

guozhangwang 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!

Uh oh!

guozhangwang commented Mar 9, 2022

Uh oh!

guozhangwang commented Mar 9, 2022

Uh oh!

guozhangwang commented Mar 16, 2022

Uh oh!

vamossagar12 commented Mar 21, 2022

Uh oh!

guozhangwang commented Mar 22, 2022

Uh oh!

vamossagar12 commented Mar 22, 2022

Uh oh!

ableegoldman 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 29, 2022

Uh oh!

vamossagar12 commented Mar 29, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

vamossagar12 commented Feb 22, 2022 •

edited by mjsax

Loading

vamossagar12 commented Feb 23, 2022 •

edited

Loading