kirktrue commented on code in PR #15585:
URL: https://github.com/apache/kafka/pull/15585#discussion_r1539435022
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -550,6 +550,11 @@ public ConsumerGroupHeartbeatRequestData
buildRequestData() {
} else {
// SubscribedTopicRegex - only sent if has changed since the
last heartbeat
// - not supported yet
Review Comment:
This comment would go away, if we do end up keeping a separate `else` block.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java:
##########
@@ -550,6 +550,11 @@ public ConsumerGroupHeartbeatRequestData
buildRequestData() {
} else {
// SubscribedTopicRegex - only sent if has changed since the
last heartbeat
// - not supported yet
+ TreeSet<String> subscribedTopicNames = new
TreeSet<>(this.subscriptions.subscription());
+ if (sendAllFields ||
!subscribedTopicNames.equals(sentFields.subscribedTopicNames)) {
+ data.setSubscribedTopicNames(new
ArrayList<>(this.subscriptions.subscription()));
+ sentFields.subscribedTopicNames = subscribedTopicNames;
+ }
Review Comment:
Thanks for catching this, @lianetm. I agree that if the logic is identical,
there's no need for duplication. Any speculation as to the reason the `if`
statement existed in the first place?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1667,6 +1667,9 @@ private void updateLastSeenEpochIfNewer(TopicPartition
topicPartition, OffsetAnd
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
maybeInvokeCommitCallbacks();
+ if (subscriptions.hasPatternSubscription()) {
+ updatePatternSubscription(metadata.fetch());
+ }
Review Comment:
Good call, @lianetm.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1750,8 +1753,14 @@ private void subscribeInternal(Pattern pattern,
Optional<ConsumerRebalanceListen
throwIfNoAssignorsConfigured();
log.info("Subscribed to pattern: '{}'", pattern);
subscriptions.subscribe(pattern, listener);
- updatePatternSubscription(metadata.fetch());
metadata.requestUpdateForNewTopics();
+ Cluster cache = metadata.fetch();
+
+ while (cache == metadata.fetch()) {
+ log.info("Waiting for new metadata update");
+ }
Review Comment:
@Phuc-Hong-Tran—why was this `while` loop added? Did you see issues in your
testing, or were you preemptively wanting to make the code more robust?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1750,8 +1753,14 @@ private void subscribeInternal(Pattern pattern,
Optional<ConsumerRebalanceListen
throwIfNoAssignorsConfigured();
log.info("Subscribed to pattern: '{}'", pattern);
subscriptions.subscribe(pattern, listener);
- updatePatternSubscription(metadata.fetch());
metadata.requestUpdateForNewTopics();
+ Cluster cache = metadata.fetch();
+
+ while (cache == metadata.fetch()) {
+ log.info("Waiting for new metadata update");
+ }
Review Comment:
I'm wondering why we call `metadata.requestUpdateForNewTopics` here when
that's already in `updatePatternSubscription`? 🤔
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
