func (be *brokerExecutor) runExecutor() {
	var lastErr error
	var lastErrTime time.Time
	for requestBatch := range be.requestBatchesCh {
		if time.Since(lastErrTime) < be.cfg.Consumer.RetryBackoff {
			for _, fr := range requestBatch {
				fr.ReplyToCh <- fetchRs{nil, lastErr}
			}
			continue
		}
		kafkaFetchRq := &sarama.FetchRequest{
			MinBytes:    1,
			MaxWaitTime: int32(be.cfg.Consumer.FetchMaxWait / time.Millisecond),
		}
		if be.cfg.Kafka.Version.IsAtLeast(sarama.V0_10_0_0) {
			kafkaFetchRq.Version = 2
		}
		if be.cfg.Kafka.Version.IsAtLeast(sarama.V0_10_1_0) {
			kafkaFetchRq.Version = 3
			kafkaFetchRq.MaxBytes = sarama.MaxResponseSize
		}
		if be.cfg.Kafka.Version.IsAtLeast(sarama.V0_11_0_0) {
			kafkaFetchRq.Version = 4
			kafkaFetchRq.Isolation = sarama.ReadUncommitted
		}
		for _, fr := range requestBatch {
			kafkaFetchRq.AddBlock(fr.Topic, fr.Partition, fr.Offset, int32(be.cfg.Consumer.FetchMaxBytes))
		}
		var kafkaFetchRs *sarama.FetchResponse
		kafkaFetchRs, lastErr = be.conn.Fetch(kafkaFetchRq)
		if lastErr != nil {
			lastErrTime = time.Now()
			be.conn.Close()
			be.execActDesc.Log().WithError(lastErr).Info("Connection reset")
		}
		for _, fr := range requestBatch {
			fr.ReplyToCh <- fetchRs{kafkaFetchRs, lastErr}
		}
	}
}
runExecutor executes fetch request aggregated into batches by the aggregator
goroutine of the broker executor.
package msgfetcher
import (
	"sync"
	"time"
	"github.com/Shopify/sarama"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/config"
	"github.com/mailgun/kafka-pixy/consumer"
	"github.com/mailgun/kafka-pixy/mapper"
	"github.com/mailgun/kafka-pixy/none"
	"github.com/pkg/errors"
)
type Factory interface {
	Spawn(parentActDesc *actor.Descriptor, topic string, partition int32, offset int64) (T, int64, error)
	Stop()
}
type T interface {
	Messages() <-chan consumer.Message
	Stop()
}
var (
	testReportErrors bool
	errMessageTooLarge    = errors.New("message is larger than consumer.fetch_max_bytes")
	errIncompleteResponse = errors.New("response did not contain the expected topic/partition block")
)
type factory struct {
	actDesc  *actor.Descriptor
	cfg      *config.Proxy
	kafkaClt sarama.Client
	mapper   *mapper.T
	childrenMu sync.Mutex
	children   map[instanceID]*msgFetcher
}
type instanceID struct {
	topic     string
	partition int32
}
func SpawnFactory(parentActDesc *actor.Descriptor, cfg *config.Proxy, kafkaClt sarama.Client) Factory {
	f := &factory{
		actDesc:  parentActDesc.NewChild("msg_fetcher_f"),
		cfg:      cfg,
		kafkaClt: kafkaClt,
		children: make(map[instanceID]*msgFetcher),
	}
	f.mapper = mapper.Spawn(f.actDesc, cfg, f)
	return f
}
func (f *factory) Spawn(parentActDesc *actor.Descriptor, topic string, partition int32, offset int64) (T, int64, error) {
	realOffset, err := f.chooseStartingOffset(topic, partition, offset)
	if err != nil {
		return nil, sarama.OffsetNewest, err
	}
	f.childrenMu.Lock()
	defer f.childrenMu.Unlock()
	id := instanceID{topic, partition}
	if _, ok := f.children[id]; ok {
		return nil, sarama.OffsetNewest, sarama.ConfigurationError("That topic/partition is already being consumed")
	}
	actDesc := parentActDesc.NewChild("msg_fetcher")
	actDesc.AddLogField("kafka.topic", topic)
	actDesc.AddLogField("kafka.partition", partition)
	mf := &msgFetcher{
		actDesc:      actDesc,
		f:            f,
		id:           id,
		assignmentCh: make(chan mapper.Executor, 1),
		messagesCh:   make(chan consumer.Message, f.cfg.Consumer.ChannelBufferSize),
		stopCh:       make(chan none.T, 1),
		offset:       realOffset,
	}
	if testReportErrors {
		mf.errorsCh = make(chan error, f.cfg.Consumer.ChannelBufferSize)
	}
	f.children[id] = mf
	actor.Spawn(mf.actDesc, &mf.wg, mf.run)
	return mf, realOffset, nil
}
func (f *factory) Stop() {
	f.mapper.Stop()
}
func (f *factory) ResolveBroker(worker mapper.Worker) (*sarama.Broker, error) {
	ms := worker.(*msgFetcher)
	if err := f.kafkaClt.RefreshMetadata(ms.id.topic); err != nil {
		return nil, err
	}
	return f.kafkaClt.Leader(ms.id.topic, ms.id.partition)
}
func (f *factory) SpawnExecutor(brokerConn *sarama.Broker) mapper.Executor {
	be := &brokerExecutor{
		aggrActDesc:      f.actDesc.NewChild("broker", brokerConn.ID(), "aggr"),
		execActDesc:      f.actDesc.NewChild("broker", brokerConn.ID(), "exec"),
		cfg:              f.cfg,
		conn:             brokerConn,
		requestsCh:       make(chan fetchRq),
		requestBatchesCh: make(chan []fetchRq),
	}
	actor.Spawn(be.aggrActDesc, &be.wg, be.runAggregator)
	actor.Spawn(be.execActDesc, &be.wg, be.runExecutor)
	return be
}
func (f *factory) chooseStartingOffset(topic string, partition int32, offset int64) (int64, error) {
	newestOffset, err := f.kafkaClt.GetOffset(topic, partition, sarama.OffsetNewest)
	if err != nil {
		return 0, err
	}
	oldestOffset, err := f.kafkaClt.GetOffset(topic, partition, sarama.OffsetOldest)
	if err != nil {
		return 0, err
	}
	switch {
	case offset == sarama.OffsetNewest || offset > newestOffset:
		return newestOffset, nil
	case offset == sarama.OffsetOldest || offset < oldestOffset:
		return oldestOffset, nil
	default:
		return offset, nil
	}
}
func (f *factory) onMsgIStreamSpawned(mf *msgFetcher) {
	f.mapper.OnWorkerSpawned(mf)
}
func (f *factory) onMsgIStreamStopped(mf *msgFetcher) {
	f.childrenMu.Lock()
	delete(f.children, mf.id)
	f.childrenMu.Unlock()
	f.mapper.OnWorkerStopped(mf)
}
type msgFetcher struct {
	actDesc               *actor.Descriptor
	f                     *factory
	id                    instanceID
	offset                int64
	assignmentCh          chan mapper.Executor
	messagesCh            chan consumer.Message
	errorsCh              chan error
	brokerRequestCh       chan<- fetchRq
	nilOrBrokerRequestsCh chan<- fetchRq
	stopCh                chan none.T
	wg                    sync.WaitGroup
}
func (mf *msgFetcher) Messages() <-chan consumer.Message {
	return mf.messagesCh
}
func (mf *msgFetcher) Stop() {
	close(mf.stopCh)
	mf.wg.Wait()
}
func (mf *msgFetcher) Assignment() chan<- mapper.Executor {
	return mf.assignmentCh
}
func (mf *msgFetcher) run() {
	defer close(mf.messagesCh)
	if mf.errorsCh != nil {
		defer close(mf.errorsCh)
	}
	mf.f.onMsgIStreamSpawned(mf)
	defer mf.f.onMsgIStreamStopped(mf)
	var (
		fetchResultCh       = make(chan fetchRs, 1)
		nilOrFetchResultsCh <-chan fetchRs
		nilOrMessagesCh     chan<- consumer.Message
		fetchedMessages     []consumer.Message
		err                 error
		currMessage         consumer.Message
		currMessageIdx      int
	)
	for {
		select {
		case bw := <-mf.assignmentCh:
			mf.actDesc.Log().Infof("Assigned executor: %s", bw)
			be := bw.(*brokerExecutor)
			mf.brokerRequestCh = be.requestsCh
			if nilOrFetchResultsCh == nil && nilOrMessagesCh == nil {
				mf.nilOrBrokerRequestsCh = mf.brokerRequestCh
			}
		case mf.nilOrBrokerRequestsCh <- fetchRq{mf.id.topic, mf.id.partition, mf.offset, fetchResultCh}:
			mf.nilOrBrokerRequestsCh = nil
			nilOrFetchResultsCh = fetchResultCh
		case fetchRs := <-nilOrFetchResultsCh:
			nilOrFetchResultsCh = nil
			if fetchedMessages, err = mf.parseFetchResponse(fetchRs); err != nil {
				mf.reportError(err)
				if err == sarama.ErrOffsetOutOfRange {
					mf.actDesc.Log().WithError(err).Error("Fatal request failure")
					return
				}
				mf.actDesc.Log().WithError(err).Error("Request failed")
				mf.brokerRequestCh = nil
				mf.f.mapper.TriggerReassign(mf)
				continue
			}
			if len(fetchedMessages) == 0 {
				mf.nilOrBrokerRequestsCh = mf.brokerRequestCh
				continue
			}
			currMessageIdx = 0
			currMessage = fetchedMessages[currMessageIdx]
			nilOrMessagesCh = mf.messagesCh
		case nilOrMessagesCh <- currMessage:
			mf.offset = currMessage.Offset + 1
			currMessageIdx++
			if currMessageIdx < len(fetchedMessages) {
				currMessage = fetchedMessages[currMessageIdx]
				continue
			}
			nilOrMessagesCh = nil
			mf.nilOrBrokerRequestsCh = mf.brokerRequestCh
		case <-mf.stopCh:
			return
		}
	}
}
func (mf *msgFetcher) parseFetchResponse(fetchRs fetchRs) ([]consumer.Message, error) {
	if fetchRs.connErr != nil {
		return nil, fetchRs.connErr
	}
	kafkaFetchRs := fetchRs.kafkaRs
	if kafkaFetchRs == nil {
		return nil, errIncompleteResponse
	}
	fetchRsBlock := kafkaFetchRs.GetBlock(mf.id.topic, mf.id.partition)
	if fetchRsBlock == nil {
		return nil, errIncompleteResponse
	}
	if fetchRsBlock.Err != sarama.ErrNoError {
		return nil, fetchRsBlock.Err
	}
	highWaterMarkOffset := fetchRsBlock.HighWaterMarkOffset
	var fetchedMessages []consumer.Message
	for _, recordsSet := range fetchRsBlock.RecordsSet {
		recordBatch := recordsSet.RecordBatch
		if recordBatch != nil {
			fetchedMessages = append(fetchedMessages, mf.parseRecordBatch(recordBatch, highWaterMarkOffset)...)
			continue
		}
		messageSet := recordsSet.MsgSet
		if messageSet != nil {
			fetchedMessages = append(fetchedMessages, mf.parseMessageSet(messageSet, highWaterMarkOffset)...)
		}
	}
	return fetchedMessages, nil
}
func (mf *msgFetcher) parseMessageSet(messageSet *sarama.MessageSet, highWaterMarkOffset int64) []consumer.Message {
	if len(messageSet.Messages) == 0 && messageSet.PartialTrailingMessage {
		mf.actDesc.Log().Errorf("Skipped partial trailing message in set: offset=%d", mf.offset)
		mf.reportError(errMessageTooLarge)
		return nil
	}
	var fetchedMessages []consumer.Message
	for _, msgBlock := range messageSet.Messages {
		lastMsgIdx := len(msgBlock.Messages()) - 1
		baseOffset := msgBlock.Offset - msgBlock.Messages()[lastMsgIdx].Offset
		for _, msg := range msgBlock.Messages() {
			offset := msg.Offset
			if msg.Msg.Version >= 1 {
				offset += baseOffset
			}
			if offset < mf.offset {
				continue
			}
			consumerMsg := consumer.Message{
				ConsumerMessage: sarama.ConsumerMessage{
					Topic:     mf.id.topic,
					Partition: mf.id.partition,
					Key:       msg.Msg.Key,
					Value:     msg.Msg.Value,
					Offset:    offset,
					Timestamp: msg.Msg.Timestamp,
				},
				HighWaterMark: highWaterMarkOffset,
			}
			fetchedMessages = append(fetchedMessages, consumerMsg)
		}
	}
	return fetchedMessages
}
func (mf *msgFetcher) parseRecordBatch(recordBatch *sarama.RecordBatch, highWaterMarkOffset int64) []consumer.Message {
	if recordBatch.Control {
		mf.actDesc.Log().Warn("Control record batch ignored")
		return nil
	}
	if len(recordBatch.Records) == 0 && recordBatch.PartialTrailingRecord {
		mf.actDesc.Log().Errorf("Skipped partial trailing record in batch: offset=%d", mf.offset)
		mf.reportError(errMessageTooLarge)
		return nil
	}
	var fetchedMessages []consumer.Message
	for _, record := range recordBatch.Records {
		offset := recordBatch.FirstOffset + record.OffsetDelta
		if offset < mf.offset {
			continue
		}
		consumerMsg := consumer.Message{
			ConsumerMessage: sarama.ConsumerMessage{
				Topic:     mf.id.topic,
				Partition: mf.id.partition,
				Key:       record.Key,
				Value:     record.Value,
				Headers:   record.Headers,
				Offset:    offset,
				Timestamp: recordBatch.FirstTimestamp.Add(record.TimestampDelta),
			},
			HighWaterMark: highWaterMarkOffset,
		}
		fetchedMessages = append(fetchedMessages, consumerMsg)
	}
	return fetchedMessages
}
func (mf *msgFetcher) reportError(err error) {
	if mf.errorsCh == nil {
		return
	}
	select {
	case mf.errorsCh <- err:
	default:
	}
}
func (mf *msgFetcher) String() string {
	return mf.actDesc.String()
}
type brokerExecutor struct {
	aggrActDesc      *actor.Descriptor
	execActDesc      *actor.Descriptor
	cfg              *config.Proxy
	conn             *sarama.Broker
	requestsCh       chan fetchRq
	requestBatchesCh chan []fetchRq
	wg               sync.WaitGroup
}
type fetchRq struct {
	Topic     string
	Partition int32
	Offset    int64
	ReplyToCh chan<- fetchRs
}
type fetchRs struct {
	kafkaRs *sarama.FetchResponse
	connErr error
}
func (be *brokerExecutor) BrokerConn() *sarama.Broker {
	return be.conn
}
func (be *brokerExecutor) Stop() {
	close(be.requestsCh)
	be.wg.Wait()
}
func (be *brokerExecutor) runAggregator() {
	defer close(be.requestBatchesCh)
	var nilOrRequestBatchesCh chan<- []fetchRq
	var requestBatch []fetchRq
	for {
		select {
		case fr, ok := <-be.requestsCh:
			if !ok {
				return
			}
			requestBatch = append(requestBatch, fr)
			nilOrRequestBatchesCh = be.requestBatchesCh
		case nilOrRequestBatchesCh <- requestBatch:
			requestBatch = nil
			nilOrRequestBatchesCh = nil
		}
	}
}