fix graceful shutdown

batch_consumer.go

@@ -76,6 +76,8 @@ func (b *batchConsumer) startBatch() {
 			b.consume(&messages, &commitMessages)
 		case msg, ok := <-b.incomingMessageStream:
 			if !ok {
+				close(b.batchConsumingStream)
+				close(b.messageProcessedStream)
 				return
 			}
 

consumer.go

@@ -68,6 +68,8 @@ func (c *consumer) startBatch() {
 			c.consume(&messages, &commitMessages)
 		case msg, ok := <-c.incomingMessageStream:
 			if !ok {
+				close(c.singleConsumingStream)
+				close(c.messageProcessedStream)
 				return
 			}
 

consumer_base.go

@@ -160,10 +160,8 @@ func (c *base) Stop() error {
 		c.cancelFn()
 		c.quit <- struct{}{}
 		close(c.incomingMessageStream)
-		close(c.singleConsumingStream)
-		close(c.batchConsumingStream)
-		close(c.messageProcessedStream)
 		c.wg.Wait()
+
 		err = c.r.Close()
 	})