KAFKA-19214: Clean up use of Optionals in RequestManagers.entries() #19609

kirktrue · 2025-05-01T00:25:00Z

Change:

public List<Optional<? extends RequestManager>> entries();

to:

public List<RequestManager> entries();

and clean up the callers.

Reviewers: TengYao Chi kitingiao@gmail.com, Andrew Schofield
aschofield@confluent.io, Chia-Ping Tsai chia7712@gmail.com

Change: public List<Optional<? extends RequestManager>>> entries(); to: public List<RequestManager> entries(); and clean up the callers.

frankvicky

@kirktrue: Thanks for the clean up!

frankvicky · 2025-05-01T09:46:00Z

...nts/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java

-        List<Optional<? extends RequestManager>> list = new ArrayList<>();
-        list.add(Optional.of(coordinatorRequestManager));
-        list.add(Optional.of(heartbeatRequestManager));
+        List<RequestManager> list = new ArrayList<>();


Could we use List.of here?

Not easily, no. I made two different attempts (see commit history), and both just added a bunch of ugly code just to make this look cleaner.

The problem is that the list contains both RequestManagers and Optional<RequestManager>s, so the type of the list would have to be defined as List<Object>. The idea of this PR is to make the entries list only include the succinct set of RequestManagers that are actually in use.

I'm open to suggestions, though 😄

@kirktrue I think I'm missing something here. Now that the list only contains RequestManager instances, can't you just write:

List<RequestManager> list = List.of(coordinatorRequestManager, heartbeatRequestManager);

I tried that locally and it looked good to me.

Oof! I was looking at the lists being built up in RequestManagers’ constructors 🤦‍♂️ I've updated ConsumerNetworkThreadTest’s list building to use List.of() as suggested.

frankvicky · 2025-05-01T09:46:19Z

...nts/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java

-        list.add(Optional.of(coordinatorRequestManager));
-        list.add(Optional.of(heartbeatRequestManager));
-        list.add(Optional.of(offsetsRequestManager));
+        List<RequestManager> list = new ArrayList<>();


List<RequestManager> list = List.of(coordinatorRequestManager, heartbeatRequestManager, offsetsRequestManager);

Yes, fixed this as well. Thank you for your patience 😄

AndrewJSchofield

Thanks for the PR. This piece of code was always a bit messy so this is a nice improvement.

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java

...nts/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java

chia7712 · 2025-05-01T17:08:09Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java

