func (c *claim) Terminate() bool {
	return c.teardown(false)
}
Terminate will invoke commit offsets, terminate the claim, but does NOT release the partition.
Does not return until the message pump has exited and termination has finished.
package marshal
import (
	"fmt"
	"math/rand"
	"sort"
	"sync"
	"sync/atomic"
	"time"
	"github.com/zorkian/kafka"
	"github.com/zorkian/kafka/proto"
	"github.com/jpillora/backoff"
)
type int64slice []int64
func (a int64slice) Len() int           { return len(a) }
func (a int64slice) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a int64slice) Less(i, j int) bool { return a[i] < a[j] }
type claim struct {
	topic  string
	partID int
	lock            *sync.RWMutex
	messagesLock    *sync.Mutex
	offsets         PartitionOffsets
	marshal         *Marshaler
	consumer        *Consumer
	rand            *rand.Rand
	terminated      *int32
	beatCounter     int32
	lastHeartbeat   int64
	lastMessageTime time.Time
	options         ConsumerOptions
	kafkaConsumer   kafka.Consumer
	messages        chan *Message
	stopChan        chan struct{}
	doneChan        chan struct{}
	tracking            map[int64]bool
	outstandingMessages int
	cyclesBehind int
	offsetCurrentHistory [10]int64
	offsetLatestHistory  [10]int64
}
func newClaim(topic string, partID int, marshal *Marshaler, consumer *Consumer,
	messages chan *Message, options ConsumerOptions) *claim {
	offsets, err := marshal.GetPartitionOffsets(topic, partID)
	if err != nil {
		log.Errorf("[%s:%d] failed to get offsets: %s", topic, partID, err)
		return nil
	}
	log.Debugf("[%s:%d] consumer offsets: early = %d, cur/comm = %d/%d, late = %d",
		topic, partID, offsets.Earliest, offsets.Current, offsets.Committed, offsets.Latest)
	if offsets.Current > 0 {
	} else if offsets.Committed > 0 {
		log.Infof("[%s:%d] no Marshal offset found, using committed offset %d",
			topic, partID, offsets.Committed)
		offsets.Current = offsets.Committed
	} else {
		log.Infof("[%s:%d] no Marshal or committed offset found, using earliest offset %d",
			topic, partID, offsets.Earliest)
		offsets.Current = offsets.Earliest
	}
	obj := &claim{
		lock:            &sync.RWMutex{},
		messagesLock:    &sync.Mutex{},
		stopChan:        make(chan struct{}),
		doneChan:        make(chan struct{}),
		marshal:         marshal,
		consumer:        consumer,
		topic:           topic,
		partID:          partID,
		terminated:      new(int32),
		offsets:         offsets,
		messages:        messages,
		options:         options,
		tracking:        make(map[int64]bool),
		rand:            rand.New(rand.NewSource(time.Now().UnixNano())),
		lastMessageTime: time.Now(),
	}
	log.Infof("[%s:%d] consumer attempting to claim", topic, partID)
	if !marshal.ClaimPartition(topic, partID) {
		log.Infof("[%s:%d] consumer failed to claim", topic, partID)
		return nil
	}
	obj.setup()
	return obj
}
func (c *claim) setup() {
	c.lock.Lock()
	defer c.lock.Unlock()
	if c.offsets.Current < c.offsets.Earliest {
		log.Warningf("[%s:%d] consumer fast-forwarding from %d to %d",
			c.topic, c.partID, c.offsets.Current, c.offsets.Earliest)
		c.offsets.Current = c.offsets.Earliest
	}
	err := c.marshal.Heartbeat(c.topic, c.partID, c.offsets.Current)
	if err != nil {
		log.Errorf("[%s:%d] consumer failed to heartbeat: %s", c.topic, c.partID, err)
		go c.Release()
		return
	}
	c.lastHeartbeat = time.Now().Unix()
	consumerConf := kafka.NewConsumerConf(c.topic, int32(c.partID))
	consumerConf.StartOffset = c.offsets.Current
	consumerConf.MaxFetchSize = c.marshal.cluster.options.MaxMessageSize
	consumerConf.RequestTimeout = c.marshal.cluster.options.ConsumeRequestTimeout
	consumerConf.RetryLimit = 0
	kafkaConsumer, err := c.marshal.cluster.broker.Consumer(consumerConf)
	if err != nil {
		log.Errorf("[%s:%d] consumer failed to create Kafka Consumer: %s",
			c.topic, c.partID, err)
		go c.Release()
		return
	}
	c.kafkaConsumer = kafkaConsumer
	go c.messagePump()
	log.Infof("[%s:%d] consumer %s claimed at offset %d (is %d behind)",
		c.topic, c.partID, c.marshal.clientID, c.offsets.Current, c.offsets.Latest-c.offsets.Current)
}
func (c *claim) Commit(offset int64) error {
	if c.Terminated() {
		return fmt.Errorf("[%s:%d] is no longer claimed; can't commit offset %d",
			c.topic, c.partID, offset)
	}
	c.lock.Lock()
	defer c.lock.Unlock()
	_, ok := c.tracking[offset]
	if !ok {
		return fmt.Errorf("[%s:%d] committing offset %d but we've never seen it",
			c.topic, c.partID, offset)
	}
	c.tracking[offset] = true
	c.outstandingMessages--
	return nil
}
func (c *claim) Terminated() bool {
	return atomic.LoadInt32(c.terminated) == 1
}
func (c *claim) GetCurrentLag() int64 {
	c.lock.RLock()
	defer c.lock.RUnlock()
	if c.offsets.Current < c.offsets.Latest {
		return c.offsets.Latest - c.offsets.Current
	}
	return 0
}
func (c *claim) Flush() error {
	if c.Terminated() {
		return nil
	}
	didAdvance, currentOffset := c.updateCurrentOffsets()
	if !didAdvance {
		return nil
	}
	if err := c.marshal.Heartbeat(c.topic, c.partID, currentOffset); err != nil {
		go c.Release()
		return fmt.Errorf("[%s:%d] failed to flush, releasing: %s", c.topic, c.partID, err)
	}
	return nil
}
func (c *claim) Release() bool {
	return c.teardown(true)
}