func (client *client) RefreshController() (*Broker, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	client.deregisterController()
	if err := client.refreshMetadata(); err != nil {
		return nil, err
	}
	controller := client.cachedController()
	if controller == nil {
		return nil, ErrControllerNotAvailable
	}
	_ = controller.Open(client.conf)
	return controller, nil
}
RefreshController retrieves the cluster controller from fresh metadata
and stores it in the local cache. Requires Kafka 0.10 or higher.
package sarama
import (
	"math/rand"
	"sort"
	"sync"
	"time"
)
type Client interface {
	Config() *Config
	Controller() (*Broker, error)
	RefreshController() (*Broker, error)
	Brokers() []*Broker
	Broker(brokerID int32) (*Broker, error)
	Topics() ([]string, error)
	Partitions(topic string) ([]int32, error)
	WritablePartitions(topic string) ([]int32, error)
	Leader(topic string, partitionID int32) (*Broker, error)
	Replicas(topic string, partitionID int32) ([]int32, error)
	InSyncReplicas(topic string, partitionID int32) ([]int32, error)
	OfflineReplicas(topic string, partitionID int32) ([]int32, error)
	RefreshBrokers(addrs []string) error
	RefreshMetadata(topics ...string) error
	GetOffset(topic string, partitionID int32, time int64) (int64, error)
	Coordinator(consumerGroup string) (*Broker, error)
	RefreshCoordinator(consumerGroup string) error
	InitProducerID() (*InitProducerIDResponse, error)
	Close() error
	Closed() bool
}
const (
	OffsetNewest int64 = -1
	OffsetOldest int64 = -2
)
type client struct {
	conf           *Config
	closer, closed chan none  
	seedBrokers []*Broker
	deadSeeds   []*Broker
	controllerID   int32                                    
	brokers        map[int32]*Broker                        
	metadata       map[string]map[int32]*PartitionMetadata  
	metadataTopics map[string]none                          
	coordinators   map[string]int32                         
	cachedPartitionsResults map[string][maxPartitionIndex][]int32
	lock sync.RWMutex  
}
func NewClient(addrs []string, conf *Config) (Client, error) {
	Logger.Println("Initializing new client")
	if conf == nil {
		conf = NewConfig()
	}
	if err := conf.Validate(); err != nil {
		return nil, err
	}
	if len(addrs) < 1 {
		return nil, ConfigurationError("You must provide at least one broker address")
	}
	client := &client{
		conf:                    conf,
		closer:                  make(chan none),
		closed:                  make(chan none),
		brokers:                 make(map[int32]*Broker),
		metadata:                make(map[string]map[int32]*PartitionMetadata),
		metadataTopics:          make(map[string]none),
		cachedPartitionsResults: make(map[string][maxPartitionIndex][]int32),
		coordinators:            make(map[string]int32),
	}
	client.randomizeSeedBrokers(addrs)
	if conf.Metadata.Full {
		err := client.RefreshMetadata()
		switch err {
		case nil:
			break
		case ErrLeaderNotAvailable, ErrReplicaNotAvailable, ErrTopicAuthorizationFailed, ErrClusterAuthorizationFailed:
			Logger.Println(err)
		default:
			close(client.closed)  
			_ = client.Close()
			return nil, err
		}
	}
	go withRecover(client.backgroundMetadataUpdater)
	Logger.Println("Successfully initialized new client")
	return client, nil
}
func (client *client) Config() *Config {
	return client.conf
}
func (client *client) Brokers() []*Broker {
	client.lock.RLock()
	defer client.lock.RUnlock()
	brokers := make([]*Broker, 0, len(client.brokers))
	for _, broker := range client.brokers {
		brokers = append(brokers, broker)
	}
	return brokers
}
func (client *client) Broker(brokerID int32) (*Broker, error) {
	client.lock.RLock()
	defer client.lock.RUnlock()
	broker, ok := client.brokers[brokerID]
	if !ok {
		return nil, ErrBrokerNotFound
	}
	_ = broker.Open(client.conf)
	return broker, nil
}
func (client *client) InitProducerID() (*InitProducerIDResponse, error) {
	var err error
	for broker := client.any(); broker != nil; broker = client.any() {
		req := &InitProducerIDRequest{}
		response, err := broker.InitProducerID(req)
		switch err.(type) {
		case nil:
			return response, nil
		default:
			Logger.Printf("Client got error from broker %d when issuing InitProducerID : %v\n", broker.ID(), err)
			_ = broker.Close()
			client.deregisterBroker(broker)
		}
	}
	return nil, err
}
func (client *client) Close() error {
	if client.Closed() {
		Logger.Printf("Close() called on already closed client")
		return ErrClosedClient
	}
	close(client.closer)
	<-client.closed
	client.lock.Lock()
	defer client.lock.Unlock()
	Logger.Println("Closing Client")
	for _, broker := range client.brokers {
		safeAsyncClose(broker)
	}
	for _, broker := range client.seedBrokers {
		safeAsyncClose(broker)
	}
	client.brokers = nil
	client.metadata = nil
	client.metadataTopics = nil
	return nil
}
func (client *client) Closed() bool {
	client.lock.RLock()
	defer client.lock.RUnlock()
	return client.brokers == nil
}
func (client *client) Topics() ([]string, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	client.lock.RLock()
	defer client.lock.RUnlock()
	ret := make([]string, 0, len(client.metadata))
	for topic := range client.metadata {
		ret = append(ret, topic)
	}
	return ret, nil
}
func (client *client) MetadataTopics() ([]string, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	client.lock.RLock()
	defer client.lock.RUnlock()
	ret := make([]string, 0, len(client.metadataTopics))
	for topic := range client.metadataTopics {
		ret = append(ret, topic)
	}
	return ret, nil
}
func (client *client) Partitions(topic string) ([]int32, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	partitions := client.cachedPartitions(topic, allPartitions)
	if len(partitions) == 0 {
		err := client.RefreshMetadata(topic)
		if err != nil {
			return nil, err
		}
		partitions = client.cachedPartitions(topic, allPartitions)
	}
	if len(partitions) == 0 {
		return nil, ErrUnknownTopicOrPartition
	}
	return partitions, nil
}
func (client *client) WritablePartitions(topic string) ([]int32, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	partitions := client.cachedPartitions(topic, writablePartitions)
	if len(partitions) == 0 {
		err := client.RefreshMetadata(topic)
		if err != nil {
			return nil, err
		}
		partitions = client.cachedPartitions(topic, writablePartitions)
	}
	if partitions == nil {
		return nil, ErrUnknownTopicOrPartition
	}
	return partitions, nil
}
func (client *client) Replicas(topic string, partitionID int32) ([]int32, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	metadata := client.cachedMetadata(topic, partitionID)
	if metadata == nil {
		err := client.RefreshMetadata(topic)
		if err != nil {
			return nil, err
		}
		metadata = client.cachedMetadata(topic, partitionID)
	}
	if metadata == nil {
		return nil, ErrUnknownTopicOrPartition
	}
	if metadata.Err == ErrReplicaNotAvailable {
		return dupInt32Slice(metadata.Replicas), metadata.Err
	}
	return dupInt32Slice(metadata.Replicas), nil
}
func (client *client) InSyncReplicas(topic string, partitionID int32) ([]int32, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	metadata := client.cachedMetadata(topic, partitionID)
	if metadata == nil {
		err := client.RefreshMetadata(topic)
		if err != nil {
			return nil, err
		}
		metadata = client.cachedMetadata(topic, partitionID)
	}
	if metadata == nil {
		return nil, ErrUnknownTopicOrPartition
	}
	if metadata.Err == ErrReplicaNotAvailable {
		return dupInt32Slice(metadata.Isr), metadata.Err
	}
	return dupInt32Slice(metadata.Isr), nil
}
func (client *client) OfflineReplicas(topic string, partitionID int32) ([]int32, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	metadata := client.cachedMetadata(topic, partitionID)
	if metadata == nil {
		err := client.RefreshMetadata(topic)
		if err != nil {
			return nil, err
		}
		metadata = client.cachedMetadata(topic, partitionID)
	}
	if metadata == nil {
		return nil, ErrUnknownTopicOrPartition
	}
	if metadata.Err == ErrReplicaNotAvailable {
		return dupInt32Slice(metadata.OfflineReplicas), metadata.Err
	}
	return dupInt32Slice(metadata.OfflineReplicas), nil
}
func (client *client) Leader(topic string, partitionID int32) (*Broker, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	leader, err := client.cachedLeader(topic, partitionID)
	if leader == nil {
		err = client.RefreshMetadata(topic)
		if err != nil {
			return nil, err
		}
		leader, err = client.cachedLeader(topic, partitionID)
	}
	return leader, err
}
func (client *client) RefreshBrokers(addrs []string) error {
	if client.Closed() {
		return ErrClosedClient
	}
	client.lock.Lock()
	defer client.lock.Unlock()
	for _, broker := range client.brokers {
		_ = broker.Close()
		delete(client.brokers, broker.ID())
	}
	client.seedBrokers = nil
	client.deadSeeds = nil
	client.randomizeSeedBrokers(addrs)
	return nil
}
func (client *client) RefreshMetadata(topics ...string) error {
	if client.Closed() {
		return ErrClosedClient
	}
	for _, topic := range topics {
		if topic == "" {
			return ErrInvalidTopic  
		}
	}
	deadline := time.Time{}
	if client.conf.Metadata.Timeout > 0 {
		deadline = time.Now().Add(client.conf.Metadata.Timeout)
	}
	return client.tryRefreshMetadata(topics, client.conf.Metadata.Retry.Max, deadline)
}
func (client *client) GetOffset(topic string, partitionID int32, time int64) (int64, error) {
	if client.Closed() {
		return -1, ErrClosedClient
	}
	offset, err := client.getOffset(topic, partitionID, time)
	if err != nil {
		if err := client.RefreshMetadata(topic); err != nil {
			return -1, err
		}
		return client.getOffset(topic, partitionID, time)
	}
	return offset, err
}
func (client *client) Controller() (*Broker, error) {
	if client.Closed() {
		return nil, ErrClosedClient
	}
	if !client.conf.Version.IsAtLeast(V0_10_0_0) {
		return nil, ErrUnsupportedVersion
	}
	controller := client.cachedController()
	if controller == nil {
		if err := client.refreshMetadata(); err != nil {
			return nil, err
		}
		controller = client.cachedController()
	}
	if controller == nil {
		return nil, ErrControllerNotAvailable
	}
	_ = controller.Open(client.conf)
	return controller, nil
}
func (client *client) deregisterController() {
	client.lock.Lock()
	defer client.lock.Unlock()
	delete(client.brokers, client.controllerID)
}