func (s *SubscriberSuite) TestReleasePartitionNotOwner(c *C) {
	cfg := newConfig("m1")
	ss := Spawn(s.ns.NewChild("m1"), "g1", cfg, s.zkConn)
	defer ss.Stop()
	cancelCh := make(chan none.T)
	claim := ss.ClaimPartition(s.ns, "foo", 1, cancelCh)
	_, err := s.zkConn.Set(chroot+"/consumers/g1/owners/foo/1", []byte("m2"), -1)
	c.Assert(err, IsNil)
	owner, err := ss.kazooModel.GetPartitionOwner("foo", 1)
	c.Assert(err, IsNil)
	c.Assert(owner, Equals, "m2")
	doneCh := make(chan struct{})
	go func() {
		defer close(doneCh)
		claim()
	}()
	select {
	case <-doneCh:
	case <-time.After(3 * time.Second):
		c.Error("Timeout waiting for claim to be release")
	}
}
Claim release should not fail if it is not this instance that owns the
partition. This situation can happen when a Kafka-Pixy looses connection
with ZooKeeper for longer then the ZooKeeper session timeout, so the rest
of Kafka-Pixy has a chance to rebalance.
package subscriber
import (
	"reflect"
	"sync"
	"testing"
	"time"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/config"
	"github.com/mailgun/kafka-pixy/none"
	"github.com/mailgun/kafka-pixy/testhelpers"
	"github.com/samuel/go-zookeeper/zk"
	. "gopkg.in/check.v1"
)
const (
	chroot = "/test"
)
func Test(t *testing.T) {
	TestingT(t)
}
type SubscriberSuite struct {
	ns     *actor.Descriptor
	zkConn *zk.Conn
}
var _ = Suite(&SubscriberSuite{})
func (s *SubscriberSuite) SetUpSuite(c *C) {
	testhelpers.InitLogging()
	var err error
	sessionTimeout := 300 * time.Second
	s.zkConn, _, err = zk.Connect(testhelpers.ZookeeperPeers, sessionTimeout)
	c.Assert(err, IsNil)
}
func (s *SubscriberSuite) SetUpTest(c *C) {
	s.ns = actor.Root().NewChild("T")
}
func (s *SubscriberSuite) TestSimpleSubscribe(c *C) {
	cfg := newConfig("m1")
	ss := Spawn(s.ns.NewChild("m1"), "g1", cfg, s.zkConn)
	defer ss.Stop()
	ss.Topics() <- []string{"foo", "bar"}
	c.Assert(<-ss.Subscriptions(), DeepEquals,
		map[string][]string{"m1": {"bar", "foo"}})
}
func (s *SubscriberSuite) TestSubscribeSequence(c *C) {
	cfg := newConfig("m1")
	ss := Spawn(s.ns.NewChild("m1"), "g1", cfg, s.zkConn)
	defer ss.Stop()
	ss.Topics() <- []string{"foo", "bar"}
	ss.Topics() <- []string{"blah", "bazz"}
	assertSubscription(c, ss.Subscriptions(),
		map[string][]string{"m1": {"bazz", "blah"}}, 3*time.Second)
}
func (s *SubscriberSuite) TestReSubscribe(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	ss1.Topics() <- []string{"foo", "bar"}
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	ss2.Topics() <- []string{"bazz", "bar"}
	membership := map[string][]string{
		"m1": {"bar", "foo"},
		"m2": {"bar", "bazz"},
	}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
	ss1.Topics() <- []string{"foo", "bar"}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
}
func (s *SubscriberSuite) TestSubscribeToNothing(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	ss1.Topics() <- []string{"foo", "bar"}
	ss2.Topics() <- []string{"foo"}
	membership := map[string][]string{"m1": {"bar", "foo"}, "m2": {"foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
	ss1.Topics() <- []string{}
	membership = map[string][]string{"m2": {"foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
}
func (s *SubscriberSuite) TestSubscribeToNil(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	ss1.Topics() <- []string{"foo", "bar"}
	ss2.Topics() <- []string{"foo"}
	membership := map[string][]string{"m1": {"bar", "foo"}, "m2": {"foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
	ss1.Topics() <- nil
	membership = map[string][]string{"m2": {"foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
}
func (s *SubscriberSuite) TestSomethingAfterNothingBug(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	ss1.Topics() <- []string{"foo"}
	c.Assert(<-ss1.Subscriptions(), DeepEquals, map[string][]string{"m1": {"foo"}})
	ss1.Topics() <- []string{}
	c.Assert(<-ss1.Subscriptions(), DeepEquals, map[string][]string{})
	ss1.Topics() <- []string{"foo"}
	c.Assert(<-ss1.Subscriptions(), DeepEquals,
		map[string][]string{"m1": {"foo"}})
}
func (s *SubscriberSuite) TestMembershipChanges(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	cfg3 := newConfig("m3")
	ss3 := Spawn(s.ns.NewChild("m3"), "g1", cfg3, s.zkConn)
	defer ss3.Stop()
	ss1.Topics() <- []string{"foo", "bar"}
	ss2.Topics() <- []string{"foo"}
	ss3.Topics() <- []string{"foo", "bazz", "blah"}
	membership := map[string][]string{
		"m1": {"bar", "foo"},
		"m2": {"foo"},
		"m3": {"bazz", "blah", "foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 5*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 5*time.Second)
	assertSubscription(c, ss3.Subscriptions(), membership, 5*time.Second)
}
func (s *SubscriberSuite) TestRedundantUpdateBug(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	ss1.Topics() <- []string{"foo", "bar"}
	ss2.Topics() <- []string{"foo", "bazz", "blah"}
	membership := map[string][]string{
		"m1": {"bar", "foo"},
		"m2": {"bazz", "blah", "foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	ss2.Topics() <- []string{"bar"}
	ss2.Topics() <- []string{"foo", "bazz", "blah"}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
}
func (s *SubscriberSuite) TestMissingSubscriptionBug(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	ss1.Topics() <- []string{"foo", "bar"}
	ss2.Topics() <- []string{"foo", "bazz", "blah"}
	membership := map[string][]string{
		"m1": {"bar", "foo"},
		"m2": {"bazz", "blah", "foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
	ss1.kazooModel.EnsureMemberSubscription(nil)
	membership = map[string][]string{
		"m2": {"bazz", "blah", "foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
	membership = map[string][]string{
		"m1": {"bar", "foo"},
		"m2": {"bazz", "blah", "foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
}
func (s *SubscriberSuite) TestMissingOutdatedSubscription(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	ss1.Topics() <- []string{"foo", "bar"}
	ss2.Topics() <- []string{"foo", "bazz", "blah"}
	membership := map[string][]string{
		"m1": {"bar", "foo"},
		"m2": {"bazz", "blah", "foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
	ss1.kazooModel.EnsureMemberSubscription([]string{"foo", "bazz"})
	membership = map[string][]string{
		"m1": {"bazz", "foo"},
		"m2": {"bazz", "blah", "foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
	membership = map[string][]string{
		"m1": {"bar", "foo"},
		"m2": {"bazz", "blah", "foo"}}
	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
}
func (s *SubscriberSuite) TestClaimPartition(c *C) {
	cfg := newConfig("m1")
	ss := Spawn(s.ns.NewChild("m1"), "g1", cfg, s.zkConn)
	defer ss.Stop()
	cancelCh := make(chan none.T)
	owner, err := ss.kazooModel.GetPartitionOwner("foo", 1)
	c.Assert(err, IsNil)
	c.Assert(owner, Equals, "")
	claim1 := ss.ClaimPartition(s.ns, "foo", 1, cancelCh)
	defer claim1()
	owner, err = ss.kazooModel.GetPartitionOwner("foo", 1)
	c.Assert(err, IsNil)
	c.Assert(owner, Equals, "m1")
}
func (s *SubscriberSuite) TestClaimPartitionClaimed(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	cancelCh := make(chan none.T)
	claim1 := ss1.ClaimPartition(s.ns, "foo", 1, cancelCh)
	defer claim1()
	close(cancelCh)
	claim2 := ss2.ClaimPartition(s.ns, "foo", 1, cancelCh)
	defer claim2()
	owner, err := ss1.kazooModel.GetPartitionOwner("foo", 1)
	c.Assert(err, IsNil)
	c.Assert(owner, Equals, "m1")
}
func (s *SubscriberSuite) TestClaimPartitionTwice(c *C) {
	cfg := newConfig("m1")
	ss := Spawn(s.ns.NewChild("m1"), "g1", cfg, s.zkConn)
	defer ss.Stop()
	cancelCh := make(chan none.T)
	claim1 := ss.ClaimPartition(s.ns, "foo", 1, cancelCh)
	defer claim1()
	claim2 := ss.ClaimPartition(s.ns, "foo", 1, cancelCh)
	defer claim2()
	owner, err := ss.kazooModel.GetPartitionOwner("foo", 1)
	c.Assert(err, IsNil)
	c.Assert(owner, Equals, "m1")
}
func (s *SubscriberSuite) TestReleasePartition(c *C) {
	cfg := newConfig("m1")
	ss := Spawn(s.ns.NewChild("m1"), "g1", cfg, s.zkConn)
	defer ss.Stop()
	cancelCh := make(chan none.T)
	claim1 := ss.ClaimPartition(s.ns, "foo", 1, cancelCh)
	claim2 := ss.ClaimPartition(s.ns, "foo", 1, cancelCh)
	claim2()  
	owner, err := ss.kazooModel.GetPartitionOwner("foo", 1)
	c.Assert(err, IsNil)
	c.Assert(owner, Equals, "")
	claim1()
}
func (s *SubscriberSuite) TestClaimPartitionParallel(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	cancelCh := make(chan none.T)
	claim1 := ss1.ClaimPartition(s.ns, "foo", 1, cancelCh)
	go func() {
		time.Sleep(300 * time.Millisecond)
		claim1()
		print("*** m1 released\n")
	}()
	claim2 := ss2.ClaimPartition(s.ns, "foo", 1, cancelCh)
	defer claim2()
	owner, err := ss2.kazooModel.GetPartitionOwner("foo", 1)
	c.Assert(err, IsNil)
	c.Assert(owner, Equals, "m2")
}
func (s *SubscriberSuite) TestClaimPartitionCanceled(c *C) {
	cfg1 := newConfig("m1")
	ss1 := Spawn(s.ns.NewChild("m1"), "g1", cfg1, s.zkConn)
	defer ss1.Stop()
	cfg2 := newConfig("m2")
	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.zkConn)
	defer ss2.Stop()
	cancelCh1 := make(chan none.T)
	cancelCh2 := make(chan none.T)
	wg := &sync.WaitGroup{}
	claim1 := ss1.ClaimPartition(s.ns, "foo", 1, cancelCh1)
	wg.Add(1)
	go func() {
		defer wg.Done()
		time.Sleep(300 * time.Millisecond)
		claim1()
	}()
	wg.Add(1)
	go func() {
		defer wg.Done()
		time.Sleep(150 * time.Millisecond)
		close(cancelCh2)
	}()
	claim2 := ss2.ClaimPartition(s.ns, "foo", 1, cancelCh2)
	defer claim2()
	owner, err := ss2.kazooModel.GetPartitionOwner("foo", 1)
	c.Assert(err, IsNil)
	c.Assert(owner, Equals, "m1")
	wg.Wait()
}