func isErrNoController(err error) bool {
	switch e := err.(type) {
	case *TopicError:
		return e.Err == ErrNotController
	case *TopicPartitionError:
		return e.Err == ErrNotController
	case KError:
		return e == ErrNotController
	}
	return false
}
isErrNoController returns `true` if the given error type unwraps to an
`ErrNotController` response from Kafka
package sarama
import (
	"errors"
	"fmt"
	"math/rand"
	"strconv"
	"sync"
	"time"
)
type ClusterAdmin interface {
	CreateTopic(topic string, detail *TopicDetail, validateOnly bool) error
	ListTopics() (map[string]TopicDetail, error)
	DescribeTopics(topics []string) (metadata []*TopicMetadata, err error)
	DeleteTopic(topic string) error
	CreatePartitions(topic string, count int32, assignment [][]int32, validateOnly bool) error
	AlterPartitionReassignments(topic string, assignment [][]int32) error
	ListPartitionReassignments(topics string, partitions []int32) (topicStatus map[string]map[int32]*PartitionReplicaReassignmentsStatus, err error)
	DeleteRecords(topic string, partitionOffsets map[int32]int64) error
	DescribeConfig(resource ConfigResource) ([]ConfigEntry, error)
	AlterConfig(resourceType ConfigResourceType, name string, entries map[string]*string, validateOnly bool) error
	CreateACL(resource Resource, acl Acl) error
	ListAcls(filter AclFilter) ([]ResourceAcls, error)
	DeleteACL(filter AclFilter, validateOnly bool) ([]MatchingAcl, error)
	ListConsumerGroups() (map[string]string, error)
	DescribeConsumerGroups(groups []string) ([]*GroupDescription, error)
	ListConsumerGroupOffsets(group string, topicPartitions map[string][]int32) (*OffsetFetchResponse, error)
	DeleteConsumerGroup(group string) error
	DescribeCluster() (brokers []*Broker, controllerID int32, err error)
	DescribeLogDirs(brokers []int32) (map[int32][]DescribeLogDirsResponseDirMetadata, error)
	DescribeUserScramCredentials(users []string) ([]*DescribeUserScramCredentialsResult, error)
	DeleteUserScramCredentials(delete []AlterUserScramCredentialsDelete) ([]*AlterUserScramCredentialsResult, error)
	UpsertUserScramCredentials(upsert []AlterUserScramCredentialsUpsert) ([]*AlterUserScramCredentialsResult, error)
	Close() error
}
type clusterAdmin struct {
	client Client
	conf   *Config
}
func NewClusterAdmin(addrs []string, conf *Config) (ClusterAdmin, error) {
	client, err := NewClient(addrs, conf)
	if err != nil {
		return nil, err
	}
	return NewClusterAdminFromClient(client)
}
func NewClusterAdminFromClient(client Client) (ClusterAdmin, error) {
	_, err := client.Controller()
	if err != nil {
		return nil, err
	}
	ca := &clusterAdmin{
		client: client,
		conf:   client.Config(),
	}
	return ca, nil
}
func (ca *clusterAdmin) Close() error {
	return ca.client.Close()
}
func (ca *clusterAdmin) Controller() (*Broker, error) {
	return ca.client.Controller()
}
func (ca *clusterAdmin) refreshController() (*Broker, error) {
	return ca.client.RefreshController()
}