KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections #19917

kirktrue · 2025-06-06T19:24:40Z

Profiling has shown that using the Collections Streams API approach adds
unnecessary overhead compared to a traditional for loop. Minor revisions
to the code have been made to use simpler constructs to improve
performance.

Reviewers: Lianet Magrans lmagrans@confluent.io, Andrew Schofield
aschofield@confluent.io

…n critical sections

AndrewJSchofield

Thanks for the PR. A few comments to address.

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

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

AndrewJSchofield · 2025-06-09T14:48:52Z

...s/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java

+            if (requestState.isExpired())
+                expiredRequests.add(requestState);
+        }
+
        expiredRequests.forEach(TopicMetadataRequestState::expire);


Couldn't you just not build expiredRequests and call requestState.expire() at L91?

Because the request’s expire() method removes itself from inflightRequests, won't that throw an exception because the for loop's underlying Iterator wasn't used to remove the entry?

Well, I take it back. Apparently it's not an error to remove it directly from the collection while iterating over the collection 🤔

I've removed the superfluous list buildup.

Apparently it's not an error to remove it directly from the collection while iterating over the collection

Agree it doesn't seem to fail with the tests we have, but still I would expect it may not be safe to modify the collection from within the loop (and may lead to ConcurrentModification exception at some point?).

I would say we play safe and use an iterator, makes sense?

…se String.join()

kirktrue · 2025-06-11T19:03:23Z

@AndrewJSchofield @lianetm—Anything else I need to fix before we can merge? Thanks!

lianetm

Thanks @kirktrue!

lianetm · 2025-06-11T20:56:30Z

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

+            Node node = entry.getKey();
+            FetchSessionHandler.FetchRequestData fetchRequestData = entry.getValue().build();
+            map.put(node, fetchRequestData);


Suggested change

Node node = entry.getKey();

FetchSessionHandler.FetchRequestData fetchRequestData = entry.getValue().build();

map.put(node, fetchRequestData);

map.put(entry.getKey(), entry.getValue().build());

This was just to make the code a little clearer about the exact processing and types in use. I'll make it more succinct.

Changed as requested.

lianetm · 2025-06-11T20:59:37Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java

