package consumer

import (
	"sort"
	"sync"
	"sync/atomic"
	"time"

	"github.com/Shopify/sarama"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	ether "github.com/pingcap/wormhole/pkg/ether-client"
	"github.com/pingcap/wormhole/pkg/mq/zk"
	gozk "github.com/samuel/go-zookeeper/zk"
)

// Consumer is a cluster group consumer
type Consumer struct {
	client  *Client
	manager *etcdOffsetManager
	ec      ether.Client

	csmr sarama.Consumer
	subs *partitionMap

	consumerID   string
	generationID int32
	groupID      string
	memberID     string

	coreTopics  []string
	extraTopics []string

	dying, dead chan none

	consuming     int32
	errors        chan error
	messages      chan Message
	notifications chan *Notification

	commitMu sync.Mutex

	zkClient *zk.Client
}

//Message contains consumer message options.
type Message struct {
	Message *sarama.ConsumerMessage
	Sync    bool
	Done    chan struct{}
}

// NewConsumerFromClient initializes a new consumer from an existing client
// Use wormhole jobID as consumer groupID
func NewConsumerFromClient(client *Client, jobID string, topics []string, etcdClient *etcdutil.Client, etherClient ether.Client, zkClient *zk.Client) (*Consumer, error) {
	csmr, err := sarama.NewConsumerFromClient(client.Client)
	if err != nil {
		return nil, errors.Trace(err)
	}

	log.Debugf("[consumer] comsume topics %+v", topics)

	sort.Strings(topics)

	c := &Consumer{
		client:  client,
		csmr:    csmr,
		subs:    newPartitionMap(),
		groupID: jobID,

		coreTopics: topics,

		dying: make(chan none),
		dead:  make(chan none),

		errors:        make(chan error, client.config.ChannelBufferSize),
		messages:      make(chan Message),
		notifications: make(chan *Notification, 1),
		manager:       newEtcdOffsetManager(etcdClient),
		ec:            etherClient,
		zkClient:      zkClient,
	}
	if err := c.client.RefreshCoordinator(jobID); err != nil {
		return nil, errors.Trace(err)
	}

	go c.mainLoop()
	return c, nil
}

// NewConsumer initializes a new consumer
func NewConsumer(addrs []string, jobID string, topics []string, config *Config, etcdClient *etcdutil.Client, etherClient ether.Client, zkClient *zk.Client) (*Consumer, error) {
	client, err := NewClient(addrs, config)
	if err != nil {
		return nil, errors.Trace(err)
	}

	consumer, err := NewConsumerFromClient(client, jobID, topics, etcdClient, etherClient, zkClient)
	if err != nil {
		_ = client.Close()
		return nil, errors.Trace(err)
	}
	consumer.client.own = true
	return consumer, nil
}

// Messages returns the read channel for the messages that are returned by
// the broker.
func (c *Consumer) Messages() <-chan Message { return c.messages }

// Errors returns a read channel of errors that occur during offset management, if
// enabled. By default, errors are logged and not returned over this channel. If
// you want to implement any custom error handling, set your config's
// Consumer.Return.Errors setting to true, and read from this channel.
func (c *Consumer) Errors() <-chan error { return c.errors }

// Notifications returns a channel of Notifications that occur during consumer
// rebalancing. Notifications will only be emitted over this channel, if your config's
// Group.Return.Notifications setting to true.
func (c *Consumer) Notifications() <-chan *Notification { return c.notifications }

// HighWaterMarks returns the current high water marks for each topic and partition
// Consistency between partitions is not guaranteed since high water marks are updated separately.
func (c *Consumer) HighWaterMarks() map[string]map[int32]int64 { return c.csmr.HighWaterMarks() }

// MarkOffset marks the provided message as processed, alongside a metadata string
// that represents the state of the partition consumer at that point in time. The
// metadata string can be used by another consumer to restore that state, so it
// can resume consumption.
//
// Note: calling MarkOffset does not necessarily commit the offset to the backend
// store immediately for efficiency reasons, and it may never be committed if
// your application crashes. This means that you may end up processing the same
// message twice, and your processing should ideally be idempotent.
func (c *Consumer) MarkOffset(msg *sarama.ConsumerMessage, metadata string) {
	c.subs.Fetch(msg.Topic, msg.Partition).MarkOffset(msg.Offset+1, metadata)
}

