func (cons *Kafka) readFromPartition(partitionID int32) {
	cons.AddWorker()
	defer cons.WorkerDone()
	partCons := cons.startConsumerForPartition(partitionID)
	spin := tsync.NewSpinner(tsync.SpinPriorityLow)
	for !cons.client.Closed() {
		select {
		case event := <-partCons.Messages():
			if event == nil || cons.offsets == nil || cons.offsets[partitionID] == nil {
				cons.Logger.Errorf("Kafka consumer failed to store offset. Trace : event : %+v, cons.partCons: %+v, partitionID: %d\n",
					event, cons.offsets, partitionID)
				partCons.Close()
				partCons = cons.startConsumerForPartition(partitionID)
				continue
			}
			atomic.StoreInt64(cons.offsets[partitionID], event.Offset)
			cons.enqueueEvent(event)
		case err := <-partCons.Errors():
			cons.Logger.Error("Kafka consumer error:", err)
			if !cons.client.Closed() {
				partCons.Close()
			}
			partCons = cons.startConsumerForPartition(partitionID)
		default:
			spin.Yield()
		}
	}
}
Main fetch loop for kafka events
package consumer
import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"io/ioutil"
	"os"
	"path"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"gollum/core"
	kafka "github.com/Shopify/sarama"
	cluster "github.com/bsm/sarama-cluster"
	"github.com/trivago/tgo/tsync"
)
const (
	kafkaOffsetNewest = "newest"
	kafkaOffsetOldest = "oldest"
)
type Kafka struct {
	core.SimpleConsumer `gollumdoc:"embed_type"`
	client              kafka.Client
	consumer            kafka.Consumer
	config              *kafka.Config
	groupClient         *cluster.Client
	groupConfig         *cluster.Config
	offsets             map[int32]*int64
	servers             []string `config:"Servers"`
	topic               string   `config:"Topic" default:"default"`
	group               string   `config:"GroupId"`
	offsetFile          string   `config:"OffsetFile"`
	defaultOffset       int64
	persistTimeout      time.Duration `config:"PresistTimoutMs" default:"5000" metric:"ms"`
	folderPermissions   os.FileMode   `config:"FolderPermissions" default:"0755"`
	MaxPartitionID      int32
	orderedRead         bool `config:"Ordered"`
	hasToSetMetadata    bool `config:"SetMetadata" default:"false"`
}
func init() {
	core.TypeRegistry.Register(Kafka{})
}
func (cons *Kafka) Configure(conf core.PluginConfigReader) {
	cons.offsets = make(map[int32]*int64)
	cons.MaxPartitionID = 0
	cons.config = kafka.NewConfig()
	cons.config.ClientID = conf.GetString("ClientId", "gollum")
	cons.config.ChannelBufferSize = int(conf.GetInt("MessageBufferCount", 8192))
	switch ver := conf.GetString("Version", "0.8.2"); ver {
	case "0.8.2.0":
		cons.config.Version = kafka.V0_8_2_0
	case "0.8.2.1":
		cons.config.Version = kafka.V0_8_2_1
	case "0.8", "0.8.2", "0.8.2.2":
		cons.config.Version = kafka.V0_8_2_2
	case "0.9.0", "0.9.0.0":
		cons.config.Version = kafka.V0_9_0_0
	case "0.9", "0.9.0.1":
		cons.config.Version = kafka.V0_9_0_1
	case "0.10", "0.10.0", "0.10.0.0":
		cons.config.Version = kafka.V0_10_0_0
	case "0.10.0.1":
		cons.config.Version = kafka.V0_10_0_1
	case "0.10.1", "0.10.1.0":
		cons.config.Version = kafka.V0_10_1_0
	case "0.10.2", "0.10.2.0":
		cons.config.Version = kafka.V0_10_2_0
	case "0.11", "0.11.0", "0.11.0.0":
		cons.config.Version = kafka.V0_11_0_0
	case "1", "1.0", "1.0.0", "1.0.0.0":
		cons.config.Version = kafka.V1_0_0_0
	default:
		cons.Logger.Warningf("Unknown kafka version given: %s. Falling back to 0.8.2", ver)
		cons.config.Version = kafka.V0_8_2_2
	}
	cons.config.Net.MaxOpenRequests = int(conf.GetInt("MaxOpenRequests", 5))
	cons.config.Net.DialTimeout = time.Duration(conf.GetInt("ServerTimeoutSec", 30)) * time.Second
	cons.config.Net.ReadTimeout = cons.config.Net.DialTimeout
	cons.config.Net.WriteTimeout = cons.config.Net.DialTimeout
	cons.config.Net.TLS.Enable = conf.GetBool("TlsEnable", false)
	if cons.config.Net.TLS.Enable {
		cons.config.Net.TLS.Config = &tls.Config{}
		keyFile := conf.GetString("TlsKeyLocation", "")
		certFile := conf.GetString("TlsCertificateLocation", "")
		if keyFile != "" && certFile != "" {
			cert, err := tls.LoadX509KeyPair(certFile, keyFile)
			if conf.Errors.Push(err) {
				return
			}
			cons.config.Net.TLS.Config.Certificates = []tls.Certificate{cert}
		} else if certFile == "" {
			conf.Errors.Pushf("Cannot specify TlsKeyLocation without TlsCertificateLocation")
			return
		} else if keyFile == "" {
			conf.Errors.Pushf("Cannot specify TlsCertificateLocation without TlsKeyLocation")
			return
		}
		caFile := conf.GetString("TlsCaLocation", "")
		if caFile == "" {
			conf.Errors.Pushf("TlsEnable is set to true, but no TlsCaLocation was specified")
			return
		}
		caCert, err := ioutil.ReadFile(caFile)
		if conf.Errors.Push(err) {
			return
		}
		caCertPool := x509.NewCertPool()
		caCertPool.AppendCertsFromPEM(caCert)
		cons.config.Net.TLS.Config.RootCAs = caCertPool
		serverName := conf.GetString("TlsServerName", "")
		if serverName != "" {
			cons.config.Net.TLS.Config.ServerName = serverName
		}
		cons.config.Net.TLS.Config.InsecureSkipVerify = conf.GetBool("TlsInsecureSkipVerify", false)
	}
	cons.config.Net.SASL.Enable = conf.GetBool("SaslEnable", false)
	if cons.config.Net.SASL.Enable {
		cons.config.Net.SASL.User = conf.GetString("SaslUser", "gollum")
		cons.config.Net.SASL.Password = conf.GetString("SaslPassword", "")
	}
	cons.config.Metadata.Retry.Max = int(conf.GetInt("ElectRetries", 3))
	cons.config.Metadata.Retry.Backoff = time.Duration(conf.GetInt("ElectTimeoutMs", 250)) * time.Millisecond
	cons.config.Metadata.RefreshFrequency = time.Duration(conf.GetInt("MetadataRefreshMs", 10000)) * time.Millisecond
	cons.config.Consumer.Fetch.Min = int32(conf.GetInt("MinFetchSizeByte", 1))
	cons.config.Consumer.Fetch.Max = int32(conf.GetInt("MaxFetchSizeByte", 0))
	cons.config.Consumer.Fetch.Default = int32(conf.GetInt("DefaultFetchSizeByte", 32768))
	cons.config.Consumer.MaxWaitTime = time.Duration(conf.GetInt("FetchTimeoutMs", 250)) * time.Millisecond
	if cons.group != "" {
		cons.offsetFile = ""  
		switch cons.config.Version {
		case kafka.V0_8_2_0, kafka.V0_8_2_1, kafka.V0_8_2_2:
			cons.Logger.Warning("Invalid kafka version 0.8.x given, minimum is 0.9 for consumer groups, defaulting to 0.9.0.1")
			cons.config.Version = kafka.V0_9_0_1
		}
		cons.groupConfig = cluster.NewConfig()
		cons.groupConfig.Config = *cons.config
	}
	offsetValue := strings.ToLower(conf.GetString("DefaultOffset", kafkaOffsetNewest))
	switch offsetValue {
	case kafkaOffsetNewest:
		cons.defaultOffset = kafka.OffsetNewest
	case kafkaOffsetOldest:
		cons.defaultOffset = kafka.OffsetOldest
	default:
		cons.defaultOffset, _ = strconv.ParseInt(offsetValue, 10, 64)
	}
	if cons.offsetFile != "" {
		fileContents, err := ioutil.ReadFile(cons.offsetFile)
		if err != nil {
			cons.Logger.Warningf("Failed to open kafka offset file: %s", err.Error())
		} else {
			encodedOffsets := make(map[string]int64)
			err = json.Unmarshal(fileContents, &encodedOffsets)
			if conf.Errors.Push(err) {
				return
			}
			for k, v := range encodedOffsets {
				id, err := strconv.Atoi(k)
				if conf.Errors.Push(err) {
					return
				}
				startOffset := v
				cons.offsets[int32(id)] = &startOffset
			}
		}
	}
	kafka.Logger = cons.Logger.WithField("Scope", "Sarama")
}
func (cons *Kafka) restartGroup() {
	time.Sleep(cons.persistTimeout)
	cons.readFromGroup()
}
func (cons *Kafka) readFromGroup() {
	consumer, err := cluster.NewConsumerFromClient(cons.groupClient, cons.group, []string{cons.topic})
	if err != nil {
		defer cons.restartGroup()
		cons.Logger.Errorf("Restarting kafka consumer (%s:%s) - %s", cons.topic, cons.group, err.Error())
		return  
	}
	cons.AddWorker()
	defer func() {
		if !cons.groupClient.Closed() {
			consumer.Close()
		}
		cons.WorkerDone()
	}()
	spin := tsync.NewSpinner(tsync.SpinPriorityLow)
	for !cons.groupClient.Closed() {
		select {
		case event, ok := <-consumer.Messages():
			if ok {
				cons.enqueueEvent(event)
				consumer.MarkOffset(event, "")
			}
		case err := <-consumer.Errors():
			defer cons.restartGroup()
			cons.Logger.Error("Kafka consumer error:", err)
			return  
		default:
			spin.Yield()
		}
	}
}
func (cons *Kafka) startConsumerForPartition(partitionID int32) kafka.PartitionConsumer {
	for !cons.client.Closed() {
		startOffset := atomic.LoadInt64(cons.offsets[partitionID])
		consumer, err := cons.consumer.ConsumePartition(cons.topic, partitionID, startOffset)
		if err == nil {
			return consumer  
		}
		cons.Logger.Errorf("Failed to start kafka consumer (%s:%d) - %s", cons.topic, startOffset, err.Error())
		if err == kafka.ErrOffsetOutOfRange {
			startOffset = kafka.OffsetOldest
			atomic.StoreInt64(cons.offsets[partitionID], startOffset)
		} else {
			time.Sleep(cons.persistTimeout)
		}
	}
	return nil
}