Drop dependency on wvanbergen/kazoo-go

Author: horkhe

Gopkg.lock

@@ -41,10 +41,6 @@
   name = "github.com/sirupsen/logrus"
   version = "1.0.1"
 
-[[constraint]]
-  branch = "master"
-  name = "github.com/wvanbergen/kazoo-go"
-
 [[constraint]]
   name = "google.golang.org/grpc"
   version = "1.15.0"

Gopkg.toml

@@ -11,7 +11,6 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/pkg/errors"
-	"github.com/wvanbergen/kazoo-go"
 	"gopkg.in/yaml.v2"
 )
 
@@ -271,18 +270,6 @@ func (pc PartitionerConstructor) ToPartitionerConstructor() (sarama.PartitionerC
 	return v, nil
 }
 
-func (p *Proxy) KazooCfg() *kazoo.Config {
-	kazooCfg := kazoo.NewConfig()
-	kazooCfg.Chroot = p.ZooKeeper.Chroot
-	// ZooKeeper documentation says following about the session timeout: "The
-	// current (ZooKeeper) implementation requires that the timeout be a
-	// minimum of 2 times the tickTime (as set in the server configuration) and
-	// a maximum of 20 times the tickTime". The default tickTime is 2 seconds.
-	// See http://zookeeper.apache.org/doc/trunk/zookeeperProgrammers.html#ch_zkSessions
-	kazooCfg.Timeout = 15 * time.Second
-	return kazooCfg
-}
-
 // SaramaProducerCfg returns a config for sarama producer.
 func (p *Proxy) SaramaProducerCfg() *sarama.Config {
 	saramaCfg := sarama.NewConfig()

config/config.go

@@ -1,6 +1,8 @@
 package consumerimpl
 
 import (
+	"time"
+
 	"github.com/Shopify/sarama"
 	"github.com/mailgun/kafka-pixy/actor"
 	"github.com/mailgun/kafka-pixy/config"
@@ -9,7 +11,7 @@ import (
 	"github.com/mailgun/kafka-pixy/consumer/groupcsm"
 	"github.com/mailgun/kafka-pixy/offsetmgr"
 	"github.com/pkg/errors"
-	"github.com/wvanbergen/kazoo-go"
+	"github.com/samuel/go-zookeeper/zk"
 )
 
 // T is a Kafka consumer implementation that automatically maintains consumer
@@ -29,7 +31,7 @@ type t struct {
 	cfg        *config.Proxy
 	dispatcher *dispatcher.T
 	kafkaClt   sarama.Client
-	kazooClt   *kazoo.Kazoo
+	zkConn     *zk.Conn
 	offsetMgrF offsetmgr.Factory
 }
 
@@ -41,7 +43,13 @@ func Spawn(parentActDesc *actor.Descriptor, cfg *config.Proxy, offsetMgrF offset
 		return nil, errors.Wrap(err, "failed to create Kafka client for message streams")
 	}
 
-	kazooClt, err := kazoo.NewKazoo(cfg.ZooKeeper.SeedPeers, cfg.KazooCfg())
+	// ZooKeeper documentation says following about the session timeout: "The
+	// current (ZooKeeper) implementation requires that the timeout be a
+	// minimum of 2 times the tickTime (as set in the server configuration) and
+	// a maximum of 20 times the tickTime". The default tickTime is 2 seconds.
+	// See http://zookeeper.apache.org/doc/trunk/zookeeperProgrammers.html#ch_zkSessions
+	sessionTimeout := 15 * time.Second
+	zkConn, _, err := zk.Connect(cfg.ZooKeeper.SeedPeers, sessionTimeout)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create kazoo.Kazoo")
 	}
@@ -51,7 +59,7 @@ func Spawn(parentActDesc *actor.Descriptor, cfg *config.Proxy, offsetMgrF offset
 		cfg:        cfg,
 		kafkaClt:   kafkaClt,
 		offsetMgrF: offsetMgrF,
-		kazooClt:   kazooClt,
+		zkConn:     zkConn,
 	}
 	c.dispatcher = dispatcher.Spawn(c.actDesc, c, c.cfg)
 	return c, nil
@@ -73,7 +81,7 @@ func (c *t) AsyncConsume(group, topic string) <-chan consumer.Response {
 // implements `consumer.T`
 func (c *t) Stop() {
 	c.dispatcher.Stop()
-	c.kazooClt.Close()
+	c.zkConn.Close()
 	c.kafkaClt.Close()
 }
 
@@ -84,7 +92,7 @@ func (c *t) KeyOf(rq consumer.Request) dispatcher.Key {
 
 // implements `dispatcher.Factory`.
 func (c *t) SpawnChild(childSpec dispatcher.ChildSpec) {
-	groupcsm.Spawn(c.actDesc, childSpec, c.cfg, c.kafkaClt, c.kazooClt, c.offsetMgrF)
+	groupcsm.Spawn(c.actDesc, childSpec, c.cfg, c.kafkaClt, c.zkConn, c.offsetMgrF)
 }
 
 // String returns a string ID of this instance to be used in logs.

consumer/consumerimpl/consumerimpl.go

@@ -19,7 +19,7 @@ import (
 	"github.com/mailgun/kafka-pixy/offsetmgr"
 	"github.com/mailgun/kafka-pixy/prettyfmt"
 	"github.com/pkg/errors"
-	"github.com/wvanbergen/kazoo-go"
+	"github.com/samuel/go-zookeeper/zk"
 )
 
 // groupConsumer manages a fleet of topic consumers and disposes of those that
@@ -33,7 +33,7 @@ type T struct {
 	group       string
 	dispatcher  *dispatcher.T
 	kafkaClt    sarama.Client
-	kazooClt    *kazoo.Kazoo
+	zkConn      *zk.Conn
 	msgFetcherF msgfetcher.Factory
 	offsetMgrF  offsetmgr.Factory
 	subscriber  *subscriber.T
@@ -45,8 +45,7 @@ type T struct {
 }
 
 func Spawn(parentActDesc *actor.Descriptor, childSpec dispatcher.ChildSpec,
-	cfg *config.Proxy, kafkaClt sarama.Client, kazooClt *kazoo.Kazoo,
-	offsetMgrF offsetmgr.Factory,
+	cfg *config.Proxy, kafkaClt sarama.Client, zkConn *zk.Conn, offsetMgrF offsetmgr.Factory,
 ) *T {
 	group := string(childSpec.Key())
 	actDesc := parentActDesc.NewChild(fmt.Sprintf("%s", group))
@@ -56,13 +55,13 @@ func Spawn(parentActDesc *actor.Descriptor, childSpec dispatcher.ChildSpec,
 		cfg:          cfg,
 		group:        group,
 		kafkaClt:     kafkaClt,
-		kazooClt:     kazooClt,
+		zkConn:       zkConn,
 		offsetMgrF:   offsetMgrF,
 		multiplexers: make(map[string]*multiplexer.T),
 		topicCsmCh:   make(chan *topiccsm.T, cfg.Consumer.ChannelBufferSize),
 	}
 
-	gc.subscriber = subscriber.Spawn(gc.actDesc, gc.group, gc.cfg, gc.kazooClt)
+	gc.subscriber = subscriber.Spawn(gc.actDesc, gc.group, gc.cfg, gc.zkConn)
 	gc.msgFetcherF = msgfetcher.SpawnFactory(gc.actDesc, gc.cfg, gc.kafkaClt)
 	actor.Spawn(gc.actDesc, &gc.wg, gc.run)
 

consumer/groupcsm/groupcsm.go

@@ -15,7 +15,6 @@ import (
 	"github.com/mailgun/kafka-pixy/testhelpers"
 	"github.com/mailgun/kafka-pixy/testhelpers/kafkahelper"
 	log "github.com/sirupsen/logrus"
-	"github.com/wvanbergen/kazoo-go"
 	. "gopkg.in/check.v1"
 )
 
@@ -58,10 +57,7 @@ func (s *PartitionCsmSuite) SetUpTest(c *C) {
 
 	s.ns = actor.Root().NewChild("T")
 
-	kazooClt, err := kazoo.NewKazoo(testhelpers.ZookeeperPeers, kazoo.NewConfig())
-	c.Assert(err, IsNil)
-
-	s.groupMember = subscriber.Spawn(s.ns, group, s.cfg, kazooClt)
+	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())