Repository: knative-sandbox/eventing-kafka
Pull request #724 (2021-06-18 17:15:48)

Create stop/start ConsumerGroup Manager with Control-Protocol
Subtask of #588 to implement Stop/Start ConsumerGroups in response to control-protocol messaging.

## Proposed Changes

- Created controlprotocol/controlprotocol.go, a small wrapper around the knative.dev/control-protocol ControlServer
- Created consumer/consumer_manager.go, a container that allows temporary stop/start of managed Sarama ConsumerGroups.
- Split the KafkaConsumerGroupFactory's StartConsumerGroup function into createConsumerGroup and startExistingConsumerGroup (no change to the way external callers use StartConsumerGroup)
- Changed the distributed dispatcher to use the new KafkaConsumerGroupManager

Note that the "stop" and "start" of a managed ConsumerGroup is an abstraction of value to the user of the manager; internally this is accomplished by closing and recreating groups with the same Group ID.

If no control-protocol messages are sent to the dispatcher, visible behavior is unchanged .


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/724#" title="Author self-approved">eric-sap</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/main/OWNERS)~~ [eric-sap]

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":[]} -->
knative-prow-robot: @eric-sap: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 2c413de342b1e2ef48c00d76c7d907963f1993fe | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-unit-tests/1405937310452682752) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/724). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405937310490431488/artifacts/line-cov.html#file1) | 87.1% | 82.7% | -4.3
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405937310490431488/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405937310490431488/artifacts/line-cov.html#file3) | Do not exist | 0.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405937310490431488/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405937310490431488/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

knative-prow-robot: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 2c413de342b1e2ef48c00d76c7d907963f1993fe | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-unit-tests/1405937310452682752) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | 2c413de342b1e2ef48c00d76c7d907963f1993fe | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405937310490431488) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/724). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405946135679340544/artifacts/line-cov.html#file1) | 87.1% | 82.7% | -4.3
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405946135679340544/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405946135679340544/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405946135679340544/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405946135679340544/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

knative-prow-robot: @eric-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | 2c413de342b1e2ef48c00d76c7d907963f1993fe | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405937310490431488) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-unit-tests | e93d04f4389440bfcb3f77d66a54fcec5c5700ee | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-unit-tests/1405946135641591808) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/724). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405991498859679744/artifacts/line-cov.html#file1) | 87.1% | 97.5% | 10.5
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405991498859679744/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405991498859679744/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405991498859679744/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1405991498859679744/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

knative-prow-robot: @eric-sap: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | ddbd19fbd083b3928e1dc3048ab3ef67efe33fda | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-unit-tests/1405991498025013248) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/724). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406646869697761280/artifacts/line-cov.html#file1) | 87.1% | 97.5% | 10.5
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406646869697761280/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406646869697761280/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406646869697761280/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406646869697761280/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#724](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (1352ba3) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/11214e9bc178aa95b00057ab8fc0e0c2e81834c7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (11214e9) will **increase** coverage by `1.08%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@            Coverage Diff             @@
##             main     #724      +/-   ##
==========================================
+ Coverage   73.80%   74.89%   +1.08%     
==========================================
  Files         133      136       +3     
  Lines        5848     6010     +162     
==========================================
+ Hits         4316     4501     +185     
+ Misses       1311     1297      -14     
+ Partials      221      212       -9     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `96.66% <100.00%> (+18.17%)` | :arrow_up: |
| [pkg/common/consumer/consumer\_factory.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb25zdW1lci9jb25zdW1lcl9mYWN0b3J5Lmdv) | `85.36% <100.00%> (+16.94%)` | :arrow_up: |
| [pkg/common/consumer/consumer\_manager.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb25zdW1lci9jb25zdW1lcl9tYW5hZ2VyLmdv) | `100.00% <100.00%> (ø)` | |
| [...g/common/controlprotocol/commands/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb250cm9scHJvdG9jb2wvY29tbWFuZHMvY29uc3VtZXJncm91cC5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/common/controlprotocol/controlprotocol.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb250cm9scHJvdG9jb2wvY29udHJvbHByb3RvY29sLmdv) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [11214e9...1352ba3](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/724?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406652896493178880/artifacts/line-cov.html#file1) | 87.1% | 97.5% | 10.5
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406652896493178880/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406652896493178880/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406652896493178880/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406652896493178880/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406655690591703040/artifacts/line-cov.html#file1) | 87.1% | 97.5% | 10.5
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406655690591703040/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406655690591703040/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406655690591703040/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1406655690591703040/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407013572348743680/artifacts/line-cov.html#file1) | 87.1% | 97.5% | 10.5
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407013572348743680/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407013572348743680/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407013572348743680/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407013572348743680/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

travis-minke-sap: Looking good Eric - I added a few questions / nit-picky things ; )
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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407132806865227776/artifacts/line-cov.html#file1) | 87.1% | 97.5% | 10.5
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407132806865227776/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407132806865227776/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/consumer/managed_group.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407132806865227776/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407132806865227776/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407132806865227776/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 

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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407698112000036864/artifacts/line-cov.html#file1) | 87.1% | 97.5% | 10.5
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407698112000036864/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407698112000036864/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/consumer/managed_group.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407698112000036864/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407698112000036864/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1407698112000036864/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 

