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)
		} else {
			pc.multiplex(stopper, c.messages, c.errors)
		}
	})

	if c.client.config.Group.Mode == ConsumerModePartitions {
		select {
		case c.partitions <- pc:
		case <-c.dying:
			pc.Close()
		}
	}
	return nil
}
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)
		} else {
			pc.multiplex(stopper, c.messages, c.errors)
		}
	})

	if c.client.config.Group.Mode == ConsumerModePartitions {
		select {
		case c.partitions <- pc:
		case <-c.dying:
			pc.Close()
		}
	}
	return nil
}
// --------------------------------------------------------------------