You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kirktrue (via GitHub)" <gi...@apache.org> on 2025/06/06 19:24:42 UTC
[PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
kirktrue opened a new pull request, #19917:
URL: https://github.com/apache/kafka/pull/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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2150944024
##########
clients/src/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()));
Review Comment:
True, it's just cleaner now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2153170900
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -84,16 +84,21 @@ public TopicMetadataRequestManager(final LogContext context, final Time time, fi
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
// Prune any requests which have timed out
- List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream()
- .filter(TimedRequestState::isExpired)
- .collect(Collectors.toList());
+ List<TopicMetadataRequestState> expiredRequests = new ArrayList<>();
+
+ for (TopicMetadataRequestState requestState : inflightRequests) {
+ if (requestState.isExpired())
+ expiredRequests.add(requestState);
+ }
+
expiredRequests.forEach(TopicMetadataRequestState::expire);
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2150887174
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -470,7 +475,19 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests()
}
}
- return fetchable.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().build()));
+ return convert(fetchable);
+ }
+
+ private Map<Node, FetchSessionHandler.FetchRequestData> convert(Map<Node, FetchSessionHandler.Builder> fetchable) {
+ Map<Node, FetchSessionHandler.FetchRequestData> map = new HashMap<>(fetchable.size());
+
+ for (Map.Entry<Node, FetchSessionHandler.Builder> entry : fetchable.entrySet()) {
+ Node node = entry.getKey();
+ FetchSessionHandler.FetchRequestData fetchRequestData = entry.getValue().build();
+ map.put(node, fetchRequestData);
Review Comment:
This was just to make the code a little clearer about the exact processing and types in use. I'll make it more succinct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2136340952
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -84,16 +84,21 @@ public TopicMetadataRequestManager(final LogContext context, final Time time, fi
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
// Prune any requests which have timed out
- List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream()
- .filter(TimedRequestState::isExpired)
- .collect(Collectors.toList());
+ List<TopicMetadataRequestState> expiredRequests = new ArrayList<>();
+
+ for (TopicMetadataRequestState requestState : inflightRequests) {
+ if (requestState.isExpired())
+ expiredRequests.add(requestState);
+ }
+
expiredRequests.forEach(TopicMetadataRequestState::expire);
Review Comment:
Because the `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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #19917:
URL: https://github.com/apache/kafka/pull/19917
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2136347808
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -84,16 +84,21 @@ public TopicMetadataRequestManager(final LogContext context, final Time time, fi
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
// Prune any requests which have timed out
- List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream()
- .filter(TimedRequestState::isExpired)
- .collect(Collectors.toList());
+ List<TopicMetadataRequestState> expiredRequests = new ArrayList<>();
+
+ for (TopicMetadataRequestState requestState : inflightRequests) {
+ if (requestState.isExpired())
+ expiredRequests.add(requestState);
+ }
+
expiredRequests.forEach(TopicMetadataRequestState::expire);
Review Comment:
Well, I take it back. Apparently it's not an error to remove it directly from the collection while iterating over the collection 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2150942549
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java:
##########
@@ -414,8 +413,8 @@ public CompletableFuture<Void> signalPartitionsRevoked(Set<TopicPartition> parti
private void logPausedPartitionsBeingRevoked(Set<TopicPartition> partitionsToRevoke) {
Set<TopicPartition> revokePausedPartitions = subscriptions.pausedPartitions();
revokePausedPartitions.retainAll(partitionsToRevoke);
- 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));
Review Comment:
You're right. Changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #19917:
URL: https://github.com/apache/kafka/pull/19917#issuecomment-2963869852
@AndrewJSchofield @lianetm—Anything else I need to fix before we can merge? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2135851674
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1135,7 +1134,9 @@ CompletableFuture<Void> revokePartitions(Set<TopicPartition> partitionsToRevoke)
// Ensure the set of partitions to revoke are still assigned
Set<TopicPartition> revokedPartitions = new HashSet<>(partitionsToRevoke);
revokedPartitions.retainAll(subscriptions.assignedPartitions());
- log.info("Revoking previously assigned partitions {}", revokedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+
+ if (log.isInfoEnabled())
+ log.info("Revoking previously assigned partitions {}", Utils.sortedTopicPartitions(revokedPartitions));
Review Comment:
This is actually sorting the list of partitions whereas it previously was not. That seems accidental in a PR that is trying to reduce CPU usage.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Utils.java:
##########
@@ -20,12 +20,41 @@
import org.apache.kafka.common.TopicPartition;
import java.io.Serializable;
+import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
public final class Utils {
+ public static String sortedTopicPartitions(Collection<TopicPartition> partitions) {
+ SortedSet<TopicPartition> sortedPartitions;
+
+ if (partitions instanceof SortedSet) {
+ sortedPartitions = (SortedSet<TopicPartition>) partitions;
+ } else {
+ sortedPartitions = new TreeSet<>(new TopicPartitionComparator());
+ sortedPartitions.addAll(partitions);
+ }
+
+ return join(sortedPartitions);
+ }
+
+ private static String join(SortedSet<?> elements) {
+ StringBuilder b = new StringBuilder();
Review Comment:
If you constructed an implementation of `Iterator` which executed `Object.toString()` on its elements, you could then use `String.join` directly. I wonder what you think about using the Java library to do the string concatenation.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -84,16 +84,21 @@ public TopicMetadataRequestManager(final LogContext context, final Time time, fi
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
// Prune any requests which have timed out
- List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream()
- .filter(TimedRequestState::isExpired)
- .collect(Collectors.toList());
+ List<TopicMetadataRequestState> expiredRequests = new ArrayList<>();
+
+ for (TopicMetadataRequestState requestState : inflightRequests) {
+ if (requestState.isExpired())
+ expiredRequests.add(requestState);
+ }
+
expiredRequests.forEach(TopicMetadataRequestState::expire);
Review Comment:
Couldn't you just not build `expiredRequests` and call `requestState.expire()` at L91?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2141084110
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -470,7 +475,19 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests()
}
}
- return fetchable.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().build()));
+ return convert(fetchable);
+ }
+
+ private Map<Node, FetchSessionHandler.FetchRequestData> convert(Map<Node, FetchSessionHandler.Builder> fetchable) {
+ Map<Node, FetchSessionHandler.FetchRequestData> map = new HashMap<>(fetchable.size());
+
+ for (Map.Entry<Node, FetchSessionHandler.Builder> entry : fetchable.entrySet()) {
+ Node node = entry.getKey();
+ FetchSessionHandler.FetchRequestData fetchRequestData = entry.getValue().build();
+ map.put(node, fetchRequestData);
Review Comment:
```suggestion
map.put(entry.getKey(), entry.getValue().build());
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1135,7 +1134,9 @@ CompletableFuture<Void> revokePartitions(Set<TopicPartition> partitionsToRevoke)
// Ensure the set of partitions to revoke are still assigned
Set<TopicPartition> revokedPartitions = new HashSet<>(partitionsToRevoke);
revokedPartitions.retainAll(subscriptions.assignedPartitions());
- 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));
Review Comment:
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.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java:
##########
@@ -166,9 +161,48 @@ public boolean contains(CompletableEvent<?> event) {
}
public List<CompletableEvent<?>> uncompletedEvents() {
- return tracked.stream()
- .filter(e -> !e.future().isDone())
- .collect(Collectors.toList());
+ List<CompletableEvent<?>> events = new ArrayList<>();
+
+ for (CompletableEvent<?> event : tracked) {
Review Comment:
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)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java:
##########
@@ -77,11 +81,14 @@ public Exception invokePartitionsAssigned(final SortedSet<TopicPartition> assign
}
public Exception invokePartitionsRevoked(final SortedSet<TopicPartition> revokedPartitions) {
- log.info("Revoke previously assigned partitions {}", revokedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+ if (log.isInfoEnabled())
+ log.info("Revoke previously assigned partitions {}", Utils.topicPartitionString(revokedPartitions));
+
Set<TopicPartition> revokePausedPartitions = subscriptions.pausedPartitions();
revokePausedPartitions.retainAll(revokedPartitions);
- 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));
Review Comment:
similar to above, if we are adding [] to the string of partitions, couldn't we just print the set?
##########
clients/src/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()));
Review Comment:
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).
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java:
##########
@@ -166,9 +161,48 @@ public boolean contains(CompletableEvent<?> event) {
}
public List<CompletableEvent<?>> uncompletedEvents() {
- return tracked.stream()
- .filter(e -> !e.future().isDone())
- .collect(Collectors.toList());
+ List<CompletableEvent<?>> events = new ArrayList<>();
+
+ for (CompletableEvent<?> event : tracked) {
+ if (!event.future().isDone())
+ events.add(event);
+ }
+
+ return events;
+ }
+
+ /**
+ * For all the {@link CompletableEvent}s in the collection, if they're not already complete, invoke
+ * {@link CompletableFuture#completeExceptionally(Throwable)}.
+ *
+ * @param events Collection of objects, assumed to be subclasses of {@link ApplicationEvent} or
+ * {@link BackgroundEvent}, but will only perform completion for any
+ * unfinished {@link CompletableEvent}s
+ *
+ * @return Number of events closed
+ */
+ private long completeEventsExceptionallyOnClose(Collection<?> events) {
+ long count = 0;
+
+ for (Object o : events) {
+ if (!(o instanceof CompletableEvent))
+ continue;
+
+ CompletableEvent<?> event = (CompletableEvent<?>) o;
+
+ if (event.future().isDone())
+ continue;
+
+ TimeoutException error = new TimeoutException(String.format("%s could not be completed before the consumer closed", event.getClass().getSimpleName()));
+
+ if (event.future().completeExceptionally(error)) {
+ log.debug("Event {} completed exceptionally since the consumer is closing", event);
+ count++;
Review Comment:
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) https://github.com/apache/kafka/blob/2b2552da1fbcdb687c99a03a81034c8f78e86871/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java#L115
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerMembershipManager.java:
##########
@@ -414,8 +413,8 @@ public CompletableFuture<Void> signalPartitionsRevoked(Set<TopicPartition> parti
private void logPausedPartitionsBeingRevoked(Set<TopicPartition> partitionsToRevoke) {
Set<TopicPartition> revokePausedPartitions = subscriptions.pausedPartitions();
revokePausedPartitions.retainAll(partitionsToRevoke);
- 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));
Review Comment:
Given that er end up printing the string in between [], can't we simply print the set directly?
```suggestion
log.info("The pause flag in partitions {} will be removed due to revocation.", revokePausedPartitions);
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java:
##########
@@ -103,11 +114,14 @@ public Exception invokePartitionsRevoked(final SortedSet<TopicPartition> revoked
}
public Exception invokePartitionsLost(final SortedSet<TopicPartition> lostPartitions) {
- log.info("Lost previously assigned partitions {}", lostPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+ if (log.isInfoEnabled())
+ log.info("Lost previously assigned partitions {}", Utils.topicPartitionString(lostPartitions));
+
Set<TopicPartition> lostPausedPartitions = subscriptions.pausedPartitions();
lostPausedPartitions.retainAll(lostPartitions);
- if (!lostPausedPartitions.isEmpty())
- 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));
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -362,12 +368,18 @@ void cleanup() {
* If there is a metadata error, complete all uncompleted events that require subscription metadata.
*/
private void maybeFailOnMetadataError(List<CompletableEvent<?>> events) {
- List<? extends CompletableApplicationEvent<?>> subscriptionMetadataEvent = events.stream()
- .filter(e -> e instanceof CompletableApplicationEvent<?>)
- .map(e -> (CompletableApplicationEvent<?>) e)
- .filter(CompletableApplicationEvent::requireSubscriptionMetadata)
- .collect(Collectors.toList());
-
+ List<CompletableApplicationEvent<?>> subscriptionMetadataEvent = new ArrayList<>();
+
+ for (CompletableEvent<?> ce : events) {
+ if (!(ce instanceof CompletableApplicationEvent))
+ continue;
+
+ CompletableApplicationEvent<?> cae = (CompletableApplicationEvent<?>) ce;
+
+ if (cae.requireSubscriptionMetadata())
+ subscriptionMetadataEvent.add(cae);
Review Comment:
maybe simplify ?
```suggestion
if (ce instanceof CompletableApplicationEvent && ((CompletableApplicationEvent<?>) ce).requireSubscriptionMetadata()) {
subscriptionMetadataEvent.add((CompletableApplicationEvent<?>) ce);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2136355942
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1135,7 +1134,9 @@ CompletableFuture<Void> revokePartitions(Set<TopicPartition> partitionsToRevoke)
// Ensure the set of partitions to revoke are still assigned
Set<TopicPartition> revokedPartitions = new HashSet<>(partitionsToRevoke);
revokedPartitions.retainAll(subscriptions.assignedPartitions());
- log.info("Revoking previously assigned partitions {}", revokedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+
+ if (log.isInfoEnabled())
+ log.info("Revoking previously assigned partitions {}", Utils.sortedTopicPartitions(revokedPartitions));
Review Comment:
Reverted change to make everything sorted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2150942923
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1135,7 +1134,9 @@ CompletableFuture<Void> revokePartitions(Set<TopicPartition> partitionsToRevoke)
// Ensure the set of partitions to revoke are still assigned
Set<TopicPartition> revokedPartitions = new HashSet<>(partitionsToRevoke);
revokedPartitions.retainAll(subscriptions.assignedPartitions());
- 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));
Review Comment:
Yep. Changed.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -362,12 +368,18 @@ void cleanup() {
* If there is a metadata error, complete all uncompleted events that require subscription metadata.
*/
private void maybeFailOnMetadataError(List<CompletableEvent<?>> events) {
- List<? extends CompletableApplicationEvent<?>> subscriptionMetadataEvent = events.stream()
- .filter(e -> e instanceof CompletableApplicationEvent<?>)
- .map(e -> (CompletableApplicationEvent<?>) e)
- .filter(CompletableApplicationEvent::requireSubscriptionMetadata)
- .collect(Collectors.toList());
-
+ List<CompletableApplicationEvent<?>> subscriptionMetadataEvent = new ArrayList<>();
+
+ for (CompletableEvent<?> ce : events) {
+ if (!(ce instanceof CompletableApplicationEvent))
+ continue;
+
+ CompletableApplicationEvent<?> cae = (CompletableApplicationEvent<?>) ce;
+
+ if (cae.requireSubscriptionMetadata())
+ subscriptionMetadataEvent.add(cae);
Review Comment:
Done.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java:
##########
@@ -77,11 +81,14 @@ public Exception invokePartitionsAssigned(final SortedSet<TopicPartition> assign
}
public Exception invokePartitionsRevoked(final SortedSet<TopicPartition> revokedPartitions) {
- log.info("Revoke previously assigned partitions {}", revokedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+ if (log.isInfoEnabled())
+ log.info("Revoke previously assigned partitions {}", Utils.topicPartitionString(revokedPartitions));
+
Set<TopicPartition> revokePausedPartitions = subscriptions.pausedPartitions();
revokePausedPartitions.retainAll(revokedPartitions);
- 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));
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2150944599
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java:
##########
@@ -166,9 +161,48 @@ public boolean contains(CompletableEvent<?> event) {
}
public List<CompletableEvent<?>> uncompletedEvents() {
- return tracked.stream()
- .filter(e -> !e.future().isDone())
- .collect(Collectors.toList());
+ List<CompletableEvent<?>> events = new ArrayList<>();
+
+ for (CompletableEvent<?> event : tracked) {
Review Comment:
Agreed. I added one here and similar comments in a few other places.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2136213734
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1135,7 +1134,9 @@ CompletableFuture<Void> revokePartitions(Set<TopicPartition> partitionsToRevoke)
// Ensure the set of partitions to revoke are still assigned
Set<TopicPartition> revokedPartitions = new HashSet<>(partitionsToRevoke);
revokedPartitions.retainAll(subscriptions.assignedPartitions());
- log.info("Revoking previously assigned partitions {}", revokedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+
+ if (log.isInfoEnabled())
+ log.info("Revoking previously assigned partitions {}", Utils.sortedTopicPartitions(revokedPartitions));
Review Comment:
agree, and btw, this should be already sorted I expect (all these partition sets are internally `SortedSet` exactly for this, better logs). We can probably change the input param type to `SortedSet<TopicPartition>` in this func
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2136329921
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1135,7 +1134,9 @@ CompletableFuture<Void> revokePartitions(Set<TopicPartition> partitionsToRevoke)
// Ensure the set of partitions to revoke are still assigned
Set<TopicPartition> revokedPartitions = new HashSet<>(partitionsToRevoke);
revokedPartitions.retainAll(subscriptions.assignedPartitions());
- log.info("Revoking previously assigned partitions {}", revokedPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+
+ if (log.isInfoEnabled())
+ log.info("Revoking previously assigned partitions {}", Utils.sortedTopicPartitions(revokedPartitions));
Review Comment:
I will revert the change to make everything sorted, if it wasn't already.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2136357518
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -84,16 +84,21 @@ public TopicMetadataRequestManager(final LogContext context, final Time time, fi
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
// Prune any requests which have timed out
- List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream()
- .filter(TimedRequestState::isExpired)
- .collect(Collectors.toList());
+ List<TopicMetadataRequestState> expiredRequests = new ArrayList<>();
+
+ for (TopicMetadataRequestState requestState : inflightRequests) {
+ if (requestState.isExpired())
+ expiredRequests.add(requestState);
+ }
+
expiredRequests.forEach(TopicMetadataRequestState::expire);
Review Comment:
I've removed the superfluous list buildup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2141104339
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -1135,7 +1134,9 @@ CompletableFuture<Void> revokePartitions(Set<TopicPartition> partitionsToRevoke)
// Ensure the set of partitions to revoke are still assigned
Set<TopicPartition> revokedPartitions = new HashSet<>(partitionsToRevoke);
revokedPartitions.retainAll(subscriptions.assignedPartitions());
- 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));
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2150942035
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -470,7 +475,19 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests()
}
}
- return fetchable.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().build()));
+ return convert(fetchable);
+ }
+
+ private Map<Node, FetchSessionHandler.FetchRequestData> convert(Map<Node, FetchSessionHandler.Builder> fetchable) {
+ Map<Node, FetchSessionHandler.FetchRequestData> map = new HashMap<>(fetchable.size());
+
+ for (Map.Entry<Node, FetchSessionHandler.Builder> entry : fetchable.entrySet()) {
+ Node node = entry.getKey();
+ FetchSessionHandler.FetchRequestData fetchRequestData = entry.getValue().build();
+ map.put(node, fetchRequestData);
Review Comment:
Changed as requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2150949379
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CompletableEventReaper.java:
##########
@@ -166,9 +161,48 @@ public boolean contains(CompletableEvent<?> event) {
}
public List<CompletableEvent<?>> uncompletedEvents() {
- return tracked.stream()
- .filter(e -> !e.future().isDone())
- .collect(Collectors.toList());
+ List<CompletableEvent<?>> events = new ArrayList<>();
+
+ for (CompletableEvent<?> event : tracked) {
+ if (!event.future().isDone())
+ events.add(event);
+ }
+
+ return events;
+ }
+
+ /**
+ * For all the {@link CompletableEvent}s in the collection, if they're not already complete, invoke
+ * {@link CompletableFuture#completeExceptionally(Throwable)}.
+ *
+ * @param events Collection of objects, assumed to be subclasses of {@link ApplicationEvent} or
+ * {@link BackgroundEvent}, but will only perform completion for any
+ * unfinished {@link CompletableEvent}s
+ *
+ * @return Number of events closed
+ */
+ private long completeEventsExceptionallyOnClose(Collection<?> events) {
+ long count = 0;
+
+ for (Object o : events) {
+ if (!(o instanceof CompletableEvent))
+ continue;
+
+ CompletableEvent<?> event = (CompletableEvent<?>) o;
+
+ if (event.future().isDone())
+ continue;
+
+ TimeoutException error = new TimeoutException(String.format("%s could not be completed before the consumer closed", event.getClass().getSimpleName()));
+
+ if (event.future().completeExceptionally(error)) {
+ log.debug("Event {} completed exceptionally since the consumer is closing", event);
+ count++;
Review Comment:
Yes, it's different. I assert it's more accurate, but I changed it to mimic the existing counting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2152538609
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerRebalanceListenerInvoker.java:
##########
@@ -103,11 +114,14 @@ public Exception invokePartitionsRevoked(final SortedSet<TopicPartition> revoked
}
public Exception invokePartitionsLost(final SortedSet<TopicPartition> lostPartitions) {
- log.info("Lost previously assigned partitions {}", lostPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", ")));
+ if (log.isInfoEnabled())
+ log.info("Lost previously assigned partitions {}", Utils.topicPartitionString(lostPartitions));
+
Set<TopicPartition> lostPausedPartitions = subscriptions.pausedPartitions();
lostPausedPartitions.retainAll(lostPartitions);
- if (!lostPausedPartitions.isEmpty())
- 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));
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2136356737
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Utils.java:
##########
@@ -20,12 +20,41 @@
import org.apache.kafka.common.TopicPartition;
import java.io.Serializable;
+import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
public final class Utils {
+ public static String sortedTopicPartitions(Collection<TopicPartition> partitions) {
+ SortedSet<TopicPartition> sortedPartitions;
+
+ if (partitions instanceof SortedSet) {
+ sortedPartitions = (SortedSet<TopicPartition>) partitions;
+ } else {
+ sortedPartitions = new TreeSet<>(new TopicPartitionComparator());
+ sortedPartitions.addAll(partitions);
+ }
+
+ return join(sortedPartitions);
+ }
+
+ private static String join(SortedSet<?> elements) {
+ StringBuilder b = new StringBuilder();
Review Comment:
I made a change to use an `Iterable`/`Iterator` under the covers to leverage `String.join()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2136340952
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -84,16 +84,21 @@ public TopicMetadataRequestManager(final LogContext context, final Time time, fi
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
// Prune any requests which have timed out
- List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream()
- .filter(TimedRequestState::isExpired)
- .collect(Collectors.toList());
+ List<TopicMetadataRequestState> expiredRequests = new ArrayList<>();
+
+ for (TopicMetadataRequestState requestState : inflightRequests) {
+ if (requestState.isExpired())
+ expiredRequests.add(requestState);
+ }
+
expiredRequests.forEach(TopicMetadataRequestState::expire);
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19297: Refactor AsyncKafkaConsumer's use of Java Streams APIs in critical sections [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19917:
URL: https://github.com/apache/kafka/pull/19917#discussion_r2153086039
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:
##########
@@ -84,16 +84,21 @@ public TopicMetadataRequestManager(final LogContext context, final Time time, fi
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
// Prune any requests which have timed out
- List<TopicMetadataRequestState> expiredRequests = inflightRequests.stream()
- .filter(TimedRequestState::isExpired)
- .collect(Collectors.toList());
+ List<TopicMetadataRequestState> expiredRequests = new ArrayList<>();
+
+ for (TopicMetadataRequestState requestState : inflightRequests) {
+ if (requestState.isExpired())
+ expiredRequests.add(requestState);
+ }
+
expiredRequests.forEach(TopicMetadataRequestState::expire);
Review Comment:
> 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org