CalvinConfluent commented on code in PR #14612:
URL: https://github.com/apache/kafka/pull/14612#discussion_r1451042152
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -155,72 +153,78 @@ class KRaftMetadataCache(val brokerId: Int) extends
MetadataCache with Logging w
* @param upperIndex The upper limit of the index of the
partitions to be included in the result.
* Note that, the upper index can be
larger than the largest partition index in
* this topic.
- * @return A collection of topic partition
metadata and whether there are more partitions.
+ * @return A collection of topic partition
metadata and next partition index (-1 means
+ * no next partition).
*/
private def getPartitionMetadataForDescribeTopicResponse(
image: MetadataImage,
topicName: String,
listenerName: ListenerName,
startIndex: Int,
- upperIndex: Int
- ): (Option[List[DescribeTopicPartitionsResponsePartition]], Boolean) = {
+ maxCount: Int
+ ): (Option[List[DescribeTopicPartitionsResponsePartition]], Int) = {
Option(image.topics().getTopic(topicName)) match {
- case None => (None, false)
+ case None => (None, -1)
case Some(topic) => {
val result = new ListBuffer[DescribeTopicPartitionsResponsePartition]()
- val endIndex = upperIndex.min(topic.partitions().size())
- for (partitionId <- startIndex until endIndex) {
- val partition = topic.partitions().get(partitionId)
- val filteredReplicas = maybeFilterAliveReplicas(image,
partition.replicas,
- listenerName, false)
- val filteredIsr = maybeFilterAliveReplicas(image, partition.isr,
listenerName,
- false)
- val offlineReplicas = getOfflineReplicas(image, partition,
listenerName)
- val maybeLeader = getAliveEndpoint(image, partition.leader,
listenerName)
- maybeLeader match {
- case None =>
- val error = if
(!image.cluster().brokers.containsKey(partition.leader)) {
- debug(s"Error while fetching metadata for
$topicName-$partitionId: leader not available")
- Errors.LEADER_NOT_AVAILABLE
- } else {
- debug(s"Error while fetching metadata for
$topicName-$partitionId: listener $listenerName " +
- s"not found on leader ${partition.leader}")
- Errors.LISTENER_NOT_FOUND
- }
- result.addOne(new DescribeTopicPartitionsResponsePartition()
- .setErrorCode(error.code)
- .setPartitionIndex(partitionId)
- .setLeaderId(MetadataResponse.NO_LEADER_ID)
- .setLeaderEpoch(partition.leaderEpoch)
- .setReplicaNodes(filteredReplicas)
- .setIsrNodes(filteredIsr)
- .setOfflineReplicas(offlineReplicas))
- case Some(leader) =>
- val error = if (filteredReplicas.size <
partition.replicas.length) {
- debug(s"Error while fetching metadata for
$topicName-$partitionId: replica information not available for " +
- s"following brokers
${partition.replicas.filterNot(filteredReplicas.contains).mkString(",")}")
- Errors.REPLICA_NOT_AVAILABLE
- } else if (filteredIsr.size < partition.isr.length) {
- debug(s"Error while fetching metadata for
$topicName-$partitionId: in sync replica information not available for " +
- s"following brokers
${partition.isr.filterNot(filteredIsr.contains).mkString(",")}")
- Errors.REPLICA_NOT_AVAILABLE
- } else {
- Errors.NONE
- }
-
- result.addOne(new DescribeTopicPartitionsResponsePartition()
- .setErrorCode(error.code)
- .setPartitionIndex(partitionId)
- .setLeaderId(leader.id())
- .setLeaderEpoch(partition.leaderEpoch)
- .setReplicaNodes(filteredReplicas)
- .setIsrNodes(filteredIsr)
- .setOfflineReplicas(offlineReplicas)
- .setEligibleLeaderReplicas(Replicas.toList(partition.elr))
- .setLastKnownElr(Replicas.toList(partition.lastKnownElr)))
+ // The partition id may not be consecutive.
+ val partitions =
topic.partitions().keySet().stream().sorted().iterator()
Review Comment:
I am not sure I get it. The partition IDs can be random like the cases in
UT, I don't have an O(n) with no extra space simple solution off the top of my
head. Maybe running the quick select can do the trick but it is not generically
supported by Java.
Instead, I use a tree set to maintain the top K smallest partitions larger
than the start index. This is better than the original sorting.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