-        if (!revokePausedPartitions.isEmpty()) {
-            log.info("The pause flag in partitions [{}] will be removed due to revocation.", revokePausedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+        if (!revokePausedPartitions.isEmpty() && log.isInfoEnabled()) {
+            log.info("The pause flag in partitions [{}] will be removed due to revocation.", Utils.topicPartitionString(revokePausedPartitions));


Given that er end up printing the string in between [], can't we simply print the set directly?

Suggested change

log.info("The pause flag in partitions [{}] will be removed due to revocation.", Utils.topicPartitionString(revokePausedPartitions));

log.info("The pause flag in partitions {} will be removed due to revocation.", revokePausedPartitions);

You're right. Changed.

lianetm · 2025-06-11T21:02:47Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

-        log.info("Revoking previously assigned partitions {}", revokedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+
+        if (log.isInfoEnabled())
+            log.info("Revoking previously assigned partitions {}", Utils.topicPartitionString(revokedPartitions));


Here also I expect we can simplify and just print the set? Only difference is that it will include the brackets, which is ok I expect?

Yep. Changed.

lianetm · 2025-06-11T21:08:50Z

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

+            if (!(ce instanceof CompletableApplicationEvent))
+                continue;
+
+            CompletableApplicationEvent<?> cae = (CompletableApplicationEvent<?>) ce;
+
+            if (cae.requireSubscriptionMetadata())
+                subscriptionMetadataEvent.add(cae);


maybe simplify ?

Suggested change

if (!(ce instanceof CompletableApplicationEvent))

continue;

CompletableApplicationEvent<?> cae = (CompletableApplicationEvent<?>) ce;

if (cae.requireSubscriptionMetadata())

subscriptionMetadataEvent.add(cae);

if (ce instanceof CompletableApplicationEvent && ((CompletableApplicationEvent<?>) ce).requireSubscriptionMetadata()) {

subscriptionMetadataEvent.add((CompletableApplicationEvent<?>) ce);

}

lianetm · 2025-06-11T21:12:51Z

.../main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java

-                log.info("The pause flag in partitions [{}] will be removed due to revocation.", revokePausedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+
+        if (!revokePausedPartitions.isEmpty() && log.isInfoEnabled())
+            log.info("The pause flag in partitions [{}] will be removed due to revocation.", Utils.topicPartitionString(revokePausedPartitions));


similar to above, if we are adding [] to the string of partitions, couldn't we just print the set?

lianetm · 2025-06-11T21:13:04Z

.../main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java

-            log.info("The pause flag in partitions [{}] will be removed due to partition lost.", lostPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+
+        if (!lostPausedPartitions.isEmpty() && log.isInfoEnabled())
+            log.info("The pause flag in partitions [{}] will be removed due to partition lost.", Utils.topicPartitionString(lostPartitions));


lianetm · 2025-06-13T13:28:17Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

@@ -310,7 +311,9 @@ private void process(final AssignmentChangeEvent event) {
            manager.updateTimerAndMaybeCommit(event.currentTimeMs());
        }

-        log.info("Assigned to partition(s): {}", event.partitions().stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+        if (log.isInfoEnabled())
+            log.info("Assigned to partition(s): {}", Utils.topicPartitionString(event.partitions()));


ok with the change, even though this one is not on any hot-path I expect, right? (just API calls to consumer.assign that do a single trip to the background, no requests).

True, it's just cleaner now.

lianetm · 2025-06-13T15:56:23Z

...src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java

-                .collect(Collectors.toList());
+        List<CompletableEvent<?>> events = new ArrayList<>();
+
+        for (CompletableEvent<?> event : tracked) {


Should we add a comment saying that we're intentionally using for loop over the Streams API for perf, as this runs on every iteration of the background thread? (I'm afraid folks will be tempted to change this simple loops back to streams api)

Agreed. I added one here and similar comments in a few other places.

lianetm · 2025-06-13T16:15:44Z

...src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java

+
+            if (event.future().completeExceptionally(error)) {
+                log.debug("Event {} completed exceptionally since the consumer is closing", event);
+                count++;


incrementing here is not exactly the same that it used to happen before, right?

Before the PR, the count inc would happen after the peek(expireEvent), so it would count events that were expired but not completedExceptionally (the ones going through line 202, where now we don't increment the count)

This count doesn't seem used anyways, but better to be accurate and consistent with what it's done in the reap func (there the count includes all expired events, no matter if we completed them exceptionally or not)

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java

Line 115 in 2b2552d

count++;

Yes, it's different. I assert it's more accurate, but I changed it to mimic the existing counting.

lianetm

Thanks @kirktrue! Took another pass, just one comment

lianetm · 2025-06-17T20:12:02Z

...s/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java

+            if (requestState.isExpired())
+                expiredRequests.add(requestState);
+        }
+
        expiredRequests.forEach(TopicMetadataRequestState::expire);


Apparently it's not an error to remove it directly from the collection while iterating over the collection

Agree it doesn't seem to fail with the tests we have, but still I would expect it may not be safe to modify the collection from within the loop (and may lead to ConcurrentModification exception at some point?).

I would say we play safe and use an iterator, makes sense?

… via an iterator This is to prevent possible state issues that could arise from removing elements of the list as it's being iterated over.

lianetm

Thanks! LGTM.

…n critical sections (#19917) Profiling has shown that using the Collections Streams API approach adds unnecessary overhead compared to a traditional for loop. Minor revisions to the code have been made to use simpler constructs to improve performance. Reviewers: Lianet Magrans <lmagrans@confluent.io>, Andrew Schofield <aschofield@confluent.io>

KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs i…

6264cae

…n critical sections

github-actions bot added triage PRs from the community consumer clients labels Jun 6, 2025

Reverting some overzealous changes

10468b4

AndrewJSchofield added ci-approved and removed triage PRs from the community labels Jun 7, 2025

AndrewJSchofield requested changes Jun 9, 2025

View reviewed changes

kirktrue added 2 commits June 9, 2025 12:28

No longer sorting topic partitions and coercing objects in order to u…

3768767

…se String.join()

Removing unnecessary list for expiration; just expiring directly

0612cd9

kirktrue requested review from AndrewJSchofield and lianetm June 9, 2025 22:34

kirktrue added 2 commits June 12, 2025 10:14

Merge branch 'trunk' into KAFKA-19297-refactor-away-streams

28a8772

Merge branch 'trunk' into KAFKA-19297-refactor-away-streams

2b2552d

lianetm reviewed Jun 13, 2025

View reviewed changes

kirktrue added 5 commits June 16, 2025 14:03

Merge branch 'trunk' into KAFKA-19297-refactor-away-streams

7388c2e

PR review cleanup

453cf5f

Reverted some whitespace

2a338a7

Whitespace change

6f2531a

Added comments and reverted more whitespace changes

cba0845

kirktrue requested a review from lianetm June 17, 2025 12:25

lianetm reviewed Jun 17, 2025

View reviewed changes

Update TopicMetadataRequestManager.poll() to remove the request state…

1d4da91

… via an iterator This is to prevent possible state issues that could arise from removing elements of the list as it's being iterated over.

lianetm approved these changes Jun 18, 2025

View reviewed changes

AndrewJSchofield approved these changes Jun 18, 2025

View reviewed changes

lianetm merged commit adcf10c into apache:trunk Jun 18, 2025
25 checks passed

kirktrue deleted the KAFKA-19297-refactor-away-streams branch June 18, 2025 15:24

	log.info("The pause flag in partitions [{}] will be removed due to revocation.", Utils.topicPartitionString(revokePausedPartitions));
	log.info("The pause flag in partitions {} will be removed due to revocation.", revokePausedPartitions);

KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections #19917

KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections #19917

Uh oh!

Conversation

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

Uh oh!

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kirktrue Jun 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

kirktrue commented Jun 11, 2025

Uh oh!

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

lianetm Jun 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

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!

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

kirktrue commented Jun 6, 2025 •

edited by github-actions bot

Loading

kirktrue Jun 9, 2025 •

edited

Loading

lianetm Jun 11, 2025 •

edited

Loading