// MarkPartitionOffset marks an offset of the provided topic/partition as processed.
// See MarkOffset for additional explanation.
func (c *Consumer) MarkPartitionOffset(topic string, partition int32, offset int64, metadata string) {
	c.subs.Fetch(topic, partition).MarkOffset(offset+1, metadata)
}

// MarkOffsets marks stashed offsets as processed.
// See MarkOffset for additional explanation.
func (c *Consumer) MarkOffsets(s *OffsetStash) {
	s.mu.Lock()
	defer s.mu.Unlock()

	for tp, info := range s.offsets {
		c.subs.Fetch(tp.Topic, tp.Partition).MarkOffset(info.Offset+1, info.Metadata)
		delete(s.offsets, tp)
	}
}

// Subscriptions returns the consumed topics and partitions
func (c *Consumer) Subscriptions() map[string][]int32 {
	return c.subs.Info()
}

// CommitOffsets manually commits marked offsets.
func (c *Consumer) CommitOffsets() error {
	c.commitMu.Lock()
	defer c.commitMu.Unlock()

	req := &OffsetCommitRequest{
		ConsumerGroup:           c.groupID,
		ConsumerGroupGeneration: c.generationID,
		ConsumerID:              c.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
		}
		// ensure adding every partition, no matter it's offset changed or not.
		req.AddBlock(tp.Topic, tp.Partition, state.Info.Offset, 0, state.Info.Metadata)
	}
	if !dirty {
		return nil
	}

	resp, err := c.manager.CommitOffset(req)
	if err != nil {
		log.Errorf("[offset] commmit offset err: %s", err)
		return errors.Trace(err)
	}

	for topic, errs := range resp.Errors {
		for partition, e := range errs {
			if e != nil {
				err = e
			} else if state, ok := snap[topicPartition{topic, partition}]; ok {
				c.subs.Fetch(topic, partition).MarkCommitted(state.Info.Offset)
			}
		}
	}
	return errors.Trace(err)
}

// Snapshot returns consumer offset snapshot.
func (c *Consumer) Snapshot() map[string]map[int32]int64 {
	snap := make(map[string]map[int32]int64)
	if c.subs == nil {
		return snap
	}
	for k, v := range c.subs.Snapshot() {
		if _, ok := snap[k.Topic]; !ok {
			snap[k.Topic] = make(map[int32]int64)
		}
		snap[k.Topic][k.Partition] = v.Info.Offset
	}
	return snap
}

// Close safely closes the consumer and releases all resources
func (c *Consumer) Close() (err error) {
	select {
	case <-c.dying:
		return
	default:
		close(c.dying)
	}
	<-c.dead

	if e := c.release(); e != nil {
		err = errors.Trace(e)
	}
	if e := c.csmr.Close(); e != nil {
		err = errors.Trace(e)
	}
	close(c.messages)
	close(c.errors)

	if e := c.leaveGroup(); e != nil {
		err = errors.Trace(e)
	}
	close(c.notifications)

	if c.client.own {
		if e := c.client.Close(); e != nil {
			err = errors.Trace(e)
		}
	}

	return
}

func (c *Consumer) mainLoop() {
	defer close(c.dead)
	defer atomic.StoreInt32(&c.consuming, 0)

	for {
		atomic.StoreInt32(&c.consuming, 0)

		// Check if close was requested
		select {
		case <-c.dying:
			return
		default:
		}

		// Remember previous subscriptions
		var notification *Notification
		if c.client.config.Group.Return.Notifications {
			notification = newNotification(c.subs.Info())
		}

		// Rebalance, fetch new subscriptions
		subs, err := c.rebalance()
		if err != nil {
			c.rebalanceError(err, notification)
			continue
		}
		log.Infof("[comsumer] rebalance get subscriptions: subs=%+v", subs)
		// go to next cycle
		if c.shouldContinue(subs) {
			time.Sleep(time.Second * 1)
			continue
		}

		// Start the heartbeat
		hbStop, hbDone := make(chan none), make(chan none)
		go c.hbLoop(hbStop, hbDone)

		// Subscribe to topic/partitions
		if err := c.subscribe(subs); err != nil {
			close(hbStop)
			<-hbDone
			c.rebalanceError(err, notification)
			continue
		}

		// Start topic watcher loop
		twStop, twDone := make(chan none), make(chan none)
		go c.twLoop(twStop, twDone)

		// Start consuming and committing offsets
		cmStop, cmDone := make(chan none), make(chan none)
		go c.cmLoop(cmStop, cmDone)
		atomic.StoreInt32(&c.consuming, 1)

		// Update notification with new claims
		if c.client.config.Group.Return.Notifications {
			notification.claim(subs)
			c.notifications <- notification
		}

		// Wait for signals
		select {
		case <-hbDone:
			close(cmStop)
			close(twStop)
			<-cmDone
			<-twDone
		case <-twDone:
			close(cmStop)
			close(hbStop)
			<-cmDone
			<-hbDone
		case <-cmDone:
			close(twStop)
			close(hbStop)
			<-twDone
			<-hbDone
		case <-c.dying:
			close(cmStop)
			close(twStop)
			close(hbStop)
			<-cmDone
			<-twDone
			<-hbDone
			return
		}
	}
}

