Repository: knative-sandbox/eventing-kafka
Pull request #264 (2020-12-08 16:55:19)

[release-0.19] fix custom consumer group error channel lifecycle
This is an automated cherry-pick of #263

```release-note
🐛 Fix crash in Kafka consumer when a rebalance occurs
```

google-cla[bot]: All (the pull request submitter and all commit authors) CLAs are signed, **but** one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project.  Please have them confirm that by leaving a comment that contains only `@googlebot I consent.` in this pull request.

*Note to project maintainer:* There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent.  In those cases, you can manually confirm consent of the commit author(s), and set the `cla` label to `yes` (if enabled on your project).

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka%2Fpull%2F264) for more info**.

<!-- need_author_consent -->
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/264#" title="Author self-approved">knative-prow-robot</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **travis-minke-sap** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @travis-minke-sap` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/release-0.19/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["travis-minke-sap"]} -->
pierDipi: /assign @lionelvillard 
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/264?src=pr&el=h1) Report
> :exclamation: No coverage uploaded for pull request base (`release-0.19@40d3d06`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/264/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/264?src=pr&el=tree)

```diff
@@               Coverage Diff               @@
##             release-0.19     #264   +/-   ##
===============================================
  Coverage                ?   75.35%           
===============================================
  Files                   ?      114           
  Lines                   ?     4410           
  Branches                ?        0           
===============================================
  Hits                    ?     3323           
  Misses                  ?      883           
  Partials                ?      204           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/264?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/264?src=pr&el=footer). Last update [40d3d06...7e20d78](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/264?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

pierDipi: It needs consent.
matzew: Funny, why is this cherry picked w/ the prow-bot ? :-) 

@lionelvillard I think you need to `@googlebot I consent.` ? 

/approve 
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/264#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/264#issuecomment-741599816" title="Approved">matzew</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/release-0.19/OWNERS)~~ [matzew]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
google-cla[bot]: All (the pull request submitter and all commit authors) CLAs are signed, **but** one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project.  Please have them confirm that by leaving a comment that contains only `@googlebot I consent.` in this pull request.

*Note to project maintainer:* There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent.  In those cases, you can manually confirm consent of the commit author(s), and set the `cla` label to `yes` (if enabled on your project).

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka%2Fpull%2F264) for more info**.

<!-- need_author_consent -->
lionelvillard: @googlebot I consent

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)
 
 		})
 	}