fix custom consumer group error channel lifecycle

[lionelvillard]

Fixes #174

Proposed Changes

• Scope consumer group handler to a session.
• Scope error channel to a consumer group

Release Note

🐛 Fix crash in Kafka consumer when a rebalance occurs

Docs

[codecov]

Codecov Report

Merging #263 (1a1f566) into master (042279b) will decrease coverage by 0.01%.
The diff coverage is 87.50%.

@@            Coverage Diff             @@
##           master     #263      +/-   ##
==========================================
- Coverage   75.68%   75.67%   -0.02%     
==========================================
  Files         114      114              
  Lines        4479     4481       +2     
==========================================
+ Hits         3390     3391       +1     
- Misses        887      888       +1     
  Partials      202      202              

Impacted Files	Coverage Δ
pkg/common/consumer/consumer_factory.go	75.75% <80.00%> (-1.67%)	⬇️
pkg/common/consumer/consumer_handler.go	80.95% <100.00%> (ø)

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 042279b...db90916. Read the comment docs.

[lberk]

awesome! thanks for putting this together. Given the nature of the crash -- is there any way we could add an e2e test to make sure we catch this moving forward? Delete the sink during a session?

[lionelvillard]

I still need to find a good way to test this. The problem with deleting the sink is it takes 4-5mn for the buffer to be depleted (not sure exactly why).

The way to reproduce is:

1. send events to a broken sink
2. trigger a rebalance by changing the number of replicas on the adapter
3. wait a bit for ConsumerHandler to Cleanup
4. a new consumer handler is created (no crash)

[knative-metrics-robot]

The following is the coverage report on the affected files.
Say /test pull-knative-sandbox-eventing-kafka-go-coverage to re-run this coverage report

File	Old Coverage	New Coverage	Delta
pkg/common/consumer/consumer_factory.go	83.9%	81.8%	-2.1

[lberk]

/lgtm
/approve

[knative-prow-robot]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: lberk, lionelvillard

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

• OWNERS [lberk,lionelvillard]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[pierDipi]

/cherrypick release-0.19

[knative-prow-robot]

@pierDipi: new pull request created: #264

In response to this:

/cherrypick release-0.19

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.