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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