func (c *Consumer) shouldContinue(subs map[string][]int32) bool {
	if len(subs) == 0 {
		log.Infof("[consumer] got 0 subs, star over again.")
		return true
	}
	for _, t := range c.extraTopics {
		if _, ok := subs[t]; !ok {
			log.Infof("extra topic %s not in subs, continue", t)
			return true
		}
	}
	return false
}

// heartbeat loop, triggered by the mainLoop
func (c *Consumer) hbLoop(stop <-chan none, done chan<- none) {
	defer close(done)

	ticker := time.NewTicker(c.client.config.Group.Heartbeat.Interval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			switch err := c.heartbeat(); errors.Cause(err) {
			case nil, sarama.ErrNoError:
			case sarama.ErrNotCoordinatorForConsumer, sarama.ErrRebalanceInProgress:
				log.Errorf("[consumer] err %v", err)
				return
			default:
				log.Errorf("[consumer] err %v", err)
				c.handleError(&Error{Ctx: "heartbeat", error: errors.Trace(err)})
				return
			}
		case <-stop:
			return
		}
	}
}

// topic watcher loop, triggered by the mainLoop
func (c *Consumer) twLoop(stop <-chan none, done chan<- none) {
	defer close(done)

Watch:
	_, event, err := c.zkClient.WatchTopics()
	if err != nil {
		c.handleError(&Error{Ctx: "topics", error: errors.Trace(err)})
		return
	}

	// minimum guarantee for refresh metadata in case something wrong with zookeeper watcher.
	ticker := time.NewTicker(c.client.config.Metadata.RefreshFrequency / 2)
	defer ticker.Stop()

	findPotentialExtraTopic := func() (breakL bool) {
		topics, err1 := c.zkClient.Topics()
		if err1 != nil {
			c.handleError(&Error{Ctx: "topics", error: errors.Trace(err1)})
			return true
		}

		// if found potential extra topic(at least one), then exit twLoop, the mainLoop would go to next round(noted by @chendahui)
		for _, topic := range topics {
			if !c.isKnownCoreTopic(topic) &&
				!c.isKnownExtraTopic(topic) &&
				c.isPotentialExtraTopic(topic) {
				log.Infof("found potentialExtracTopic %v", topic)
				return true
			}
		}
		return false
	}

	for {
		select {
		case e := <-event:
			if e.Err != nil {
				c.handleError(&Error{Ctx: "topics", error: errors.Trace(err)})
				return
			}
			log.Infof("[consumer] zk event %+v", e)
			if e.Type != gozk.EventNodeChildrenChanged {
				if e.Type == 0 && e.State == gozk.StateDisconnected {
					log.Debug("go to watch again")
					goto Watch
				}
				time.Sleep(time.Second * 1)
				continue
			}
			if findPotentialExtraTopic() {
				return
			}

		case <-ticker.C:
			if findPotentialExtraTopic() {
				return
			}

		case <-stop:
			return
		}
	}
}

// commit loop, triggered by the mainLoop
func (c *Consumer) cmLoop(stop <-chan none, done chan<- none) {
	defer close(done)

	ticker := time.NewTicker(c.client.config.Consumer.Offsets.CommitInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			if err := c.commitOffsetsWithRetry(c.client.config.Group.Offsets.Retry.Max); err != nil {
				c.handleError(&Error{Ctx: "commit", error: errors.Trace(err)})
				return
			}
		case <-stop:
			return
		}
	}
}

