ClusterAdmin DescribeCluster api return controllerID invalid

[antsbean]

Versions

Please specify real version numbers or git SHAs, not just "Latest" since that changes fairly regularly.

Sarama	Kafka	Go
v1.24	2.0.0 or 2.2.1	1.13

Configuration

What configuration values are you using for Sarama and Kafka?

	cfg := sarama.NewConfig()
	var err error
	cfg.Version, err = sarama.ParseKafkaVersion("2.0.0")

Logs

none

Problem Description

type KafkaUtil struct {
	ca    sarama.ClusterAdmin
	addrs []string
	cfg   *sarama.Config
}

func NewKafkaUtil(addrs []string, cfg *sarama.Config) (*KafkaUtil, error) {
	if !cfg.Version.IsAtLeast(sarama.V2_0_0_0) {
		log.Warn("at least kafka version 2.0")
		return nil, sarama.ErrUnsupportedVersion
	}
	ca, err := sarama.NewClusterAdmin(addrs, cfg)
	if err != nil {
		return nil, err
	}
	kafkaUtil := &KafkaUtil{ca: ca, addrs: addrs, cfg: cfg}

	return kafkaUtil, nil
}



// GetDefaultReplicationFactor get kafka cluster default replicate factor
func (k *KafkaUtil) GetDefaultReplicationFactor() (int16, error) {
	_, controllerID, err := k.ca.DescribeCluster()
	if err != nil {
		return 0, err
	}
	var defaultReplicateFactor int
	configs, err := k.ca.DescribeConfig(sarama.ConfigResource{Type: sarama.ClusterResource,
		Name:        fmt.Sprintf("%d", controllerID),
		ConfigNames: []string{"default.replication.factor"}})
	if err != nil {
		return 0, err
	}
	if defaultReplicateFactor, err = strconv.Atoi(configs[0].Value); err != nil {
		return 0, err
	}
	return int16(defaultReplicateFactor), nil
}

k.ca.DescribeConfig return controllerID always is -1, this will lead to get kafka configs failed by DescribeConfig, but k.getController() get controllerID is ok

func (k *KafkaUtil) getController() (*sarama.Broker, error) {
	client, err := sarama.NewClient(k.addrs, k.cfg)
	if err != nil {
		return nil, err
	}
	//make sure we can retrieve the controller
	controller, err := client.Controller()
	if err != nil {
                client.Close()
		return nil, err
	}
	return controller, nil
}
// GetDefaultReplicationFactor get kafka cluster default replicate factor
func (k *KafkaUtil) GetDefaultReplicationFactor() (int16, error) {
	b, err := k.getController()
	if err != nil {
		return 0, err
	}
	defer b.Close()
	controllerID := b.ID()
	var defaultReplicateFactor int
	configs, err := k.ca.DescribeConfig(sarama.ConfigResource{Type: sarama.ClusterResource,
		Name:        fmt.Sprintf("%d", controllerID),
		ConfigNames: []string{"default.replication.factor"}})
	if err != nil {
		return 0, err
	}
	if defaultReplicateFactor, err = strconv.Atoi(configs[0].Value); err != nil {
		return 0, err
	}
	return int16(defaultReplicateFactor), nil
}

[weeco]

I observed the same. Addtionally the RackID seems to be always empty, even though they are actually set for my Brokers.

[antsbean]

I observed the same. Addtionally the RackID seems to be always empty, even though they are actually set for my Brokers.

I had fixed this problem, please see pr