Account for missing registration (fixes #124)

CHANGELOG.md

@@ -11,7 +11,9 @@ Fixed:
 * [#120](https://github.com/mailgun/kafka-pixy/issues/120) Consumption from a
   topic stopped for a group.
 * [#123](https://github.com/mailgun/kafka-pixy/issues/123) Inexplicable offset
-  manager timeouts
+  manager timeouts.
+* [#124](https://github.com/mailgun/kafka-pixy/issues/124) Subscription to a 
+  topic fails indefinitely after ZooKeeper connection loss.
 
 #### Version 0.14.0 (2017-09-11)
 

Gopkg.toml

@@ -42,7 +42,7 @@
   version = "1.0.1"
 
 [[constraint]]
-  branch = "master"
+  branch = "maxim/develop"
   name = "github.com/mailgun/kazoo-go"
 
 [[constraint]]

consumer/subscriber/subscriber.go

@@ -276,13 +276,17 @@ func (ss *T) submitTopics(topics []string) error {
 		ss.registered = false
 		return nil
 	}
-	var err error
+
 	if ss.registered {
-		err = ss.groupMemberZNode.UpdateRegistration(topics)
-	} else {
-		err = ss.groupMemberZNode.Register(topics)
+		err := ss.groupMemberZNode.UpdateRegistration(topics)
+		if err != kazoo.ErrInstanceNotRegistered {
+			return errors.Wrap(err, "failed to update registration")
+		}
+		ss.registered = false
+		ss.actDesc.Log().Errorf("Registration disappeared")
 	}
-	for err != nil {
+
+	if err := ss.groupMemberZNode.Register(topics); err != nil {
 		return errors.Wrap(err, "failed to register")
 	}
 	ss.registered = true

consumer/subscriber/subscriber_test.go

@@ -221,6 +221,38 @@ func (s *SubscriberSuite) TestRedundantUpdateBug(c *C) {
 	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.kazooClt)
+	defer ss1.Stop()
+	cfg2 := newConfig("m2")
+	ss2 := Spawn(s.ns.NewChild("m2"), "g1", cfg2, s.kazooClt)
+	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)
+
+	// Simulate subscription ephemeral node expiration by removing it manually.
+	gm := s.kazooClt.Consumergroup("g1").Instance("m1")
+	gm.Deregister()
+
+	// When
+	ss1.Topics() <- []string{"foo", "kaboom"}
+
+	// Then
+	membership = map[string][]string{
+		"m1": {"foo", "kaboom"},
+		"m2": {"bazz", "blah", "foo"}}
+	assertSubscription(c, ss1.Subscriptions(), membership, 3*time.Second)
+	assertSubscription(c, ss2.Subscriptions(), membership, 3*time.Second)
+}
+
 // When a group registrator claims a topic partitions it becomes its owner.
 func (s *SubscriberSuite) TestClaimPartition(c *C) {
 	cfg := newConfig("m1")