func (c *Consumer) rebalanceError(err error, notification *Notification) {
	if c.client.config.Group.Return.Notifications {
		c.notifications <- notification
	}

	switch errors.Cause(err) {
	case sarama.ErrRebalanceInProgress:
	default:
		c.handleError(&Error{Ctx: "rebalance", error: errors.Trace(err)})
	}

	select {
	case <-c.dying:
	case <-time.After(c.client.config.Metadata.Retry.Backoff):
	}
}

func (c *Consumer) handleError(e *Error) {
	if c.client.config.Consumer.Return.Errors {
		select {
		case c.errors <- e:
		case <-c.dying:
			return
		}
	} else {
		log.Debugf("[consumer] %s error: %s\n", e.Ctx, e.Error())
	}
}

// Releases the consumer and commits offsets, called from rebalance() and Close()
func (c *Consumer) release() (err error) {
	// Stop all consumers
	c.subs.Stop()

	// Clear subscriptions on exit
	defer c.subs.Clear()

	// Wait for messages to be processed
	select {
	case <-c.dying:
	case <-time.After(c.client.config.Group.Offsets.Synchronization.DwellTime):
	}

	// Commit offsets, continue on errors
	if e := c.commitOffsetsWithRetry(c.client.config.Group.Offsets.Retry.Max); e != nil {
		err = errors.Trace(e)
	}

	return
}

// --------------------------------------------------------------------

// Performs a heartbeat, part of the mainLoop()
func (c *Consumer) heartbeat() error {
	broker, err := c.client.Coordinator(c.groupID)
	if err != nil {
		c.closeCoordinator(broker, err)
		return errors.Trace(err)
	}

	resp, err := broker.Heartbeat(&sarama.HeartbeatRequest{
		GroupId:      c.groupID,
		MemberId:     c.memberID,
		GenerationId: c.generationID,
	})
	if err != nil {
		c.closeCoordinator(broker, err)
		return errors.Trace(err)
	}
	return errors.Trace(resp.Err)
}

// Performs a rebalance, part of the mainLoop()
func (c *Consumer) rebalance() (map[string][]int32, error) {
	log.Debugf("[consumer] rebalance memberID=%+v", c.memberID)

	if err := c.refreshMetadata(); err != nil {
		return nil, errors.Trace(err)
	}

	if err := c.client.RefreshCoordinator(c.groupID); err != nil {
		return nil, errors.Trace(err)
	}

	// Use zk client to get topics, instead of consumer.Client.Topics(), because the later can not
	// get the latest metadata(topics).
	allTopics, err := c.zkClient.Topics()
	if err != nil {
		return nil, errors.Trace(err)
	}
	c.extraTopics = c.selectExtraTopics(allTopics)
	sort.Strings(c.extraTopics)
	log.Debugf("[consumer] all topics %+v, extra %+v", allTopics, c.extraTopics)

	// Release subscriptions
	if err = c.release(); err != nil {
		return nil, errors.Trace(err)
	}

	// Re-join consumer group
	strategy, err := c.joinGroup()
	err = errors.Cause(err)
	switch {
	case err == sarama.ErrUnknownMemberId:
		c.memberID = ""
		return nil, errors.Trace(err)
	case err != nil:
		return nil, errors.Trace(err)
	}
	log.Debugf("[consumer] %s/%d joined group %s", c.memberID, c.generationID, c.groupID)

	// Sync consumer group state, fetch subscriptions
	subs, err := c.syncGroup(strategy)
	err = errors.Cause(err)
	switch {
	case err == sarama.ErrRebalanceInProgress:
		return nil, errors.Trace(err)
	case err != nil:
		_ = c.leaveGroup()
		return nil, errors.Trace(err)
	}
	return subs, nil
}

// Performs the subscription, part of the mainLoop()
func (c *Consumer) subscribe(subs map[string][]int32) error {
	// fetch offsets
	offsets, err := c.fetchOffsets(subs)
	if err != nil {
		_ = c.leaveGroup()
		return errors.Trace(err)
	}

	// create consumers in parallel
	var mu sync.Mutex
	var wg sync.WaitGroup

	for topic, partitions := range subs {
		for _, partition := range partitions {
			wg.Add(1)

			info := offsets[topic][partition]
			log.Debugf("[consumer] create consumer. topic: %v, partition: %v, offset:%+v", topic, partition, info)
			go func(t string, p int32) {
				if e := c.createConsumer(t, p, info); e != nil {
					mu.Lock()
					err = e
					mu.Unlock()
				}
				wg.Done()
			}(topic, partition)
		}
	}
	wg.Wait()

	if err != nil {
		_ = c.release()
		_ = c.leaveGroup()
	}
	return errors.Trace(err)
}

