func (s *PartitionCsmSuite) TestZeroRetries(c *C) {
	offsetsBefore := s.kh.GetOldestOffsets(topic)
	s.cfg.Consumer.AckTimeout = 100 * time.Millisecond
	s.cfg.Consumer.MaxRetries = 0
	s.kh.SetOffsetValues(group, topic, offsetsBefore)
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	msg0 := <-pc.Messages()
	log.Infof("*** First: offset=%v", msg0.Offset)
	sendEvOffered(msg0)
	time.Sleep(200 * time.Millisecond)
	for i := 0; i < 5; i++ {
		msg := <-pc.Messages()
		log.Infof("*** Next: offset=%v", msg.Offset)
		c.Assert(msg.Offset, Equals, msg0.Offset+int64(1+i), Commentf("i=%d", i))
		sendEvOffered(msg)
		sendEvAcked(msg)
	}
	pc.Stop()
	offsetsAfter := s.kh.GetCommittedOffsets(group, topic)
	c.Assert(offsetsAfter[partition].Val, Equals, offsetsBefore[partition]+6)
	c.Assert(offsettrk.SparseAcks2Str(offsetsAfter[partition]), Equals, "")
}
If the max retries limit set to 0, then messages are offered only once.
package partitioncsm
import (
	"testing"
	"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/consumer/msgfetcher"
	"github.com/mailgun/kafka-pixy/consumer/offsettrk"
	"github.com/mailgun/kafka-pixy/consumer/subscriber"
	"github.com/mailgun/kafka-pixy/offsetmgr"
	"github.com/mailgun/kafka-pixy/testhelpers"
	"github.com/mailgun/kafka-pixy/testhelpers/kafkahelper"
	log "github.com/sirupsen/logrus"
	. "gopkg.in/check.v1"
)
const (
	group     = "test_group"
	topic     = "test.1"
	partition = 0
	memberID  = "test_member"
)
type PartitionCsmSuite struct {
	cfg          *config.Proxy
	ns           *actor.Descriptor
	groupMember  *subscriber.T
	msgFetcherF  msgfetcher.Factory
	offsetMgrF   offsetmgr.Factory
	kh           *kafkahelper.T
	initOffsetCh chan offsetmgr.Offset
}
var _ = Suite(&PartitionCsmSuite{})
func Test(t *testing.T) {
	TestingT(t)
}
func (s *PartitionCsmSuite) SetUpSuite(c *C) {
	testhelpers.InitLogging()
	s.kh = kafkahelper.New(c)
	s.kh.PutMessages("pc", topic, map[string]int{"": 100})
}
func (s *PartitionCsmSuite) SetUpTest(c *C) {
	s.cfg = testhelpers.NewTestProxyCfg("test")
	s.cfg.Consumer.MaxPendingMessages = 100
	s.cfg.Consumer.MaxRetries = 1
	check4RetryInterval = 50 * time.Millisecond
	s.ns = actor.Root().NewChild("T")
	s.groupMember = subscriber.Spawn(s.ns, group, s.cfg, s.kh.ZKConn())
	s.msgFetcherF = msgfetcher.SpawnFactory(s.ns, s.cfg, s.kh.KafkaClt())
	s.offsetMgrF = offsetmgr.SpawnFactory(s.ns, s.cfg, s.kh.KafkaClt())
	s.initOffsetCh = make(chan offsetmgr.Offset, 1)
	initialOffsetCh = s.initOffsetCh
}
func (s *PartitionCsmSuite) TearDownTest(c *C) {
}
func (s *PartitionCsmSuite) TestOldestOffset(c *C) {
	oldestOffsets := s.kh.GetOldestOffsets(topic)
	s.kh.SetOffsets(group, topic, []offsetmgr.Offset{{Val: sarama.OffsetOldest, Meta: ""}})
	offsets := s.kh.GetCommittedOffsets(group, topic)
	c.Assert(offsets[partition], Equals, offsetmgr.Offset{Val: sarama.OffsetOldest, Meta: ""})
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	<-pc.Messages()
	pc.Stop()
	offsets = s.kh.GetCommittedOffsets(group, topic)
	c.Assert(offsets[partition].Val, Equals, oldestOffsets[partition])
}
func (s *PartitionCsmSuite) TestInitialOffsetTooLarge(c *C) {
	oldestOffsets := s.kh.GetOldestOffsets(topic)
	newestOffsets := s.kh.GetNewestOffsets(topic)
	log.Infof("*** test.1 offsets: oldest=%v, newest=%v", oldestOffsets, newestOffsets)
	s.kh.SetOffsets(group, topic, []offsetmgr.Offset{{Val: newestOffsets[partition] + 3}})
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	defer pc.Stop()
	initialOffset := <-s.initOffsetCh
	messages := s.kh.PutMessages("pc", topic, map[string]int{"": 4})
	var msg consumer.Message
	select {
	case msg = <-pc.Messages():
	case <-time.After(time.Second):
		c.Errorf("Message is not consumed")
	}
	c.Assert(msg.Offset, Equals, messages[""][3].Offset)
	c.Assert(msg.Offset, Equals, initialOffset.Val)
}
func (s *PartitionCsmSuite) TestMustBeOfferedToProceed(c *C) {
	s.kh.SetOffsetValues(group, topic, s.kh.GetOldestOffsets(topic))
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	defer pc.Stop()
	msg := <-pc.Messages()
	select {
	case <-pc.Messages():
		c.Error("Message must not be available until previous is offered")
	case <-time.After(200 * time.Millisecond):
	}
	sendEvOffered(msg)
	<-pc.Messages()
}
func (s *PartitionCsmSuite) TestSparseAckedNotRead(c *C) {
	ackedDlts := []bool{
		  false,
		  true,
		  true,
		  true,
		  false,
		  false,
		  true,
		  false,
		  false,
	}
	oldestOffsets := s.kh.GetOldestOffsets(topic)
	base := oldestOffsets[partition]
	ot := offsettrk.New(s.ns, offsetmgr.Offset{Val: base}, -1)
	var initOffset offsetmgr.Offset
	for i, acked := range ackedDlts {
		if acked {
			initOffset, _ = ot.OnAcked(base + int64(i))
		}
	}
	c.Assert(offsettrk.SparseAcks2Str(initOffset), Equals, "1-4,6-7")
	s.kh.SetOffsets(group, topic, []offsetmgr.Offset{initOffset})
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	defer pc.Stop()
	for i, acked := range ackedDlts {
		if acked {
			continue
		}
		msg := <-pc.Messages()
		c.Assert(msg.Offset, Equals, base+int64(i))
		sendEvOffered(msg)
		sendEvAcked(msg)
	}
}
func (s *PartitionCsmSuite) TestOfferInvalid(c *C) {
	s.kh.SetOffsetValues(group, topic, s.kh.GetOldestOffsets(topic))
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	defer pc.Stop()
	msg, ok := <-pc.Messages()
	c.Assert(ok, Equals, true)
	msg.EventsCh <- consumer.Event{T: consumer.EvOffered, Offset: msg.Offset + 1}
	msg.EventsCh <- consumer.Event{T: consumer.EvOffered, Offset: msg.Offset - 1}
	msg.EventsCh <- consumer.Event{T: consumer.EvOffered, Offset: msg.Offset}
	msg2, ok := <-pc.Messages()
	c.Assert(msg2.Offset, Equals, msg.Offset+1)
	c.Assert(ok, Equals, true)
}
func (s *PartitionCsmSuite) TestOfferedTooMany(c *C) {
	s.cfg.Consumer.AckTimeout = 500 * time.Millisecond
	s.cfg.Consumer.MaxPendingMessages = 3
	s.kh.SetOffsetValues(group, topic, s.kh.GetOldestOffsets(topic))
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	defer pc.Stop()
	var msg consumer.Message
	var messages []consumer.Message
	for i := 0; i < 4; i++ {
		msg = <-pc.Messages()
		messages = append(messages, msg)
		sendEvOffered(msg)
	}
	select {
	case <-pc.Messages():
		c.Error("No messages should be available above HWM limit")
	case <-time.After(200 * time.Millisecond):
	}
	sendEvAcked(messages[1])
	msg = <-pc.Messages()
	messages = append(messages, msg)
	sendEvOffered(msg)
	select {
	case msg := <-pc.Messages():
		c.Errorf("No messages should be available above HWM limit: %v", msg)
	case <-time.After(200 * time.Millisecond):
	}
}
func (s *PartitionCsmSuite) TestSparseAckedCommitted(c *C) {
	offsetsBefore := s.kh.GetOldestOffsets(topic)
	acks := []bool{
		  true,
		  false,
		  true,
		  true,
		  true,
		  false,
		  false,
		  true,
		  true,
		  false,
	}
	s.kh.SetOffsetValues(group, topic, offsetsBefore)
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	for _, shouldAck := range acks {
		msg := <-pc.Messages()
		sendEvOffered(msg)
		if shouldAck {
			sendEvAcked(msg)
		}
	}
	pc.Stop()
	offsetsAfter := s.kh.GetCommittedOffsets(group, topic)
	c.Assert(offsetsAfter[partition].Val, Equals, offsetsBefore[partition]+1)
	c.Assert(offsettrk.SparseAcks2Str(offsetsAfter[partition]), Equals, "1-4,6-8")
}
func (s *PartitionCsmSuite) TestSparseAckedAfterStop(c *C) {
	offsetsBefore := s.kh.GetOldestOffsets(topic)
	s.cfg.Consumer.AckTimeout = 300 * time.Millisecond
	s.kh.SetOffsetValues(group, topic, offsetsBefore)
	pc := Spawn(s.ns, group, topic, partition, s.cfg, s.groupMember, s.msgFetcherF, s.offsetMgrF)
	var messages []consumer.Message
	for i := 0; i < 10; i++ {
		msg := <-pc.Messages()
		sendEvOffered(msg)
		messages = append(messages, msg)
	}
	sendEvAcked(messages[7])
	sendEvAcked(messages[1])
	sendEvAcked(messages[8])
	go pc.Stop()  
	time.Sleep(100 * time.Millisecond)
	sendEvAcked(messages[3])
	sendEvAcked(messages[4])
	sendEvAcked(messages[6])
	sendEvAcked(messages[0])
	for {
		if _, ok := <-pc.Messages(); !ok {
			break
		}
	}
	sendEvAcked(messages[1])
	sendEvAcked(messages[4])
	offsetsAfter := s.kh.GetCommittedOffsets(group, topic)
	c.Assert(offsetsAfter[partition].Val, Equals, offsetsBefore[partition]+2)
	c.Assert(offsettrk.SparseAcks2Str(offsetsAfter[partition]), Equals, "1-3,4-7")
}