KAFKA-14247: Consumer background thread base implementation by philipnee · Pull Request #12672 · apache/kafka

philipnee · 2022-09-21T18:32:24Z

1-pager: https://cwiki.apache.org/confluence/display/KAFKA/Proposal%3A+Consumer+Threading+Model+Refactor

EventHandler runs a task to ingest the ApplicationEvents. The task runs on a background thread, and consumes events from the ApplicationEventQueue, and produces events to the BackgroundEventQueue.

The PR consist of the basic skeleton of the background thread with

network client
basic demonstration of the consumption logic

Continuation of #12663

Committer Checklist (excluded from commit message)

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

kirktrue

Are ConsumerRequestEvent/ConsumerResponseEvent and ApplicationEvent/BackgroundEvent the same thing?

philipnee · 2022-09-30T18:57:52Z

@mhowlett @kirktrue Mind reviewing this PR? This is actually stack on top of : #12663 So some code might be outdated

Documentation on the beahvior

Stubbed event handler

This reverts commit 625b16c.

This reverts commit 816b285.

clean up documentation implemented poll and commitSync clean up clean up

background thread

initialize backgorund thread. refactor stuff a bit Added unit tests more unit tests renaming and documentation clean up clean up clean up

network IO thread

guozhangwang

Thanks @philipnee for the PR. I made a first pass and left some comments.

guozhangwang · 2022-10-07T20:56:02Z

+            log.debug("{} started", getClass());
+            while (running) {
+                runOnce();
+                time.sleep(retryBackoffMs);


Instead of setting a sleep here, I think it's better to have a waitOnEmptyInputQueue function with a condition. Where the caller thread would notify when enqueuing a new item.

@guozhangwang - I thought we wanted to maintain an active loop because we want to keep sending fetches and rebalance requests despite an empty ApplicationQueue.

Yeah I think so -- please see my other comment below: #12672 (comment). But I still feel that instead of having a sleep for each iteration, we should just consider poll on the network client with a timeout, i.e. supposing our loop would look like:

try to get some event from the queue, if yes, then handle it and potentially send the corresponding request;

if the queue is empty, then try to send a fetch request.

poll the network client in order to receive any responses ready-to-read on the socket.

Then our logic could be: if there's no new actions taken at step 1/2), i.e. we do not have any new items from the queue, and we do not yet need to send any new fetch/rebalance-related requests, then at step 3) we poll for a bit longer time until being notified by the caller that there's new items in the queue; otherwise, at step 3) we just poll without timeout and then immediately move on to the next iteration.

guozhangwang · 2022-10-07T22:27:14Z

+                applicationEventQueue,
+                backgroundEventQueue);
+        assertTrue(eventHandler.add(new NoopApplicationEvent("hello-world")));
+        while (eventHandler.isEmpty()) { }


This is a general comment: I think it worth adding a poll(timeout) in the EventHandler interface since 1) for testing, we do not need the while loop any more, and 2) in production code, for the caller thread scenarios I think there are cases where a poll(timeout) would be useful as well.

guozhangwang · 2022-10-07T22:31:06Z