// --------------------------------------------------------------------

// Send a request to the broker to join group on rebalance()
func (c *Consumer) joinGroup() (*balancer, error) {
	req := &sarama.JoinGroupRequest{
		GroupId:        c.groupID,
		MemberId:       c.memberID,
		SessionTimeout: int32(c.client.config.Group.Session.Timeout / time.Millisecond),
		ProtocolType:   "consumer",
	}

	log.Debugf("[consumer] join group. groupid %s, memberid %s", c.groupID, c.memberID)

	meta := &sarama.ConsumerGroupMemberMetadata{
		Version:  1,
		Topics:   append(c.coreTopics, c.extraTopics...),
		UserData: c.client.config.Group.Member.UserData,
	}
	log.Debugf("[consumer] join group meta.topics %+v", meta.Topics)
	err := req.AddGroupProtocolMetadata(string(StrategyRange), meta)
	if err != nil {
		return nil, errors.Trace(err)
	}
	err = req.AddGroupProtocolMetadata(string(StrategyRoundRobin), meta)
	if err != nil {
		return nil, errors.Trace(err)
	}

	broker, err := c.client.Coordinator(c.groupID)
	if err != nil {
		c.closeCoordinator(broker, err)
		return nil, errors.Trace(err)
	}

	resp, err := broker.JoinGroup(req)
	if err != nil {
		c.closeCoordinator(broker, err)
		return nil, errors.Trace(err)
	} else if resp.Err != sarama.ErrNoError {
		c.closeCoordinator(broker, resp.Err)
		return nil, errors.Trace(resp.Err)
	}

	var strategy *balancer
	if resp.LeaderId == resp.MemberId {
		members, err := resp.GetMembers()
		if err != nil {
			return nil, errors.Trace(err)
		}

		strategy, err = newBalancerFromMeta(c.client, members)
		if err != nil {
			return nil, errors.Trace(err)
		}
	}

	c.memberID = resp.MemberId
	c.generationID = resp.GenerationId

	return strategy, nil
}

// Send a request to the broker to sync the group on rebalance().
// Returns a list of topics and partitions to consume.
func (c *Consumer) syncGroup(strategy *balancer) (map[string][]int32, error) {
	req := &sarama.SyncGroupRequest{
		GroupId:      c.groupID,
		MemberId:     c.memberID,
		GenerationId: c.generationID,
	}

	for memberID, topics := range strategy.Perform(c.client.config.Group.PartitionStrategy) {
		if err := req.AddGroupAssignmentMember(memberID, &sarama.ConsumerGroupMemberAssignment{
			Version: 1,
			Topics:  topics,
		}); err != nil {
			return nil, errors.Trace(err)
		}
	}

	broker, err := c.client.Coordinator(c.groupID)
	if err != nil {
		c.closeCoordinator(broker, err)
		return nil, errors.Trace(err)
	}

	resp, err := broker.SyncGroup(req)
	if err != nil {
		c.closeCoordinator(broker, err)
		return nil, errors.Trace(err)
	} else if resp.Err != sarama.ErrNoError {
		c.closeCoordinator(broker, resp.Err)
		return nil, errors.Trace(resp.Err)
	}

	// Return if there is nothing to subscribe to
	if len(resp.MemberAssignment) == 0 {
		return nil, nil
	}

	// Get assigned subscriptions
	members, err := resp.GetMemberAssignment()
	if err != nil {
		return nil, errors.Trace(err)
	}

	// Sort partitions, for each topic
	for topic := range members.Topics {
		sort.Sort(int32Slice(members.Topics[topic]))
	}
	return members.Topics, nil
}

