func (client *KafkaClient) Stop() {
	client.brokerOffsetTicker.Stop()
	client.importer.stop()
}
Stop the client
package monitor
import (
	"crypto/tls"
	"crypto/x509"
	"io/ioutil"
	"regexp"
	"strings"
	"sync"
	"time"
	"github.com/Shopify/sarama"
	log "github.com/cihub/seelog"
	"github.com/sundy-li/burrowx/config"
)
type KafkaClient struct {
	cluster        string
	cfg            *config.Config
	client         sarama.Client
	topicMap       map[string]int
	topic2Consumer map[string][]string
	schemaUpdateMtx *sync.RWMutex
	brokerOffsetTicker *time.Ticker
	topicOffsetMapLock *sync.RWMutex
	topicOffset map[string]map[int32]int64
	importer *Importer
	topicFilterRegexps []*regexp.Regexp
	groupFilterRegexps []*regexp.Regexp
}
type BrokerTopicRequest struct {
	Result chan int
	Topic  string
}
var (
	METRIC_FETCH_INTERVAL_SECOND = 10
	META_UPDATE_INTERVAL_SECOND  = 60
)
func NewKafkaClient(cfg *config.Config, cluster string) (*KafkaClient, error) {
	clientConfig := sarama.NewConfig()
	clientConfig.Version = sarama.V0_10_2_0
	profile := cfg.ClientProfile[cfg.Kafka[cluster].ClientProfile]
	clientConfig.ClientID = profile.ClientId
	clientConfig.Net.TLS.Enable = profile.TLS
	if profile.TLSCertFilePath == "" || profile.TLSKeyFilePath == "" || profile.TLSCAFilePath == "" {
		clientConfig.Net.TLS.Config = &tls.Config{}
	} else {
		caCert, err := ioutil.ReadFile(profile.TLSCAFilePath)
		if err != nil {
			return nil, err
		}
		cert, err := tls.LoadX509KeyPair(profile.TLSCertFilePath, profile.TLSKeyFilePath)
		if err != nil {
			return nil, err
		}
		caCertPool := x509.NewCertPool()
		caCertPool.AppendCertsFromPEM(caCert)
		clientConfig.Net.TLS.Config = &tls.Config{
			Certificates: []tls.Certificate{cert},
			RootCAs:      caCertPool,
		}
		clientConfig.Net.TLS.Config.BuildNameToCertificate()
	}
	clientConfig.Net.TLS.Config.InsecureSkipVerify = profile.TLSNoVerify
	if cfg.Kafka[cluster].Sasl.Username != "" {
		clientConfig.Net.SASL.Enable = true
		clientConfig.Net.SASL.User = cfg.Kafka[cluster].Sasl.Username
		clientConfig.Net.SASL.Password = cfg.Kafka[cluster].Sasl.Password
	}
	sclient, err := sarama.NewClient(strings.Split(cfg.Kafka[cluster].Brokers, ","), clientConfig)
	if err != nil {
		return nil, err
	}
	importer, err := NewImporter(cfg)
	if err != nil {
		return nil, err
	}
	client := &KafkaClient{
		cluster:        cluster,
		cfg:            cfg,
		client:         sclient,
		topicMap:       make(map[string]int),
		topic2Consumer: make(map[string][]string),
		schemaUpdateMtx: &sync.RWMutex{},
		topicOffset:        make(map[string]map[int32]int64),
		topicOffsetMapLock: &sync.RWMutex{},
		importer: importer,
	}
	{
		if cfg.General.TopicFilter == "" {
			cfg.General.TopicFilter = ".*"
		}
		patterns := strings.Split(cfg.General.TopicFilter, ",")
		client.topicFilterRegexps = make([]*regexp.Regexp, 0, 10)
		for _, p := range patterns {
			client.topicFilterRegexps = append(client.topicFilterRegexps, regexp.MustCompile(p))
		}
	}
	{
		if cfg.General.GroupFilter == "" {
			cfg.General.GroupFilter = ".*"
		}
		patterns := strings.Split(cfg.General.GroupFilter, ",")
		client.groupFilterRegexps = make([]*regexp.Regexp, 0, 10)
		for _, p := range patterns {
			client.groupFilterRegexps = append(client.groupFilterRegexps, regexp.MustCompile(p))
		}
	}
	return client, nil
}
func (client *KafkaClient) Start() {
	client.importer.start()
	client.RefreshMetaData()
	client.getOffsets()
	client.brokerOffsetTicker = time.NewTicker(time.Duration(METRIC_FETCH_INTERVAL_SECOND) * time.Second)
	go func() {
		for _ = range client.brokerOffsetTicker.C {
			client.getOffsets()
		}
	}()
	go func() {
		ticker := time.NewTicker(time.Duration(META_UPDATE_INTERVAL_SECOND) * time.Second)
		for _ = range ticker.C {
			client.RefreshMetaData()
		}
	}()
}