func NewConsumer(client *Client, topic string, partition int32, group string, config *ConsumerConfig) (*Consumer, error) {
	if config == nil {
		config = new(ConsumerConfig)
	}
	if config.DefaultFetchSize < 0 {
		return nil, ConfigurationError("Invalid DefaultFetchSize")
	} else if config.DefaultFetchSize == 0 {
		config.DefaultFetchSize = 1024
	}
	if config.MinFetchSize < 0 {
		return nil, ConfigurationError("Invalid MinFetchSize")
	} else if config.MinFetchSize == 0 {
		config.MinFetchSize = 1
	}
	if config.MaxMessageSize < 0 {
		return nil, ConfigurationError("Invalid MaxMessageSize")
	}
	if config.MaxWaitTime <= 0 {
		return nil, ConfigurationError("Invalid MaxWaitTime")
	} else if config.MaxWaitTime < 100 {
		Logger.Println("ConsumerConfig.MaxWaitTime is very low, which can cause high CPU and network usage. See documentation for details.")
	}
	if config.EventBufferSize < 0 {
		return nil, ConfigurationError("Invalid EventBufferSize")
	}
	if topic == "" {
		return nil, ConfigurationError("Empty topic")
	}
	broker, err := client.Leader(topic, partition)
	if err != nil {
		return nil, err
	}
	c := &Consumer{
		client:    client,
		topic:     topic,
		partition: partition,
		group:     group,
		config:    *config,
		broker:    broker,
		stopper:   make(chan bool),
		done:      make(chan bool),
		events:    make(chan *ConsumerEvent, config.EventBufferSize),
	}
	switch config.OffsetMethod {
	case OffsetMethodManual:
		if config.OffsetValue < 0 {
			return nil, ConfigurationError("OffsetValue cannot be < 0 when OffsetMethod is MANUAL")
		}
		c.offset = config.OffsetValue
	case OffsetMethodNewest:
		c.offset, err = c.getOffset(LatestOffsets, true)
		if err != nil {
			return nil, err
		}
	case OffsetMethodOldest:
		c.offset, err = c.getOffset(EarliestOffset, true)
		if err != nil {
			return nil, err
		}
	default:
		return nil, ConfigurationError("Invalid OffsetMethod")
	}
	go withRecover(c.fetchMessages)
	return c, nil
}
NewConsumer creates a new consumer attached to the given client. It will read messages from the given topic and partition, as
part of the named consumer group.
package sarama
type OffsetMethod int
const (
	OffsetMethodManual OffsetMethod = iota
	OffsetMethodNewest
	OffsetMethodOldest
)
type ConsumerConfig struct {
	DefaultFetchSize int32
	MinFetchSize int32
	MaxMessageSize int32
	MaxWaitTime int32
	OffsetMethod OffsetMethod
	OffsetValue int64
	EventBufferSize int
}
type ConsumerEvent struct {
	Key, Value []byte
	Offset     int64
	Err        error
}
type Consumer struct {
	client *Client
	topic     string
	partition int32
	group     string
	config    ConsumerConfig
	offset        int64
	broker        *Broker
	stopper, done chan bool
	events        chan *ConsumerEvent
}