consumer.go
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 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
}
// 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.