func (s *ConsumerSuite) TestLotsOfPartitions(c *C) {
	s.kh.ResetOffsets("g1", "test.64")
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	msg, err := cons.Consume("g1", "test.64")
	c.Assert(err, Equals, consumer.ErrRequestTimeout, Commentf("Unexpected message consumed, %v", msg))
	s.kh.PutMessages("lots", "test.64", map[string]int{"A": 7, "B": 13, "C": 169})
	log.Infof("*** WHEN")
	consumed := consume(c, cons, "g1", "test.64", 189, 10*time.Second)
	log.Infof("*** THEN")
	c.Assert(7, Equals, len(consumed["A"]))
	c.Assert(13, Equals, len(consumed["B"]))
	c.Assert(169, Equals, len(consumed["C"]))
}
A topic that has a lot of partitions can be consumed.
package consumerimpl
import (
	"fmt"
	"sync"
	"sync/atomic"
	"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/partitioncsm"
	"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"
)
func Test(t *testing.T) {
	TestingT(t)
}
type ConsumerSuite struct {
	ns  *actor.Descriptor
	cfg *config.Proxy
	kh  *kafkahelper.T
	omf offsetmgr.Factory
}
var _ = Suite(&ConsumerSuite{})
func (s *ConsumerSuite) SetUpSuite(c *C) {
	testhelpers.InitLogging()
	s.kh = kafkahelper.New(c)
	cfg := testhelpers.NewTestProxyCfg("omf")
	tid := actor.Root().NewChild("omf")
	s.omf = offsetmgr.SpawnFactory(tid, cfg, s.kh.KafkaClt())
}
func (s *ConsumerSuite) TearDownSuite(*C) {
	s.omf.Stop()
	s.kh.Close()
}
func (s *ConsumerSuite) SetUpTest(*C) {
	s.ns = actor.Root().NewChild("T")
	s.cfg = testhelpers.NewTestProxyCfg("c1")
	partitioncsm.FirstMessageFetchedCh = make(chan *partitioncsm.T, 100)
}
func (s *ConsumerSuite) TestInitialOffsetTooLarge(c *C) {
	oldestOffsets := s.kh.GetOldestOffsets("test.1")
	newestOffsets := s.kh.GetNewestOffsets("test.1")
	log.Infof("*** test.1 offsets: oldest=%v, newest=%v", oldestOffsets, newestOffsets)
	om, err := s.omf.Spawn(s.ns, "g1", "test.1", 0)
	c.Assert(err, IsNil)
	om.SubmitOffset(offsetmgr.Offset{Val: newestOffsets[0] + 3, Meta: ""})
	om.Stop()
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	_, err = cons.Consume("g1", "test.1")
	c.Assert(err, Equals, consumer.ErrRequestTimeout)
	produced := s.kh.PutMessages("offset-too-large", "test.1", map[string]int{"key": 5})
	consumed := consume(c, cons, "g1", "test.1", 1, 5*time.Second)
	c.Assert(consumed["key"][0].Offset, Equals, newestOffsets[0]+3)
	assertMsg(c, consumed["key"][0], produced["key"][3])
}
func (s *ConsumerSuite) TestSinglePartitionTopic(c *C) {
	s.kh.ResetOffsets("g1", "test.1")
	produced := s.kh.PutMessages("single", "test.1", map[string]int{"": 3})
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	consumed := consume(c, cons, "g1", "test.1", 1, 5*time.Second)
	assertMsg(c, consumed[""][0], produced[""][0])
}
func (s *ConsumerSuite) TestSequentialConsume(c *C) {
	s.kh.ResetOffsets("g1", "test.1")
	produced := s.kh.PutMessages("sequencial", "test.1", map[string]int{"": 3})
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	log.Infof("*** GIVEN 1")
	consumed := consume(c, cons, "g1", "test.1", 2, 5*time.Second)
	assertMsg(c, consumed[""][0], produced[""][0])
	assertMsg(c, consumed[""][1], produced[""][1])
	log.Infof("*** WHEN")
	cons.Stop()
	cons, err = Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	log.Infof("*** THEN")
	consumed = consume(c, cons, "g1", "test.1", 1, 5*time.Second, consumed)
	assertMsg(c, consumed[""][2], produced[""][2])
}
func (s *ConsumerSuite) TestMultiplePartitions(c *C) {
	s.kh.ResetOffsets("g1", "test.4")
	s.kh.PutMessages("multiple.partitions", "test.4", map[string]int{"A": 100, "B": 100})
	log.Infof("*** GIVEN 1")
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	log.Infof("*** WHEN")
	consumed := consume(c, cons, "g1", "test.4", 1, 5*time.Second)
	waitFirstFetched(cons, 2)
	consumed = consume(c, cons, "g1", "test.4", 99, 20*time.Second, consumed)
	log.Infof("*** THEN")
	if len(consumed["A"]) == 0 || len(consumed["A"]) == 0 {
		c.Errorf("Not distributed: consumed[A]=%d, consumed[B]=%d", len(consumed["A"]), len(consumed["B"]))
	}
}
func (s *ConsumerSuite) TestMultipleTopics(c *C) {
	s.kh.ResetOffsets("g1", "test.1")
	s.kh.ResetOffsets("g1", "test.4")
	produced1 := s.kh.PutMessages("multiple.topics", "test.1", map[string]int{"A": 1})
	produced4 := s.kh.PutMessages("multiple.topics", "test.4", map[string]int{"B": 1, "C": 1})
	log.Infof("*** GIVEN 1")
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	log.Infof("*** WHEN")
	consumed := consume(c, cons, "g1", "test.4", 1, 5*time.Second)
	consumed = consume(c, cons, "g1", "test.1", 1, 5*time.Second, consumed)
	consumed = consume(c, cons, "g1", "test.4", 1, 5*time.Second, consumed)
	log.Infof("*** THEN")
	assertMsg(c, consumed["A"][0], produced1["A"][0])
	assertMsg(c, consumed["B"][0], produced4["B"][0])
	assertMsg(c, consumed["C"][0], produced4["C"][0])
}
func (s *ConsumerSuite) TestMultipleGroups(c *C) {
	s.kh.ResetOffsets("g1", "test.4")
	s.kh.ResetOffsets("g2", "test.4")
	s.kh.PutMessages("multi", "test.4", map[string]int{"A": 10, "B": 10, "C": 10})
	log.Infof("*** GIVEN 1")
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	log.Infof("*** WHEN")
	consumed1 := consume(c, cons, "g1", "test.4", 10, 5*time.Second)
	consumed2 := consume(c, cons, "g2", "test.4", 20, 5*time.Second)
	consumed1 = consume(c, cons, "g1", "test.4", 20, 5*time.Second, consumed1)
	consumed2 = consume(c, cons, "g2", "test.4", 10, 5*time.Second, consumed2)
	log.Infof("*** THEN")
	c.Assert(consumed1, DeepEquals, consumed2)
}
func (s *ConsumerSuite) TestTooFewPartitions(c *C) {
	s.kh.ResetOffsets("g1", "test.1")
	produced := s.kh.PutMessages("few", "test.1", map[string]int{"": 3})
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	log.Infof("*** GIVEN 1")
	consumed := consume(c, cons, "g1", "test.1", 2, 5*time.Second)
	assertMsg(c, consumed[""][0], produced[""][0])
	log.Infof("*** WHEN")
	cfg1 := testhelpers.NewTestProxyCfg("c2")
	omf1 := offsetmgr.SpawnFactory(s.ns, cfg1, s.kh.KafkaClt())
	defer omf1.Stop()
	cons1, err := Spawn(s.ns, cfg1, omf1)
	c.Assert(err, IsNil)
	defer cons1.Stop()
	_, err = cons1.Consume("g1", "test.1")
	log.Infof("*** THEN")
	c.Assert(err, Equals, consumer.ErrRequestTimeout)
	consume(c, cons, "g1", "test.1", 1, 5*time.Second, consumed)
	assertMsg(c, consumed[""][1], produced[""][1])
}
func (s *ConsumerSuite) TestRebalanceOnJoin(c *C) {
	s.kh.ResetOffsets("g1", "test.4")
	s.kh.PutMessages("join", "test.4", map[string]int{"A": 10, "B": 10})
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	log.Infof("*** GIVEN 1")
	consumed1 := consume(c, cons, "g1", "test.4", 1, 5*time.Second)
	waitFirstFetched(cons, 2)
	log.Infof("*** GIVEN 2")
	consumed1 = consume(c, cons, "g1", "test.4", 4, 5*time.Second, consumed1)
	c.Assert(len(consumed1["A"]), Not(Equals), 0)
	c.Assert(len(consumed1["B"]), Not(Equals), 0)
	consumedBeforeJoin := len(consumed1["B"])
	log.Infof("*** WHEN")
	cfg1 := testhelpers.NewTestProxyCfg("c2")
	omf1 := offsetmgr.SpawnFactory(s.ns, cfg1, s.kh.KafkaClt())
	defer omf1.Stop()
	cons1, err := Spawn(s.ns, cfg1, omf1)
	c.Assert(err, IsNil)
	defer cons1.Stop()
	log.Infof("*** THEN")
	consumed2 := consume(c, cons1, "g1", "test.4", consumeAll, 5*time.Second)
	consumed1 = consume(c, cons, "g1", "test.4", consumeAll, 5*time.Second, consumed1)
	c.Assert(len(consumed1["A"]), Equals, 10)
	c.Assert(len(consumed2["A"]), Equals, 0)
	c.Assert(len(consumed1["B"]), Equals, consumedBeforeJoin)
	c.Assert(len(consumed2["B"]), Not(Equals), 0)
	c.Assert(len(consumed1["B"])+len(consumed2["B"]), Equals, 10)
	c.Assert(consumed2["B"][0].Offset, Equals, consumed1["B"][len(consumed1["B"])-1].Offset+1)
}
func (s *ConsumerSuite) TestRebalanceOnLeave(c *C) {
	s.kh.ResetOffsets("g1", "test.4")
	produced := s.kh.PutMessages("leave", "test.4", map[string]int{"A": 10, "B": 10, "C": 10})
	var err error
	consumers := make([]*t, 3)
	for i := 0; i < 3; i++ {
		cfg := testhelpers.NewTestProxyCfg(fmt.Sprintf("c%d", i))
		omf := offsetmgr.SpawnFactory(s.ns, cfg, s.kh.KafkaClt())
		defer omf.Stop()
		consumers[i], err = Spawn(s.ns, cfg, omf)
		c.Assert(err, IsNil)
	}
	defer consumers[0].Stop()
	defer consumers[1].Stop()
	log.Infof("*** GIVEN 1")
	consumed := make([]map[string][]consumer.Message, 3)
	for i := 0; i < 3; i++ {
		consumed[i] = consume(c, consumers[i], "g1", "test.4", 1, 5*time.Second)
	}
	log.Infof("*** GIVEN 2")
	if len(consumed[0]["A"]) == 1 {
		if len(consumed[1]["B"]) == 1 {
			assertMsg(c, consumed[2]["B"][0], produced["B"][1])
		} else {  
			assertMsg(c, consumed[2]["B"][0], produced["B"][0])
		}
	} else if len(consumed[0]["B"]) == 1 {
		if len(consumed[1]["B"]) == 1 {
			assertMsg(c, consumed[2]["B"][0], produced["B"][2])
		} else {  
			assertMsg(c, consumed[2]["B"][0], produced["B"][1])
		}
	} else {  
		if len(consumed[1]["B"]) == 1 {
			assertMsg(c, consumed[2]["B"][0], produced["B"][1])
		} else {  
			assertMsg(c, consumed[2]["B"][0], produced["B"][0])
		}
	}
	consume(c, consumers[2], "g1", "test.4", 4, 5*time.Second, consumed[2])
	c.Assert(len(consumed[2]["B"]), Equals, 5)
	lastConsumedFromBby2 := consumed[2]["B"][4]
	for _, consumer := range consumers {
		drainFirstFetched(consumer)
	}
	log.Infof("*** WHEN")
	consumers[2].Stop()
	waitFirstFetched(consumers[1], 1)
	log.Infof("*** THEN")
	consumedSoFar := make(map[string]int)
	for _, consumedByOne := range consumed {
		for key, consumedWithKey := range consumedByOne {
			consumedSoFar[key] = consumedSoFar[key] + len(consumedWithKey)
		}
	}
	yetToBeConsumedBy1 := (len(produced["B"]) + len(produced["C"])) - (consumedSoFar["B"] + consumedSoFar["C"])
	log.Infof("*** Consumed so far: %v", consumedSoFar)
	log.Infof("*** Yet to be consumed by cons[1]: %v", yetToBeConsumedBy1)
	consumedBy1 := consume(c, consumers[1], "g1", "test.4", yetToBeConsumedBy1, 5*time.Second)
	c.Assert(len(consumedBy1["B"]), Equals, len(produced["B"])-consumedSoFar["B"])
	c.Assert(consumedBy1["B"][0].Offset, Equals, lastConsumedFromBby2.Offset+1)
}
func (s *ConsumerSuite) TestRebalanceOnTimeout(c *C) {
	var wg sync.WaitGroup
	consumed := make([]map[string][]consumer.Message, 2)
	s.kh.ResetOffsets("g1", "test.4")
	s.kh.PutMessages("timeout", "test.4", map[string]int{"A": 10, "B": 10})
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	cfg1 := testhelpers.NewTestProxyCfg("c2")
	cfg1.Consumer.SubscriptionTimeout = 500 * time.Millisecond
	omf1 := offsetmgr.SpawnFactory(s.ns, cfg1, s.kh.KafkaClt())
	defer omf1.Stop()
	sc1, err := Spawn(s.ns, cfg1, omf1)
	c.Assert(err, IsNil)
	defer sc1.Stop()
	log.Infof("*** GIVEN 1")
	consumed[0] = consume(c, cons, "g1", "test.4", 1, 5*time.Second)
	consumed[1] = consume(c, sc1, "g1", "test.4", 1, 5*time.Second)
	if len(consumed[0]["B"]) == 0 {
		c.Assert(len(consumed[0]["A"]), Equals, 1)
	} else {
		c.Assert(len(consumed[0]["A"]), Equals, 0)
	}
	c.Assert(len(consumed[1]["A"]), Equals, 0)
	c.Assert(len(consumed[1]["B"]), Equals, 1)
	log.Infof("*** GIVEN 2")
	actor.Spawn(s.ns.NewChild("consume[0]"), &wg, func() {
		consumed[0] = consume(c, cons, "g1", "test.4", 4, 5*time.Second, consumed[0])
	})
	actor.Spawn(s.ns.NewChild("consume[1]"), &wg, func() {
		consumed[1] = consume(c, sc1, "g1", "test.4", 4, 5*time.Second, consumed[1])
	})
	wg.Wait()
	if len(consumed[0]["B"]) == 1 {
		c.Assert(len(consumed[0]["A"]), Equals, 4)
	} else {
		c.Assert(len(consumed[0]["A"]), Equals, 5)
	}
	c.Assert(len(consumed[1]["A"]), Equals, 0)
	c.Assert(len(consumed[1]["B"]), Equals, 5)
	drainFirstFetched(cons)
	log.Infof("*** WHEN")
	waitFirstFetched(cons, 1)
	log.Infof("*** THEN")
	consumed[0] = consume(c, cons, "g1", "test.4", 10, 5*time.Second, consumed[0])
	c.Assert(len(consumed[0]["A"]), Equals, 10)
	c.Assert(len(consumed[0]["B"]), Equals, 5)
	c.Assert(len(consumed[1]["A"]), Equals, 0)
	c.Assert(len(consumed[1]["B"]), Equals, 5)
}
func (s *ConsumerSuite) TestTooManyRequestsError(c *C) {
	s.kh.ResetOffsets("g1", "test.1")
	s.kh.PutMessages("join", "test.1", map[string]int{"A": 30})
	s.cfg.Consumer.ChannelBufferSize = 1
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	var tooManyRequestsCount int32
	var wg sync.WaitGroup
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < 10; i++ {
				_, err := cons.Consume("g1", "test.1")
				if err == consumer.ErrTooManyRequests {
					atomic.AddInt32(&tooManyRequestsCount, 1)
				}
			}
		}()
	}
	wg.Wait()
	c.Assert(tooManyRequestsCount, Not(Equals), 0)
	log.Infof("*** ErrTooMannyRequests was returned %d times", tooManyRequestsCount)
}
func (s *ConsumerSuite) TestInvalidTopic(c *C) {
	s.cfg.Consumer.LongPollingTimeout = 1 * time.Second
	cons, err := Spawn(s.ns, s.cfg, s.omf)
	c.Assert(err, IsNil)
	defer cons.Stop()
	_, err = cons.Consume("g1", "no-such-topic")
	c.Assert(err, Equals, consumer.ErrRequestTimeout)
}