func (c *Consumer) twLoop(stopped <-chan none) {
	ticker := time.NewTicker(c.client.config.Metadata.RefreshFrequency / 2)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			topics, err := c.client.Topics()
			if err != nil {
				c.handleError(&Error{Ctx: "topics", error: err})
				return
			}
			for _, topic := range topics {
				if !c.isKnownCoreTopic(topic) &&
					!c.isKnownExtraTopic(topic) &&
					c.isPotentialExtraTopic(topic) {
					return
				}
			}
		case <-stopped:
			return
		case <-c.dying:
			return
		}
	}
}
topic watcher loop, triggered by the mainLoop
package cluster
import (
	"sort"
	"sync"
	"sync/atomic"
	"time"
	"github.com/Shopify/sarama"
)
type Consumer struct {
	client    *Client
	ownClient bool
	consumer sarama.Consumer
	subs     *partitionMap
	consumerID string
	groupID    string
	memberID     string
	generationID int32
	membershipMu sync.RWMutex
	coreTopics  []string
	extraTopics []string
	dying, dead chan none
	closeOnce   sync.Once
	consuming     int32
	messages      chan *sarama.ConsumerMessage
	errors        chan error
	partitions    chan PartitionConsumer
	notifications chan *Notification
	commitMu sync.Mutex
}
func NewConsumer(addrs []string, groupID string, topics []string, config *Config) (*Consumer, error) {
	client, err := NewClient(addrs, config)
	if err != nil {
		return nil, err
	}
	consumer, err := NewConsumerFromClient(client, groupID, topics)
	if err != nil {
		return nil, err
	}
	consumer.ownClient = true
	return consumer, nil
}
func NewConsumerFromClient(client *Client, groupID string, topics []string) (*Consumer, error) {
	if !client.claim() {
		return nil, errClientInUse
	}
	consumer, err := sarama.NewConsumerFromClient(client.Client)
	if err != nil {
		client.release()
		return nil, err
	}
	sort.Strings(topics)
	c := &Consumer{
		client:   client,
		consumer: consumer,
		subs:     newPartitionMap(),
		groupID:  groupID,
		coreTopics: topics,
		dying: make(chan none),
		dead:  make(chan none),
		messages:      make(chan *sarama.ConsumerMessage),
		errors:        make(chan error, client.config.ChannelBufferSize),
		partitions:    make(chan PartitionConsumer, 1),
		notifications: make(chan *Notification),
	}
	if err := c.client.RefreshCoordinator(groupID); err != nil {
		client.release()
		return nil, err
	}
	go c.mainLoop()
	return c, nil
}
func (c *Consumer) Messages() <-chan *sarama.ConsumerMessage { return c.messages }
func (c *Consumer) Partitions() <-chan PartitionConsumer { return c.partitions }
func (c *Consumer) Errors() <-chan error { return c.errors }
func (c *Consumer) Notifications() <-chan *Notification { return c.notifications }
func (c *Consumer) HighWaterMarks() map[string]map[int32]int64 { return c.consumer.HighWaterMarks() }
func (c *Consumer) MarkOffset(msg *sarama.ConsumerMessage, metadata string) {
	if sub := c.subs.Fetch(msg.Topic, msg.Partition); sub != nil {
		sub.MarkOffset(msg.Offset, metadata)
	}
}
func (c *Consumer) MarkPartitionOffset(topic string, partition int32, offset int64, metadata string) {
	if sub := c.subs.Fetch(topic, partition); sub != nil {
		sub.MarkOffset(offset, metadata)
	}
}
func (c *Consumer) MarkOffsets(s *OffsetStash) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for tp, info := range s.offsets {
		if sub := c.subs.Fetch(tp.Topic, tp.Partition); sub != nil {
			sub.MarkOffset(info.Offset, info.Metadata)
		}
		delete(s.offsets, tp)
	}
}
func (c *Consumer) ResetOffset(msg *sarama.ConsumerMessage, metadata string) {
	if sub := c.subs.Fetch(msg.Topic, msg.Partition); sub != nil {
		sub.ResetOffset(msg.Offset, metadata)
	}
}
func (c *Consumer) ResetPartitionOffset(topic string, partition int32, offset int64, metadata string) {
	sub := c.subs.Fetch(topic, partition)
	if sub != nil {
		sub.ResetOffset(offset, metadata)
	}
}
func (c *Consumer) ResetOffsets(s *OffsetStash) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for tp, info := range s.offsets {
		if sub := c.subs.Fetch(tp.Topic, tp.Partition); sub != nil {
			sub.ResetOffset(info.Offset, info.Metadata)
		}
		delete(s.offsets, tp)
	}
}
func (c *Consumer) Subscriptions() map[string][]int32 {
	return c.subs.Info()
}
func (c *Consumer) CommitOffsets() error {
	c.commitMu.Lock()
	defer c.commitMu.Unlock()
	memberID, generationID := c.membership()
	req := &sarama.OffsetCommitRequest{
		Version:                 2,
		ConsumerGroup:           c.groupID,
		ConsumerGroupGeneration: generationID,
		ConsumerID:              memberID,
		RetentionTime:           -1,
	}
	if ns := c.client.config.Consumer.Offsets.Retention; ns != 0 {
		req.RetentionTime = int64(ns / time.Millisecond)
	}
	snap := c.subs.Snapshot()
	dirty := false
	for tp, state := range snap {
		if state.Dirty {
			dirty = true
			req.AddBlock(tp.Topic, tp.Partition, state.Info.Offset, 0, state.Info.Metadata)
		}
	}
	if !dirty {
		return nil
	}
	broker, err := c.client.Coordinator(c.groupID)
	if err != nil {
		c.closeCoordinator(broker, err)
		return err
	}
	resp, err := broker.CommitOffset(req)
	if err != nil {
		c.closeCoordinator(broker, err)
		return err
	}
	for topic, errs := range resp.Errors {
		for partition, kerr := range errs {
			if kerr != sarama.ErrNoError {
				err = kerr
			} else if state, ok := snap[topicPartition{topic, partition}]; ok {
				if sub := c.subs.Fetch(topic, partition); sub != nil {
					sub.markCommitted(state.Info.Offset)
				}
			}
		}
	}
	return err
}
func (c *Consumer) Close() (err error) {
	c.closeOnce.Do(func() {
		close(c.dying)
		<-c.dead
		if e := c.release(); e != nil {
			err = e
		}
		if e := c.consumer.Close(); e != nil {
			err = e
		}
		close(c.messages)
		close(c.errors)
		if e := c.leaveGroup(); e != nil {
			err = e
		}
		close(c.partitions)
		close(c.notifications)
		for range c.messages {
		}
		for range c.errors {
		}
		for p := range c.partitions {
			_ = p.Close()
		}
		for range c.notifications {
		}
		c.client.release()
		if c.ownClient {
			if e := c.client.Close(); e != nil {
				err = e
			}
		}
	})
	return
}
func (c *Consumer) mainLoop() {
	defer close(c.dead)
	defer atomic.StoreInt32(&c.consuming, 0)
	for {
		atomic.StoreInt32(&c.consuming, 0)
		select {
		case <-c.dying:
			return
		default:
		}
		c.nextTick()
	}
}
func (c *Consumer) nextTick() {
	var notification *Notification
	if c.client.config.Group.Return.Notifications {
		notification = newNotification(c.subs.Info())
	}
	if err := c.refreshCoordinator(); err != nil {
		c.rebalanceError(err, nil)
		return
	}
	if err := c.release(); err != nil {
		c.rebalanceError(err, nil)
		return
	}
	if c.client.config.Group.Return.Notifications {
		c.handleNotification(notification)
	}
	subs, err := c.rebalance()
	if err != nil {
		c.rebalanceError(err, notification)
		return
	}
	tomb := newLoopTomb()
	defer tomb.Close()
	tomb.Go(c.hbLoop)
	if err := c.subscribe(tomb, subs); err != nil {
		c.rebalanceError(err, notification)
		return
	}
	if c.client.config.Group.Return.Notifications {
		notification = notification.success(subs)
		c.handleNotification(notification)
	}
	tomb.Go(c.twLoop)
	tomb.Go(c.cmLoop)
	atomic.StoreInt32(&c.consuming, 1)
	select {
	case <-tomb.Dying():
	case <-c.dying:
	}
}
func (c *Consumer) hbLoop(stopped <-chan none) {
	ticker := time.NewTicker(c.client.config.Group.Heartbeat.Interval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			switch err := c.heartbeat(); err {
			case nil, sarama.ErrNoError:
			case sarama.ErrNotCoordinatorForConsumer, sarama.ErrRebalanceInProgress:
				return
			default:
				c.handleError(&Error{Ctx: "heartbeat", error: err})
				return
			}
		case <-stopped:
			return
		case <-c.dying:
			return
		}
	}
}