code: func (b *Broker) Close() {
  	b.mu.Lock()
  	defer b.mu.Unlock()
  	for nodeID, conn := range b.conns {
  		if err := conn.Close(); err != nil {
  			b.conf.Logger.Info("cannot close node connection",
  				"nodeID", nodeID,
  				"error", err)
  		}
  	}
  }
docs: // Close closes the broker and all active kafka nodes connections.