Repository: knative-sandbox/eventing-kafka (go)

Issue:
KafkaSource dispatcher crashes when sink is deleted (and session closed?)
**Describe the bug**

Getting this error:

```
{"level":"debug","ts":"2020-11-06T21:37:51.590Z","caller":"adapter/adapter.go:136","msg":"Error while sending the message{error 26 0  Post \"http://event-display.sources-kafka-keda.svc.cluster.local/\": dial tcp 10.96.140.159:80: connect: connection refused}"}
{"level":"info","ts":"2020-11-06T21:37:51.590Z","caller":"consumer/consumer_handler.go:83","msg":"Failure while handling a message","topic":"products","partition":7,"offset":14373,"error":"Post \"http://event-display.sources-kafka-keda.svc.cluster.local/\": dial tcp 10.96.140.159:80: connect: connection refused"}
panic: send on closed channel

goroutine 349 [running]:
knative.dev/eventing-kafka/pkg/common/consumer.(*SaramaConsumerHandler).ConsumeClaim(0xc000706480, 0x20eb100, 0xc0007a9c00, 0x20dd6a0, 0xc000a3ef30, 0x8, 0x7)
        knative.dev/eventing-kafka/pkg/common/consumer/consumer_handler.go:84 +0xbf3
github.com/Shopify/sarama.(*consumerGroupSession).consume(0xc0007a9c00, 0xc0008601f0, 0x8, 0x7)
        github.com/Shopify/sarama@v1.27.0/consumer_group.go:690 +0x27f
github.com/Shopify/sarama.newConsumerGroupSession.func2(0xc0007a9c00, 0xc0008601f0, 0x8, 0xc000000007)
        github.com/Shopify/sarama@v1.27.0/consumer_group.go:615 +0x87
created by github.com/Shopify/sarama.newConsumerGroupSession
        github.com/Shopify/sarama@v1.27.0/consumer_group.go:607 +0x526
```

 **Expected behavior**
A clear and concise description of what you expected to happen.

no crash

 **To Reproduce**
Steps to reproduce the behavior.

TDB. In a nutshell, create a KafkaSource, scale it to few pods, kill the sink and observe the crash. 

 **Knative release version**

 **Additional context**
Add any other context about the problem here such as proposed priority




