cancel consumer control messages when channel is closed

bwNetFlow · Nov 7, 2018 · 4356a0f · 4356a0f
1 parent ec8dce0
commit 4356a0f
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/kafka-listener.go b/kafka-listener.go
@@ -82,6 +82,7 @@ func handleControlMessages() {
 	for {
 		ctrlMsg, ok := <-ctrlChan
 		if !ok {
+			kafkaConn.CancelConsumerControlMessages()
 			return
 		}
 		partition := ctrlMsg.Partition