Fix PartitionConsumer race condition on Close. (#258)

* Use Sarama AsyncClose for Sarama Cluster AsyncClose * Async Close PartitionConsumer in Consumer.Close()
bsm · Aug 15, 2018 · 6230ffb · 6230ffb
1 parent 7e67d87
commit 6230ffb
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/consumer.go b/consumer.go
@@ -295,7 +295,7 @@ func (c *Consumer) Close() (err error) {
 		for range c.errors {
 		}
 		for p := range c.partitions {
-			_ = p.Close()
+			p.AsyncClose()
 		}
 		for range c.notifications {
 		}

diff --git a/partitions.go b/partitions.go
@@ -86,7 +86,7 @@ func (c *partitionConsumer) InitialOffset() int64 { return c.initialOffset }
 // AsyncClose implements PartitionConsumer
 func (c *partitionConsumer) AsyncClose() {
 	c.closeOnce.Do(func() {
-		c.closeErr = c.PartitionConsumer.Close()
+		c.PartitionConsumer.AsyncClose()
 		close(c.dying)
 	})
 }