You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "msn-tldr (via GitHub)" <gi...@apache.org> on 2023/10/17 17:32:38 UTC
[PR] Kip951 client changes [kafka]
msn-tldr opened a new pull request, #14564:
URL: https://github.com/apache/kafka/pull/14564
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1370531714
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -171,6 +179,15 @@ protected void handleFetchResponse(final Node fetchTarget,
log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
fetchConfig.isolationLevel, fetchOffset, partition, partitionData);
+ Errors partitionError = Errors.forCode(partitionData.errorCode());
+ if (requestVersion >= 16 && (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH)) {
Review Comment:
It wasn't needed, so removed(from the Sender code path too). As new leader is coming through tagged fields, for version < 16, tagged fields would be initialised with default values that would be ignored.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1378662641
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -198,6 +215,20 @@ protected void handleFetchSuccess(final Node fetchTarget,
fetchBuffer.add(completedFetch);
}
+ if (!partitionsWithUpdatedLeaderInfo.isEmpty()) {
Review Comment:
`partitionsWithUpdatedLeaderInfo` are collected above to update metadata once with all updates via `metadata.updatePartitionLeadership`.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1370368421
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
this was introduced here
Net effect of the bug was that in the merged cache, the IDs of retained topics(from pre-existing metadata) would be lost in the newly built cache(via merging).
As the current comment explains the intention of the code is to get merged list of topic-ids.
> We start with the previous ID stored for retained topics and then update with newest information from the MetadataResponse.
This should be done by initialising the merged list of topic-ids with retained topic(`this.topicIds`). And then updating with newest information(`topicIds`). But the code uses `topicIds` even to get retained topic-ids. The bug got introduced inadvertently [here](https://github.com/apache/kafka/pull/11004/files#diff-5e4d9713b6e5a386e5cac7e81215160948859243a71b439c3a990876e56b3ec5).
--
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] Kip951 client changes [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1362521496
##########
clients/src/test/java/org/apache/kafka/clients/MetadataCacheTest.java:
##########
@@ -83,4 +84,69 @@ public void testMissingLeaderEndpoint() {
assertEquals(nodesById.get(7), replicas.get(7));
}
+ @Test
+ public void testMergeWithThatPreExistingPartitionIsRetainedPostMerge() {
Review Comment:
test for bug fixed in `MetadataCache`
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
this was a bug, now has a test in `MetadataCachTest.java`
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1371479753
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -614,13 +618,28 @@ private void handleProduceResponse(ClientResponse response, Map<TopicPartition,
.collect(Collectors.toList()),
p.errorMessage());
ProducerBatch batch = batches.get(tp);
- completeBatch(batch, partResp, correlationId, now);
+ completeBatch(batch, partResp, correlationId, now, partitionsWithUpdatedLeaderInfo);
+
+ final short requestVersion = response.requestHeader().apiVersion();
+ if (requestVersion >= 16 && !partitionsWithUpdatedLeaderInfo.isEmpty()) {
+ List<Node> leaderNodes = produceResponse.data().nodeEndpoints().stream()
+ .map(e -> new Node(e.nodeId(), e.host(), e.port(), e.rack()))
+ .filter(e -> !e.equals(Node.noNode()))
+ .collect(
+ Collectors.toList());
+ Set<TopicPartition> updatedPartitions = metadata.updatePartially(partitionsWithUpdatedLeaderInfo, leaderNodes);
+ if (log.isTraceEnabled()) {
+ updatedPartitions.forEach(
+ part -> log.trace("For {} leader was updated.", part)
+ );
+ }
+ }
Review Comment:
No, as there are subtle differences as in different response types & Fetcher has extra actions after leader is updated. So the common part is really is just 1-2 statements, so not worth it IMO.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1371441595
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -685,6 +711,13 @@ private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionRespons
"to request metadata update now", batch.topicPartition,
error.exception(response.errorMessage).toString());
}
+ if (error.exception() instanceof NotLeaderOrFollowerException || error.exception() instanceof FencedLeaderEpochException) {
+ log.debug("For {}, received error {}, with leaderIdAndEpoch {}", batch.topicPartition, error, response.currentLeader);
+ if (partitionsWithUpdatedLeaderInfo != null && response.currentLeader.leaderId() != -1 && response.currentLeader.leaderEpoch() != -1) {
Review Comment:
This line could be more legible with some parentheses and splitting the condition across multiple lines.
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
I would rename the argument currently called `topicIds` too, just to try to prevent a similar bug being re-introduced by accident.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1378153127
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,91 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Updates the partition-leadership info in the metadata. Update is done by merging existing metadata with the input leader information and nodes.
+ * This is called whenever partition-leadership updates are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartitionLeadership(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ for (Entry<TopicPartition, Metadata.LeaderIdAndEpoch> partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.debug("For {}, incoming leader information is incomplete {}", partition, newLeader);
Review Comment:
Should we only log these if debug is enabled? Or should we move them to INFO
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -807,12 +807,18 @@ public boolean hasUndrained() {
private boolean shouldBackoff(boolean hasLeaderChanged, final ProducerBatch batch, final long waitedTimeMs) {
boolean shouldWaitMore = batch.attempts() > 0 && waitedTimeMs < retryBackoff.backoff(batch.attempts() - 1);
boolean shouldBackoff = !hasLeaderChanged && shouldWaitMore;
- if (shouldBackoff) {
- log.trace(
- "For {}, will backoff", batch);
- } else {
- log.trace(
- "For {}, will not backoff, shouldWaitMore {}, hasLeaderChanged {}", batch, shouldWaitMore, hasLeaderChanged);
+ if (log.isTraceEnabled()) {
Review Comment:
Love seeing logging improvements! 🙏
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -807,12 +807,18 @@ public boolean hasUndrained() {
private boolean shouldBackoff(boolean hasLeaderChanged, final ProducerBatch batch, final long waitedTimeMs) {
boolean shouldWaitMore = batch.attempts() > 0 && waitedTimeMs < retryBackoff.backoff(batch.attempts() - 1);
boolean shouldBackoff = !hasLeaderChanged && shouldWaitMore;
- if (shouldBackoff) {
- log.trace(
- "For {}, will backoff", batch);
- } else {
- log.trace(
- "For {}, will not backoff, shouldWaitMore {}, hasLeaderChanged {}", batch, shouldWaitMore, hasLeaderChanged);
+ if (log.isTraceEnabled()) {
Review Comment:
Love seeing logging improvements! 🙏
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -198,6 +215,20 @@ protected void handleFetchSuccess(final Node fetchTarget,
fetchBuffer.add(completedFetch);
}
+ if (!partitionsWithUpdatedLeaderInfo.isEmpty()) {
Review Comment:
I'm a bit confused why we collect the `partitionsWithUpdatedLeaderInfo` above when it looks like all we do with them is validate them to the subscriptions later. Is there any other use for having it out of the loop?
--
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] Kip951 client changes [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1362664111
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
Review Comment:
Can't you use a type argument here to avoid the casts?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -582,25 +584,27 @@ private void handleProduceResponse(ClientResponse response, Map<TopicPartition,
requestHeader, response.destination());
for (ProducerBatch batch : batches.values())
completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.REQUEST_TIMED_OUT, String.format("Disconnected from node %s due to timeout", response.destination())),
- correlationId, now);
+ correlationId, now, new HashMap<>());
Review Comment:
It does seem wasteful constructing an unnecessary new HashMap in all these cases.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -183,6 +200,24 @@ protected void handleFetchResponse(final Node fetchTarget,
fetchBuffer.add(completedFetch);
}
+ if (requestVersion >= 16 && !partitionsWithUpdatedLeaderInfo.isEmpty()) {
+ List<Node> leaderNodes = response.data().nodeEndpoints().stream()
+ .map(e -> new Node(e.nodeId(), e.host(), e.port(), e.rack()))
+ .filter(e -> !e.equals(Node.noNode()))
+ .collect(
+ Collectors.toList());
+ Set<TopicPartition> updatedPartitions = metadata.updatePartially(partitionsWithUpdatedLeaderInfo, leaderNodes);
+ updatedPartitions.forEach(
+ tp -> {
+ log.trace("For {}, the leader was updated, so clear preferred replica & validate position.");
+ // Called via FetchCollector.handleInitializeErrors -> requestMetadataUpdate
+ // TBD - assert in test maybe, no needed otherwise.
Review Comment:
This seems worth resolving in this PR.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -685,6 +705,13 @@ private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionRespons
"to request metadata update now", batch.topicPartition,
error.exception(response.errorMessage).toString());
}
+ if (error == Errors.NOT_LEADER_OR_FOLLOWER || error == Errors.FENCED_LEADER_EPOCH) {
Review Comment:
It's strange that there's a mixture of checking errors with `instanceof` and `Errors.ARBITRARY_KAFKA_ERROR`. Perhaps this line ought to use `instanceof` too.
--
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] Kip951 client changes [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1364438419
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
😱 What was the issue?
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
😱 What was the issue?
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr closed pull request #14564: KAFKA-15627: KIP-951's Leader discovery optimisations on the client
URL: https://github.com/apache/kafka/pull/14564
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on PR #14564:
URL: https://github.com/apache/kafka/pull/14564#issuecomment-1788826662
Rebasing my fork seems to have removed the PR commits from the remote branch `kip951_client_changes`, that has *automatically* closed this PR 😞
Something to be careful of in future, this github is also discussed in https://stackoverflow.com/questions/46053530/pull-request-closed-unexpectedly.
@wcarlson5 opening a new PR, don't see a way around it 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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1370531714
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -171,6 +179,15 @@ protected void handleFetchResponse(final Node fetchTarget,
log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
fetchConfig.isolationLevel, fetchOffset, partition, partitionData);
+ Errors partitionError = Errors.forCode(partitionData.errorCode());
+ if (requestVersion >= 16 && (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH)) {
Review Comment:
It wasn't needed, so removed. As new leader is coming through tagged fields, for version < 16, tagged fields would be initialised with default values that would be ignored.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1370368421
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
this was introduced here
https://github.com/apache/kafka/pull/11004/files#diff-5e4d9713b6e5a386e5cac7e81215160948859243a71b439c3a990876e56b3ec5
Net effect of the bug was that in the merged cache, the IDs of retained topics(from pre-existing metadata) would be lost in the newly built cache(via merging).
As the comment explains the intention of the code is to get merged list of topic-ids. This is done by initialising the merged list of topic-ids with retained topic(`this.topicIds`). And then updating with newest information(`this.topicIds`). But the code used to use `topicIds` even to get retained topic-ids.
> We start with the previous ID stored for retained topics and then update with newest information from the MetadataResponse.
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
Review Comment:
You are right, i see everywhere else is at `DEBUG` level so i have changed the logging level.
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
Review Comment:
> if it makes sense to throw these into a separate method shouldUpdatePartitionLeader(. . .)?
Right now conditions to not update are fairly contained, so i will skip adding the method now.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -171,6 +179,15 @@ protected void handleFetchResponse(final Node fetchTarget,
log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
fetchConfig.isolationLevel, fetchOffset, partition, partitionData);
+ Errors partitionError = Errors.forCode(partitionData.errorCode());
+ if (requestVersion >= 16 && (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH)) {
Review Comment:
It wasn't needed. As new leader is coming through tagged fields, for version < 16, tagged fields would be initialised with default values that would be ignored.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -685,6 +705,13 @@ private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionRespons
"to request metadata update now", batch.topicPartition,
error.exception(response.errorMessage).toString());
}
+ if (error == Errors.NOT_LEADER_OR_FOLLOWER || error == Errors.FENCED_LEADER_EPOCH) {
Review Comment:
Either is OK, just different styles. But get ur point, sticking to 1 style makes sense, so going with current one.
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
+
+ MetadataResponse.PartitionMetadata existingMetadata = this.cache.partitionMetadata(partition).get();
Review Comment:
Good call-out, but no, due to check at line 392 making sure `optional.isPresent()`
```
if (!this.cache.partitionMetadata(partition).isPresent()) {
log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
continue;
}
```
--
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] Kip951 client changes [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1364448430
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
Review Comment:
Ditto on the use of unnecessary casting.
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
Review Comment:
Off topic, but I'm not sure I see the utility in using `TRACE`-level debugging. I just don't think it'll ever be enabled in production.
Please correct me if there's a use case where it could be helpful.
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
+
+ MetadataResponse.PartitionMetadata existingMetadata = this.cache.partitionMetadata(partition).get();
+ MetadataResponse.PartitionMetadata updatedMetadata = new MetadataResponse.PartitionMetadata(
+ existingMetadata.error,
+ partition,
+ newLeader.leaderId,
+ newLeader.epoch,
+ existingMetadata.replicaIds,
+ existingMetadata.inSyncReplicaIds,
+ existingMetadata.offlineReplicaIds
+ );
+ updatePartitionMetadata.add(updatedMetadata);
+
+ lastSeenLeaderEpochs.put(partition, newLeader.epoch.get());
+
+ updatedPartitions.add(partition);
+ }
+
+ if (updatedPartitions.isEmpty()) {
+ log.debug("No relevant metadata updates.");
+ return updatedPartitions;
+ }
+
+ // Get topic-ids for filtered partitions from existing topic-ids.
+ Map<String, Uuid> existingTopicIds = this.cache.topicIds();
+ Map<String, Uuid> filteredTopicIds = updatePartitionMetadata.stream()
+ .filter(e -> existingTopicIds.containsKey(e.topic()))
+ .collect(Collectors.toMap(e -> e.topic(), e -> existingTopicIds.get(e.topic())));
+
+ Set<String> updatedTopics = updatePartitionMetadata.stream().map(MetadataResponse.PartitionMetadata::topic).collect(Collectors.toSet());
+
+ if (log.isTraceEnabled()) {
+ updatePartitionMetadata.forEach(
+ partMetadata -> log.trace("For {} updating to leader-id {}, leader-epoch {}.", partMetadata, partMetadata.leaderId.get(), partMetadata.leaderEpoch.get())
+ );
+ }
+
+ this.cache = cache.mergeWith(cache.clusterResource().clusterId(), newNodes,
+ updatePartitionMetadata, Collections.emptySet(), Collections.emptySet(),
+ Collections.emptySet(), cache.cluster().controller(), filteredTopicIds,
+ (topic, isInternal) -> !updatedTopics.contains(topic));
+ clusterResourceListeners.onUpdate(cache.clusterResource());
+
+ return updatedPartitions;
Review Comment:
Can we just pull the set of partitions out like so:
```suggestion
return updatePartitionMetadata.stream()
.map(MetadataResponse.PartitionMetadata::topicPartition)
.collect(Collectors.toSet());
```
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
+
+ MetadataResponse.PartitionMetadata existingMetadata = this.cache.partitionMetadata(partition).get();
+ MetadataResponse.PartitionMetadata updatedMetadata = new MetadataResponse.PartitionMetadata(
+ existingMetadata.error,
+ partition,
+ newLeader.leaderId,
+ newLeader.epoch,
+ existingMetadata.replicaIds,
+ existingMetadata.inSyncReplicaIds,
+ existingMetadata.offlineReplicaIds
+ );
+ updatePartitionMetadata.add(updatedMetadata);
+
+ lastSeenLeaderEpochs.put(partition, newLeader.epoch.get());
+
+ updatedPartitions.add(partition);
+ }
+
+ if (updatedPartitions.isEmpty()) {
+ log.debug("No relevant metadata updates.");
+ return updatedPartitions;
+ }
+
+ // Get topic-ids for filtered partitions from existing topic-ids.
+ Map<String, Uuid> existingTopicIds = this.cache.topicIds();
+ Map<String, Uuid> filteredTopicIds = updatePartitionMetadata.stream()
+ .filter(e -> existingTopicIds.containsKey(e.topic()))
+ .collect(Collectors.toMap(e -> e.topic(), e -> existingTopicIds.get(e.topic())));
+
+ Set<String> updatedTopics = updatePartitionMetadata.stream().map(MetadataResponse.PartitionMetadata::topic).collect(Collectors.toSet());
+
+ if (log.isTraceEnabled()) {
+ updatePartitionMetadata.forEach(
+ partMetadata -> log.trace("For {} updating to leader-id {}, leader-epoch {}.", partMetadata, partMetadata.leaderId.get(), partMetadata.leaderEpoch.get())
+ );
+ }
+
+ this.cache = cache.mergeWith(cache.clusterResource().clusterId(), newNodes,
+ updatePartitionMetadata, Collections.emptySet(), Collections.emptySet(),
+ Collections.emptySet(), cache.cluster().controller(), filteredTopicIds,
+ (topic, isInternal) -> !updatedTopics.contains(topic));
Review Comment:
This method call is very important, but it is quite hard to read. Can we prettify it a bit? Perhaps...
```suggestion
this.cache = cache.mergeWith(
cache.clusterResource().clusterId(),
newNodes,
updatePartitionMetadata,
Collections.emptySet(),
Collections.emptySet(),
Collections.emptySet(),
cache.cluster().controller(),
filteredTopicIds,
(topic, isInternal) -> !updatedTopics.contains(topic)
);
```
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
Review Comment:
It's a shame that use of `Optional` is so noisy, what with its `isPresent()`s and `get()`s everywhere 😦
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
Review Comment:
It's a shame that use of `Optional` is so noisy, what with its `isPresent()`s and `get()`s everywhere 😦
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
+
+ MetadataResponse.PartitionMetadata existingMetadata = this.cache.partitionMetadata(partition).get();
+ MetadataResponse.PartitionMetadata updatedMetadata = new MetadataResponse.PartitionMetadata(
+ existingMetadata.error,
+ partition,
+ newLeader.leaderId,
+ newLeader.epoch,
+ existingMetadata.replicaIds,
+ existingMetadata.inSyncReplicaIds,
+ existingMetadata.offlineReplicaIds
+ );
+ updatePartitionMetadata.add(updatedMetadata);
+
+ lastSeenLeaderEpochs.put(partition, newLeader.epoch.get());
+
+ updatedPartitions.add(partition);
+ }
+
+ if (updatedPartitions.isEmpty()) {
Review Comment:
`updatedPartitions` is updated alongside `updatePartitionMetadata`, but I'm wondering if we can use the size of `updatePartitionMetadata` to make this short circuit check?
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
Review Comment:
All of these checks are to handle cases where the incoming leader information is not usable, right? I'm wondering—and this is a stylistic choice—if it makes sense to throw these into a separate method `shouldUpdatePartitionLeader(. . .)`? Might make it more easily accessible for a dedicated unit test, too.
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -706,4 +795,22 @@ public String toString() {
'}';
}
}
+
+ public static class LeaderIdAndEpoch {
+ public final Optional<Integer> leaderId;
+ public final Optional<Integer> epoch;
+
+ public LeaderIdAndEpoch(Optional<Integer> leaderId, Optional<Integer> epoch) {
+ this.leaderId = Objects.requireNonNull(leaderId);
+ this.epoch = Objects.requireNonNull(epoch);
+ }
+
+ @Override
+ public String toString() {
+ return "LeaderIdAndEpoch{" +
+ "leaderId=" + leaderId.map(Number::toString).orElse("absent") +
+ ", epoch=" + epoch.map(Number::toString).orElse("absent") +
Review Comment:
Interesting. I've not seen that usage of `Optional.map()` before. Cool!
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
+
+ MetadataResponse.PartitionMetadata existingMetadata = this.cache.partitionMetadata(partition).get();
Review Comment:
Is there any way that `existingMetadata` could be `null` at this point? I assume not, due to the use of the `Optional` wrapper here.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -183,6 +200,24 @@ protected void handleFetchResponse(final Node fetchTarget,
fetchBuffer.add(completedFetch);
}
+ if (requestVersion >= 16 && !partitionsWithUpdatedLeaderInfo.isEmpty()) {
+ List<Node> leaderNodes = response.data().nodeEndpoints().stream()
+ .map(e -> new Node(e.nodeId(), e.host(), e.port(), e.rack()))
+ .filter(e -> !e.equals(Node.noNode()))
+ .collect(
+ Collectors.toList());
Review Comment:
Minor nit:
```suggestion
List<Node> leaderNodes = response.data().nodeEndpoints().stream()
.map(e -> new Node(e.nodeId(), e.host(), e.port(), e.rack()))
.filter(e -> !e.equals(Node.noNode()))
.collect(Collectors.toList());
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -171,6 +179,15 @@ protected void handleFetchResponse(final Node fetchTarget,
log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
fetchConfig.isolationLevel, fetchOffset, partition, partitionData);
+ Errors partitionError = Errors.forCode(partitionData.errorCode());
+ if (requestVersion >= 16 && (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH)) {
Review Comment:
Can we add a constant, with some comments, for the magic value of `16`?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -614,13 +618,28 @@ private void handleProduceResponse(ClientResponse response, Map<TopicPartition,
.collect(Collectors.toList()),
p.errorMessage());
ProducerBatch batch = batches.get(tp);
- completeBatch(batch, partResp, correlationId, now);
+ completeBatch(batch, partResp, correlationId, now, partitionsWithUpdatedLeaderInfo);
+
+ final short requestVersion = response.requestHeader().apiVersion();
+ if (requestVersion >= 16 && !partitionsWithUpdatedLeaderInfo.isEmpty()) {
+ List<Node> leaderNodes = produceResponse.data().nodeEndpoints().stream()
+ .map(e -> new Node(e.nodeId(), e.host(), e.port(), e.rack()))
+ .filter(e -> !e.equals(Node.noNode()))
+ .collect(
+ Collectors.toList());
+ Set<TopicPartition> updatedPartitions = metadata.updatePartially(partitionsWithUpdatedLeaderInfo, leaderNodes);
+ if (log.isTraceEnabled()) {
+ updatedPartitions.forEach(
+ part -> log.trace("For {} leader was updated.", part)
+ );
+ }
+ }
Review Comment:
Is this worth the effort to generalize and share among `Sender` and `AbstractFetch`?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##########
@@ -633,9 +652,10 @@ private void handleProduceResponse(ClientResponse response, Map<TopicPartition,
* @param response The produce response
* @param correlationId The correlation id for the request
* @param now The current POSIX timestamp in milliseconds
+ * @param partitionsWithUpdatedLeaderInfo This will be populated with partitions that have updated leader info.
*/
private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response, long correlationId,
- long now) {
+ long now, Map<TopicPartition, Metadata.LeaderIdAndEpoch> partitionsWithUpdatedLeaderInfo) {
Review Comment:
Can we wrap this `Map` in an `Optional` or explicitly allow for `null` since four out of the five callers aren't interested in the updates to the map?
##########
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##########
@@ -350,6 +353,92 @@ public synchronized void update(int requestVersion, MetadataResponse response, b
log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
+ /**
+ * Update the metadata by merging existing metadata with the input leader information and nodes. This is called whenever
+ * partial updates to metadata are returned in a response from broker(ex - ProduceResponse & FetchResponse).
+ * Note that the updates via Metadata RPC are handled separately in ({@link #update}).
+ * Both partitionLeader and leaderNodes override the existing metadata. Non-overlapping metadata is kept as it is.
+ * @param partitionLeaders map of new leadership information for partitions.
+ * @param leaderNodes a list of nodes for leaders in the above map.
+ * @return a set of partitions, for which leaders were updated.
+ */
+ public synchronized Set<TopicPartition> updatePartially(Map<TopicPartition, LeaderIdAndEpoch> partitionLeaders, List<Node> leaderNodes) {
+ Map<Integer, Node> newNodes = leaderNodes.stream().collect(Collectors.toMap(Node::id, node -> node));
+ // Insert non-overlapping nodes from existing-nodes into new-nodes.
+ this.cache.cluster().nodes().stream().forEach(node -> newNodes.putIfAbsent(node.id(), node));
+
+ // Create partition-metadata for all updated partitions. Exclude updates for partitions -
+ // 1. for which the corresponding partition has newer leader in existing metadata.
+ // 2. for which corresponding leader's node is missing in the new-nodes.
+ // 3. for which the existing metadata doesn't know about the partition.
+ List<PartitionMetadata> updatePartitionMetadata = new ArrayList<>();
+ Set<TopicPartition> updatedPartitions = new HashSet<>();
+ for (Entry partitionLeader: partitionLeaders.entrySet()) {
+ TopicPartition partition = (TopicPartition) partitionLeader.getKey();
+ Metadata.LeaderAndEpoch currentLeader = currentLeader(partition);
+ Metadata.LeaderIdAndEpoch newLeader = (LeaderIdAndEpoch) partitionLeader.getValue();
+ if (!newLeader.epoch.isPresent() || !newLeader.leaderId.isPresent()) {
+ log.trace("For {}, incoming leader information is incomplete {}", partition, newLeader);
+ continue;
+ }
+ if (currentLeader.epoch.isPresent() && newLeader.epoch.get() <= currentLeader.epoch.get()) {
+ log.trace("For {}, incoming leader({}) is not-newer than the one in the existing metadata {}, so ignoring.", partition, newLeader, currentLeader);
+ continue;
+ }
+ if (!newNodes.containsKey(newLeader.leaderId.get())) {
+ log.trace("For {}, incoming leader({}), the corresponding node information for node-id {} is missing, so ignoring.", partition, newLeader, newLeader.leaderId.get());
+ continue;
+ }
+ if (!this.cache.partitionMetadata(partition).isPresent()) {
+ log.trace("For {}, incoming leader({}), no longer has cached metadata so ignoring.", partition, newLeader);
+ continue;
+ }
Review Comment:
All of these checks are to handle cases where the incoming leader information is not usable, right? I'm wondering—and this is a stylistic choice—if it makes sense to throw these into a separate method `shouldUpdatePartitionLeader(. . .)`? Might make it more easily accessible for a dedicated unit test, too.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1370368421
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
this was introduced here
https://github.com/apache/kafka/pull/11004/files#diff-5e4d9713b6e5a386e5cac7e81215160948859243a71b439c3a990876e56b3ec5
Net effect of the bug was that in the merged cache, the IDs of retained topics(from pre-existing metadata) would be lost in the newly built cache(via merging).
As the comment explains the intention of the code is to get merged list of topic-ids. This should be done by initialising the merged list of topic-ids with retained topic(`this.topicIds`). And then updating with newest information(`topicIds`). But the code uses `topicIds` even to get retained topic-ids.
> We start with the previous ID stored for retained topics and then update with newest information from the MetadataResponse.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on PR #14564:
URL: https://github.com/apache/kafka/pull/14564#issuecomment-1779551983
@AndrewJSchofield & @kirktrue addressed the feedback so far!
thanks for the reviews.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1370368421
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
Net effect of the bug was that in the merged cache, the IDs of retained topics(from pre-existing metadata) would be lost in the newly built cache(via merging).
As the current comment explains the intention of the code is to get merged list of topic-ids.
> We start with the previous ID stored for retained topics and then update with newest information from the MetadataResponse.
This should be done by initialising the merged list of topic-ids with retained topic(`this.topicIds`). And then updating with newest information(`topicIds`). But the code uses `topicIds` even to get retained topic-ids. The bug got introduced inadvertently [here](https://github.com/apache/kafka/pull/11004/files#diff-5e4d9713b6e5a386e5cac7e81215160948859243a71b439c3a990876e56b3ec5), due to same-named variables at object Vs method scope. Now it has a test to catch regressions in behaviour.
--
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-15627: KIP-951's Leader discovery optimisations on the client [kafka]
Posted by "msn-tldr (via GitHub)" <gi...@apache.org>.
msn-tldr commented on code in PR #14564:
URL: https://github.com/apache/kafka/pull/14564#discussion_r1370368421
##########
clients/src/main/java/org/apache/kafka/clients/MetadataCache.java:
##########
@@ -150,7 +150,7 @@ MetadataCache mergeWith(String newClusterId,
// 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 = topicIds.entrySet().stream()
+ Map<String, Uuid> newTopicIds = this.topicIds.entrySet().stream()
Review Comment:
Net effect of the bug was that in the merged cache, the IDs of retained topics(from pre-existing metadata) would be lost in the newly built cache(via merging).
As the current comment explains the intention of the code is to get merged list of topic-ids.
> We start with the previous ID stored for retained topics and then update with newest information from the MetadataResponse.
This should be done by initialising the merged list of topic-ids with retained topic(`this.topicIds`). And then updating with newest information(`topicIds`). But the code uses `topicIds` even to get retained topic-ids. The bug got introduced inadvertently [here](https://github.com/apache/kafka/pull/11004/files#diff-5e4d9713b6e5a386e5cac7e81215160948859243a71b439c3a990876e56b3ec5). Now it has a test to catch regressions in behaviour.
--
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