knative-prow-robot: @eric-sap: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-upgrade-tests | e7828d7e80943c0182afdfe1a7bc142929a25757 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-upgrade-tests/1407698111895179264) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/724). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
eric-sap: /test pull-knative-sandbox-eventing-kafka-upgrade-tests
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/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1408439135458299904/artifacts/line-cov.html#file1) | 87.1% | 97.5% | 10.5
[pkg/common/consumer/consumer_factory.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1408439135458299904/artifacts/line-cov.html#file2) | 78.4% | 87.2% | 8.8
[pkg/common/consumer/consumer_manager.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1408439135458299904/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/common/consumer/managed_group.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1408439135458299904/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/controlprotocol/controlprotocol.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/724/pull-knative-sandbox-eventing-kafka-go-coverage/1408439135458299904/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

eric-sap: /test pull-knative-sandbox-eventing-kafka-upgrade-tests
This test seems to have an error in VerifyContinualTests/SourceContinualTests but never finishes after that.  Not sure how that test works.. guess I'll dig into it if a retest doesn't work
travis-minke-sap: /lgtm

Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index d71ea2578..19ecd000e 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"strconv"
 	"strings"
+	"time"
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
@@ -44,6 +45,7 @@ import (
 	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	"knative.dev/eventing-kafka/pkg/client/informers/externalversions"
 	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol"
 	"knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
 	"knative.dev/eventing/pkg/kncloudevents"
@@ -137,6 +139,13 @@ func main() {
 		MaxIdleConnsPerHost: ekConfig.CloudEvents.MaxIdleConnsPerHost,
 	})
 
+	logger.Info("Initializing Control-Protocol Server")
+	controlProtocolServer, err := controlprotocol.NewServerHandler(ctx, controlprotocol.ServerPort)
+	if err != nil {
+		logger.Fatal("Failed To Initialize Control-Protocol Server - Terminating", zap.Error(err))
+	}
+	defer controlProtocolServer.Shutdown(5 * time.Second)
+
 	// Create The Dispatcher With Specified Configuration
 	dispatcherConfig := dispatch.DispatcherConfig{
 		Logger:          logger,
@@ -148,7 +157,7 @@ func main() {
 		MetricsRegistry: ekConfig.Sarama.Config.MetricRegistry,
 		SaramaConfig:    ekConfig.Sarama.Config,
 	}
-	dispatcher = dispatch.NewDispatcher(dispatcherConfig)
+	dispatcher = dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer)
 
 	// Watch The Secret For Changes
 	err = distributedcommonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, environment.ResyncPeriod, secretObserver)
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index e1dbf54fc..07c780ac3 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -34,6 +34,7 @@ import (
 	"knative.dev/eventing-kafka/pkg/common/client"
 	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
 	commonconsumer "knative.dev/eventing-kafka/pkg/common/consumer"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
 )
 
@@ -50,16 +51,15 @@ type DispatcherConfig struct {
 	SubscriberSpecs []eventingduck.SubscriberSpec
 }
 
-// SubscriberWrapper Defines A Knative Eventing SubscriberSpec Wrapper Enhanced With Sarama ConsumerGroup
+// SubscriberWrapper Defines A Knative Eventing SubscriberSpec Wrapper Enhanced With Sarama ConsumerGroup ID
 type SubscriberWrapper struct {
 	eventingduck.SubscriberSpec
-	GroupId       string
-	ConsumerGroup sarama.ConsumerGroup
+	GroupId string
 }
 
 // NewSubscriberWrapper Is The SubscriberWrapper Constructor
-func NewSubscriberWrapper(subscriberSpec eventingduck.SubscriberSpec, groupId string, consumerGroup sarama.ConsumerGroup) *SubscriberWrapper {
-	return &SubscriberWrapper{subscriberSpec, groupId, consumerGroup}
+func NewSubscriberWrapper(subscriberSpec eventingduck.SubscriberSpec, groupId string) *SubscriberWrapper {
+	return &SubscriberWrapper{subscriberSpec, groupId}
 }
 
 // Dispatcher Interface
@@ -72,28 +72,30 @@ type Dispatcher interface {
 // DispatcherImpl Is A Struct With Configuration & ConsumerGroup State
 type DispatcherImpl struct {
 	DispatcherConfig
-	subscribers          map[types.UID]*SubscriberWrapper
-	consumerGroupFactory commonconsumer.KafkaConsumerGroupFactory
-	consumerUpdateLock   sync.Mutex
-	messageDispatcher    channel.MessageDispatcher
-	MetricsStopChan      chan struct{}
-	MetricsStoppedChan   chan struct{}
+	subscribers        map[types.UID]*SubscriberWrapper
+	consumerUpdateLock sync.Mutex
+	messageDispatcher  channel.MessageDispatcher
+	MetricsStopChan    chan struct{}
+	MetricsStoppedChan chan struct{}
+	consumerMgr        commonconsumer.KafkaConsumerGroupManager
 }
 
 // Verify The DispatcherImpl Implements The Dispatcher Interface
 var _ Dispatcher = &DispatcherImpl{}
 
 // NewDispatcher Is The Dispatcher Constructor
-func NewDispatcher(dispatcherConfig DispatcherConfig) Dispatcher {
+func NewDispatcher(dispatcherConfig DispatcherConfig, controlServer controlprotocol.ServerHandler) Dispatcher {
+
+	consumerGroupManager := commonconsumer.NewConsumerGroupManager(dispatcherConfig.Logger, controlServer, dispatcherConfig.Brokers, dispatcherConfig.SaramaConfig)
 
 	// Create The DispatcherImpl With Specified Configuration
 	dispatcher := &DispatcherImpl{
-		DispatcherConfig:     dispatcherConfig,
-		subscribers:          make(map[types.UID]*SubscriberWrapper),
-		consumerGroupFactory: commonconsumer.NewConsumerGroupFactory(dispatcherConfig.Brokers, dispatcherConfig.SaramaConfig),
-		messageDispatcher:    channel.NewMessageDispatcher(dispatcherConfig.Logger),
-		MetricsStopChan:      make(chan struct{}),
-		MetricsStoppedChan:   make(chan struct{}),
+		DispatcherConfig:   dispatcherConfig,
+		subscribers:        make(map[types.UID]*SubscriberWrapper),
+		messageDispatcher:  channel.NewMessageDispatcher(dispatcherConfig.Logger),
+		MetricsStopChan:    make(chan struct{}),
+		MetricsStoppedChan: make(chan struct{}),
+		consumerMgr:        consumerGroupManager,
 	}
 
 	// Start Observing Metrics
@@ -148,7 +150,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 
 			// Create/Start A New ConsumerGroup With Custom Handler
 			handler := NewHandler(logger, groupId, &subscriberSpec)
-			consumerGroup, err := d.consumerGroupFactory.StartConsumerGroup(groupId, []string{d.Topic}, d.Logger.Sugar(), handler)
+			err := d.consumerMgr.StartConsumerGroup(groupId, []string{d.Topic}, d.Logger.Sugar(), handler)
 			if err != nil {
 
 				// Log & Return Failure
@@ -158,12 +160,12 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 			} else {
 
 				// Create A New SubscriberWrapper With The ConsumerGroup
-				subscriber := NewSubscriberWrapper(subscriberSpec, groupId, consumerGroup)
+				subscriber := NewSubscriberWrapper(subscriberSpec, groupId)
 
 				// Asynchronously Process ConsumerGroup's Error Channel
 				go func() {
 					logger.Info("ConsumerGroup Error Processing Initiated")
-					for err := range subscriber.ConsumerGroup.Errors() { // Closing ConsumerGroup Will Break Out Of This
+					for err := range d.consumerMgr.Errors(subscriber.GroupId) { // Closing ConsumerGroup Will Break Out Of This
 						logger.Error("ConsumerGroup Error", zap.Error(err))
 					}
 					logger.Info("ConsumerGroup Error Processing Terminated")
@@ -201,17 +203,14 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 // closeConsumerGroup closes the ConsumerGroup associated with a single Subscriber
 func (d *DispatcherImpl) closeConsumerGroup(subscriber *SubscriberWrapper) {
 
-	// Get The ConsumerGroup Associated with The Specified Subscriber
-	consumerGroup := subscriber.ConsumerGroup
-
 	// Create Logger With GroupId & Subscriber URI
 	logger := d.Logger.With(zap.String("GroupId", subscriber.GroupId), zap.String("URI", subscriber.SubscriberURI.String()))
 
 	// If The ConsumerGroup Is Valid
-	if consumerGroup != nil {
+	if d.consumerMgr.IsManaged(subscriber.GroupId) {
 
 		// Close The ConsumerGroup
-		err := consumerGroup.Close()
+		err := d.consumerMgr.CloseConsumerGroup(subscriber.GroupId)
 		if err != nil {
 			// Simply Log ConsumerGroup Close Failures
 			//   - Don't include in failedSubscriptions response as that is used to update Subscription Status.
@@ -261,24 +260,15 @@ func (d *DispatcherImpl) SecretChanged(ctx context.Context, secret *corev1.Secre
 		return
 	}
 
-	// Close all of the ConsumerGroups so that the settings can be changed
-	// Note: Not calling d.Shutdown because we don't want to close the metrics channel here
-	d.Logger.Info("Changes Detected In New Secret - Closing & Recreating Consumer Groups")
-	for _, subscriber := range d.subscribers {
-		d.closeConsumerGroup(subscriber)
-	}
-
 	// The only values in the secret that matter here are the username, password, and SASL type, which are all part
 	// of the SaramaConfig, so that's all that needs to be modified
 	d.DispatcherConfig.SaramaConfig = newConfig
 
 	// Replace The Dispatcher's ConsumerGroupFactory With Updated Version Using New Config
-	d.consumerGroupFactory = commonconsumer.NewConsumerGroupFactory(d.DispatcherConfig.Brokers, d.DispatcherConfig.SaramaConfig)
-
-	// Recreate the ConsumerGroups For Active Subscriptions
-	failedSubscriptions := d.UpdateSubscriptions(d.SubscriberSpecs)
-	if len(failedSubscriptions) > 0 {
-		d.Logger.Fatal("Failed To Subscribe Kafka Subscriptions Using Updated Secret", zap.Int("Count", len(failedSubscriptions)))
+	// Note:  This will close and recreate all of the managed consumer groups
+	err = d.consumerMgr.Reconfigure(d.DispatcherConfig.Brokers, d.DispatcherConfig.SaramaConfig)
+	if err != nil {
+		d.Logger.Error("Failed To Reconfigure Consumer Group Manager Using Updated Secret", zap.Error(err))
 	}
 }
 
diff --git a/pkg/common/consumer/consumer_factory.go b/pkg/common/consumer/consumer_factory.go
index 1ff21beba..0efe1eb36 100644
--- a/pkg/common/consumer/consumer_factory.go
+++ b/pkg/common/consumer/consumer_factory.go
@@ -13,6 +13,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 */
+
 package consumer
 
 import (
@@ -23,9 +24,13 @@ import (
 	"go.uber.org/zap"
 )
 
+// newConsumerGroup is a wrapper for the Sarama NewConsumerGroup function, to facilitate unit testing
 var newConsumerGroup = sarama.NewConsumerGroup
 
-// Kafka consumer factory creates the ConsumerGroup and start consuming the specified topic
+// consumeFunc is a function type that matches the Sarama ConsumerGroup's Consume function
+type consumeFunc func(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) error
+
+// KafkaConsumerGroupFactory creates the ConsumerGroup and start consuming the specified topic
 type KafkaConsumerGroupFactory interface {
 	StartConsumerGroup(groupID string, topics []string, logger *zap.SugaredLogger, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error)
 }
@@ -42,7 +47,7 @@ type customConsumerGroup struct {
 	releasedCh chan bool
 }
 
-// Merge handler errors chan and consumer group error chan
+// Errors merges handler errors chan and consumer group error chan
 func (c *customConsumerGroup) Errors() <-chan error {
 	return mergeErrorChannels(c.ConsumerGroup.Errors(), c.handlerErrorChannel)
 }
@@ -58,11 +63,31 @@ func (c *customConsumerGroup) Close() error {
 
 var _ sarama.ConsumerGroup = (*customConsumerGroup)(nil)
 
+// StartConsumerGroup creates a new customConsumerGroup and starts a Consume goroutine on it
 func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(groupID string, topics []string, logger *zap.SugaredLogger, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
-	consumerGroup, err := newConsumerGroup(c.addrs, groupID, c.config)
+	consumerGroup, err := c.createConsumerGroup(groupID)
 	if err != nil {
 		return nil, err
 	}
+	// Start the consumerGroup.Consume function in a separate goroutine
+	return c.startExistingConsumerGroup(consumerGroup, consumerGroup.Consume, topics, logger, handler, options...), nil
+}
+
+// createConsumerGroup creates a Sarama ConsumerGroup using the newConsumerGroup wrapper, with the
+// factory's internal brokers and sarama config.
+func (c kafkaConsumerGroupFactoryImpl) createConsumerGroup(groupID string) (sarama.ConsumerGroup, error) {
+	return newConsumerGroup(c.addrs, groupID, c.config)
+}
+
+// startExistingConsumerGroup creates a goroutine that begins a custom Consume loop on the provided ConsumerGroup
+// This loop is cancelable via the function provided in the returned customConsumerGroup.
+func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
+	saramaGroup sarama.ConsumerGroup,
+	consume consumeFunc,
+	topics []string,
+	logger *zap.SugaredLogger,
+	handler KafkaConsumerHandler,
+	options ...SaramaConsumerHandlerOption) *customConsumerGroup {
 
 	errorCh := make(chan error, 10)
 	releasedCh := make(chan bool)
@@ -76,7 +101,7 @@ func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(groupID string, topics
 		for {
 			consumerHandler := NewConsumerHandler(logger, handler, errorCh, options...)
 
-			err := consumerGroup.Consume(ctx, topics, &consumerHandler)
+			err := consume(ctx, topics, &consumerHandler)
 			if err == sarama.ErrClosedConsumerGroup {
 				return
 			}
@@ -91,8 +116,7 @@ func (c kafkaConsumerGroupFactoryImpl) StartConsumerGroup(groupID string, topics
 			}
 		}
 	}()
-
-	return &customConsumerGroup{cancel, errorCh, consumerGroup, releasedCh}, err
+	return &customConsumerGroup{cancel, errorCh, saramaGroup, releasedCh}
 }
 
 func NewConsumerGroupFactory(addrs []string, config *sarama.Config) KafkaConsumerGroupFactory {
diff --git a/pkg/common/consumer/consumer_manager.go b/pkg/common/consumer/consumer_manager.go
new file mode 100644
index 000000000..7510317e4
--- /dev/null
+++ b/pkg/common/consumer/consumer_manager.go
@@ -0,0 +1,309 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/*
+The KafkaConsumerGroupManager is an optional piece of support infrastructure that may be used to
+automatically manage the starting and stopping (or "pausing and resuming") of Sarama ConsumerGroups
+in response to messages sent via the control-protocol.
+The manager uses the KafkaConsumerGroupFactory for underlying ConsumerGroup creation.
+
+Usage:
+- Create a ServerHandler and call NewConsumerGroupManager()
+- Use the manager's StartConsumerGroup() and CloseConsumerGroup() functions instead of creating
+  and closing sarama ConsumerGroups directly
+- Errors() obtains an error channel for the managed group (persists between stop/start actions)
+- IsManaged() returns true if a given GroupId is under management
+- Reconfigure() allows you to change consumer factory settings (automatically stopping and
+  restarting all managed ConsumerGroups)
+*/
+
+package consumer
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/Shopify/sarama"
+	"go.uber.org/multierr"
+	"go.uber.org/zap"
+	ctrlservice "knative.dev/control-protocol/pkg/service"
+
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+)
+
+// KafkaConsumerGroupManager keeps track of Sarama consumer groups and handles messages from control-protocol clients
+type KafkaConsumerGroupManager interface {
+	Reconfigure(brokers []string, config *sarama.Config) error
+	StartConsumerGroup(groupId string, topics []string, logger *zap.SugaredLogger, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) error
+	CloseConsumerGroup(groupId string) error
+	Errors(groupId string) <-chan error
+	IsManaged(groupId string) bool
+}
+
+// groupMap is a mapping of GroupIDs to managed Consumer Group pointers (fields in the managedGroup
+// are modified directly by the kafkaConsumerGroupManagerImpl receiver functions; the pointer makes
+// removing and re-adding them to the map unnecessary in those cases)
+type groupMap map[string]*managedGroup
+
+// kafkaConsumerGroupManagerImpl is the primary implementation of a KafkaConsumerGroupManager, which
+// handles control protocol messages and stopping/starting ("pausing/resuming") of ConsumerGroups.
+type kafkaConsumerGroupManagerImpl struct {
+	logger    *zap.Logger
+	server    controlprotocol.ServerHandler
+	factory   *kafkaConsumerGroupFactoryImpl
+	groups    groupMap
+	groupLock sync.RWMutex // Synchronizes write access to the groupMap
+}
+
+// Verify that the kafkaConsumerGroupManagerImpl satisfies the KafkaConsumerGroupManager interface
+var _ KafkaConsumerGroupManager = (*kafkaConsumerGroupManagerImpl)(nil)
+
+// NewConsumerGroupManager returns a new kafkaConsumerGroupManagerImpl as a KafkaConsumerGroupManager interface
+func NewConsumerGroupManager(logger *zap.Logger, serverHandler controlprotocol.ServerHandler, brokers []string, config *sarama.Config) KafkaConsumerGroupManager {
+
+	manager := &kafkaConsumerGroupManagerImpl{
+		logger:    logger,
+		server:    serverHandler,
+		groups:    make(groupMap),
+		factory:   &kafkaConsumerGroupFactoryImpl{addrs: brokers, config: config},
+		groupLock: sync.RWMutex{},
+	}
+
+	logger.Info("Registering Consumer Group Manager Control-Protocol Handlers")
+
+	// Add a handler that understands the StopConsumerGroupOpCode and stops the requested group
+	serverHandler.AddAsyncHandler(
+		commands.StopConsumerGroupOpCode,
+		commands.StopConsumerGroupResultOpCode,
+		&commands.ConsumerGroupAsyncCommand{},
+		func(ctx context.Context, commandMessage ctrlservice.AsyncCommandMessage) {
+			processAsyncGroupNotification(commandMessage, manager.stopConsumerGroup)
+		})
+
+	// Add a handler that understands the StartConsumerGroupOpCode and starts the requested group
+	serverHandler.AddAsyncHandler(
+		commands.StartConsumerGroupOpCode,
+		commands.StartConsumerGroupResultOpCode,
+		&commands.ConsumerGroupAsyncCommand{},
+		func(ctx context.Context, commandMessage ctrlservice.AsyncCommandMessage) {
+			processAsyncGroupNotification(commandMessage, manager.startConsumerGroup)
+		})
+
+	return manager
+}
+
+// Reconfigure will incorporate a new set of brokers and Sarama config settings into the manager
+// without requiring a new control-protocol server or losing the current map of managed groups.
+// It will stop and start all of the managed groups in the group map.
+func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sarama.Config) error {
+	m.logger.Info("Reconfigure Consumer Group Manager - Stopping All Managed Consumer Groups")
+	var multiErr error
+	for groupId := range m.groups {
+		err := m.stopConsumerGroup(groupId)
+		if err != nil {
+			multierr.AppendInto(&multiErr, err)
+		}
+	}
+	m.factory = &kafkaConsumerGroupFactoryImpl{addrs: brokers, config: config}
+	m.logger.Info("Reconfigure Consumer Group Manager - Starting All Managed Consumer Groups")
+	for groupId := range m.groups {
+		err := m.startConsumerGroup(groupId)
+		if err != nil {
+			multierr.AppendInto(&multiErr, err)
+		}
+	}
+	return multiErr
+}
+
+// StartConsumerGroup uses the consumer factory to create a new ConsumerGroup, add it to the list
+// of managed groups (for start/stop functionality) and start the Consume loop.
+func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(groupId string, topics []string, logger *zap.SugaredLogger, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) error {
+	groupLogger := m.logger.With(zap.String("GroupId", groupId))
+	groupLogger.Info("Creating New Managed ConsumerGroup")
+	group, err := m.factory.createConsumerGroup(groupId)
+	if err != nil {
+		groupLogger.Error("Failed To Create New Managed ConsumerGroup")
+		return err
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	// Add the Sarama ConsumerGroup we obtained from the factory to the managed group map,
+	// so that it can be stopped and started via control-protocol messages.
+	m.groupLock.Lock()
+	m.groups[groupId] = createManagedGroup(group, cancel)
+	m.groupLock.Unlock()
+
+	// Begin listening on the group's Errors() channel and write them to the managedGroup's errors channel
+	m.groups[groupId].transferErrors(ctx)
+
+	// consume is passed in to the KafkaConsumerGroupFactory so that it will call the manager's
+	// consume() function instead of the one on the internal sarama ConsumerGroup.  This allows the
+	// manager to continue to block in the Consume call while a group goes through a stop/start cycle.
+	consume := func(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) error {
+		logger.Debug("Consuming Messages On managed Consumer Group", zap.String("GroupId", groupId))
+		return m.consume(ctx, groupId, topics, handler)
+	}
+
+	// The only thing we really want from the factory is the cancel function for the customConsumerGroup
+	customGroup := m.factory.startExistingConsumerGroup(group, consume, topics, logger, handler, options...)
+	m.groups[groupId].cancelConsume = customGroup.cancel
+	return nil
+}
+
+// CloseConsumerGroup calls the Close function on the ConsumerGroup embedded in the managedGroup
+// associated with the given groupId, and also closes its managed errors channel.  It then removes the
+// group from management.
+func (m *kafkaConsumerGroupManagerImpl) CloseConsumerGroup(groupId string) error {
+	groupLogger := m.logger.With(zap.String("GroupId", groupId))
+	groupLogger.Info("Closing ConsumerGroup and removing from management")
+	managedGrp, ok := m.groups[groupId]
+	if !ok {
+		groupLogger.Warn("CloseConsumerGroup called on unmanaged group")
+		return fmt.Errorf("could not close consumer group with id '%s' - group is not present in the managed map", groupId)
+	}
+	// Make sure a managed group is "started" before closing the inner ConsumerGroup; otherwise anything
+	// waiting for the manager to restart the group will never return.
+	managedGrp.closeRestartChannel()
+	if managedGrp.cancelErrors != nil {
+		m.logger.Warn("The cancelErrors function of the managed group is nil")
+		managedGrp.cancelErrors() // This will terminate any transferError call that is waiting for a restart
+	}
+	if managedGrp.cancelConsume != nil {
+		m.logger.Warn("The cancelConsume function of the managed group is nil")
+		managedGrp.cancelConsume() // This will stop the factory's consume loop after the ConsumerGroup is closed
+	}
+	err := managedGrp.group.Close()
+	if err != nil {
+		groupLogger.Error("Failed To Close Managed ConsumerGroup", zap.Error(err))
+		return err
+	}
+
+	// Remove this groupId from the map so that manager functions may not be called on it
+	m.groupLock.Lock()
+	delete(m.groups, groupId)
+	m.groupLock.Unlock()
+
+	return nil
+}
+
+// Errors returns the errors channel of the managedGroup associated with the given groupId.  This channel
+// is different than using the Errors() channel of a ConsumerGroup directly, as it will remain open during
+//  a stop/start ("pause/resume") cycle
+func (m *kafkaConsumerGroupManagerImpl) Errors(groupId string) <-chan error {
+	if !m.IsManaged(groupId) {
+		return nil
+	}
+	return m.groups[groupId].errors
+}
+
+// IsManaged returns true if the given groupId corresponds to a managed ConsumerGroup
+func (m *kafkaConsumerGroupManagerImpl) IsManaged(groupId string) bool {
+	if _, ok := m.groups[groupId]; ok {
+		return true
+	}
+	return false
+}
+
+// Consume calls the Consume method of a managed consumer group, using a loop to call it again if that
+// group is restarted by the manager.  If the Consume call is terminated by some other mechanism, the
+// result will be returned to the caller.
+func (m *kafkaConsumerGroupManagerImpl) consume(ctx context.Context, groupId string, topics []string, handler sarama.ConsumerGroupHandler) error {
+	managedGrp, ok := m.groups[groupId]
+	if !ok {
+		return fmt.Errorf("consume called on nonexistent groupId '%s'", groupId)
+	}
+	for {
+		// Call the internal sarama ConsumerGroup's Consume function directly
+		err := managedGrp.group.Consume(ctx, topics, handler)
+		if !managedGrp.isStopped() {
+			m.logger.Debug("Managed Consume Finished Without Stop", zap.String("GroupId", groupId), zap.Error(err))
+			// This ConsumerGroup wasn't stopped by the manager, so pass the error along to the caller
+			return err
+		}
+		// Wait for the managed ConsumerGroup to be restarted
+		if !managedGrp.waitForStart(ctx) {
+			// Context was canceled; abort
+			m.logger.Debug("Managed Consume Canceled", zap.String("GroupId", groupId))
+			return fmt.Errorf("context was canceled waiting for group '%s' to start", groupId)
+		}
+	}
+}
+
+// stopConsumerGroups closes the managed ConsumerGroup identified by the provided groupId, and marks it
+// as "stopped" (that is, "able to be restarted" as opposed to being closed by something outside the manager)
+func (m *kafkaConsumerGroupManagerImpl) stopConsumerGroup(groupId string) error {
+	groupLogger := m.logger.With(zap.String("GroupId", groupId))
+	groupLogger.Info("Stopping Managed ConsumerGroup")
+	managedGrp, ok := m.groups[groupId]
+	if ok {
+		// The managedGroup's start channel must be created (that is, the managedGroup must be marked
+		// as "stopped") before closing the internal ConsumerGroup, otherwise the consume function would
+		// return control to the factory.
+		managedGrp.createRestartChannel()
+		// Close the inner sarama ConsumerGroup, which will cause our consume() function to stop
+		// and wait for the managedGroup to start again.
+		err := managedGrp.group.Close()
+		if err != nil {
+			groupLogger.Error("Failed To Close Managed ConsumerGroup", zap.Error(err))
+			// Don't leave the start channel open if the group.Close() call failed; that would be misleading
+			managedGrp.closeRestartChannel()
+			return err
+		}
+		return nil
+	}
+	groupLogger.Info("ConsumerGroup Not Managed - Ignoring Stop Request")
+	return fmt.Errorf("stop requested for consumer group not in managed list: %s", groupId)
+}
+
+// startConsumerGroups creates a new Consumer Group based on the groupId provided
+func (m *kafkaConsumerGroupManagerImpl) startConsumerGroup(groupId string) error {
+	groupLogger := m.logger.With(zap.String("GroupId", groupId))
+	groupLogger.Info("Starting Managed ConsumerGroup")
+	m.groupLock.Lock() // Don't allow m.groups to be modified while processing a start request
+	defer m.groupLock.Unlock()
+	if _, ok := m.groups[groupId]; ok {
+		group, err := m.factory.createConsumerGroup(groupId)
+		if err != nil {
+			groupLogger.Error("Failed To Restart Managed ConsumerGroup", zap.Error(err))
+			return err
+		}
+		m.groups[groupId].group = group
+		m.groups[groupId].closeRestartChannel() // Closing this allows the waitForStart function to finish
+		return nil
+	}
+	groupLogger.Info("ConsumerGroup Not Managed - Ignoring Start Request")
+	return fmt.Errorf("start requested for consumer group not in managed list: %s", groupId)
+}
+
+// processAsyncGroupNotification calls the provided groupFunction with whatever GroupId is contained
+// in the commandMessage, after verifying that the command version is correct.  It then calls the
+// appropriate Async response function on the commandMessage (NotifyFailed or NotifySuccess)
+func processAsyncGroupNotification(commandMessage ctrlservice.AsyncCommandMessage, groupFunction func(groupId string) error) {
+	if cmd, ok := commandMessage.ParsedCommand().(*commands.ConsumerGroupAsyncCommand); ok {
+		if cmd.Version != commands.ConsumerGroupAsyncCommandVersion {
+			commandMessage.NotifyFailed(fmt.Errorf("version mismatch; expected %d but got %d", commands.ConsumerGroupAsyncCommandVersion, cmd.Version))
+		} else {
+			err := groupFunction(cmd.GroupId)
+			if err != nil {
+				commandMessage.NotifyFailed(err)
+			} else {
+				commandMessage.NotifySuccess()
+			}
+		}
+	}
+}
diff --git a/pkg/common/consumer/managed_group.go b/pkg/common/consumer/managed_group.go
new file mode 100644
index 000000000..539444073
--- /dev/null
+++ b/pkg/common/consumer/managed_group.go
@@ -0,0 +1,115 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package consumer
+
+import (
+	"context"
+	"sync/atomic"
+
+	"github.com/Shopify/sarama"
+)
+
+// managedGroup contains information about a Sarama ConsumerGroup that is required to start (i.e. re-create) it
+type managedGroup struct {
+	group              sarama.ConsumerGroup
+	errors             chan error
+	restartWaitChannel chan struct{}
+	stopped            atomic.Value // Boolean value indicating channel is stopped
+	cancelErrors       func()
+	cancelConsume      func()
+}
+
+// createManagedGroup associates a Sarama ConsumerGroup and cancel function (usually from the factory)
+// inside a new managedGroup struct
+func createManagedGroup(group sarama.ConsumerGroup, cancel func()) *managedGroup {
+	managedGrp := managedGroup{
+		group:        group,
+		errors:       make(chan error),
+		cancelErrors: cancel,
+	}
+	// Initialize the atomic boolean value to false, otherwise it will be a nil interface
+	// (restartWaitChannel is only accessed when "stopped" is true)
+	managedGrp.stopped.Store(false)
+	return &managedGrp
+}
+
+// isStopped is an accessor for the restartWaitChannel channel's status, which if non-nil indicates
+// that the KafkaConsumerGroupManager stopped ("paused") the managed ConsumerGroup (versus it having
+// been closed outside the manager)
+func (m *managedGroup) isStopped() bool {
+	return m.stopped.Load().(bool)
+}
+
+// createRestartChannel sets the state of the managed group to "stopped" by creating the restartWaitChannel
+// channel that will be closed when the group is restarted.
+func (m *managedGroup) createRestartChannel() {
+	// Don't re-create the channel if it already exists (the managed group is already stopped)
+	if !m.isStopped() {
+		m.restartWaitChannel = make(chan struct{})
+		m.stopped.Store(true)
+	}
+}
+
+// closeRestartChannel sets the state of the managed group to "started" by closing the restartWaitChannel channel.
+func (m *managedGroup) closeRestartChannel() {
+	// If the managed group is already started, don't try to close the restart wait channel
+	if m.isStopped() {
+		m.stopped.Store(false)
+		close(m.restartWaitChannel)
+	}
+}
+
+// waitForStart will block until a stopped ("paused") ConsumerGroup has been restarted by the
+// KafkaConsumerGroupManager, returning true in that case or false if the context's cancel function is called
+func (m *managedGroup) waitForStart(ctx context.Context) bool {
+	if !m.isStopped() {
+		return true // group is already started; don't try to read from closed channel
+	}
+	// Wait for either the restartWaitChannel to be closed, or for the provided context to have its cancel function called.
+	select {
+	case <-m.restartWaitChannel:
+		return true
+	case <-ctx.Done():
+		return false
+	}
+}
+
+// transferErrors starts a goroutine that reads errors from the managedGroup's internal group.Errors() channel
+// and sends them to the m.errors channel.  This is done so that when the group.Errors() channel is closed during
+// a stop ("pause") of the group, the m.errors channel can remain open (so that users of the manager do not
+// receive a closed error channel during stop/start events).
+func (m *managedGroup) transferErrors(ctx context.Context) {
+	go func() {
+		for {
+			for groupErr := range m.group.Errors() {
+				m.errors <- groupErr
+			}
+			if !m.isStopped() {
+				// If the error channel was closed without the consumergroup being marked as stopped,
+				// or if we were unable to wait for the group to be restarted, that is outside
+				// of the manager's responsibility, so we are finished transferring errors.
+				close(m.errors)
+				return
+			}
+			// Wait for the manager to restart the Consumer Group before calling m.group.Errors() again
+			if !m.waitForStart(ctx) {
+				// Abort if the context was canceled
+				return
+			}
+		}
+	}()
+}
diff --git a/pkg/common/controlprotocol/controlprotocol.go b/pkg/common/controlprotocol/controlprotocol.go
new file mode 100644
index 000000000..e88c7cc39
--- /dev/null
+++ b/pkg/common/controlprotocol/controlprotocol.go
@@ -0,0 +1,130 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controlprotocol
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	ctrl "knative.dev/control-protocol/pkg"
+	"knative.dev/control-protocol/pkg/message"
+	"knative.dev/control-protocol/pkg/network"
+	ctrlservice "knative.dev/control-protocol/pkg/service"
+)
+
+const (
+	// ServerPort is the port used when initializing the control-protocol server
+	ServerPort = 8085
+)
+
+// AsyncHandlerFunc is an alias that matches the type used in the control-protocol's NewAsyncCommandHandler function
+type AsyncHandlerFunc = func(ctx context.Context, commandMessage ctrlservice.AsyncCommandMessage)
+
+// startServerWrapper wraps the Control Protocol initialization call to facilitate
+// unit testing without needing to start live TCP servers
+var startServerWrapper = network.StartInsecureControlServer
+
+// ServerHandler defines the interface for adding and removing sync or async handlers from a
+// control-protocol server
+type ServerHandler interface {
+	Shutdown(timeout time.Duration)
+	AddAsyncHandler(opcode ctrl.OpCode, resultOpcode ctrl.OpCode, payloadType message.AsyncCommand, handler AsyncHandlerFunc)
+	AddSyncHandler(opcode ctrl.OpCode, handler ctrl.MessageHandlerFunc)
+	RemoveHandler(opcode ctrl.OpCode)
+}
+
+// serverHandlerImpl is the primary implementation of a ServerHandler
+type serverHandlerImpl struct {
+	router       ctrlservice.MessageRouter
+	server       *network.ControlServer
+	cancelServer context.CancelFunc
+	routerLock   sync.RWMutex
+}
+
+// Verify that the serverHandlerImpl implements the ServerHandler interface
+var _ ServerHandler = (*serverHandlerImpl)(nil)
+
+// NewServerHandler starts a control-protocol server on the specified port and returns
+// the serverHandlerImpl as a ServerHandler interface
+func NewServerHandler(ctx context.Context, port int) (ServerHandler, error) {
+	serverCtx, serverCancelFn := context.WithCancel(ctx)
+
+	// Create a new control-protocol server that will listen on the given port
+	controlServer, err := startServerWrapper(serverCtx, network.WithPort(port))
+
+	if err != nil {
+		serverCancelFn()
+		return nil, err
+	}
+
+	serverHandler := &serverHandlerImpl{
+		server:       controlServer,
+		cancelServer: serverCancelFn,
+		router:       make(ctrlservice.MessageRouter),
+		routerLock:   sync.RWMutex{},
+	}
+	serverHandler.setHandler()
+	return serverHandler, nil
+}
+
+// Shutdown cancels the control-protocol server and wait for it to stop
+func (s *serverHandlerImpl) Shutdown(timeout time.Duration) {
+	s.cancelServer()
+	select {
+	case <-s.server.ClosedCh():
+	case <-time.After(timeout):
+	}
+}
+
+// AddAsyncHandler will add an async handler for the given opcode to the message router used by the
+// control-protocol server, which will send a message using the resultOpcode when it finishes.
+func (s *serverHandlerImpl) AddAsyncHandler(opcode ctrl.OpCode, resultOpcode ctrl.OpCode, payloadType message.AsyncCommand, handler AsyncHandlerFunc) {
+	s.routerLock.Lock()
+	s.router[opcode] = ctrlservice.NewAsyncCommandHandler(s.server, payloadType, resultOpcode, handler)
+	s.routerLock.Unlock()
+	s.setHandler()
+}
+
+// AddSyncHandler will add a handler to the control-protocol server for the given opcode
+func (s *serverHandlerImpl) AddSyncHandler(opcode ctrl.OpCode, handler ctrl.MessageHandlerFunc) {
+	s.routerLock.Lock()
+	s.router[opcode] = handler
+	s.routerLock.Unlock()
+	s.setHandler()
+}
+
+// RemoveHandler will delete a handler (sync or async) from the internal message router
+func (s *serverHandlerImpl) RemoveHandler(opcode ctrl.OpCode) {
+	s.routerLock.Lock()
+	delete(s.router, opcode)
+	s.routerLock.Unlock()
+	s.setHandler()
+}
+
+// setHandler re-sets the MessageHandler on the internal control-protocol service to a copy of the router
+func (s *serverHandlerImpl) setHandler() {
+	// Invoke the MessageHandler on the control-protocol service with a copy of our router map, to avoid it being
+	// changed in this ServerHandlerImpl while in active use as a MessageHandler.
+	routerCopy := make(ctrlservice.MessageRouter, len(s.router))
+	s.routerLock.RLock()
+	for opcode := range s.router {
+		routerCopy[opcode] = s.router[opcode]
+	}
+	s.routerLock.RUnlock()
+	s.server.MessageHandler(routerCopy)
+}


Test patch:
diff --git a/pkg/channel/distributed/common/kafka/consumer/consumer_test.go b/pkg/channel/distributed/common/kafka/consumer/consumer_test.go
index 956817ba4..6857fd770 100644
--- a/pkg/channel/distributed/common/kafka/consumer/consumer_test.go
+++ b/pkg/channel/distributed/common/kafka/consumer/consumer_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	consumertesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer/testing"
+	kafkatesting "knative.dev/eventing-kafka/pkg/common/kafka/testing"
 )
 
 // Test The CreateConsumerGroup() Functionality
@@ -33,7 +34,7 @@ func TestCreateConsumerGroup(t *testing.T) {
 	config := sarama.NewConfig()
 
 	// Create A Mock ConsumerGroup, Stub NewConsumerGroupWrapper() & Restore After Test
-	mockConsumerGroup := consumertesting.NewMockConsumerGroup()
+	mockConsumerGroup := kafkatesting.NewMockConsumerGroup()
 	consumertesting.StubNewConsumerGroupFn(consumertesting.ValidatingNewConsumerGroupFn(t, brokers, groupId, config, mockConsumerGroup))
 	defer consumertesting.RestoreNewConsumerGroupFn()
 
diff --git a/pkg/channel/distributed/common/kafka/testing/mocks.go b/pkg/channel/distributed/common/kafka/testing/mocks.go
deleted file mode 100644
index 57aa0662f..000000000
--- a/pkg/channel/distributed/common/kafka/testing/mocks.go
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
-Copyright 2020 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package testing
-
-import (
-	"context"
-	"testing"
-
-	"github.com/Shopify/sarama"
-)
-
-//
-// Mock ConsumerGroup Implementation
-//
-
-// Verify The Mock ConsumerGroup Implements The Interface
-var _ sarama.ConsumerGroup = &MockConsumerGroup{}
-
-// Define The Mock ConsumerGroup
-type MockConsumerGroup struct {
-	t           *testing.T
-	errorChan   chan error
-	consumeChan chan struct{}
-	Closed      bool
-}
-
-// Mock ConsumerGroup Constructor
-func NewMockConsumerGroup(t *testing.T) *MockConsumerGroup {
-	return &MockConsumerGroup{
-		t:           t,
-		errorChan:   make(chan error),
-		consumeChan: make(chan struct{}),
-		Closed:      false,
-	}
-}
-
-func (m *MockConsumerGroup) Consume(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) error {
-	<-m.consumeChan                      // Block To Simulate Real Execution
-	return sarama.ErrClosedConsumerGroup // Return ConsumerGroup Closed "Error" For Clean Shutdown
-}
-
-func (m *MockConsumerGroup) Errors() <-chan error {
-	return m.errorChan
-}
-
-func (m *MockConsumerGroup) Close() error {
-	close(m.errorChan)
-	close(m.consumeChan)
-	m.Closed = true
-	return nil
-}
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index b5f24eff7..164c2b7a4 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
@@ -35,6 +36,8 @@ import (
 	clienttesting "knative.dev/eventing-kafka/pkg/common/client/testing"
 	configtesting "knative.dev/eventing-kafka/pkg/common/config/testing"
 	consumertesting "knative.dev/eventing-kafka/pkg/common/consumer/testing"
+	controltesting "knative.dev/eventing-kafka/pkg/common/controlprotocol/testing"
+	kafkatesting "knative.dev/eventing-kafka/pkg/common/kafka/testing"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
 )
@@ -55,15 +58,13 @@ func TestNewSubscriberWrapper(t *testing.T) {
 	// Test Data
 	subscriber := eventingduck.SubscriberSpec{UID: uid123}
 	groupId := "TestGroupId"
-	consumerGroup := consumertesting.NewMockConsumerGroup()
 
 	// Perform The Test
-	subscriberWrapper := NewSubscriberWrapper(subscriber, groupId, consumerGroup)
+	subscriberWrapper := NewSubscriberWrapper(subscriber, groupId)
 
 	// Verify Results
 	assert.NotNil(t, subscriberWrapper)
 	assert.Equal(t, subscriber.UID, subscriberWrapper.UID)
-	assert.Equal(t, consumerGroup, subscriberWrapper.ConsumerGroup)
 	assert.Equal(t, groupId, subscriberWrapper.GroupId)
 }
 
@@ -73,7 +74,7 @@ func TestNewDispatcher(t *testing.T) {
 	baseSaramaConfig, err := commonclient.NewConfigBuilder().
 		WithDefaults().
 		WithVersion(&sarama.V2_0_0_0).
-		Build(context.TODO())
+		Build(context.Background())
 	assert.Nil(t, err)
 
 	// Perform The Test & Verify Results (Not Much To See Due To Interface)
@@ -82,11 +83,12 @@ func TestNewDispatcher(t *testing.T) {
 
 // Test The Dispatcher's Shutdown() Functionality
 func TestShutdown(t *testing.T) {
+	mockManager := consumertesting.NewMockConsumerGroupManager()
 
 	// Create Mock ConsumerGroups To Register Close() Requests
-	consumerGroup1 := consumertesting.NewMockConsumerGroup()
-	consumerGroup2 := consumertesting.NewMockConsumerGroup()
-	consumerGroup3 := consumertesting.NewMockConsumerGroup()
+	consumerGroup1 := kafkatesting.NewMockConsumerGroup()
+	consumerGroup2 := kafkatesting.NewMockConsumerGroup()
+	consumerGroup3 := kafkatesting.NewMockConsumerGroup()
 
 	// Create Test Subscribers To Close The ConsumerGroups Of
 	subscriber1 := eventingduck.SubscriberSpec{UID: id123}
@@ -96,28 +98,50 @@ func TestShutdown(t *testing.T) {
 	groupId2 := fmt.Sprintf("kafka.%s", subscriber2.UID)
 	groupId3 := fmt.Sprintf("kafka.%s", subscriber3.UID)
 
+	mockManager.Groups[groupId1] = consumerGroup1
+	mockManager.Groups[groupId2] = consumerGroup2
+	mockManager.Groups[groupId3] = consumerGroup3
+
 	// Create The Dispatcher To Test With Existing Subscribers
 	dispatcher := &DispatcherImpl{
 		DispatcherConfig: DispatcherConfig{
 			Logger: logtesting.TestLogger(t).Desugar(),
 		},
+		consumerMgr: mockManager,
 		subscribers: map[types.UID]*SubscriberWrapper{
-			subscriber1.UID: NewSubscriberWrapper(subscriber1, groupId1, consumerGroup1),
-			subscriber2.UID: NewSubscriberWrapper(subscriber2, groupId2, consumerGroup2),
-			subscriber3.UID: NewSubscriberWrapper(subscriber3, groupId3, consumerGroup3),
+			subscriber1.UID: NewSubscriberWrapper(subscriber1, groupId1),
+			subscriber2.UID: NewSubscriberWrapper(subscriber2, groupId2),
+			subscriber3.UID: NewSubscriberWrapper(subscriber3, groupId3),
 		},
 	}
 
+	consumerGroup1.On("Close").Return(nil)
+	// Note that consumerGroup2 does not have Close called because we are returning false for IsManaged()
+	consumerGroup3.On("Close").Return(nil)
+
+	mockManager.On("IsManaged", groupId1).Return(true)
+	mockManager.On("IsManaged", groupId2).Return(false)
+	mockManager.On("IsManaged", groupId3).Return(true)
+	mockManager.On("CloseConsumerGroup", groupId1).Return(nil)
+	// Group 2 is not managed, so CloseConsumerGroup is not called
+	mockManager.On("CloseConsumerGroup", groupId3).Return(fmt.Errorf("close error"))
+
 	// Perform The Test
 	dispatcher.Shutdown()
 
 	// Verify The Results
-	assert.True(t, consumerGroup1.Closed)
-	assert.True(t, consumerGroup2.Closed)
-	assert.True(t, consumerGroup3.Closed)
-	assert.Len(t, dispatcher.subscribers, 0)
-
-	// Verify that calling Shutdown a second time does not cause a panic
+	consumerGroup1.AssertExpectations(t)
+	consumerGroup2.AssertExpectations(t)
+	consumerGroup3.AssertExpectations(t)
+	mockManager.AssertExpectations(t)
+
+	assert.Len(t, dispatcher.subscribers, 1) // One was unmanaged and should be ignored
+
+	// Verify that calling Shutdown a second time does not cause a panic, and that the
+	// metrics channel is properly stopped if it exists
+	dispatcher.MetricsStopChan = make(chan struct{})
+	dispatcher.MetricsStoppedChan = make(chan struct{})
+	close(dispatcher.MetricsStoppedChan)
 	dispatcher.Shutdown()
 }
 
@@ -125,7 +149,7 @@ func TestShutdown(t *testing.T) {
 func TestUpdateSubscriptions(t *testing.T) {
 
 	logger := logtesting.TestLogger(t)
-	ctx := logging.WithLogger(context.TODO(), logger)
+	ctx := logging.WithLogger(context.Background(), logger)
 
 	// Test Data
 	brokers := []string{configtesting.DefaultKafkaBroker}
@@ -138,6 +162,12 @@ func TestUpdateSubscriptions(t *testing.T) {
 		SaramaConfig: config,
 	}
 
+	badDispatcherConfig := DispatcherConfig{
+		Logger:       logger.Desugar(),
+		Brokers:      brokers,
+		SaramaConfig: nil,
+	}
+
 	// Define The TestCase Struct
 	type fields struct {
 		DispatcherConfig DispatcherConfig
@@ -149,11 +179,16 @@ func TestUpdateSubscriptions(t *testing.T) {
 
 	// Define The TestCase Struct
 	type TestCase struct {
-		only   bool
-		name   string
-		fields fields
-		args   args
-		want   map[eventingduck.SubscriberSpec]error
+		name          string
+		fields        fields
+		createErr     error
+		consumeErr    bool
+		args          args
+		wantErrors    int
+		wantFailure   bool
+		expectStarted []string
+		expectErrors  []string
+		expectManaged []string
 	}
 
 	// Create The Test Cases
@@ -169,7 +204,53 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid123},
 				},
 			},
-			want: map[eventingduck.SubscriberSpec]error{},
+			expectStarted: []string{id123},
+			expectErrors:  []string{id123},
+			expectManaged: []string{id123},
+		},
+		{
+			name: "Create ConsumerGroup Error",
+			fields: fields{
+				DispatcherConfig: dispatcherConfig,
+				subscribers:      map[types.UID]*SubscriberWrapper{},
+			},
+			args: args{
+				subscriberSpecs: []eventingduck.SubscriberSpec{
+					{UID: uid123},
+				},
+			},
+			wantErrors:    1,
+			createErr:     fmt.Errorf("create error"),
+			expectStarted: []string{id123},
+		},
+		{
+			name: "Error During Consume",
+			fields: fields{
+				DispatcherConfig: dispatcherConfig,
+				subscribers:      map[types.UID]*SubscriberWrapper{},
+			},
+			args: args{
+				subscriberSpecs: []eventingduck.SubscriberSpec{
+					{UID: uid123},
+				},
+			},
+			consumeErr:    true,
+			expectStarted: []string{id123},
+			expectErrors:  []string{id123},
+			expectManaged: []string{id123},
+		},
+		{
+			name: "Invalid Config",
+			fields: fields{
+				DispatcherConfig: badDispatcherConfig,
+				subscribers:      map[types.UID]*SubscriberWrapper{},
+			},
+			args: args{
+				subscriberSpecs: []eventingduck.SubscriberSpec{
+					{UID: uid123},
+				},
+			},
+			wantFailure: true,
 		},
 		{
 			name: "Add Second Subscription",
@@ -185,7 +266,9 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid456},
 				},
 			},
-			want: map[eventingduck.SubscriberSpec]error{},
+			expectStarted: []string{id456},
+			expectErrors:  []string{id456},
+			expectManaged: []string{id123, id456},
 		},
 		{
 			name: "Add And Remove Subscriptions",
@@ -202,7 +285,9 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid789},
 				},
 			},
-			want: map[eventingduck.SubscriberSpec]error{},
+			expectStarted: []string{id789},
+			expectErrors:  []string{id789},
+			expectManaged: []string{id123, id456, id789},
 		},
 		{
 			name: "Remove Penultimate Subscription",
@@ -218,7 +303,7 @@ func TestUpdateSubscriptions(t *testing.T) {
 					{UID: uid123},
 				},
 			},
-			want: map[eventingduck.SubscriberSpec]error{},
+			expectManaged: []string{id123, id456},
 		},
 		{
 			name: "Remove Last Subscription",
@@ -231,50 +316,67 @@ func TestUpdateSubscriptions(t *testing.T) {
 			args: args{
 				subscriberSpecs: []eventingduck.SubscriberSpec{},
 			},
-			want: map[eventingduck.SubscriberSpec]error{},
+			expectManaged: []string{id123},
 		},
 	}
 
-	// Filter To Those With "only" Flag (If Any Specified)
-	filteredTestCases := make([]TestCase, 0)
-	for _, testCase := range testCases {
-		if testCase.only {
-			filteredTestCases = append(filteredTestCases, testCase)
-		}
-	}
-	if len(filteredTestCases) == 0 {
-		filteredTestCases = testCases
-	}
-
 	// Execute The Test Cases (Create A DispatcherImpl & UpdateSubscriptions() :)
-	for _, testCase := range filteredTestCases {
+	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
 
+			mockManager := consumertesting.NewMockConsumerGroupManager()
 			// Create A New DispatcherImpl To Test
 			dispatcher := &DispatcherImpl{
-				DispatcherConfig:     testCase.fields.DispatcherConfig,
-				subscribers:          testCase.fields.subscribers,
-				consumerGroupFactory: &consumertesting.MockKafkaConsumerGroupFactory{},
+				DispatcherConfig: testCase.fields.DispatcherConfig,
+				subscribers:      testCase.fields.subscribers,
+				consumerMgr:      mockManager,
 			}
 
-			// Perform The Test
-			got := dispatcher.UpdateSubscriptions(testCase.args.subscriberSpecs)
-
-			// Verify Results
-			assert.Equal(t, testCase.want, got)
-
-			// Verify The Dispatcher's Tracking Of Subscribers Matches Specified State
-			assert.Len(t, dispatcher.subscribers, len(testCase.args.subscriberSpecs))
-			for _, subscriber := range testCase.args.subscriberSpecs {
-				assert.NotNil(t, dispatcher.subscribers[subscriber.UID])
+			errorSource := make(chan error, 1)
+			if testCase.consumeErr {
+				errorSource <- fmt.Errorf("consume error")
 			}
 
-			// Shutdown The Dispatcher to Cleanup Resources
-			dispatcher.Shutdown()
-			assert.Len(t, dispatcher.subscribers, 0)
+			for _, id := range testCase.expectStarted {
+				mockManager.On("StartConsumerGroup", "kafka."+id, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(testCase.createErr)
+			}
+			for _, id := range testCase.expectErrors {
+				mockManager.On("Errors", "kafka."+id).Return((<-chan error)(errorSource))
+			}
+			for _, id := range testCase.expectManaged {
+				mockManager.On("IsManaged", "kafka."+id).Return(true)
+				mockManager.On("CloseConsumerGroup", "kafka."+id).Return(nil)
+			}
 
-			// Pause Briefly To Let Any Async Shutdown Finish (Lame But Only For Visual Confirmation Of Logging ;)
-			time.Sleep(500 * time.Millisecond)
+			// Perform The Test
+			result := dispatcher.UpdateSubscriptions(testCase.args.subscriberSpecs)
+
+			close(errorSource)
+
+			if testCase.wantFailure {
+				assert.Nil(t, result)
+			} else {
+				// Verify Results
+				assert.Equal(t, testCase.wantErrors, len(result))
+
+				// Verify The Dispatcher's Tracking Of Subscribers Matches Specified State
+				assert.Equal(t, len(testCase.args.subscriberSpecs)-testCase.wantErrors, len(dispatcher.subscribers))
+				for _, subscriber := range testCase.args.subscriberSpecs {
+					if testCase.createErr != nil {
+						assert.Nil(t, dispatcher.subscribers[subscriber.UID])
+					} else {
+						assert.NotNil(t, dispatcher.subscribers[subscriber.UID])
+					}
+				}
+
+				// Shutdown The Dispatcher to Cleanup Resources
+				dispatcher.Shutdown()
+				assert.Len(t, dispatcher.subscribers, 0)
+
+				// Pause Briefly To Let Any Async Shutdown Finish (Lame But Only For Visual Confirmation Of Logging ;)
+				time.Sleep(500 * time.Millisecond)
+			}
+			mockManager.AssertExpectations(t)
 		})
 	}
 }
@@ -283,7 +385,7 @@ func TestUpdateSubscriptions(t *testing.T) {
 func TestSecretChanged(t *testing.T) {
 
 	logger := logtesting.TestLogger(t)
-	ctx := logging.WithLogger(context.TODO(), logger)
+	ctx := logging.WithLogger(context.Background(), logger)
 
 	// Setup Test Environment Namespaces
 	commontesting.SetTestEnvironment(t)
@@ -313,6 +415,7 @@ func TestSecretChanged(t *testing.T) {
 		expectNewUsername   string
 		expectNewPassword   string
 		expectNewSaslType   string
+		reconfigureErr      bool
 	}
 
 	// Create The TestCases
@@ -349,16 +452,27 @@ func TestSecretChanged(t *testing.T) {
 			name:      "No Auth Config In Secret (No Modifications)",
 			newSecret: configtesting.NewKafkaSecret(configtesting.WithMissingConfig),
 		},
+		{
+			name:              "Reconfigure Error",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithModifiedSaslType),
+			expectNewSaslType: configtesting.ModifiedSecretSaslType,
+			reconfigureErr:    true,
+		},
 	}
 
 	// Run The Filtered TestCases
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
 
+			mockManager := consumertesting.NewMockConsumerGroupManager()
 			// Create A Test Dispatcher To Perform Tests Against
 			dispatcher := createTestDispatcher(t, brokers, baseSaramaConfig)
 			impl := dispatcher.(*DispatcherImpl)
 			impl.subscribers = map[types.UID]*SubscriberWrapper{uid123: createSubscriberWrapper(uid123)}
+			if testCase.reconfigureErr {
+				mockManager.On("Reconfigure", mock.Anything, mock.Anything).Return(fmt.Errorf("reconfigure error"))
+				impl.consumerMgr = mockManager
+			}
 
 			// Perform The Test
 			dispatcher.SecretChanged(ctx, testCase.newSecret)
@@ -378,13 +492,15 @@ func TestSecretChanged(t *testing.T) {
 			if testCase.expectNewSaslType != "" {
 				assert.Equal(t, testCase.expectNewSaslType, string(impl.SaramaConfig.Net.SASL.Mechanism))
 			}
+
+			mockManager.AssertExpectations(t)
 		})
 	}
 }
 
 // Utility Function For Creating A SubscriberWrapper With Specified UID & Mock ConsumerGroup
 func createSubscriberWrapper(uid types.UID) *SubscriberWrapper {
-	return NewSubscriberWrapper(eventingduck.SubscriberSpec{UID: uid}, fmt.Sprintf("kafka.%s", string(uid)), nil)
+	return NewSubscriberWrapper(eventingduck.SubscriberSpec{UID: uid}, fmt.Sprintf("kafka.%s", string(uid)))
 }
 
 // Utility Function For Creating A Dispatcher With Specified Configuration
@@ -409,8 +525,13 @@ func createTestDispatcher(t *testing.T, brokers []string, config *sarama.Config)
 		SubscriberSpecs: subscriberSpecs,
 	}
 
+	serverHandler := controltesting.GetMockServerHandler()
+	serverHandler.On("AddAsyncHandler", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
+	serverHandler.Service.On("SendAndWaitForAck", mock.Anything, mock.Anything).Return(nil)
+
 	// Create The Dispatcher
-	dispatcher := NewDispatcher(dispatcherConfig)
+	dispatcher := NewDispatcher(dispatcherConfig, serverHandler)
+	serverHandler.AssertExpectations(t)
 
 	// Verify State
 	assert.NotNil(t, dispatcher)
diff --git a/pkg/common/consumer/consumer_factory_test.go b/pkg/common/consumer/consumer_factory_test.go
index 78b329a67..acfb03549 100644
--- a/pkg/common/consumer/consumer_factory_test.go
+++ b/pkg/common/consumer/consumer_factory_test.go
@@ -41,13 +41,13 @@ func (m *mockConsumerGroup) Consume(ctx context.Context, topics []string, handle
 		go func() {
 			m.generateErrorOnce.Do(func() {
 				h := handler.(*SaramaConsumerHandler)
-				h.errors <- errors.New("cgh")
-				close(h.errors)
+				h.errors <- errors.New("consumer group handler error")
+				// Don't close h.errors here; the deferred shutdown in startExistingConsumerGroup already does that
 			})
 		}()
 	}
 	if m.consumeMustReturnError {
-		return errors.New("boom!")
+		return errors.New("consume error")
 	}
 	return nil
 }
@@ -56,7 +56,7 @@ func (m *mockConsumerGroup) Errors() <-chan error {
 	ch := make(chan error)
 	go func() {
 		if m.mustGenerateConsumerGroupError {
-			ch <- errors.New("cg")
+			ch <- errors.New("consumer group error")
 		}
 		close(ch)
 	}()
@@ -103,16 +103,28 @@ func TestErrorPropagationCustomConsumerGroup(t *testing.T) {
 
 	errorsSlice := make([]error, 0)
 
-	for e := range consumerGroup.Errors() {
-		errorsSlice = append(errorsSlice, e)
-	}
+	errorsWait := sync.WaitGroup{}
+	errorsWait.Add(2)
+	go func() {
+		for e := range consumerGroup.Errors() {
+			errorsSlice = append(errorsSlice, e)
+			errorsWait.Done() // Should be called twice, once for the ConsumerGroup error and once for the Handler error
+		}
+	}()
+
+	// Wait for the mock to send the errors
+	errorsWait.Wait()
+	consumerGroup.(*customConsumerGroup).cancel() // Stop the consume loop and close the error channel
 
 	if len(errorsSlice) != 2 {
 		t.Errorf("len(errorsSlice) != 2")
 	}
 
-	assertContainsError(t, errorsSlice, "cgh")
-	assertContainsError(t, errorsSlice, "cg")
+	// Wait for the goroutine inside of startExistingConsumerGroup to finish
+	<-consumerGroup.(*customConsumerGroup).releasedCh
+
+	assertContainsError(t, errorsSlice, "consumer group handler error")
+	assertContainsError(t, errorsSlice, "consumer group error")
 }
 
 func assertContainsError(t *testing.T, collection []error, errorStr string) {
@@ -147,11 +159,14 @@ func TestErrorWhileNewConsumerGroup(t *testing.T) {
 		config: sarama.NewConfig(),
 		addrs:  []string{"b1", "b2"},
 	}
-	cg, _ := factory.StartConsumerGroup("bla", []string{}, zap.L().Sugar(), nil)
+	consumerGroup, _ := factory.StartConsumerGroup("bla", []string{}, zap.L().Sugar(), nil)
 
-	err := <-cg.Errors()
+	consumerGroup.(*customConsumerGroup).cancel() // Stop the consume loop from spinning after the error is generated
+	err := <-consumerGroup.Errors()
+	// Wait for the goroutine inside of startExistingConsumerGroup to finish
+	<-consumerGroup.(*customConsumerGroup).releasedCh
 
-	if err == nil || err.Error() != "boom!" {
-		t.Errorf("Should contain an error with message boom!. Got %v", err)
+	if err == nil || err.Error() != "consume error" {
+		t.Errorf("Should contain an error with message consume error. Got %v", err)
 	}
 }
diff --git a/pkg/common/consumer/consumer_manager_test.go b/pkg/common/consumer/consumer_manager_test.go
new file mode 100644
index 000000000..6c7120612
--- /dev/null
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -0,0 +1,421 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package consumer
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+
+	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
+	ctrl "knative.dev/control-protocol/pkg"
+	logtesting "knative.dev/pkg/logging/testing"
+
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+	controltesting "knative.dev/eventing-kafka/pkg/common/controlprotocol/testing"
+	kafkatesting "knative.dev/eventing-kafka/pkg/common/kafka/testing"
+)
+
+func TestNewConsumerGroupManager(t *testing.T) {
+	server := getMockServerHandler()
+	manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), server, []string{}, &sarama.Config{})
+	assert.NotNil(t, manager)
+	assert.NotNil(t, server.Router[commands.StopConsumerGroupOpCode])
+	assert.NotNil(t, server.Router[commands.StartConsumerGroupOpCode])
+	server.AssertExpectations(t)
+}
+
+func TestReconfigure(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup)
+
+	for _, testCase := range []struct {
+		name       string
+		groupId    string
+		factoryErr bool
+		closeErr   error
+		expectErr  string
+	}{
+		{
+			name: "No managed groups",
+		},
+		{
+			name:    "One managed group",
+			groupId: "test-id1",
+		},
+		{
+			name:      "Error stopping groups",
+			groupId:   "test-id1",
+			closeErr:  fmt.Errorf("error closing consumer group"),
+			expectErr: "error closing consumer group",
+		},
+		{
+			name:       "Error starting groups",
+			groupId:    "test-id1",
+			factoryErr: true,
+			expectErr:  "factory error",
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			manager, group, _, server := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
+			if group != nil {
+				group.On("Close").Return(testCase.closeErr)
+			}
+			err := manager.Reconfigure([]string{"new-broker"}, &sarama.Config{})
+			if testCase.expectErr == "" {
+				assert.Nil(t, err)
+			} else {
+				assert.NotNil(t, err)
+				assert.Equal(t, testCase.expectErr, err.Error())
+			}
+			server.AssertExpectations(t)
+		})
+	}
+}
+
+func TestStartConsumerGroup(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup)
+
+	for _, testCase := range []struct {
+		name       string
+		factoryErr bool
+	}{
+		{
+			name: "No error",
+		},
+		{
+			name:       "Factory error",
+			factoryErr: true,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), getMockServerHandler(), []string{}, &sarama.Config{})
+			mockGroup := kafkatesting.NewMockConsumerGroup()
+			newConsumerGroup = func(addrs []string, groupID string, config *sarama.Config) (sarama.ConsumerGroup, error) {
+				if testCase.factoryErr {
+					return mockGroup, fmt.Errorf("factory error")
+				}
+				mockGroup.On("Errors").Return(mockGroup.ErrorChan)
+				return mockGroup, nil
+			}
+			err := manager.StartConsumerGroup("testid", []string{}, nil, nil)
+			assert.Equal(t, testCase.factoryErr, err != nil)
+			time.Sleep(5 * time.Millisecond) // Give the transferErrors routine a chance to call Errors()
+			mockGroup.AssertExpectations(t)
+		})
+	}
+}
+
+func TestCloseConsumerGroup(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
+
+	for _, testCase := range []struct {
+		name      string
+		groupId   string
+		expectErr bool
+		closeErr  error
+		cancel    bool
+	}{
+		{
+			name:      "Nonexistent GroupID",
+			expectErr: true,
+		},
+		{
+			name:    "Existing GroupID",
+			groupId: "test-group-id",
+		},
+		{
+			name:      "Existing GroupID, Error Closing Group",
+			groupId:   "test-group-id",
+			closeErr:  fmt.Errorf("close error"),
+			expectErr: true,
+		},
+		{
+			name:    "With Cancel Functions",
+			groupId: "test-group-id",
+			cancel:  true,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			manager, group, managedGrp, server := getManagerWithMockGroup(t, testCase.groupId, false)
+			var cancelConsumeCalled, cancelErrorsCalled bool
+			if group != nil {
+				group.On("Close").Return(testCase.closeErr)
+				if testCase.cancel {
+					managedGrp.cancelConsume = func() { cancelConsumeCalled = true }
+					managedGrp.cancelErrors = func() { cancelErrorsCalled = true }
+				}
+			}
+			err := manager.CloseConsumerGroup(testCase.groupId)
+			assert.Equal(t, testCase.expectErr, err != nil)
+			assert.Equal(t, testCase.cancel, cancelConsumeCalled)
+			assert.Equal(t, testCase.cancel, cancelErrorsCalled)
+			server.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConsume(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
+
+	for _, testCase := range []struct {
+		name      string
+		groupId   string
+		stopped   bool
+		cancel    bool
+		expectErr string
+	}{
+		{
+			name:      "Nonexistent GroupID",
+			expectErr: "consume called on nonexistent groupId ''",
+		},
+		{
+			name:      "Existing GroupID, Started",
+			groupId:   "test-group-id",
+			expectErr: "kafka: tried to use a consumer group that was closed",
+		},
+		{
+			name:      "Existing GroupID, Stopped",
+			groupId:   "test-group-id",
+			stopped:   true,
+			expectErr: "kafka: tried to use a consumer group that was closed",
+		},
+		{
+			name:      "Existing GroupID, Canceled",
+			groupId:   "test-group-id",
+			stopped:   true,
+			cancel:    true,
+			expectErr: "context was canceled waiting for group 'test-group-id' to start",
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			manager, mockGroup, mgdGroup, server := getManagerWithMockGroup(t, testCase.groupId, false)
+			if mockGroup != nil {
+				mockGroup.On("Consume", ctx, []string{"topic"}, nil).Return(sarama.ErrClosedConsumerGroup)
+				mockGroup.On("Close").Return(nil)
+			}
+
+			if testCase.stopped {
+				mgdGroup.createRestartChannel()
+			}
+			waitGroup := sync.WaitGroup{}
+			waitGroup.Add(1)
+			go func() {
+				err := manager.(*kafkaConsumerGroupManagerImpl).consume(ctx, testCase.groupId, []string{"topic"}, nil)
+				if testCase.expectErr != "" {
+					assert.NotNil(t, err)
+					assert.Equal(t, testCase.expectErr, err.Error())
+				} else {
+					assert.Nil(t, err)
+				}
+				waitGroup.Done()
+			}()
+			time.Sleep(5 * time.Millisecond) // Give Consume() a chance to call waitForStart()
+			if mgdGroup != nil {
+				if testCase.cancel {
+					cancel()
+				} else {
+					mgdGroup.closeRestartChannel()
+				}
+				assert.Nil(t, mgdGroup.group.Close()) // Stops the MockConsumerGroup's Consume() call
+			}
+			waitGroup.Wait() // Allows the goroutine with the consume call to finish
+			server.AssertExpectations(t)
+		})
+	}
+}
+
+func TestErrors(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
+
+	for _, testCase := range []struct {
+		name      string
+		groupId   string
+		expectErr bool
+	}{
+		{
+			name:    "Existing GroupID",
+			groupId: "test-group-id",
+		},
+		{
+			name:      "Nonexistent GroupID",
+			expectErr: true,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			manager, _, _, server := getManagerWithMockGroup(t, testCase.groupId, false)
+			valid := manager.IsManaged(testCase.groupId)
+			mgrErrors := manager.Errors(testCase.groupId)
+			assert.Equal(t, testCase.expectErr, mgrErrors == nil)
+			assert.Equal(t, !testCase.expectErr, valid)
+			server.AssertExpectations(t)
+		})
+	}
+}
+
+func TestNotifications(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
+	for _, testCase := range []struct {
+		name        string
+		groupId     string
+		version     int16
+		opcode      ctrl.OpCode
+		initialStop bool
+		expectStop  bool
+		factoryErr  bool
+		expectErr   bool
+		closeErr    error
+		expectClose bool
+	}{
+		{
+			name:    "Invalid Command Version",
+			opcode:  commands.StopConsumerGroupOpCode,
+			version: 0,
+		},
+		{
+			name:    "Stop Group OpCode, Nonexistent Group",
+			opcode:  commands.StopConsumerGroupOpCode,
+			version: 1,
+		},
+		{
+			name:    "Start Group OpCode, Nonexistent Group",
+			opcode:  commands.StartConsumerGroupOpCode,
+			version: 1,
+		},
+		{
+			name:        "Stop Group OpCode",
+			opcode:      commands.StopConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			version:     1,
+			expectStop:  true,
+			expectClose: true,
+		},
+		{
+			name:        "Stop Group OpCode, Group Close Error",
+			opcode:      commands.StopConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			version:     1,
+			closeErr:    fmt.Errorf("close error"),
+			expectClose: true,
+		},
+		{
+			name:    "Start Group OpCode, Group Already Started",
+			opcode:  commands.StartConsumerGroupOpCode,
+			groupId: "test-group-id",
+			version: 1,
+		},
+		{
+			name:        "Start Group OpCode, Group Not Started, Factory Error",
+			opcode:      commands.StartConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			version:     1,
+			factoryErr:  true,
+			initialStop: true,
+			expectStop:  true,
+		},
+		{
+			name:        "Start Group OpCode, Group Not Started",
+			opcode:      commands.StartConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			version:     1,
+			initialStop: true,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			manager, group, _, serverHandler := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
+			impl := manager.(*kafkaConsumerGroupManagerImpl)
+			if testCase.initialStop {
+				impl.groups[testCase.groupId].createRestartChannel()
+			}
+			if group != nil && testCase.expectClose {
+				group.On("Close").Return(testCase.closeErr)
+			}
+
+			handler, ok := serverHandler.Router[testCase.opcode]
+			assert.True(t, ok)
+
+			testCommand := commands.ConsumerGroupAsyncCommand{
+				Version:   testCase.version,
+				CommandId: 1,
+				TopicName: "test-topic-name",
+				GroupId:   testCase.groupId,
+			}
+			payload, err := testCommand.MarshalBinary()
+			assert.Nil(t, err)
+
+			msg := ctrl.NewMessage([16]byte{1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4}, uint8(testCase.opcode), payload)
+			handler.HandleServiceMessage(context.Background(), ctrl.NewServiceMessage(&msg, func(err error) {
+				assert.Equal(t, testCase.expectErr, err != nil)
+			}))
+
+			if group != nil {
+				assert.Equal(t, testCase.expectStop, impl.groups[testCase.groupId].isStopped())
+				group.AssertExpectations(t)
+			}
+			serverHandler.AssertExpectations(t)
+		})
+	}
+}
+
+// getManagerWithMockGroup creates a KafkaConsumerGroupManager and optionally seeds it with a mock consumer group
+func getManagerWithMockGroup(t *testing.T, groupId string, factoryErr bool) (KafkaConsumerGroupManager,
+	*kafkatesting.MockConsumerGroup,
+	*managedGroup,
+	*controltesting.MockServerHandler) {
+
+	serverHandler := getMockServerHandler()
+	newConsumerGroup = func(addrs []string, groupID string, config *sarama.Config) (sarama.ConsumerGroup, error) {
+		if factoryErr {
+			return nil, fmt.Errorf("factory error")
+		}
+		return kafkatesting.NewMockConsumerGroup(), nil
+	}
+	manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), serverHandler, []string{}, &sarama.Config{})
+	if groupId != "" {
+		mockGrp, managedGrp := createTestGroup()
+		manager.(*kafkaConsumerGroupManagerImpl).groups[groupId] = managedGrp
+		return manager, mockGrp, managedGrp, serverHandler
+	}
+	return manager, nil, nil, serverHandler
+}
+
+func getMockServerHandler() *controltesting.MockServerHandler {
+	server := controltesting.GetMockServerHandler()
+	server.On("AddAsyncHandler", commands.StopConsumerGroupOpCode, commands.StopConsumerGroupResultOpCode, mock.Anything, mock.Anything).Return()
+	server.On("AddAsyncHandler", commands.StartConsumerGroupOpCode, commands.StartConsumerGroupResultOpCode, mock.Anything, mock.Anything).Return()
+	server.Service.On("SendAndWaitForAck", commands.StopConsumerGroupOpCode, mock.Anything).Return(nil)
+	server.Service.On("SendAndWaitForAck", commands.StartConsumerGroupOpCode, mock.Anything).Return(nil)
+	server.Service.On("SendAndWaitForAck", commands.StopConsumerGroupResultOpCode, mock.Anything).Return(nil)
+	server.Service.On("SendAndWaitForAck", commands.StartConsumerGroupResultOpCode, mock.Anything).Return(nil)
+	return server
+}
+
+func createTestGroup() (*kafkatesting.MockConsumerGroup, *managedGroup) {
+	mockGroup := kafkatesting.NewMockConsumerGroup()
+	return mockGroup, createManagedGroup(mockGroup, func() {})
+}
+
+// restoreNewConsumerGroup allows a single defer call to be used for saving and restoring the newConsumerGroup wrapper
+func restoreNewConsumerGroup(fn func(addrs []string, groupID string, config *sarama.Config) (sarama.ConsumerGroup, error)) {
+	newConsumerGroup = fn
+}
diff --git a/pkg/common/consumer/managed_group_test.go b/pkg/common/consumer/managed_group_test.go
new file mode 100644
index 000000000..03186d557
--- /dev/null
+++ b/pkg/common/consumer/managed_group_test.go
@@ -0,0 +1,148 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package consumer
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	kafkatesting "knative.dev/eventing-kafka/pkg/common/kafka/testing"
+)
+
+func TestManagedGroup(t *testing.T) {
+	for _, testCase := range []struct {
+		name     string
+		restart  bool
+		cancel   bool
+		expected bool
+	}{
+		{
+			name:     "Wait for Started Group",
+			expected: true,
+		},
+		{
+			name:     "Wait for Restarted Group",
+			restart:  true,
+			expected: true,
+		},
+		{
+			name:     "Cancel Wait",
+			cancel:   true,
+			expected: false,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			// Test stop/start of a managedGroup
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			group := createManagedGroup(nil, cancel)
+			waitGroup := sync.WaitGroup{}
+			assert.False(t, group.isStopped())
+
+			if testCase.restart || testCase.cancel {
+				group.createRestartChannel()
+				assert.True(t, group.isStopped())
+			}
+
+			waitGroup.Add(1)
+			go func() {
+				assert.Equal(t, testCase.expected, group.waitForStart(ctx))
+				waitGroup.Done()
+			}()
+			time.Sleep(5 * time.Millisecond) // Let the waitForStart function begin
+
+			if testCase.restart {
+				group.closeRestartChannel()
+			} else if testCase.cancel {
+				cancel()
+			}
+			waitGroup.Wait() // Let the waitForStart function finish
+		})
+	}
+
+}
+
+func TestTransferErrors(t *testing.T) {
+	for _, testCase := range []struct {
+		name       string
+		stopGroup  bool
+		startGroup bool
+		cancel     bool
+	}{
+		{
+			name: "Close Channel Without Stop",
+		},
+		{
+			name:      "Close Channel After Stop",
+			stopGroup: true,
+		},
+		{
+			name:      "Cancel Context",
+			stopGroup: true,
+			cancel:    true,
+		},
+		{
+			name:       "Restart After Stop",
+			stopGroup:  true,
+			startGroup: true,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			mockGrp, managedGrp := createTestGroup()
+			mockGrp.On("Errors").Return(mockGrp.ErrorChan)
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			managedGrp.transferErrors(ctx)
+
+			mockGrp.ErrorChan <- fmt.Errorf("test-error")
+			err := <-managedGrp.errors
+			assert.NotNil(t, err)
+			assert.Equal(t, "test-error", err.Error())
+			if testCase.stopGroup {
+				managedGrp.createRestartChannel()
+			}
+			if testCase.cancel {
+				cancel()
+			}
+			close(mockGrp.ErrorChan)
+			mockGrp.AssertExpectations(t)
+
+			time.Sleep(5 * time.Millisecond) // Let the error handling loop move forward
+			if testCase.startGroup {
+				// Simulate the effects of startConsumerGroup (new ConsumerGroup, same managedConsumerGroup)
+				mockGrp = kafkatesting.NewMockConsumerGroup()
+				mockGrp.On("Errors").Return(mockGrp.ErrorChan)
+				managedGrp.group = mockGrp
+				managedGrp.closeRestartChannel()
+
+				time.Sleep(5 * time.Millisecond) // Let the waitForStart function finish
+				// Verify that errors work again after restart
+				mockGrp.ErrorChan <- fmt.Errorf("test-error-2")
+				err = <-managedGrp.errors
+				assert.NotNil(t, err)
+				assert.Equal(t, "test-error-2", err.Error())
+				close(mockGrp.ErrorChan)
+				mockGrp.AssertExpectations(t)
+			}
+		})
+	}
+}
diff --git a/pkg/common/consumer/testing/mocks.go b/pkg/common/consumer/testing/mocks.go
index 4fc874e11..86a521225 100644
--- a/pkg/common/consumer/testing/mocks.go
+++ b/pkg/common/consumer/testing/mocks.go
@@ -17,10 +17,8 @@ limitations under the License.
 package testing
 
 import (
-	"context"
-	"errors"
-
 	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/mock"
 	"go.uber.org/zap"
 
 	"knative.dev/eventing-kafka/pkg/common/consumer"
@@ -32,50 +30,58 @@ import (
 
 type MockKafkaConsumerGroupFactory struct {
 	// CreateErr will return an error when creating a consumer
-	CreateErr bool
+	mock.Mock
 }
 
-func (c MockKafkaConsumerGroupFactory) StartConsumerGroup(groupID string, topics []string, logger *zap.SugaredLogger, handler consumer.KafkaConsumerHandler, options ...consumer.SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
-	if c.CreateErr {
-		return nil, errors.New("error creating consumer")
-	}
-	return NewMockConsumerGroup(), nil
+func (c *MockKafkaConsumerGroupFactory) StartConsumerGroup(groupId string, topics []string, logger *zap.SugaredLogger, handler consumer.KafkaConsumerHandler, options ...consumer.SaramaConsumerHandlerOption) (sarama.ConsumerGroup, error) {
+	args := c.Called(groupId, topics, logger, handler, options)
+	return args.Get(0).(sarama.ConsumerGroup), args.Error(1)
 }
 
 var _ consumer.KafkaConsumerGroupFactory = (*MockKafkaConsumerGroupFactory)(nil)
 
 //
-// Mock ConsumerGroup
+// Mock KafkaConsumerGroupManager
 //
 
-var _ sarama.ConsumerGroup = &MockConsumerGroup{}
+// MockConsumerGroupManager implements the KafkaConsumerGroupManager interface
+type MockConsumerGroupManager struct {
+	mock.Mock
+	Groups map[string]sarama.ConsumerGroup
+}
 
-type MockConsumerGroup struct {
-	errorChan   chan error
-	consumeChan chan struct{}
-	Closed      bool
+func NewMockConsumerGroupManager() *MockConsumerGroupManager {
+	return &MockConsumerGroupManager{Groups: make(map[string]sarama.ConsumerGroup)}
 }
 
-func NewMockConsumerGroup() *MockConsumerGroup {
-	return &MockConsumerGroup{
-		errorChan:   make(chan error),
-		consumeChan: make(chan struct{}),
-		Closed:      false,
-	}
+var _ consumer.KafkaConsumerGroupManager = (*MockConsumerGroupManager)(nil)
+
+func (m *MockConsumerGroupManager) Reconfigure(brokers []string, config *sarama.Config) error {
+	args := m.Called(brokers, config)
+	return args.Error(0)
+}
+
+func (m *MockConsumerGroupManager) StartConsumerGroup(groupId string, topics []string, logger *zap.SugaredLogger,
+	handler consumer.KafkaConsumerHandler, options ...consumer.SaramaConsumerHandlerOption) error {
+	args := m.Called(groupId, topics, logger, handler, options)
+	return args.Error(0)
 }
 
-func (m *MockConsumerGroup) Consume(_ context.Context, _ []string, _ sarama.ConsumerGroupHandler) error {
-	<-m.consumeChan                      // Block To Simulate Real Execution
-	return sarama.ErrClosedConsumerGroup // Return ConsumerGroup Closed "Error" For Clean Shutdown
+func (m *MockConsumerGroupManager) CloseConsumerGroup(groupId string) error {
+	args := m.Called(groupId)
+	if group, ok := m.Groups[groupId]; ok {
+		_ = group.Close()
+		delete(m.Groups, groupId)
+	}
+	return args.Error(0)
 }
 
-func (m *MockConsumerGroup) Errors() <-chan error {
-	return m.errorChan
+func (m *MockConsumerGroupManager) IsManaged(groupId string) bool {
+	args := m.Called(groupId)
+	return args.Bool(0)
 }
 
-func (m *MockConsumerGroup) Close() error {
-	close(m.errorChan)
-	close(m.consumeChan)
-	m.Closed = true
-	return nil
+func (m *MockConsumerGroupManager) Errors(groupId string) <-chan error {
+	args := m.Called(groupId)
+	return args.Get(0).(<-chan error)
 }
diff --git a/pkg/common/controlprotocol/controlprotocol_test.go b/pkg/common/controlprotocol/controlprotocol_test.go
new file mode 100644
index 000000000..a2da97e59
--- /dev/null
+++ b/pkg/common/controlprotocol/controlprotocol_test.go
@@ -0,0 +1,101 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controlprotocol
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	ctrl "knative.dev/control-protocol/pkg"
+	"knative.dev/control-protocol/pkg/network"
+	ctrlservice "knative.dev/control-protocol/pkg/service"
+
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+	ctrltesting "knative.dev/eventing-kafka/pkg/common/controlprotocol/testing"
+)
+
+func TestNewServerHandler(t *testing.T) {
+
+	saveStartServer := startServerWrapper
+	defer func() { startServerWrapper = saveStartServer }()
+
+	for _, testCase := range []struct {
+		name      string
+		serverErr error
+	}{
+		{
+			name: "No Error",
+		},
+		{
+			name:      "Server Error",
+			serverErr: fmt.Errorf("test error"),
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			mockService := &ctrltesting.MockControlProtocolService{}
+			mockService.On("MessageHandler", mock.Anything).Return()
+			startServerWrapper = func(_ context.Context, _ ...network.ControlServerOption) (*network.ControlServer, error) {
+				return &network.ControlServer{Service: mockService}, testCase.serverErr
+			}
+
+			handler, err := NewServerHandler(context.Background(), 12345)
+			if testCase.serverErr == nil {
+				assert.NotNil(t, handler)
+				mockService.AssertExpectations(t)
+			}
+			assert.Equal(t, testCase.serverErr, err)
+		})
+	}
+}
+
+func TestHandlers(t *testing.T) {
+	saveStartServer := startServerWrapper
+	defer func() { startServerWrapper = saveStartServer }()
+
+	mockService := &ctrltesting.MockControlProtocolService{}
+	mockService.On("MessageHandler", mock.Anything).Return()
+
+	startServerWrapper = func(_ context.Context, _ ...network.ControlServerOption) (*network.ControlServer, error) {
+		return &network.ControlServer{Service: mockService}, nil
+	}
+	handler, err := NewServerHandler(context.Background(), 12345)
+	assert.NotNil(t, handler)
+	assert.Nil(t, err)
+
+	// Extract the unexported implementation of the ServerHandler for testing the router directly
+	impl := handler.(*serverHandlerImpl)
+
+	handler.AddSyncHandler(ctrl.OpCode(1), func(ctx context.Context, message ctrl.ServiceMessage) {})
+	assert.NotNil(t, impl.router[ctrl.OpCode(1)])
+
+	handler.AddAsyncHandler(ctrl.OpCode(2), ctrl.OpCode(3), &commands.ConsumerGroupAsyncCommand{},
+		func(ctx context.Context, commandMessage ctrlservice.AsyncCommandMessage) {})
+	assert.NotNil(t, impl.router[ctrl.OpCode(2)])
+
+	handler.RemoveHandler(ctrl.OpCode(1))
+	handler.RemoveHandler(ctrl.OpCode(2))
+	assert.Nil(t, impl.router[ctrl.OpCode(1)])
+	assert.Nil(t, impl.router[ctrl.OpCode(2)])
+
+	mockService.AssertExpectations(t)
+
+	handler.Shutdown(time.Millisecond)
+}
diff --git a/pkg/common/controlprotocol/testing/mocks.go b/pkg/common/controlprotocol/testing/mocks.go
new file mode 100644
index 000000000..d9c2ae569
--- /dev/null
+++ b/pkg/common/controlprotocol/testing/mocks.go
@@ -0,0 +1,89 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package testing
+
+import (
+	"context"
+	"encoding"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+	ctrl "knative.dev/control-protocol/pkg"
+	"knative.dev/control-protocol/pkg/message"
+	ctrlservice "knative.dev/control-protocol/pkg/service"
+)
+
+//
+// Mock Control-Protocol ServerHandler
+//
+
+// MockServerHandler is a mock of the ServerHandler that only stores results from AddAsyncHandler
+type MockServerHandler struct {
+	mock.Mock
+	// Export the Router field so that tests can verify handlers in it directly
+	Router  ctrlservice.MessageRouter
+	Service *MockControlProtocolService
+}
+
+func (s *MockServerHandler) AddAsyncHandler(opcode ctrl.OpCode, resultOpcode ctrl.OpCode, payloadType message.AsyncCommand,
+	handler func(ctx context.Context, commandMessage ctrlservice.AsyncCommandMessage)) {
+	_ = s.Called(opcode, resultOpcode, payloadType, handler)
+	s.Router[opcode] = ctrlservice.NewAsyncCommandHandler(s.Service, payloadType, resultOpcode, handler)
+}
+
+func (s *MockServerHandler) Shutdown(timeout time.Duration) {
+	_ = s.Called(timeout)
+}
+
+func (s *MockServerHandler) AddSyncHandler(opcode ctrl.OpCode, handler ctrl.MessageHandlerFunc) {
+	_ = s.Called(opcode, handler)
+}
+
+func (s *MockServerHandler) RemoveHandler(opcode ctrl.OpCode) {
+	_ = s.Called(opcode)
+}
+
+func GetMockServerHandler() *MockServerHandler {
+	return &MockServerHandler{
+		Router:  make(ctrlservice.MessageRouter),
+		Service: &MockControlProtocolService{},
+	}
+}
+
+//
+// Mock Control-Protocol Service
+//
+
+// MockControlProtocolService is a stub-only mock of the Control Protocol Service
+type MockControlProtocolService struct {
+	mock.Mock
+}
+
+var _ ctrl.Service = (*MockControlProtocolService)(nil)
+
+func (m *MockControlProtocolService) SendAndWaitForAck(opcode ctrl.OpCode, payload encoding.BinaryMarshaler) error {
+	args := m.Called(opcode, payload)
+	return args.Error(0)
+}
+
+func (m *MockControlProtocolService) MessageHandler(handler ctrl.MessageHandler) {
+	_ = m.Called(handler)
+}
+
+func (m *MockControlProtocolService) ErrorHandler(handler ctrl.ErrorHandler) {
+	_ = m.Called(handler)
+}
diff --git a/pkg/channel/distributed/common/kafka/consumer/testing/mocks.go b/pkg/common/kafka/testing/mocks.go
similarity index 66%
rename from pkg/channel/distributed/common/kafka/consumer/testing/mocks.go
rename to pkg/common/kafka/testing/mocks.go
index 139f9c5f2..0bff1ceb9 100644
--- a/pkg/channel/distributed/common/kafka/consumer/testing/mocks.go
+++ b/pkg/common/kafka/testing/mocks.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/mock"
 )
 
 //
@@ -29,31 +30,33 @@ import (
 var _ sarama.ConsumerGroup = &MockConsumerGroup{}
 
 type MockConsumerGroup struct {
-	errorChan   chan error
+	mock.Mock
+	ErrorChan   chan error
 	consumeChan chan struct{}
-	Closed      bool
 }
 
 func NewMockConsumerGroup() *MockConsumerGroup {
-	return &MockConsumerGroup{
-		errorChan:   make(chan error),
+	mockGroup := &MockConsumerGroup{
+		ErrorChan:   make(chan error),
 		consumeChan: make(chan struct{}),
-		Closed:      false,
 	}
+	return mockGroup
 }
 
-func (m *MockConsumerGroup) Consume(_ context.Context, _ []string, _ sarama.ConsumerGroupHandler) error {
-	<-m.consumeChan                      // Block To Simulate Real Execution
-	return sarama.ErrClosedConsumerGroup // Return ConsumerGroup Closed "Error" For Clean Shutdown
+func (m *MockConsumerGroup) Consume(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) error {
+	args := m.Called(ctx, topics, handler)
+	<-m.consumeChan // Block To Simulate Real Execution
+	return args.Error(0)
 }
 
 func (m *MockConsumerGroup) Errors() <-chan error {
-	return m.errorChan
+	args := m.Called()
+	return args.Get(0).(chan error)
 }
 
 func (m *MockConsumerGroup) Close() error {
-	close(m.errorChan)
+	args := m.Called()
+	close(m.ErrorChan)
 	close(m.consumeChan)
-	m.Closed = true
-	return nil
+	return args.Error(0)
 }