code: func (b *Broker) cacheMetadata(resp *proto.MetadataResp) {
  	if !b.metadata.created.IsZero() {
  		b.conf.Logger.Debug("rewriting old metadata",
  			"age", time.Now().Sub(b.metadata.created))
  	}
  	b.metadata = clusterMetadata{
  		created:    time.Now(),
  		nodes:      make(map[int32]string),
  		endpoints:  make(map[topicPartition]int32),
  		partitions: make(map[string]int32),
  	}
  	b.metadata.controllerId = resp.ControllerID
  	for _, node := range resp.Brokers {
  		addr := fmt.Sprintf("%s:%d", node.Host, node.Port)
  		b.metadata.nodes[node.NodeID] = addr
  	}
  	for _, topic := range resp.Topics {
  		for _, part := range topic.Partitions {
  			dest := topicPartition{topic.Name, part.ID}
  			b.metadata.endpoints[dest] = part.Leader
  		}
  		b.metadata.partitions[topic.Name] = int32(len(topic.Partitions))
  	}
  	b.conf.Logger.Debug("new metadata cached")
  }
docs: // cacheMetadata creates new internal metadata representation using data from
  // given response. It's call has to be protected with lock.
  //
  // Do not call with partial metadata response, this assumes we have the full
  // set of metadata in the response