You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lorcanj (via GitHub)" <gi...@apache.org> on 2025/03/13 13:34:06 UTC
[PR] Kafka-16616: refactor mergeWith and updatePartitionLeadership [kafka]
lorcanj opened a new pull request, #19199:
URL: https://github.com/apache/kafka/pull/19199
Delete this text and replace it with a detailed description of your change. The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are expected
for any behavior change and system tests should be considered for larger
changes.
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "lorcanj (via GitHub)" <gi...@apache.org>.
lorcanj commented on PR #19199:
URL: https://github.com/apache/kafka/pull/19199#issuecomment-2745095132
Hi @kirktrue, I've made the updates as per your comments.
There is a failing test testOutdatedCoordinatorAssignment under JUnit Java 17 (Flaky) that's failed twice. I don't seem to be able to hit that test when running all tests under clients locally. I also get a 'No matching tests found in any candidate test task' message when running that test individually.
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "lorcanj (via GitHub)" <gi...@apache.org>.
lorcanj commented on code in PR #19199:
URL: https://github.com/apache/kafka/pull/19199#discussion_r2006364925
##########
clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java:
##########
@@ -173,12 +173,15 @@ MetadataSnapshot mergeWith(String newClusterId,
Map<TopicPartition, PartitionMetadata> newMetadataByPartition = new HashMap<>(addPartitions.size());
- // We want the most recent topic ID. We start with the previous ID stored for retained topics and then
- // update with newest information from the MetadataResponse. We always take the latest state, removing existing
- // topic IDs if the latest state contains the topic name but not a topic ID.
- Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
- .filter(entry -> shouldRetainTopic.test(entry.getKey()))
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ Map<String, Uuid> newTopicIds = new HashMap<>(this.topicIds.size());
+
+ for (Map.Entry<TopicPartition, PartitionMetadata> entry : metadataByPartition.entrySet()) {
+ if (shouldRetainTopic.test(entry.getKey().topic())) {
+ newMetadataByPartition.put(entry.getKey(), entry.getValue());
Review Comment:
I've reverted this to use putIfAbsent
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #19199: KAFKA-16616: refactor mergeWith and updatePartitionLeadership
URL: https://github.com/apache/kafka/pull/19199
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "lorcanj (via GitHub)" <gi...@apache.org>.
lorcanj commented on code in PR #19199:
URL: https://github.com/apache/kafka/pull/19199#discussion_r2004919701
##########
clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java:
##########
@@ -173,12 +173,15 @@ MetadataSnapshot mergeWith(String newClusterId,
Map<TopicPartition, PartitionMetadata> newMetadataByPartition = new HashMap<>(addPartitions.size());
- // We want the most recent topic ID. We start with the previous ID stored for retained topics and then
- // update with newest information from the MetadataResponse. We always take the latest state, removing existing
- // topic IDs if the latest state contains the topic name but not a topic ID.
- Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
- .filter(entry -> shouldRetainTopic.test(entry.getKey()))
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ Map<String, Uuid> newTopicIds = new HashMap<>(this.topicIds.size());
+
+ for (Map.Entry<TopicPartition, PartitionMetadata> entry : metadataByPartition.entrySet()) {
+ if (shouldRetainTopic.test(entry.getKey().topic())) {
+ newMetadataByPartition.put(entry.getKey(), entry.getValue());
Review Comment:
I think I changed it because previously it was primarily using `putIfAbsent()` to ensure that the newest information from the MetadataResponse wasn't being overwritten, which is now different because of the code re-order where this is now done first.
I guess if there are duplicate entries there isn't any need to put if there is an entry for the topic partition, but I don't have a full grasp of what this map would look like populated with real-world data. I think I tested with both and it didn't seem to affect the unit tests.
Happy to go with either approach.
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19199:
URL: https://github.com/apache/kafka/pull/19199#issuecomment-3153187766
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "lorcanj (via GitHub)" <gi...@apache.org>.
lorcanj commented on code in PR #19199:
URL: https://github.com/apache/kafka/pull/19199#discussion_r2004905258
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -422,21 +423,17 @@ public synchronized Set<TopicPartition> updatePartitionLeadership(Map<TopicParti
updatePartitionMetadata.add(updatedMetadata);
lastSeenLeaderEpochs.put(partition, newLeader.epoch.get());
+
+ final String updatedTopic = updatedMetadata.topic();
+ if (this.metadataSnapshot.topicIds().containsKey(updatedTopic))
+ topicIdsForUpdatedTopics.put(updatedTopic, this.metadataSnapshot.topicIds().get(updatedTopic));
Review Comment:
Makes sense to me. I think a similar approach can be taken in MetadataSnapshot in the lines below
`if (this.topicIds.containsKey(entry.getKey().topic()))`
` newTopicIds.putIfAbsent(entry.getKey().topic(), this.topicIds.get(entry.getKey().topic()));`
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #19199:
URL: https://github.com/apache/kafka/pull/19199#issuecomment-2767702063
@lorcanj—Looking at the [test history of `CooperativeConsumerCoordinatorTest.testOutdatedCoordinatorAssignment()`](https://develocity.apache.org/scans/tests?search.relativeStartTime=P28D&search.rootProjectNames=kafka&search.timeZoneId=America%2FLos_Angeles&tests.container=org.apache.kafka.clients.consumer.internals.CooperativeConsumerCoordinatorTest&tests.test=testOutdatedCoordinatorAssignment()), it appears that the test is a teensy bit flaky. What I noted as unusual in the history was that runs of the test for this PR were marked as "FAILED" vs. "FLAKY" in other cases. So it's definitely something to dig deeper on.
When I see "No matching tests found in any candidate test task" it's because I've made a typo when specifying the test method name.
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "lorcanj (via GitHub)" <gi...@apache.org>.
lorcanj commented on PR #19199:
URL: https://github.com/apache/kafka/pull/19199#issuecomment-2770394927
Hi @kirktrue, I believe the test was failing due a falky test which some people had noticed was causing the test to outright fail too, [KAFKA-15900](https://issues.apache.org/jira/browse/KAFKA-15900)
When developing I was pulling from my fork which wasn't up to date, which meant the fix for this issue ([merged PR)](https://github.com/apache/kafka/pull/18945) wasn't included in my branch.
I also made a mistake with the rebase and then pulled, which is why there are so many unrelevant labels now showing on this PR.
I managed to rectify this by properly rebasing and the ci checks have now fully passed.
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19199:
URL: https://github.com/apache/kafka/pull/19199#discussion_r2002151201
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -422,21 +423,17 @@ public synchronized Set<TopicPartition> updatePartitionLeadership(Map<TopicParti
updatePartitionMetadata.add(updatedMetadata);
lastSeenLeaderEpochs.put(partition, newLeader.epoch.get());
+
+ final String updatedTopic = updatedMetadata.topic();
+ if (this.metadataSnapshot.topicIds().containsKey(updatedTopic))
+ topicIdsForUpdatedTopics.put(updatedTopic, this.metadataSnapshot.topicIds().get(updatedTopic));
Review Comment:
This is bit uglier, but it avoids the second hash map lookup.
```suggestion
Uuid topicUuid = this.metadataSnapshot.topicIds().getOrDefault(updatedTopic, Uuid.ZERO_UUID);
if (!topicUuid.equals(Uuid.ZERO_UUID))
topicIdsForUpdatedTopics.put(updatedTopic, topicUuid);
```
##########
clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java:
##########
@@ -173,12 +173,15 @@ MetadataSnapshot mergeWith(String newClusterId,
Map<TopicPartition, PartitionMetadata> newMetadataByPartition = new HashMap<>(addPartitions.size());
- // We want the most recent topic ID. We start with the previous ID stored for retained topics and then
- // update with newest information from the MetadataResponse. We always take the latest state, removing existing
- // topic IDs if the latest state contains the topic name but not a topic ID.
- Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
- .filter(entry -> shouldRetainTopic.test(entry.getKey()))
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ Map<String, Uuid> newTopicIds = new HashMap<>(this.topicIds.size());
+
+ for (Map.Entry<TopicPartition, PartitionMetadata> entry : metadataByPartition.entrySet()) {
+ if (shouldRetainTopic.test(entry.getKey().topic())) {
+ newMetadataByPartition.put(entry.getKey(), entry.getValue());
Review Comment:
The existing code calls `newMetadataByPartition.putIfAbsent()`. Is that something that should be retained here?
##########
clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java:
##########
@@ -173,12 +173,15 @@ MetadataSnapshot mergeWith(String newClusterId,
Map<TopicPartition, PartitionMetadata> newMetadataByPartition = new HashMap<>(addPartitions.size());
- // We want the most recent topic ID. We start with the previous ID stored for retained topics and then
- // update with newest information from the MetadataResponse. We always take the latest state, removing existing
- // topic IDs if the latest state contains the topic name but not a topic ID.
Review Comment:
Is it possible to reword the comments rather that removing them completely?
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "lorcanj (via GitHub)" <gi...@apache.org>.
lorcanj commented on code in PR #19199:
URL: https://github.com/apache/kafka/pull/19199#discussion_r2004922608
##########
clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java:
##########
@@ -173,12 +173,15 @@ MetadataSnapshot mergeWith(String newClusterId,
Map<TopicPartition, PartitionMetadata> newMetadataByPartition = new HashMap<>(addPartitions.size());
- // We want the most recent topic ID. We start with the previous ID stored for retained topics and then
- // update with newest information from the MetadataResponse. We always take the latest state, removing existing
- // topic IDs if the latest state contains the topic name but not a topic ID.
Review Comment:
I'll put this back in as it's still applicable when looking at this again.
--
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-16616: refactor mergeWith and updatePartitionLeadership [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #19199:
URL: https://github.com/apache/kafka/pull/19199#issuecomment-3040738868
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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