Pull request description:
fix custom consumer group error channel lifecycle
Fixes #174

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Scope consumer group handler to a session. 
- Scope error channel to a consumer group 
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
🐛 Fix crash in Kafka consumer when a rebalance occurs
```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/pkg/common/consumer/consumer_factory.go b/pkg/common/consumer/consumer_factory.go
index 92f2ab48a..76faf69ab 100644
--- a/pkg/common/consumer/consumer_factory.go
+++ b/pkg/common/consumer/consumer_factory.go
@@ -48,6 +48,7 @@ func (c *customConsumerGroup) Errors() <-chan error {
 
 func (c *customConsumerGroup) Close() error {
 	c.cancel()
+	close(c.handlerErrorChannel)
 	return c.ConsumerGroup.Close()
 }
 
@@ -59,18 +60,19 @@ func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(groupID string, topics
 		return nil, err
 	}
 
-	consumerHandler := NewConsumerHandler(logger, handler)
-
+	errorCh := make(chan error, 10)
 	ctx, cancel := context.WithCancel(context.Background())
 
 	go func() {
 		for {
+			consumerHandler := NewConsumerHandler(logger, handler, errorCh)
+
 			err := consumerGroup.Consume(ctx, topics, &consumerHandler)
 			if err == sarama.ErrClosedConsumerGroup {
 				return
 			}
 			if err != nil {
-				consumerHandler.errors <- err
+				errorCh <- err
 			}
 
 			select {
@@ -81,7 +83,7 @@ func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(groupID string, topics
 		}
 	}()
 
-	return &customConsumerGroup{cancel, consumerHandler.errors, consumerGroup}, err
+	return &customConsumerGroup{cancel, errorCh, consumerGroup}, err
 }
 
 func NewConsumerGroupFactory(addrs []string, config *sarama.Config) KafkaConsumerGroupFactory {
diff --git a/pkg/common/consumer/consumer_handler.go b/pkg/common/consumer/consumer_handler.go
index 5cd03eb87..7722ec0a3 100644
--- a/pkg/common/consumer/consumer_handler.go
+++ b/pkg/common/consumer/consumer_handler.go
@@ -19,7 +19,6 @@ package consumer
 import (
 	"context"
 	"fmt"
-	"sync"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -38,29 +37,28 @@ type SaramaConsumerHandler struct {
 	handler KafkaConsumerHandler
 
 	logger *zap.SugaredLogger
+
 	// Errors channel
-	closeErrors sync.Once
-	errors      chan error
+	errors chan error
 }
 
-func NewConsumerHandler(logger *zap.SugaredLogger, handler KafkaConsumerHandler) SaramaConsumerHandler {
+func NewConsumerHandler(logger *zap.SugaredLogger, handler KafkaConsumerHandler, errorsCh chan error) SaramaConsumerHandler {
 	return SaramaConsumerHandler{
 		logger:  logger,
 		handler: handler,
-		errors:  make(chan error, 10), // Some buffering to avoid blocking the message processing
+		errors:  errorsCh,
 	}
 }
 
 // Setup is run at the beginning of a new session, before ConsumeClaim
 func (consumer *SaramaConsumerHandler) Setup(sarama.ConsumerGroupSession) error {
+	consumer.logger.Info("setting up handler")
 	return nil
 }
 
 // Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
 func (consumer *SaramaConsumerHandler) Cleanup(session sarama.ConsumerGroupSession) error {
-	consumer.closeErrors.Do(func() {
-		close(consumer.errors)
-	})
+	consumer.logger.Info("cleanup handler")
 	return nil
 }
 
@@ -85,6 +83,7 @@ func (consumer *SaramaConsumerHandler) ConsumeClaim(session sarama.ConsumerGroup
 			consumer.logger.Infow("Failure while handling a message", zap.String("topic", message.Topic), zap.Int32("partition", message.Partition), zap.Int64("offset", message.Offset), zap.Error(err))
 			consumer.errors <- err
 		}
+
 		if mustMark {
 			session.MarkMessage(message, "") // Mark kafka message as processed
 			if ce := consumer.logger.Desugar().Check(zap.DebugLevel, "debugging"); ce != nil {


Test patch:
diff --git a/pkg/common/consumer/consumer_factory_test.go b/pkg/common/consumer/consumer_factory_test.go
index e975a5cf9..78b329a67 100644
--- a/pkg/common/consumer/consumer_factory_test.go
+++ b/pkg/common/consumer/consumer_factory_test.go
@@ -42,7 +42,7 @@ func (m *mockConsumerGroup) Consume(ctx context.Context, topics []string, handle
 			m.generateErrorOnce.Do(func() {
 				h := handler.(*SaramaConsumerHandler)
 				h.errors <- errors.New("cgh")
-				_ = h.Cleanup(nil)
+				close(h.errors)
 			})
 		}()
 	}
diff --git a/pkg/common/consumer/consumer_handler_test.go b/pkg/common/consumer/consumer_handler_test.go
index a755bfb2a..f1e96919b 100644
--- a/pkg/common/consumer/consumer_handler_test.go
+++ b/pkg/common/consumer/consumer_handler_test.go
@@ -133,7 +133,8 @@ func Test(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(fmt.Sprintf("shouldErr: %v, shouldMark: %v", test.shouldErr, test.shouldMark), func(t *testing.T) {
-			cgh := NewConsumerHandler(zap.NewNop().Sugar(), test)
+			errorCh := make(chan error, 1)
+			cgh := NewConsumerHandler(zap.NewNop().Sugar(), test, errorCh)
 
 			session := mockConsumerGroupSession{}
 			claim := mockConsumerGroupClaim{msg: &mockMessage}
@@ -155,6 +156,7 @@ func Test(t *testing.T) {
 			}
 
 			_ = cgh.Cleanup(&session)
+			close(errorCh)
 
 		})
 	}