@@ -154,16 +153,12 @@ void runOnce() {
        lastPollTimeMs = currentTimeMs;

        final long pollWaitTimeMs = requestManagers.entries().stream()
-                .filter(Optional::isPresent)
-                .map(Optional::get)
                .map(rm -> rm.poll(currentTimeMs))
                .map(networkClientDelegate::addAll)
                .reduce(MAX_POLL_TIMEOUT_MS, Math::min);


I made a change here, but instead of:

final long pollWaitTimeMs = requestManagers.entries().stream() .map(rm -> rm.poll(currentTimeMs)) .mapToLong(networkClientDelegate::addAll) .min() .orElse(MAX_POLL_TIMEOUT_MS);

I had to add a call to filter():

final long pollWaitTimeMs = requestManagers.entries().stream() .map(rm -> rm.poll(currentTimeMs)) .mapToLong(networkClientDelegate::addAll) .filter(ms -> ms <= MAX_POLL_TIMEOUT_MS) .min() .orElse(MAX_POLL_TIMEOUT_MS);

The stream must only contain values that are numerically less-than-or-equal-to the maximum (MAX_POLL_TIMEOUT_MS) when min() is invoked or else we could end up with value that's greater than our maximum. Fortunately there is a unit test that caught that little wrinkle 😄

thanks for the sharing. nice find!

BTW, do you know the purpose of max poll time? it seems we don't honor the timeUntilNextPollMs configured by the rm. If the value larger than 5 secs is invalid, should we throw exception in PollResult? Additionally, maybe the default timeUntilNextPollMs should be configured to 5 seconds (or optional.empty) as the default value is useless.

Yes, the polling timeout logic is a little... "hard to understand" 😞

The idea was that we want to block in poll() so that we're not just looping in runOnce() continuously. The amount of time to block is dependent on the inflight requests. Some requests, like coordinator discovery, don't want to wait at all, but others don't want to impose any timeouts on blocking in runOnce().

But yes, this is an area that could use some clean up.

run a simple benchmark

@State(Scope.Thread) @OutputTimeUnit(TimeUnit.MILLISECONDS) public class ReduceBenchmark { private final List<Long> values = IntStream.range(0, 100 * 1000).mapToObj(i -> (long) i).toList(); private final long maxValue = 100; @Benchmark public void testReduce(Blackhole blackhole) { blackhole.consume(values.stream().reduce(maxValue, Math::min)); } @Benchmark public void testMin(Blackhole blackhole) { blackhole.consume(values.stream().mapToLong(i -> i) .filter(i -> i <= maxValue) .min() .orElse(maxValue)); } }

Benchmark Mode Cnt Score Error Units ReduceBenchmark.testMin thrpt 25 7.915 ± 0.087 ops/ms ReduceBenchmark.testReduce thrpt 25 3.256 ± 0.017 ops/ms

It seems the filter could optimize it a bit.

…d clean up timeout determination logic in ConsumerNetworkThread

AndrewJSchofield

Just a couple of final comments.

AndrewJSchofield · 2025-05-05T18:32:20Z

...nts/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java

-        List<Optional<? extends RequestManager>> list = new ArrayList<>();
-        list.add(Optional.of(coordinatorRequestManager));
-        list.add(Optional.of(heartbeatRequestManager));
+        List<RequestManager> list = new ArrayList<>();


@kirktrue I think I'm missing something here. Now that the list only contains RequestManager instances, can't you just write:

List<RequestManager> list = List.of(coordinatorRequestManager, heartbeatRequestManager);

I tried that locally and it looked good to me.

AndrewJSchofield · 2025-05-05T18:32:56Z

...nts/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java

-        list.add(Optional.of(coordinatorRequestManager));
-        list.add(Optional.of(heartbeatRequestManager));
-        list.add(Optional.of(offsetsRequestManager));
+        List<RequestManager> list = new ArrayList<>();


List<RequestManager> list = List.of(coordinatorRequestManager, heartbeatRequestManager, offsetsRequestManager);

… tests

…pache#19609) Change: `public List<Optional<? extends RequestManager>> entries();` to: `public List<RequestManager> entries();` and clean up the callers. Reviewers: TengYao Chi <kitingiao@gmail.com>, Andrew Schofield <aschofield@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

KAFKA-19214: Clean up use of Optionals in RequestManagers.entries()

0df2acc

Change: public List<Optional<? extends RequestManager>>> entries(); to: public List<RequestManager> entries(); and clean up the callers.

github-actions bot added triage PRs from the community consumer clients small Small PRs labels May 1, 2025

frankvicky reviewed May 1, 2025

View reviewed changes

AndrewJSchofield requested changes May 1, 2025

View reviewed changes

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java Outdated Show resolved Hide resolved

...nts/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java Outdated Show resolved Hide resolved

AndrewJSchofield added ci-approved and removed triage PRs from the community labels May 1, 2025

chia7712 reviewed May 1, 2025

View reviewed changes

kirktrue added 2 commits May 2, 2025 11:12

Merge branch 'trunk' into KAFKA-19214-remove-optional

789d597

Use List.of() where possible, emulate listOf() in RequestManagers, an…

306c51e

…d clean up timeout determination logic in ConsumerNetworkThread

github-actions bot removed the small Small PRs label May 2, 2025

kirktrue added 2 commits May 2, 2025 12:49

Another attempt at being cleaner during RequestManagers construction

1005646

Reverting entries build up to avoid overhead

d62e4a4

github-actions bot added the small Small PRs label May 2, 2025

chia7712 approved these changes May 5, 2025

View reviewed changes

kirktrue requested a review from AndrewJSchofield May 5, 2025 17:08

AndrewJSchofield requested changes May 5, 2025

View reviewed changes

Using List.of() for RequestManager lists in ConsumerNetworkThreadTest…

ab0b687

… tests

AndrewJSchofield approved these changes May 7, 2025

View reviewed changes

chia7712 approved these changes May 7, 2025

View reviewed changes

AndrewJSchofield merged commit d3707fc into apache:trunk May 7, 2025
27 checks passed

kirktrue deleted the KAFKA-19214-remove-optional branch May 7, 2025 16:35

KAFKA-19214: Clean up use of Optionals in RequestManagers.entries() #19609

KAFKA-19214: Clean up use of Optionals in RequestManagers.entries() #19609

Uh oh!

Conversation

kirktrue commented May 1, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

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

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!

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

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

Uh oh!

Uh oh!

kirktrue commented May 1, 2025 •

edited by github-actions bot

Loading