func (c *claim) ConsumerVelocity() float64 {
	c.lock.RLock()
	defer c.lock.RUnlock()
	return average(c.offsetCurrentHistory[0:])
}
ConsumerVelocity returns the average of our consumers' velocity
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)
}
func (c *claim) Terminate() bool {
	return c.teardown(false)
}
func (c *claim) teardown(releasePartition bool) bool {
	if !atomic.CompareAndSwapInt32(c.terminated, 0, 1) {
		<-c.doneChan
		return false
	}
	close(c.stopChan)
	c.messagesLock.Lock()
	defer c.messagesLock.Unlock()
	_, currentOffset := c.updateCurrentOffsets()
	if c.consumer != nil {
		go c.consumer.claimTerminated(c, releasePartition)
	}
	var err error
	if releasePartition {
		log.Infof("[%s:%d] releasing partition claim", c.topic, c.partID)
		err = c.marshal.ReleasePartition(c.topic, c.partID, currentOffset)
	} else {
		err = c.marshal.Heartbeat(c.topic, c.partID, currentOffset)
	}
	<-c.doneChan
	if err != nil {
		log.Errorf("[%s:%d] failed to release: %s", c.topic, c.partID, err)
		return false
	}
	return true
}
func (c *claim) messagePump() {
	defer close(c.doneChan)
	retry := &backoff.Backoff{Min: 10 * time.Millisecond, Max: 1 * time.Second, Jitter: true}
	for !c.Terminated() {
		msg, err := c.kafkaConsumer.Consume()
		if err == proto.ErrOffsetOutOfRange {
			log.Warningf("[%s:%d] error consuming: out of range, abandoning partition",
				c.topic, c.partID)
			go c.Release()
			return
		} else if err == kafka.ErrNoData {
			time.Sleep(retry.Duration())
			continue
		} else if err != nil {
			log.Errorf("[%s:%d] error consuming: %s", c.topic, c.partID, err)
			time.Sleep(1 * time.Second)
			continue
		}
		retry.Reset()
		c.lock.Lock()
		c.lastMessageTime = time.Now()
		c.tracking[msg.Offset] = false
		c.outstandingMessages++
		if msg.Offset < c.offsets.Current {
			log.Errorf("[%s:%d] just consumed offset %d earlier than current %d",
				c.topic, c.partID, msg.Offset, c.offsets.Current)
		}
		c.lock.Unlock()
		c.messagesLock.Lock()
		if !c.Terminated() {
			tmp := Message(*msg)
			select {
			case c.messages <- &tmp:
			case <-c.stopChan:
			}
		}
		c.messagesLock.Unlock()
	}
	log.Debugf("[%s:%d] no longer claimed, pump exiting", c.topic, c.partID)
}
func (c *claim) heartbeat() bool {
	if c.Terminated() {
		return false
	}
	c.lock.Lock()
	defer c.lock.Unlock()
	err := c.marshal.Heartbeat(c.topic, c.partID, c.offsets.Current)
	if err != nil {
		log.Errorf("[%s:%d] failed to heartbeat, releasing: %s", c.topic, c.partID, err)
		go c.Release()
	}
	log.Infof("[%s:%d] heartbeat: Current offset is %d, partition offset range is %d..%d.",
		c.topic, c.partID, c.offsets.Current, c.offsets.Earliest, c.offsets.Latest)
	log.Infof("[%s:%d] heartbeat: There are %d messages in queue and %d messages outstanding.",
		c.topic, c.partID, len(c.messages), c.outstandingMessages)
	c.lastHeartbeat = time.Now().Unix()
	return true
}
func (c *claim) updateCurrentOffsets() (bool, int64) {
	c.lock.Lock()
	defer c.lock.Unlock()
	offsets := make(int64slice, 0, len(c.tracking))
	for key := range c.tracking {
		offsets = append(offsets, key)
	}
	sort.Sort(offsets)
	didAdvance := false
	for _, offset := range offsets {
		if !c.tracking[offset] {
			break
		}
		didAdvance = true
		if offset+1 <= c.offsets.Current  {
			log.Errorf("[%s:%d] rewinding current offset from %d to %d",
				c.topic, c.partID, c.offsets.Current, offset+1)
		}
		c.offsets.Current = offset + 1
		delete(c.tracking, offset)
	}
	if len(c.tracking) > c.marshal.cluster.options.MaxMessageQueue {
		log.Errorf("[%s:%d] has %d uncommitted offsets. You must call Commit.",
			c.topic, c.partID, len(c.tracking))
	}
	return didAdvance, c.offsets.Current
}
func (c *claim) heartbeatExpired() bool {
	c.lock.RLock()
	defer c.lock.RUnlock()
	return c.lastHeartbeat < time.Now().Unix()-HeartbeatInterval
}
func (c *claim) healthCheck() bool {
	if c.Terminated() {
		return false
	}
	consumerVelocity := c.ConsumerVelocity()
	partitionVelocity := c.PartitionVelocity()
	if c.heartbeatExpired() {
		log.Warningf("[%s:%d] consumer unhealthy by heartbeat test, releasing",
			c.topic, c.partID)
		go c.Release()
		return false
	}
	if c.marshal.cluster.IsGroupPaused(c.marshal.GroupID()) {
		log.Infof("[%s:%d] consumer group %s is paused, claim releasing",
			c.topic, c.partID, c.marshal.GroupID())
		go c.Release()
		return false
	}
	c.lock.Lock()
	defer c.lock.Unlock()
	if time.Now().After(c.lastMessageTime.Add(HeartbeatInterval * time.Second)) {
		if c.options.ReleaseClaimsIfBehind && consumerVelocity == 0 && (partitionVelocity > 0 || c.offsets.Latest > c.offsets.Current) {
			log.Warningf("[%s:%d] no messages received for %d seconds with CV=%0.2f PV=%0.2f, releasing",
				c.topic, c.partID, HeartbeatInterval, consumerVelocity, partitionVelocity)
			go c.Release()
			return false
		} else {
			log.Infof("[%s:%d] no messages received for %d seconds with CV=%0.2f PV=%0.2f",
				c.topic, c.partID, HeartbeatInterval, consumerVelocity, partitionVelocity)
		}
	}
	if c.options.ClaimEntireTopic || !c.options.ReleaseClaimsIfBehind {
		return true
	}
	testOffset := c.offsets.Current + int64(consumerVelocity*2)
	if testOffset >= c.offsets.Latest {
		c.cyclesBehind = 0
		return true
	}
	if partitionVelocity < consumerVelocity {
		log.Infof("[%s:%d] consumer catching up: consume ∆ %0.2f >= produce ∆ %0.2f",
			c.topic, c.partID, consumerVelocity, partitionVelocity)
		c.cyclesBehind = 0
		return true
	}
	c.cyclesBehind++
	if c.cyclesBehind >= 3 {
		log.Warningf("[%s:%d] consumer unhealthy for too long, releasing",
			c.topic, c.partID)
		go c.Release()
		return false
	}
	log.Warningf("[%s:%d] consumer too slow: consume ∆ %0.2f < produce ∆ %0.2f (warning #%d)",
		c.topic, c.partID, consumerVelocity, partitionVelocity, c.cyclesBehind)
	return true
}
func (c *claim) healthCheckLoop() {
	time.Sleep(<-c.marshal.cluster.jitters)
	for !c.Terminated() {
		for !c.heartbeatExpired() {
			if err := c.updateOffsets(); err != nil {
				log.Errorf("[%s:%d] health check loop failed to update offsets: %s",
					c.topic, c.partID, err)
				time.Sleep(1 * time.Second)
				continue
			}
			break
		}
		if c.healthCheck() {
			go c.heartbeat()
		}
		time.Sleep(<-c.marshal.cluster.jitters)
	}
	log.Infof("[%s:%d] health check loop exiting, claim terminated",
		c.topic, c.partID)
}
func average(vals []int64) float64 {
	min, max, ct := int64(0), int64(0), int64(0)
	for _, val := range vals {
		if val <= 0 {
			continue
		}
		if min == 0 || val < min {
			min = val
		}
		if max == 0 || val > max {
			max = val
		}
		ct++
	}
	if min == max || ct < 2 {
		return 0
	}
	return float64(max-min) / float64(ct-1)
}