+
+    public enum EventType {
+        COMMIT,
+        NOOP,


Ditto, I think we should keep the testing-related element out of the production code eventually. If we are going to remove it later then that's fine.

Thanks, I think the NOOP event will be moved to the test package in the future PR; I left it here to elucidate the purpose of a few methods in the background thread.

Thanks for clarifying this, makes sense.

guozhangwang · 2022-10-07T22:31:33Z

+/**
+ * Noop event. Intentionally left it here for demonstration purpose.
+ */
+public class NoopBackgroundEvent extends BackgroundEvent {


Ditto here, I think we should move it to test package, not main package.

refactor PR comment

philipnee · 2022-10-12T17:11:40Z

Hey @guozhangwang , much thanks for the detail reviews, I tried to address some of the comments, please review them. In particular:

poll timeout logic
InterruptException handling (A side question here, when the BT got interrupted, shouldn't we terminate instead of swallowing the exception?)
Added a couple of tests to test the network client calls in a single poll.
I also removed the interface, originally I thought it would be helpful to write stubbed tests.

I left the NOOP event there, but I agree we should move it once we've got an actual event implemented, which should happen soon.

philipnee · 2022-10-13T16:19:25Z

cc John - @vvcephei

kirktrue

I did another review of the code. I haven't gotten as far as the tests yet.

kirktrue · 2022-10-18T16:53:53Z

+            log.error("The background thread failed due to unexpected error",
+                    t);
+            if (t instanceof RuntimeException)
+                this.exception.set(Optional.of((RuntimeException) t));


Do we want to use KafkaException instead of RuntimeException?

We could, there are other places in the code that uses RTE so trying to be consistent with the existing code.

kirktrue · 2022-10-18T17:15:19Z

+    }
+
+    public void close() {
+        this.wakeup();


Does the order of these two lines (wakeup and running = false) want to be swapped?

close is called by the polling thread, right? If wakeup is called, it will throw an exception, but then the background thread might potentially loop back around to runOnce before the call to set running to false by the polling thread.

kirktrue · 2022-10-18T17:16:54Z

+                               Sensor fetcherThrottleTimeSensor) {
+        this.applicationEventQueue = applicationEventQueue;
+        this.backgroundEventQueue = backgroundEventQueue;
+        ConsumerMetadata metadata = bootstrapMetadata(logContext,


The call to bootstrapMetadata is blocking, right? Is that what we want? If so, can you add some comments (in the code) as to that behavior?

It should be fine for this call to block, I think because all it does is create a Node object with id = -1 and an IP address. The actual metadata refresh happens during the background thread loop, not during this function call, since there are no network response deps to this call, so it shouldn't be blocking for too long.

vvcephei · 2022-10-19T23:31:13Z

Hey, @philipnee, @guozhangwang isn't available for reviews right now, so I'll pick it up.

I reviewed it, and it all looks good to me. I had a large number of code convention corrections, so I just went ahead and applied the changes, rather than peppering you with nitpicks. I hope that's ok.

Generally:

please use final when it can be used
indent 4 spaces instead of 8
when a method call has too many args to fit within the line length limit, please break it up into one-arg-per-line, rather than just breaking the args at an arbitrary point.

I know that's all arbitrary, and that there are merits to doing things other ways; I'm just trying to keep these files in line with the rest of AK.

Thanks! Once the tests pass, I'll go ahead and merge.

vvcephei · 2022-10-20T01:48:36Z

Unrelated test failure: [Build / JDK 17 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testDelayedConfigurationOperations()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-12672/41/testReport/junit/org.apache.kafka.controller/QuorumControllerTest/Build___JDK_17_and_Scala_2_13___testDelayedConfigurationOperations__/)

philipnee · 2022-10-20T01:49:57Z

Thanks @vvcephei Will be better at the basic styling requirements in the future...

…2672) Adds skeleton of the background thread. 1-pager: https://cwiki.apache.org/confluence/display/KAFKA/Proposal%3A+Consumer+Threading+Model+Refactor Continuation of apache#12663 Reviewers: Guozhang Wang <guozhang@apache.org>, Kirk True <kirk@mustardgrain.com>, John Roesler <vvcephei@apache.org>

…2672) (#76) Adds skeleton of the background thread. 1-pager: https://cwiki.apache.org/confluence/display/KAFKA/Proposal%3A+Consumer+Threading+Model+Refactor Continuation of apache#12663 Reviewers: Guozhang Wang <guozhang@apache.org>, Kirk True <kirk@mustardgrain.com>, John Roesler <vvcephei@apache.org> Co-authored-by: Philip Nee <pnee@confluent.io>

kirktrue requested changes Sep 21, 2022

View reviewed changes

philipnee force-pushed the consumer-refactor-background-thread branch from 12ce01d to 7c7c072 Compare September 27, 2022 23:00

philipnee changed the title ~~KAFKA-14252: Consumer refactor background thread~~ KAFKA-14252: Consumer background thread base implementation Sep 28, 2022

philipnee commented Sep 28, 2022

View reviewed changes

Comment thread clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerBackgroundThread.java Outdated

philipnee force-pushed the consumer-refactor-background-thread branch from 0ae5431 to af2727e Compare September 30, 2022 18:55

philipnee marked this pull request as ready for review September 30, 2022 18:55

philipnee added 22 commits October 3, 2022 09:41

Basic event handler definition

a059d85

Define EventHandler interface

008ca46

lint error

5bb3362

moving packages

6551807

reduce the scope

86191c2

PR comments for naming and documentation

c0ef143

Documentation on the beahvior

564660f

Documentation on the beahvior

Add isEmpty and remove excessive public

4f0a15e

Remove unused import

4828f71

Handle capacity limitation

779c00c

clean up a comment

b18098d

A stubbed event handler to demonstrate the usage.

ce00e48

Stubbed event handler

typo

30e059b

Revert "typo"

0e5a6dd

This reverts commit 625b16c.

Revert "A stubbed event handler to demonstrate the usage."

1ed6836

This reverts commit 816b285.

Prototyping

bbabb93

clean up and documentation

e081c5f

clean up documentation implemented poll and commitSync clean up clean up

background thread

d5c8009

background thread

Testing, background thread impl.

684a0a1

initialize backgorund thread. refactor stuff a bit Added unit tests more unit tests renaming and documentation clean up clean up clean up

renaming for clarity

c80059a

fixes

480eb9a

Refactor the background to make clear it is a network io thread

422eed4

network IO thread

philipnee force-pushed the consumer-refactor-background-thread branch from 88e3d39 to 422eed4 Compare October 3, 2022 16:44

philipnee force-pushed the consumer-refactor-background-thread branch from 28a5814 to 6cfa912 Compare October 4, 2022 18:53

philipnee added 3 commits October 4, 2022 11:59

exception handling

c3d3d03

clean up unused import

7669b0a

More documentation and better exception handling

2d66bda

guozhangwang reviewed Oct 7, 2022

View reviewed changes

philipnee added 3 commits October 10, 2022 14:19

refactor based on PR comment

3a52b13

refactor PR comment

Move network client construction to the handler

45c55f9

clean up

8f2c870

philipnee force-pushed the consumer-refactor-background-thread branch from d17fe3b to 8f2c870 Compare October 11, 2022 18:57

philipnee added 4 commits October 11, 2022 18:57

wakeup logic

a343d2d

documentation

7f3424d

swallow interrupt exception

d0e8f20

test swallowed exception

5440664

philipnee added 2 commits October 12, 2022 10:20

remove the test for testing

498e06b

wakeup exception test

f3158aa

kirktrue requested changes Oct 18, 2022

View reviewed changes

philipnee and others added 3 commits October 18, 2022 11:02

refactor based on PR comment

fbb59e8

debug message

bec9391

apply AK code conventions

f10d530

vvcephei merged commit 0a045d4 into apache:trunk Oct 20, 2022

philipnee added KIP-848 The Next Generation of the Consumer Rebalance Protocol ctr Consumer Threading Refactor (KIP-848) labels May 23, 2023

Conversation

philipnee commented Sep 21, 2022 • edited by vvcephei Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

philipnee commented Sep 30, 2022

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

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!

Uh oh!

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!

philipnee commented Oct 12, 2022

Uh oh!

philipnee commented Oct 13, 2022

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

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!

vvcephei commented Oct 19, 2022

Uh oh!

vvcephei commented Oct 20, 2022

Uh oh!

philipnee commented Oct 20, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

philipnee commented Sep 21, 2022 •

edited by vvcephei

Loading