jolshan commented on a change in pull request #10049:
URL: https://github.com/apache/kafka/pull/10049#discussion_r569896329
##########
File path: core/src/main/scala/kafka/server/MetadataCache.scala
##########
@@ -157,102 +178,86 @@ class MetadataCache(brokerId: Int) extends Logging {
*
* @return None if broker is not alive or if the broker does not have a
listener named `listenerName`.
*/
- private def getAliveEndpoint(snapshot: MetadataSnapshot, brokerId: Int,
listenerName: ListenerName): Option[Node] = {
- snapshot.aliveNodes.get(brokerId).flatMap(_.get(listenerName))
+ private def getAliveEndpoint(image: MetadataImage, id: Int, listenerName:
ListenerName): Option[Node] = {
+
image.brokers.aliveBroker(id).flatMap(_.endpoints.get(listenerName.value()))
}
// errorUnavailableEndpoints exists to support v0 MetadataResponses
def getTopicMetadata(topics: Set[String],
listenerName: ListenerName,
errorUnavailableEndpoints: Boolean = false,
errorUnavailableListeners: Boolean = false):
Seq[MetadataResponseTopic] = {
- val snapshot = metadataSnapshot
+ val image = _currentImage
topics.toSeq.flatMap { topic =>
- getPartitionMetadata(snapshot, topic, listenerName,
errorUnavailableEndpoints, errorUnavailableListeners).map { partitionMetadata =>
+ getPartitionMetadata(image, topic, listenerName,
errorUnavailableEndpoints, errorUnavailableListeners).map { partitionMetadata =>
new MetadataResponseTopic()
.setErrorCode(Errors.NONE.code)
.setName(topic)
- .setTopicId(snapshot.topicIds.getOrElse(topic, Uuid.ZERO_UUID))
Review comment:
Is there a place where we are setting the topic ID in the metadata
response?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
