Fix race on error channel in consumer factory (knative-extensions#364)

* Fix knative-extensions#342

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Commit mistake

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

pkg/common/consumer/consumer_factory.go

@@ -48,7 +48,6 @@ func (c *customConsumerGroup) Errors() <-chan error {
 
 func (c *customConsumerGroup) Close() error {
  c.cancel()
- close(c.handlerErrorChannel)
  return c.ConsumerGroup.Close()
 }
 
@@ -64,6 +63,7 @@ func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(groupID string, topics
  ctx, cancel := context.WithCancel(context.Background())
 
  go func() {
+ defer close(errorCh)
  for {
  consumerHandler := NewConsumerHandler(logger, handler, errorCh)