diff --git a/kafka-listener.go b/kafka-listener.go
index a6e7af5..137dcde 100644
--- 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