diff --git a/vendor/github.com/bsm/sarama-cluster/consumer.go b/vendor/github.com/bsm/sarama-cluster/consumer.go
new file mode 100644
index 0000000..e7a67da
--- /dev/null
+++ b/vendor/github.com/bsm/sarama-cluster/consumer.go
@@ -0,0 +1,919 @@
+package cluster
+
+import (
+	"sort"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/Shopify/sarama"
+)
+
+// Consumer is a cluster group consumer
+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
+}
+
+// NewConsumer initializes a new consumer
+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
+}
+
+// NewConsumerFromClient initializes a new consumer from an existing client.
+//
+// Please note that clients cannot be shared between consumers (due to Kafka internals),
+// they can only be re-used which requires the user to call Close() on the first consumer
+// before using this method again to initialize another one. Attempts to use a client with
+// more than one consumer at a time will return errors.
+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
+}
+
+// Messages returns the read channel for the messages that are returned by
+// the broker.
+//
+// This channel will only return if Config.Group.Mode option is set to
+// ConsumerModeMultiplex (default).
+func (c *Consumer) Messages() <-chan *sarama.ConsumerMessage { return c.messages }
+
+// Partitions returns the read channels for individual partitions of this broker.
+//
+// This will channel will only return if Config.Group.Mode option is set to
+// ConsumerModePartitions.
+//
+// The Partitions() channel must be listened to for the life of this consumer;
+// when a rebalance happens old partitions will be closed (naturally come to
+// completion) and new ones will be emitted. The returned channel will only close
+// when the consumer is completely shut down.
+func (c *Consumer) Partitions() <-chan PartitionConsumer { return c.partitions }
+
+// 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.consumer.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) {
+	if sub := c.subs.Fetch(msg.Topic, msg.Partition); sub != nil {
+		sub.MarkOffset(msg.Offset, 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) {
+	if sub := c.subs.Fetch(topic, partition); sub != nil {
+		sub.MarkOffset(offset, 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 {
+		if sub := c.subs.Fetch(tp.Topic, tp.Partition); sub != nil {
+			sub.MarkOffset(info.Offset, info.Metadata)
+		}
+		delete(s.offsets, tp)
+	}
+}
+
+// ResetOffsets 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.
+//
+// Difference between ResetOffset and MarkOffset is that it allows to rewind to an earlier offset
+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)
+	}
+}
+
+// ResetPartitionOffset marks an offset of the provided topic/partition as processed.
+// See ResetOffset for additional explanation.
+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)
+	}
+}
+
+// ResetOffsets marks stashed offsets as processed.
+// See ResetOffset for additional explanation.
+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)
+	}
+}
+
+// Subscriptions returns the consumed topics and partitions
+func (c *Consumer) Subscriptions() map[string][]int32 {
+	return c.subs.Info()
+}
+
+// CommitOffsets allows to manually commit previously marked offsets. By default there is no
+// need to call this function as the consumer will commit offsets automatically
+// using the Config.Consumer.Offsets.CommitInterval setting.
+//
+// Please be aware that calling this function during an internal rebalance cycle may return
+// broker errors (e.g. sarama.ErrUnknownMemberId or sarama.ErrIllegalGeneration).
+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
+}
+
+// Close safely closes the consumer and releases all resources
+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)
+
+		// drain
+		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)
+
+		// Check if close was requested
+		select {
+		case <-c.dying:
+			return
+		default:
+		}
+
+		// Start next consume cycle
+		c.nextTick()
+	}
+}
+
+func (c *Consumer) nextTick() {
+	// Remember previous subscriptions
+	var notification *Notification
+	if c.client.config.Group.Return.Notifications {
+		notification = newNotification(c.subs.Info())
+	}
+
+	// Refresh coordinator
+	if err := c.refreshCoordinator(); err != nil {
+		c.rebalanceError(err, nil)
+		return
+	}
+
+	// Release subscriptions
+	if err := c.release(); err != nil {
+		c.rebalanceError(err, nil)
+		return
+	}
+
+	// Issue rebalance start notification
+	if c.client.config.Group.Return.Notifications {
+		c.handleNotification(notification)
+	}
+
+	// Rebalance, fetch new subscriptions
+	subs, err := c.rebalance()
+	if err != nil {
+		c.rebalanceError(err, notification)
+		return
+	}
+
+	// Coordinate loops, make sure everything is
+	// stopped on exit
+	tomb := newLoopTomb()
+	defer tomb.Close()
+
+	// Start the heartbeat
+	tomb.Go(c.hbLoop)
+
+	// Subscribe to topic/partitions
+	if err := c.subscribe(tomb, subs); err != nil {
+		c.rebalanceError(err, notification)
+		return
+	}
+
+	// Update/issue notification with new claims
+	if c.client.config.Group.Return.Notifications {
+		notification = notification.success(subs)
+		c.handleNotification(notification)
+	}
+
+	// Start topic watcher loop
+	tomb.Go(c.twLoop)
+
+	// Start consuming and committing offsets
+	tomb.Go(c.cmLoop)
+	atomic.StoreInt32(&c.consuming, 1)
+
+	// Wait for signals
+	select {
+	case <-tomb.Dying():
+	case <-c.dying:
+	}
+}
+
+// heartbeat loop, triggered by the mainLoop
+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
+		}
+	}
+}
+
+// topic watcher loop, triggered by the mainLoop
+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
+		}
+	}
+}
+
+// commit loop, triggered by the mainLoop
+func (c *Consumer) cmLoop(stopped <-chan none) {
+	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: err})
+				return
+			}
+		case <-stopped:
+			return
+		case <-c.dying:
+			return
+		}
+	}
+}
+
+func (c *Consumer) rebalanceError(err error, n *Notification) {
+	if n != nil {
+		n.Type = RebalanceError
+		c.handleNotification(n)
+	}
+
+	switch err {
+	case sarama.ErrRebalanceInProgress:
+	default:
+		c.handleError(&Error{Ctx: "rebalance", error: err})
+	}
+
+	select {
+	case <-c.dying:
+	case <-time.After(c.client.config.Metadata.Retry.Backoff):
+	}
+}
+
+func (c *Consumer) handleNotification(n *Notification) {
+	if c.client.config.Group.Return.Notifications {
+		select {
+		case c.notifications <- n:
+		case <-c.dying:
+			return
+		}
+	}
+}
+
+func (c *Consumer) handleError(e *Error) {
+	if c.client.config.Consumer.Return.Errors {
+		select {
+		case c.errors <- e:
+		case <-c.dying:
+			return
+		}
+	} else {
+		sarama.Logger.Printf("%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
+	timeout := time.NewTimer(c.client.config.Group.Offsets.Synchronization.DwellTime)
+	defer timeout.Stop()
+
+	select {
+	case <-c.dying:
+	case <-timeout.C:
+	}
+
+	// Commit offsets, continue on errors
+	if e := c.commitOffsetsWithRetry(c.client.config.Group.Offsets.Retry.Max); e != nil {
+		err = 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 err
+	}
+
+	memberID, generationID := c.membership()
+	resp, err := broker.Heartbeat(&sarama.HeartbeatRequest{
+		GroupId:      c.groupID,
+		MemberId:     memberID,
+		GenerationId: generationID,
+	})
+	if err != nil {
+		c.closeCoordinator(broker, err)
+		return err
+	}
+	return resp.Err
+}
+
+// Performs a rebalance, part of the mainLoop()
+func (c *Consumer) rebalance() (map[string][]int32, error) {
+	memberID, _ := c.membership()
+	sarama.Logger.Printf("cluster/consumer %s rebalance\n", memberID)
+
+	allTopics, err := c.client.Topics()
+	if err != nil {
+		return nil, err
+	}
+	c.extraTopics = c.selectExtraTopics(allTopics)
+	sort.Strings(c.extraTopics)
+
+	// Re-join consumer group
+	strategy, err := c.joinGroup()
+	switch {
+	case err == sarama.ErrUnknownMemberId:
+		c.membershipMu.Lock()
+		c.memberID = ""
+		c.membershipMu.Unlock()
+		return nil, err
+	case err != nil:
+		return nil, err
+	}
+
+	// Sync consumer group state, fetch subscriptions
+	subs, err := c.syncGroup(strategy)
+	switch {
+	case err == sarama.ErrRebalanceInProgress:
+		return nil, err
+	case err != nil:
+		_ = c.leaveGroup()
+		return nil, err
+	}
+	return subs, nil
+}
+
+// Performs the subscription, part of the mainLoop()
+func (c *Consumer) subscribe(tomb *loopTomb, subs map[string][]int32) error {
+	// fetch offsets
+	offsets, err := c.fetchOffsets(subs)
+	if err != nil {
+		_ = c.leaveGroup()
+		return 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]
+			go func(topic string, partition int32) {
+				if e := c.createConsumer(tomb, topic, partition, info); e != nil {
+					mu.Lock()
+					err = e
+					mu.Unlock()
+				}
+				wg.Done()
+			}(topic, partition)
+		}
+	}
+	wg.Wait()
+
+	if err != nil {
+		_ = c.release()
+		_ = c.leaveGroup()
+	}
+	return err
+}
+
+// --------------------------------------------------------------------
+
+// Send a request to the broker to join group on rebalance()
+func (c *Consumer) joinGroup() (*balancer, error) {
+	memberID, _ := c.membership()
+	req := &sarama.JoinGroupRequest{
+		GroupId:        c.groupID,
+		MemberId:       memberID,
+		SessionTimeout: int32(c.client.config.Group.Session.Timeout / time.Millisecond),
+		ProtocolType:   "consumer",
+	}
+
+	meta := &sarama.ConsumerGroupMemberMetadata{
+		Version:  1,
+		Topics:   append(c.coreTopics, c.extraTopics...),
+		UserData: c.client.config.Group.Member.UserData,
+	}
+	err := req.AddGroupProtocolMetadata(string(StrategyRange), meta)
+	if err != nil {
+		return nil, err
+	}
+	err = req.AddGroupProtocolMetadata(string(StrategyRoundRobin), meta)
+	if err != nil {
+		return nil, err
+	}
+
+	broker, err := c.client.Coordinator(c.groupID)
+	if err != nil {
+		c.closeCoordinator(broker, err)
+		return nil, err
+	}
+
+	resp, err := broker.JoinGroup(req)
+	if err != nil {
+		c.closeCoordinator(broker, err)
+		return nil, err
+	} else if resp.Err != sarama.ErrNoError {
+		c.closeCoordinator(broker, resp.Err)
+		return nil, resp.Err
+	}
+
+	var strategy *balancer
+	if resp.LeaderId == resp.MemberId {
+		members, err := resp.GetMembers()
+		if err != nil {
+			return nil, err
+		}
+
+		strategy, err = newBalancerFromMeta(c.client, members)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	c.membershipMu.Lock()
+	c.memberID = resp.MemberId
+	c.generationID = resp.GenerationId
+	c.membershipMu.Unlock()
+
+	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) {
+	memberID, generationID := c.membership()
+	req := &sarama.SyncGroupRequest{
+		GroupId:      c.groupID,
+		MemberId:     memberID,
+		GenerationId: generationID,
+	}
+
+	if strategy != nil {
+		for memberID, topics := range strategy.Perform(c.client.config.Group.PartitionStrategy) {
+			if err := req.AddGroupAssignmentMember(memberID, &sarama.ConsumerGroupMemberAssignment{
+				Topics: topics,
+			}); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	broker, err := c.client.Coordinator(c.groupID)
+	if err != nil {
+		c.closeCoordinator(broker, err)
+		return nil, err
+	}
+
+	resp, err := broker.SyncGroup(req)
+	if err != nil {
+		c.closeCoordinator(broker, err)
+		return nil, err
+	} else if resp.Err != sarama.ErrNoError {
+		c.closeCoordinator(broker, resp.Err)
+		return nil, 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, 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 := &sarama.OffsetFetchRequest{
+		Version:       1,
+		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)
+		}
+	}
+
+	broker, err := c.client.Coordinator(c.groupID)
+	if err != nil {
+		c.closeCoordinator(broker, err)
+		return nil, err
+	}
+
+	resp, err := broker.FetchOffset(req)
+	if err != nil {
+		c.closeCoordinator(broker, err)
+		return nil, err
+	}
+
+	for topic, partitions := range subs {
+		for _, partition := range partitions {
+			block := resp.GetBlock(topic, partition)
+			if block == nil {
+				return nil, sarama.ErrIncompleteResponse
+			}
+
+			if block.Err == sarama.ErrNoError {
+				offsets[topic][partition] = offsetInfo{Offset: block.Offset, Metadata: block.Metadata}
+			} else {
+				return nil, 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 err
+	}
+
+	memberID, _ := c.membership()
+	if _, err = broker.LeaveGroup(&sarama.LeaveGroupRequest{
+		GroupId:  c.groupID,
+		MemberId: memberID,
+	}); err != nil {
+		c.closeCoordinator(broker, err)
+	}
+	return err
+}
+
+// --------------------------------------------------------------------
+
+func (c *Consumer) createConsumer(tomb *loopTomb, topic string, partition int32, info offsetInfo) error {
+	memberID, _ := c.membership()
+	sarama.Logger.Printf("cluster/consumer %s consume %s/%d from %d\n", memberID, topic, partition, info.NextOffset(c.client.config.Consumer.Offsets.Initial))
+
+	// Create partitionConsumer
+	pc, err := newPartitionConsumer(c.consumer, topic, partition, info, c.client.config.Consumer.Offsets.Initial)
+	if err != nil {
+		return err
+	}
+
+	// Store in subscriptions
+	c.subs.Store(topic, partition, pc)
+
+	// Start partition consumer goroutine
+	tomb.Go(func(stopper <-chan none) {
+		if c.client.config.Group.Mode == ConsumerModePartitions {
+			pc.waitFor(stopper, c.errors)
+		} else {
+			pc.multiplex(stopper, c.messages, c.errors)
+		}
+	})
+
+	if c.client.config.Group.Mode == ConsumerModePartitions {
+		c.partitions <- pc
+	}
+	return nil
+}
+
+func (c *Consumer) commitOffsetsWithRetry(retries int) error {
+	err := c.CommitOffsets()
+	if err != nil && retries > 0 {
+		return c.commitOffsetsWithRetry(retries - 1)
+	}
+	return err
+}
+
+func (c *Consumer) closeCoordinator(broker *sarama.Broker, err error) {
+	if broker != nil {
+		_ = broker.Close()
+	}
+
+	switch 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) refreshCoordinator() error {
+	if err := c.refreshMetadata(); err != nil {
+		return err
+	}
+	return c.client.RefreshCoordinator(c.groupID)
+}
+
+func (c *Consumer) refreshMetadata() (err error) {
+	if c.client.config.Metadata.Full {
+		err = c.client.RefreshMetadata()
+	} else {
+		var topics []string
+		if topics, err = c.client.Topics(); err == nil && len(topics) != 0 {
+			err = c.client.RefreshMetadata(topics...)
+		}
+	}
+
+	// maybe we didn't have authorization to describe all topics
+	switch err {
+	case sarama.ErrTopicAuthorizationFailed:
+		err = c.client.RefreshMetadata(c.coreTopics...)
+	}
+	return
+}
+
+func (c *Consumer) membership() (memberID string, generationID int32) {
+	c.membershipMu.RLock()
+	memberID, generationID = c.memberID, c.generationID
+	c.membershipMu.RUnlock()
+	return
+}