// Fetches latest committed offsets for all subscriptions
func (c *Consumer) fetchOffsets(subs map[string][]int32) (map[string]map[int32]offsetInfo, error) {
	offsets := make(map[string]map[int32]offsetInfo, len(subs))
	req := &OffsetFetchRequest{
		ConsumerGroup: c.groupID,
	}

	for topic, partitions := range subs {
		offsets[topic] = make(map[int32]offsetInfo, len(partitions))
		for _, partition := range partitions {
			offsets[topic][partition] = offsetInfo{Offset: -1}
			req.AddPartition(topic, partition)
		}
	}

	// Wait for other cluster consumers to process, release and commit
	// Times-two so that we are less likely to "win" a race against a sarama-cluster client that is:
	// 1) losing a topic-partition in a rebalance, and therefore:
	// 2) sleeping to allow some processing to complete, before:
	// 3) committing the offsets.
	// Note that this doesn't necessarily account for the end-to-end latency of the Kafka offsets topic.
	select {
	case <-c.dying:
		log.Error("[consumer] sarama.ErrClosedClient")
		return nil, sarama.ErrClosedClient
	case <-time.After(c.client.config.Group.Offsets.Synchronization.DwellTime * 2):
	}

	resp, err := c.manager.FetchOffset(req)
	if err != nil {
		log.Errorf("[offset] fetch offset error:%s", err)
		return nil, errors.Trace(err)
	}

	for topic, partitions := range subs {
		for _, partition := range partitions {
			block := resp.GetBlock(topic, partition)
			if block == nil {
				log.Error("[consumer] sarama.ErrIncompleteResponse")
				return nil, errors.Trace(sarama.ErrIncompleteResponse)
			}
			if block.Err == nil {
				offsets[topic][partition] = offsetInfo{Offset: block.Offset, Metadata: block.Metadata}
			} else {
				log.Error(block.Err)
				return nil, errors.Trace(block.Err)
			}
		}
	}
	return offsets, nil
}

// Send a request to the broker to leave the group on failes rebalance() and on Close()
func (c *Consumer) leaveGroup() error {
	broker, err := c.client.Coordinator(c.groupID)
	if err != nil {
		c.closeCoordinator(broker, err)
		return errors.Trace(err)
	}

	if _, err = broker.LeaveGroup(&sarama.LeaveGroupRequest{
		GroupId:  c.groupID,
		MemberId: c.memberID,
	}); err != nil {
		c.closeCoordinator(broker, err)
	}
	return errors.Trace(err)
}

// --------------------------------------------------------------------

func (c *Consumer) createConsumer(topic string, partition int32, info offsetInfo) error {
	// Create partitionConsumer
	pc, err := newPartitionConsumer(c.csmr, topic, partition, info, c.client.config.Consumer.Offsets.Initial, c.ec, c.groupID)
	if err != nil {
		return errors.Trace(err)
	}

	// Store in subscriptions
	c.subs.Store(topic, partition, pc)

	// Start partition consumer goroutine
	go pc.Loop(c.messages, c.errors)

	return nil
}

func (c *Consumer) commitOffsetsWithRetry(retries int) error {
	err := c.CommitOffsets()
	if err != nil && retries > 0 {
		return c.commitOffsetsWithRetry(retries - 1)
	}
	return errors.Trace(err)
}

func (c *Consumer) closeCoordinator(broker *sarama.Broker, err error) {
	if broker != nil {
		_ = broker.Close()
	}

	switch errors.Cause(err) {
	case sarama.ErrConsumerCoordinatorNotAvailable, sarama.ErrNotCoordinatorForConsumer:
		_ = c.client.RefreshCoordinator(c.groupID)
	}
}

func (c *Consumer) selectExtraTopics(allTopics []string) []string {
	extra := allTopics[:0]
	for _, topic := range allTopics {
		if !c.isKnownCoreTopic(topic) && c.isPotentialExtraTopic(topic) {
			extra = append(extra, topic)
		}
	}
	return extra
}

func (c *Consumer) isKnownCoreTopic(topic string) bool {
	pos := sort.SearchStrings(c.coreTopics, topic)
	return pos < len(c.coreTopics) && c.coreTopics[pos] == topic
}

func (c *Consumer) isKnownExtraTopic(topic string) bool {
	pos := sort.SearchStrings(c.extraTopics, topic)
	return pos < len(c.extraTopics) && c.extraTopics[pos] == topic
}

func (c *Consumer) isPotentialExtraTopic(topic string) bool {
	rx := c.client.config.Group.Topics
	if rx.Blacklist != nil && rx.Blacklist.MatchString(topic) {
		return false
	}
	if rx.Whitelist != nil && rx.Whitelist.MatchString(topic) {
		return true
	}
	return false
}

func (c *Consumer) refreshMetadata() error {
	err := c.client.RefreshMetadata()
	if err == sarama.ErrTopicAuthorizationFailed {
		log.Errorf("err %+v", err)
		// maybe we didn't have authorization to describe all topics
		err = c.client.RefreshMetadata(c.coreTopics...)
	}
	return errors.Trace(err)
}