Repository: knative-sandbox/eventing-kafka
Pull request #746 (2021-06-30 01:52:33)

Fix stop start without locking
This fixes a few bugs that cropped up due to #742 

## Proposed Changes

- 🐛 The managedGroup lock/unlock was being allowed if no CommandLock struct was passed to processLock, even if it was already locked with a token.
- 🐛 CloseConsumerGroup was logging warnings indicating that cancelErrors and cancelConsume were nil when they, in fact, were not
- 🧽 Several places were still directly accessing the groups map instead of using the getGroup/setGroup/deleteGroup functions
- 🧽 Added an interface for the managedGroup (now has managedGroupImpl) to clear up some confusion

The lock/unlock bug was due in no small part to functionality not being very well isolated to the appropriate structs (manager vs. managedGroup), so I have changed "managedGroup" to an interface that the manager now uses instead of arbitrarily changing fields.  To that end, this interface now has "consume/start/stop/close/errors/processLock" as the lifecycle functions that the manager calls, and a bit of functionality has moved from the manager into the managedGroupImpl (particularly consume and processLock).

I realize this is a bit of a nuisance refactor right after merging the original lock/unlock code, but none of that is live yet (the resetoffset controller is not enabled which is what would send the control-protocol messages) and I believe this is a much better way to handle the managedGroup.

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/746#" 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":[]} -->
eric-sap: /hold
Would like to see what @travis-minke-sap thinks of this before merging
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 | 7c63fe1c3c25d0b6bfe779e9cafd0367caea096e | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/746/pull-knative-sandbox-eventing-kafka-unit-tests/1410053621243973632) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/746). [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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#746](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (7b5b417) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/000dd9c4955f4b3c3e2f71b35d30138c7d1b9770?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (000dd9c) will **increase** coverage by `0.42%`.
> The diff coverage is `78.80%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/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/746?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     #746      +/-   ##
==========================================
+ Coverage   75.29%   75.72%   +0.42%     
==========================================
  Files         132      140       +8     
  Lines        5785     6328     +543     
==========================================
+ Hits         4356     4792     +436     
- Misses       1208     1316     +108     
+ Partials      221      220       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...onsolidated/dispatcher/consumer\_message\_handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvY29uc3VtZXJfbWVzc2FnZV9oYW5kbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [.../channel/distributed/controller/util/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci91dGlsL2Rpc3BhdGNoZXIuZ28=) | `100.00% <ø> (ø)` | |
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `58.42% <ø> (-1.08%)` | :arrow_down: |
| [pkg/source/client/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9jbGllbnQvY2xpZW50Lmdv) | `15.66% <0.00%> (-19.48%)` | :arrow_down: |
| [pkg/source/reconciler/source/kafkasource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9yZWNvbmNpbGVyL3NvdXJjZS9rYWZrYXNvdXJjZS5nbw==) | `0.00% <0.00%> (ø)` | |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `47.82% <22.72%> (-27.36%)` | :arrow_down: |
| [pkg/apis/sources/v1beta1/kafka\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvc291cmNlcy92MWJldGExL2thZmthX2xpZmVjeWNsZS5nbw==) | `53.33% <50.00%> (-0.52%)` | :arrow_down: |
| [pkg/channel/consolidated/utils/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3V0aWxzL3V0aWwuZ28=) | `95.91% <50.00%> (-4.09%)` | :arrow_down: |
| [pkg/source/client/offsets.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9jbGllbnQvb2Zmc2V0cy5nbw==) | `60.00% <60.00%> (ø)` | |
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `61.58% <66.66%> (+0.10%)` | :arrow_up: |
| ... and [31 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746?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/746?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [580aed3...7b5b417](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/746?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-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 | 0b7622948371b08aeaeb44ea01b1fb5820031a81 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/746/pull-knative-sandbox-eventing-kafka-unit-tests/1410247763941134336) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/746). [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 -->

Patch:
diff --git a/config/command/resources/RESET_OFFSET.md b/config/command/resources/RESET_OFFSET.md
new file mode 100644
index 000000000..39a9672bd
--- /dev/null
+++ b/config/command/resources/RESET_OFFSET.md
@@ -0,0 +1,199 @@
+# ResetOffset Command CRD
+
+The ResetOffset "command" provides the ability to reposition the Kafka
+ConsumerGroup Offsets of a particular Knative resource (Subscription, Trigger,
+etc.) in any implementation that supports its use. See the documentation of the
+individual Kafka Channels to determine support.
+
+The ability to reposition ConsumerGroup Offsets to a specific timestamp (forward
+or backward), is intended to aid in failure recovery scenarios where you might
+otherwise have to resort to low-level manual Offset manipulation. It is expected
+to be an infrequently used tool in your toolbox for when a Subscriber has
+experienced unexpected downtime. As such, it requires some special handling
+compared with the traditional fire-and-forget nature of most CustomResources.
+
+This "command" CRD differs from your standard Kubernetes CustomResource in that
+it expresses the intent to perform a **one-time** repositioning of the Offsets
+for a particular ConsumerGroup, and not a desired final state of the cluster.
+
+## Usage
+
+Once you have determined that the Kafka Channel implementation you are using
+supports the ResetOffset CRDs, you can create and deploy a simple bit of YAML to
+initiate the repositioning:
+
+```yaml
+apiVersion: kafka.eventing.knative.dev/v1alpha1
+kind: ResetOffset
+metadata:
+  name: my-reset-offset
+  namespace: my-namespace
+spec:
+  offset:
+    time: "2021-06-17T15:04:39Z"
+  ref:
+    apiVersion: messaging.knative.dev/v1
+    kind: Subscription
+    namespace: my-namespace
+    name: my-subscription
+```
+
+It is intended that you create a ResetOffset CRD, monitor its execution status,
+and then remove it once complete. Successfully completed ResetOffset CRDs will
+not be re-executed, but will be re-evaluated by the Controller and are thus
+clutter that reduce efficiency and should be removed. Note, "completed" refers
+to the process of repositioning the Offsets only, and does not imply that the
+event processing has once again caught up to the head of the Topic.
+
+It is expected that only a single ResetOffset is ever present for a specific
+Subscription at any given time. As mentioned above, this resource is intended to
+aid in failure recovery and should not be frequently used. If however, multiple
+ResetOffset instances do exist for the same Subscription, there are no
+guarantees as to their execution order, which could result in unintended
+sequences of replayed events. Basic locking is in place to ensure the two
+ResetOffsets are handled sequentially and are not interleaved. This expectation
+doesn't hold in failure scenarios where one or more of the ResetOffsets is
+experiencing lengthy exponential back-offs, allowing the locks to expire and
+another instance to begin.
+
+While you can create the ResetOffset instance in any namespace, it makes sense
+to use the same namespace as the Subscription "ref" of which you wish to
+reposition the Offsets. If no `spec.ref.namespace` value is provided it will
+default to the ResetOffset namespace.
+
+The `spec.offset.time` is a string that can be one of **"earliest"**,
+**"latest"**, or a valid RFC3339 format timestamp. The **"earliest"** and
+**"latest"** keywords refer to the boundaries of the Kafka retention window,
+while the timestamp is expected to be a valid time in that window. Specifying
+times outside the retention window will result in a failure.
+
+The `spec.ref` is a standard Knative Reference which indicates the Subscription
+whose ConsumerGroup's Offsets will be repositioned. In the future, other
+implementations might choose to support others types (e.g., Brokers / Triggers).
+
+## Algorithm
+
+It will help to have a high-level understanding of the process for repositioning
+the Offsets when reviewing the following sections.
+
+```
+1 - Map the ResetOffset spec.Ref to the corresponding Kafka Topic and ConsumerGroup.
+
+2 - Update DataPlane communication infrastructure between Controller & Dispatchers.
+
+3 - Stop all related ConsumerGroups in the Dispatcher Replicas.
+
+4 - Reposition the Offsets of all ConsumerGroup Partitions.
+
+5 - Re-Start all related ConsumerGroups in the Dispatcher Replicas.
+```
+
+The first two steps are read-only and do not alter the cluster in any way.
+Stopping and Starting the ConsumerGroups require the Controller to communicate
+with the Dispatcher Replicas, and can possibly fail on one or more instances
+resulting in an inconsistent state. The Repositioning of the Offsets is atomic
+and will fail or succeed for all Partitions in the ConsumerGroup which greatly
+simplifies recovery scenarios as detailed below.
+
+Once the `OffsetsUpdated` step has succeeded the ConsumerGroups will never be
+stopped again, nor will the Offsets be repositioned, and the re-reconciliation
+will focus on attempting to restart the ConsumerGroups. This is to prevent
+repetitive Offset repositioning and restarting of any ConsumerGroups instances
+that were able to successfully restart.
+
+## Status
+
+The `Status` subresource of the ResetOffset CRD represents the steps required to
+perform a successful repositioning of the Kafka ConsumerGroup Offsets. The
+ResetOffset will be re-reconciled perpetually until it has `Succeeded` or is
+deleted.
+
+Additionally, meta-data is also provided indicating the Kafka `Topic`, `Group`,
+and the old/new partition `Offsets`. The meta-data information is intended to
+aid any manual recovery required in failure scenarios as described below.
+
+```yaml
+status:
+  conditions:
+  - lastTransitionTime: "2021-06-28T14:22:47Z"
+    status: "True"
+    type: AcquireDataPlaneServices
+  - lastTransitionTime: "2021-06-28T14:22:47Z"
+    status: "True"
+    type: ConsumerGroupsStarted
+  - lastTransitionTime: "2021-06-28T14:22:47Z"
+    status: "True"
+    type: ConsumerGroupsStopped
+  - lastTransitionTime: "2021-06-28T14:22:47Z"
+    status: "True"
+    type: OffsetsUpdated
+  - lastTransitionTime: "2021-06-28T14:22:47Z"
+    status: "True"
+    type: RefMapped
+  - lastTransitionTime: "2021-06-28T14:22:47Z"
+    status: "True"
+    type: Succeeded
+  group: kafka.114aee7c-9fa6-4315-ac78-c78f2053b69b
+  partitions:
+  - newOffset: 0
+    oldOffset: 2
+    partition: 0
+  - newOffset: 0
+    oldOffset: 2
+    partition: 1
+  - newOffset: 0
+    oldOffset: 2
+    partition: 2
+  - newOffset: 0
+    oldOffset: 2
+    partition: 3
+  topic: tenant1.sample-kafka-channel-1
+```
+
+## Failures
+
+The repositioning of Kafka ConsumerGroup Offsets should generally be a fast,
+successful operation. However, the process for repositioning Offsets requires
+all existing ConsumerGroup instances to be stopped and restarted, and so
+failures can potentially result in disabled Subscriptions. The Controller will
+attempt to re-reconcile failed ResetOffsets indefinitely until they are
+successful. In the case of unresolvable errors it is important to understand the
+state of the ConsumerGroup and the recommended manual recovery procedure. The
+following failure states align with the `Status` fields shown above and are
+described in the Algorithm section.
+
+- **Reference Mapping Error:** If the Controller is unable to map the `spec.ref`
+  field to a valid Subscription of a Kafka Topic / ConsumerGroup it will fail
+  the operation and re-queue to try again. At this point NO ConsumerGroups have
+  been stopped, and NO changes have been made to any Offsets, and the
+  ResetOffset can simply be removed.
+
+
+- **DataPlane Communication Error:** The Controller responsible for reconciling
+  the ResetOffset CRD communicates with the Dispatcher (DataPlane) in order to
+  Stop / Start all ConsumerGroup instances. If this communication infrastructure
+  could not be initiated an error will occur, and the Controller will re-queue
+  and try again. At this point NO ConsumerGroups have been stopped, and NO
+  changes have been made to any Offsets, and the ResetOffset can simply be
+  removed.
+
+
+- **ConsumerGroup Stop Error:** If there was a failure while attempting to stop
+  the Kafka ConsumerGroups, it is possible that some ConsumerGroup instances
+  were stopped while others were not. At this point NO Offsets have been
+  modified, and you can simply restart the associated Dispatcher Pods to restart
+  all ConsumerGroups from their current position.
+
+
+- **Offset Repositioning Error:**  If there is a failure while attempting to
+  reposition the ConsumerGroup Offsets it can be assumed that they have NOT been
+  changed. They are committed atomically as a set and will have been changed for
+  all Partitions or not at all. To recover you can simply restart the Dispatcher
+  Pods to resume processing events from their prior position.
+
+
+- **ConsumerGroup Start Error:** If the ConsumerGroups are failing to restart
+  then they HAVE been repositioned to the newly desired state and one or more
+  ConsumerGroups is failing to re-start. If the ResetOffset is stuck in this
+  state then the Offsets have been successfully repositioned, and you can simply
+  restart the Dispatcher Pods to start processing events from the new Offsets.
diff --git a/pkg/channel/distributed/README.md b/pkg/channel/distributed/README.md
index 4a3ef93c0..9a2db3e5b 100644
--- a/pkg/channel/distributed/README.md
+++ b/pkg/channel/distributed/README.md
@@ -1,6 +1,6 @@
 # Eventing-Kafka Distributed Channel
 
-This is a Kafka Channel implementation, originally contributed by
+This is the Kafka Channel implementation, originally contributed by
 [SAP's Kyma project](https://github.com/kyma-project).
 
 > See <https://github.com/knative/eventing-contrib/issues/1070> for discussion
@@ -10,58 +10,34 @@ This repo falls under the
 [Knative Code of Conduct](https://github.com/knative/community/blob/master/CODE-OF-CONDUCT.md)
 
 This project is a Knative Eventing implementation of a Kafka backed channel
-which provides advanced functionality and production grade qualities as an
-alternative to the "[consolidated](../consolidated)" implementation offers.
+which provides a more granular architecture as an alternative to what the
+original "[consolidated](../consolidated)" implementation offers. Specifically
+it deploys a single/separate Receiver, and one Dispatcher per KafkaChannel.
 
 ## Rationale
 
-The Knative consolidated implementation (originally from the eventing-contrib
-repository) already contains a Kafka implementation, so why invest the time in
-building another one? At the time this project was begun, and still today, the
-reference Kafka implementation does not provide the scaling characteristics
-required by a large and varied use case with many different Topics and
-Consumers. That implementation is based on a single choke point that could
-easily allow one Topic's traffic to impact the throughput of another Topic. It
-is also not horizontally scalable as it only supports a single instance of the
-dispatcher/consumer. Further, no ordering guarantees on the consumption side are
-provided which is required in certain use cases.
+The Knative "consolidated" KafkaChannel already provides a Kafka backed Channel
+implementation, so why invest the time in building another one? At the time this
+project was begun, and still today, the reference Kafka implementation does not
+provide the scaling characteristics required by a large and varied use case with
+many Topics and Consumers. That implementation is based on a single choke point
+that could allow one Topic's traffic to impact the throughput of another Topic.
 
 We also had the need to support a variety of Kafka providers, including Azure
-EventHubs in Kafka compatibility mode. Finally, the ability to expose Kafka
+EventHubs in Kafka compatibility mode as well as exposing the ability to
+customize the Kafka Topic management. Finally, the ability to expose Kafka
 configuration was very limited, and we needed the ability to customize certain
 aspects of the Kafka Topics / Producers / Consumers.
 
-## Background / Status
-
-The Knative-kafka project originated as an internal SAP implementation and was
-based on very early knative-eventing implementations. At the time this meant
-kube-builder and the controller-runtime library were used for the foundation of
-the controller. This also predated any of the more recent duck-typing and higher
-level abstractions (brokers, triggers, etc) which have since been added to
-knative-eventing.
-
-The implementation initially utilized the
-[Confluent Go Client](https://github.com/confluentinc/confluent-kafka-go), and
-the [librdkafka C library](https://github.com/edenhill/librdkafka) on which it
-is based. Initially the [Sarama Go Client](https://github.com/Shopify/sarama)
-had limitations which have since been resolved so we have refactored to that
-library instead.
-
-The internal SAP project for which this was intended also underwent several
-variations in it's requirements and approach, which sometimes meant the
-development of this project languished behind the fast moving and ever changing
-knative-eventing implementation. Further, internal corporate CI/CD constraints
-imposed some of the structure of the current project.
-
-Recently, however, the commitment to this effort was renewed, and the
-implementation is now current with knative-eventing **/master**. Work is in
-progress to align further in the hopes that this project can merge with the
-default "kafka" implementation. The project which previously existed in the open
-source [kyma-incubator](https://github.com/kyma-incubator/), but was moved into
-it's current home here in the
-[knative-sandbox](https://github.com/knative-sandbox) The focus is now on
-bringing the implementation into alignment with the other knative channel
-implementations.
+## Status
+
+Significant work has recently gone into aligning the two implementations from a
+CRD, configuration, authorization, and code-sharing perspective, in order to
+standardize the user experience as well as maximize code reuse. While the
+runtime architectures will always be different
+(the "raison d'etre" for having multiple implementations), the goal is to
+continue this sharing. The eventual goal might be to have a single KafkaChannel
+implementation that can deploy either runtime architecture as desired.
 
 ## Architecture
 
@@ -86,15 +62,8 @@ segregation / scaling.
 
 ### Project Structure
 
-**Eventing-kafka** is comprised of three distinct runtime K8S deployments as
-follows...
-
-- [channel](receiver/README.md) - The event receiver of the Channel to which
-  inbound messages are sent. An http server which accepts messages that conform
-  to the CloudEvent specification, and then writes those messages to the
-  corresponding Kafka Topic. This is the "Producer" from the Kafka perspective.
-  A separate Channel Deployment is created for each Kafka Secret detected in the
-  knative-eventing namespace.
+The "distributed" KafkaChannel consists of three distinct runtime K8S
+deployments as follows...
 
 - [controller](controller/README.md) - This component implements the
   `KafkaChannel` Controller. It is using the current knative-eventing "Shared
@@ -109,6 +78,12 @@ follows...
   Topic), and will contain a distinct Kafka Consumer Group for each Subscription
   to the `KafkaChannel`.
 
+- [receiver](receiver/README.md) - The event receiver to which all inbound
+  messages are sent. An HTTP server which accepts messages that conform to the
+  CloudEvent specification, and then writes those messages to the corresponding
+  Kafka Topic. This is the "Producer" from the Kafka perspective. A single
+  receiver Deployment is created to service all KafkaChannels in the cluster.
+
 - [config](../../../config/channel/distributed/README.md) - Eventing-kafka
   **ko** installable YAML files for installation.
 
@@ -123,22 +98,23 @@ the Data Plane is configured to support the flow of events as defined by
 to a KafkaChannel. The underlying Kafka infrastructure to be used is defined in
 a specially labeled
 [K8S Secret](../../../config/channel/distributed/README.md#Credentials) in the
-knative-eventing namespace. Eventing-kafka supports several different Kafka (and
+knative-eventing namespace. Eventing-kafka supports several Kafka (and
 Kafka-like)
-[infrastructures](../../../config/channel/distributed/README.md#Kafka%20Providers).
+[infrastructures](../../../config/channel/distributed/README.md#Kafka%20Providers)
+.
 
 ### Data Plane
 
 The data plane for all `KafkaChannels` runs in the knative-eventing namespace.
 There is a single deployment for the receiver side of all channels which accepts
-CloudEvents and sends them to Kafka. Each `KafkaChannel` uses one Kafka topic.
-This deployment supports horizontal scaling with linearly increasing performance
-characteristics through specifying the number of replicas.
+CloudEvents and writes them to Kafka Topics. Each `KafkaChannel` uses one Kafka
+Topic. This deployment supports horizontal scaling with linearly increasing
+performance characteristics through specifying the number of replicas.
 
 Each `KafkaChannel` has one deployment for the dispatcher side which reads from
-the Kafka topic and sends to subscribers. Each subscriber has its own Kafka
+the Kafka Topic and sends to subscribers. Each subscriber has its own Kafka
 consumer group. This deployment can be scaled up to a replica count equalling
-the number of partitions in the Kafka topic.
+the number of partitions in the Kafka Topic.
 
 ### Messaging Guarantees
 
@@ -148,12 +124,13 @@ a 202 response is received by the sender.
 The CloudEvent is partitioned based on the
 [CloudEvent partitioning extension](https://github.com/cloudevents/spec/blob/master/extensions/partitioning.md)
 field called `partitionkey`. If the `partitionkey` is not present, then the
-`subject` field will be used. Finally if neither is available, it will fall-back
-to random partitioning.
+`subject` field will be used. Finally, if neither is available, it will
+fall-back to random partitioning.
 
 Events in each partition are processed in order, with an **at-least-once**
 guarantee. If a full cycle of retries for a given subscription fails, the event
-is ignored and processing continues with the next event.
+is ignored, or sent to the DLQ according to the Subscription's `DeliverySpec`
+and processing continues with the next event.
 
 ## Installation
 
diff --git a/pkg/common/commands/resetoffset/README.md b/pkg/common/commands/resetoffset/README.md
new file mode 100644
index 000000000..d450be5b9
--- /dev/null
+++ b/pkg/common/commands/resetoffset/README.md
@@ -0,0 +1,112 @@
+# ResetOffset Command
+
+The common ResetOffset "command" provides the ability to reposition the Kafka
+ConsumerGroup Offsets for the Topic / ConsumerGroup associated with a particular
+Knative resources (Subscription, Trigger, etc.). Any Kafka related
+implementation can include this capability by including the reusable
+[Controller](./controller/controller.go) and
+[Manager](../../consumer/consumer_manager.go) which rely on the associated
+[ResetOffset CRD](../../../../config/command/resources/RESET_OFFSET.md).
+
+## Controller
+
+To work around the fact that the Knative `sharedmain` framework does not provide
+an easy mechanism to pass arguments to the Controller during instantiation, the
+ResetOffset implementation uses Factories and delayed instantiation to expose
+the necessary "custom" components that allow it to be re-usable.
+
+Including the ResetOffset Controller into your existing Controller will look
+something like this...
+
+```go
+package main
+
+import (
+	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/control-protocol/pkg/reconciler"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/kafkachannel"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
+)
+
+func main() {
+
+	...
+
+	// Create A Subscription RefMapper Factory With Custom Topic/Group Naming
+	subscriptionRefMapperFactory := refmappers.NewSubscriptionRefMapperFactory(
+		myTopicNameMapper,
+		myGroupIdMapper,
+		myConnectionPoolKeyMapper,
+		myDataPlaneNamespaceMapper,
+		myDataPlaneLabelsMapper,
+	)
+
+	// Create A control-protocol ControlPlaneConnectionPool
+	connectionPool := reconciler.NewInsecureControlPlaneConnectionPool()
+	defer connectionPool.Close(ctx)
+
+	// Create A ResetOffset ControllerConstructor Factory With Custom Subscription Ref Mapping
+	resetOffsetControllerConstructor := controller.NewControllerFactory(subscriptionRefMapperFactory, connectionPool)
+
+	// Create The SharedMain Instance With The Various Controllers
+	sharedmain.MainWithContext(ctx, ControllerComponentName, kafkachannel.NewController, resetOffsetControllerConstructor)
+}
+```
+
+## Reconciliation
+
+The general algorithm for reconciling ResetOffset instances is as follows...
+
+```
+1 - Map the ResetOffset spec.Ref to the corresponding Kafka Topic and ConsumerGroup.
+
+2 - Update DataPlane communication infrastructure between Controller & Dispatchers.
+
+3 - Stop all related ConsumerGroups in the Dispatcher Replicas.
+
+4 - Reposition the Offsets of all ConsumerGroup Partitions.
+
+5 - Re-Start all related ConsumerGroups in the Dispatcher Replicas.
+```
+
+The reconciler will continue to process a ResetOffset until is has `Succeeded`.
+It is careful to only ever reposition the ConsumerGroup Offsets a single time to
+prevent any back/forth repositioning in overlapping failure scenarios.
+
+## RefMappers
+
+In order to provide a reusable "common" implementation of the ResetOffset
+Controller while still allowing for differences in implementation between
+various use cases, it is necessary to expose a custom mechanism for mapping
+between various Knative resources (Subscriptions, Triggers, etc) and the Kafka
+Topic / ConsumerGroup to be repositioned.
+
+The [refmappers](./refmappers) package contains the Type definitions to allow
+for such customization, as well as the initial implementation for Subscriptions
+of KafkaChannels. This implementation includes common logic for loading the
+Channel associated with the subscription and providing it to the custom mappers
+responsible for determining the Kafka Topic and ConsumerGroup names.
+
+Once the Reconciler has the "mapped" `RefInfo` data, it is able to proceed with
+the Offset repositioning process.
+
+## DataPlane
+
+In order to Stop / Start the ConsumerGroups the Control-Plane needs to
+communicate with the Data-Plane where the ConsumerGroups are running. The
+Dispatchers which comprise the Data-Plane are likely to be horizontally scalable
+and thus must each be contacted. To facilitate this communication the
+[control-protocol](https://github.com/knative-sandbox/control-protocol) is used.
+
+The [ConsumerGroupAsyncCommand](../../controlprotocol/commands/consumergroup.go)
+defines the Start / Stop messages between the Control-Plane and Data-Plane. The
+commands are inherently asynchronous and are sent to all Data-Plane Replicas in
+parallel for efficiency. The responses are aggregated and block to be
+synchronous so that the Reconciler can deal with them as a single unit.
+
+While you are free to use the ResetOffset Controller with your own custom
+Data-Plane implementation, the intent is that the common
+[ConsumerManager](../../consumer/consumer_manager.go) will be used. This
+implementation already provides the expected ConsumerGroup lifecycle management
+and locking control.
diff --git a/pkg/common/commands/resetoffset/controller/dataplane.go b/pkg/common/commands/resetoffset/controller/dataplane.go
index 33dbaaf4a..dc56197fe 100644
--- a/pkg/common/commands/resetoffset/controller/dataplane.go
+++ b/pkg/common/commands/resetoffset/controller/dataplane.go
@@ -45,8 +45,12 @@ const (
 )
 
 var (
-	// TODO - How long should a Successful ResetOffset Reconciliation Take ???
-	asyncCommandLockTimeout = 1 * time.Minute // AsyncCommand ConsumerGroup Lock Timeout (Enough for successful ResetOffset reconciliation)
+	// asyncCommandLockTimeout defines the timeout of the lock kept in the DataPlane (Dispatcher)
+	// during stopping / starting the ConsumerGroups.  It should be long enough to safely perform
+	// a successful ResetOffset operation, but short enough that a failed ResetOffset attempt
+	// isn't blocking another attempt.  The locks are in-memory only and can be cleared by
+	// restarting the DataPlane (Dispatcher) Pods in order to recover.
+	asyncCommandLockTimeout = 1 * time.Minute
 )
 
 // reconcileDataPlaneServices updates the Reconciler ConnectionPool Services associated with the specified RefInfo.
diff --git a/pkg/common/consumer/consumer_manager.go b/pkg/common/consumer/consumer_manager.go
index 33bc40bde..088def87a 100644
--- a/pkg/common/consumer/consumer_manager.go
+++ b/pkg/common/consumer/consumer_manager.go
@@ -49,11 +49,9 @@ import (
 
 const (
 	defaultLockTimeout = 5 * time.Minute
+	internalToken      = "internal-token"
 )
 
-// GroupLockedError is the error returned if the a locked managed group is given a different token for access
-var GroupLockedError = fmt.Errorf("managed group lock failed: locked by a different token")
-
 // KafkaConsumerGroupManager keeps track of Sarama consumer groups and handles messages from control-protocol clients
 type KafkaConsumerGroupManager interface {
 	Reconfigure(brokers []string, config *sarama.Config) error
@@ -63,10 +61,8 @@ type KafkaConsumerGroupManager interface {
 	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
+// groupMap is a mapping of GroupIDs to managed Consumer Group interfaces
+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.
@@ -123,7 +119,7 @@ func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sa
 	var multiErr error
 	groupsToRestart := make([]string, 0, len(m.groups))
 	for groupId := range m.groups {
-		err := m.stopConsumerGroup(getInternalLockCommand(true), groupId)
+		err := m.stopConsumerGroup(&commands.CommandLock{Token: internalToken, LockBefore: true}, groupId)
 		if err != nil {
 			// If we couldn't stop a group, or failed to obtain a lock, note it as an error.  However,
 			// in a practical sense, the new brokers/config will be used when whatever locked the group
@@ -140,7 +136,7 @@ func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sa
 	// Restart any groups this function stopped
 	m.logger.Info("Reconfigure Consumer Group Manager - Starting All Managed Consumer Groups")
 	for _, groupId := range groupsToRestart {
-		err := m.startConsumerGroup(getInternalLockCommand(false), groupId)
+		err := m.startConsumerGroup(&commands.CommandLock{Token: internalToken, UnlockAfter: true}, groupId)
 		if err != nil {
 			multierr.AppendInto(&multiErr, err)
 		}
@@ -148,17 +144,6 @@ func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sa
 	return multiErr
 }
 
-// getInternalLockCommand returns a CommandLock object with a constant lock token used internally by
-// the consumer manager.
-func getInternalLockCommand(lock bool) *commands.CommandLock {
-	return &commands.CommandLock{
-		Token:       "consumer-manager-internal-token",
-		Timeout:     time.Minute,
-		LockBefore:  lock,
-		UnlockAfter: !lock,
-	}
-}
-
 // 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 {
@@ -171,10 +156,6 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(groupId string, topic
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
-	managedGrp := createManagedGroup(m.logger, group, cancel)
-
-	// Begin listening on the group's Errors() channel and write them to the managedGroup's errors channel
-	managedGrp.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
@@ -186,7 +167,7 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(groupId string, topic
 
 	// 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...)
-	managedGrp.cancelConsume = customGroup.cancel
+	managedGrp := createManagedGroup(ctx, m.logger, group, cancel, customGroup.cancel)
 
 	// 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.
@@ -200,24 +181,12 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(groupId string, topic
 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 {
+	managedGrp := m.getGroup(groupId)
+	if managedGrp == nil {
 		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 {
+	if err := managedGrp.close(); err != nil {
 		groupLogger.Error("Failed To Close Managed ConsumerGroup", zap.Error(err))
 		return err
 	}
@@ -232,10 +201,11 @@ func (m *kafkaConsumerGroupManagerImpl) CloseConsumerGroup(groupId string) error
 // 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) {
+	group := m.getGroup(groupId)
+	if group == nil {
 		return nil
 	}
-	return m.groups[groupId].errors
+	return group.errors()
 }
 
 // IsManaged returns true if the given groupId corresponds to a managed ConsumerGroup
@@ -251,22 +221,7 @@ func (m *kafkaConsumerGroupManagerImpl) consume(ctx context.Context, groupId str
 	if managedGrp == nil {
 		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
-		m.logger.Debug("Consume is waiting for managed group restart")
-		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)
-		}
-	}
+	return managedGrp.consume(ctx, topics, handler)
 }
 
 // stopConsumerGroups closes the managed ConsumerGroup identified by the provided groupId, and marks it
@@ -288,22 +243,13 @@ func (m *kafkaConsumerGroupManagerImpl) stopConsumerGroup(lock *commands.Command
 		return fmt.Errorf("stop requested for consumer group not in managed list: %s", groupId)
 	}
 
-	// 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()
+	if err := managedGrp.stop(); err != nil {
+		groupLogger.Error("Failed to stop managed consumer group", zap.Error(err))
 		return err
 	}
 
 	// Unlock the managedGroup after stopping it, if lock.UnlockAfter is true
-	if err = m.unlockAfter(lock, groupId); err != nil {
+	if err := m.unlockAfter(lock, groupId); err != nil {
 		groupLogger.Error("Failed to unlock consumer group after stopping", zap.Error(err))
 		return err
 	}
@@ -321,21 +267,23 @@ func (m *kafkaConsumerGroupManagerImpl) startConsumerGroup(lock *commands.Comman
 	}
 
 	groupLogger.Info("Starting Managed ConsumerGroup")
-	if !m.IsManaged(groupId) {
+	managedGrp := m.getGroup(groupId)
+	if managedGrp == nil {
 		groupLogger.Info("ConsumerGroup Not Managed - Ignoring Start Request")
 		return fmt.Errorf("start requested for consumer group not in managed list: %s", groupId)
 	}
 
-	group, err := m.factory.createConsumerGroup(groupId)
+	createGroup := func() (sarama.ConsumerGroup, error) {
+		return m.factory.createConsumerGroup(groupId)
+	}
+
+	// Instruct the managed group to use this new ConsumerGroup
+	err := managedGrp.start(createGroup)
 	if err != nil {
 		groupLogger.Error("Failed To Restart Managed ConsumerGroup", zap.Error(err))
 		return err
 	}
 
-	managedGrp := m.getGroup(groupId)
-	managedGrp.group = group
-	managedGrp.closeRestartChannel() // Closing this allows the waitForStart function to finish
-
 	// Unlock the managedGroup after starting it, if lock.UnlockAfter is true
 	if err = m.unlockAfter(lock, groupId); err != nil {
 		groupLogger.Error("Failed to unlock consumer group after starting", zap.Error(err))
@@ -345,73 +293,44 @@ func (m *kafkaConsumerGroupManagerImpl) startConsumerGroup(lock *commands.Comman
 }
 
 // getGroup returns a group from the groups map using the groupLock mutex
-func (m *kafkaConsumerGroupManagerImpl) getGroup(groupId string) *managedGroup {
+func (m *kafkaConsumerGroupManagerImpl) getGroup(groupId string) managedGroup {
 	m.groupLock.RLock()
 	defer m.groupLock.RUnlock()
 	return m.groups[groupId]
 }
 
-// getGroup associates a group with a groupId in the groups map using the groupLock mutex
-func (m *kafkaConsumerGroupManagerImpl) setGroup(groupId string, group *managedGroup) {
+// setGroup associates a group with a groupId in the groups map using the groupLock mutex
+func (m *kafkaConsumerGroupManagerImpl) setGroup(groupId string, group managedGroup) {
 	m.groupLock.Lock()
+	defer m.groupLock.Unlock()
 	m.groups[groupId] = group
-	m.groupLock.Unlock()
 }
 
 // getGroup removes a group from the groups map by groupId, using the groupLock mutex
 func (m *kafkaConsumerGroupManagerImpl) removeGroup(groupId string) {
 	m.groupLock.Lock()
+	defer m.groupLock.Unlock()
 	delete(m.groups, groupId)
-	m.groupLock.Unlock()
 }
 
 // lockBefore will lock the managedGroup corresponding to the groupId, if lock.LockBefore is true
 func (m *kafkaConsumerGroupManagerImpl) lockBefore(lock *commands.CommandLock, groupId string) error {
-	return m.processLock(lock, groupId, true)
+	group := m.getGroup(groupId)
+	if group == nil {
+		m.logger.Warn("Attempted to lock a nonexistent group ID", zap.String("GroupId", groupId))
+		return nil // Can't lock a nonexistent group
+	}
+	return group.processLock(lock, true)
 }
 
 // unlockAfter will unlock the managedGroup corresponding to the groupId, if lock.UnlockAfter is true
 func (m *kafkaConsumerGroupManagerImpl) unlockAfter(lock *commands.CommandLock, groupId string) error {
-	return m.processLock(lock, groupId, false)
-}
-
-// processLock handles setting and removing the managedGroup's lock status.
-// For the managedGroup with the given groupId, if that group exists, this function will:
-// - Lock the group, if "lock" is true and cmdLock.LockBefore is true
-// - Unlock the group, if "lock" is false and cmdLock.UnlockAfter is true
-// Returns an error if the lock.Token field doesn't match an existing token in the managedGroup (for either case), as
-// this indicates that the group is locked by a different sender
-func (m *kafkaConsumerGroupManagerImpl) processLock(cmdLock *commands.CommandLock, groupId string, lock bool) error {
-	if cmdLock == nil {
-		return nil // No lock processing was requested
-	}
-
 	group := m.getGroup(groupId)
 	if group == nil {
-		return nil // Can't lock a nonexistent group
-	}
-
-	if (lock && !cmdLock.LockBefore) || (!lock && !cmdLock.UnlockAfter) {
-		return nil // If neither a lock nor unlock were requested, no need to go any further
-	}
-
-	if !group.canUnlock(cmdLock.Token) {
-		m.logger.Info("Managed group access denied; already locked with a different token",
-			zap.String("Token", cmdLock.Token), zap.String("GroupId", groupId))
-		return GroupLockedError // Already locked by a different command token
+		m.logger.Warn("Attempted to unlock a nonexistent group ID", zap.String("GroupId", groupId))
+		return nil // Can't unlock a nonexistent group
 	}
-
-	if lock && cmdLock.LockBefore {
-		timeout := cmdLock.Timeout
-		if timeout == 0 {
-			timeout = defaultLockTimeout
-		}
-		group.resetLock(cmdLock.Token, timeout)
-	} else if cmdLock.UnlockAfter {
-		group.removeLock()
-	}
-
-	return nil // Lock succeeded
+	return group.processLock(lock, false)
 }
 
 // processAsyncGroupNotification calls the provided groupFunction with whatever GroupId is contained
diff --git a/pkg/common/consumer/managed_group.go b/pkg/common/consumer/managed_group.go
index 53a94e25d..9408553d9 100644
--- a/pkg/common/consumer/managed_group.go
+++ b/pkg/common/consumer/managed_group.go
@@ -18,36 +18,59 @@ package consumer
 
 import (
 	"context"
+	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 )
 
+// GroupLockedError is the error returned if the a locked managed group is given a different token for access
+var GroupLockedError = fmt.Errorf("managed group lock failed: locked by a different token")
+
+// createSaramaGroupFn represents the mechanism the managedGroup should use when creating a Sarama ConsumerGroup
+type createSaramaGroupFn func() (sarama.ConsumerGroup, error)
+
 // managedGroup contains information about a Sarama ConsumerGroup that is required to start (i.e. re-create) it
-type managedGroup struct {
+type managedGroup interface {
+	consume(context.Context, []string, sarama.ConsumerGroupHandler) error
+	start(createSaramaGroupFn) error
+	stop() error
+	close() error
+	errors() chan error
+	processLock(*commands.CommandLock, bool) error
+}
+
+// managedGroupImpl implements the managedGroup interface
+type managedGroupImpl struct {
 	logger             *zap.Logger
-	group              sarama.ConsumerGroup // The Sarama ConsumerGroup which is under management
-	errors             chan error           // An error channel that will replicate the errors from the Sarama ConsumerGroup
+	saramaGroup        sarama.ConsumerGroup // The Sarama ConsumerGroup which is under management
+	transferredErrors  chan error           // An error channel that will replicate the errors from the Sarama ConsumerGroup
 	restartWaitChannel chan struct{}        // A channel that will be closed when a stopped group is restarted
 	stopped            atomic.Value         // Boolean value indicating that the managed group is stopped
 	cancelErrors       func()               // Called by the manager's CloseConsumerGroup to terminate the error forwarding
 	cancelConsume      func()               // Called by the manager during CloseConsumerGroup
 	lockedBy           atomic.Value         // The LockToken of the ConsumerGroupAsyncCommand that requested the lock
 	cancelLockTimeout  func()               // Called internally to stop the lock timeout when a lock is removed
+	groupMutex         sync.RWMutex         // Used to synchronize access to the internal sarama ConsumerGroup
 }
 
 // createManagedGroup associates a Sarama ConsumerGroup and cancel function (usually from the factory)
 // inside a new managedGroup struct.  If a timeout is given (nonzero), the lockId will be reset to an
 // empty string (i.e. "unlocked") after that time has passed.
-func createManagedGroup(logger *zap.Logger, group sarama.ConsumerGroup, cancel func()) *managedGroup {
+func createManagedGroup(ctx context.Context, logger *zap.Logger, group sarama.ConsumerGroup, cancelErrors func(), cancelConsume func()) managedGroup {
 
-	managedGrp := managedGroup{
-		logger:       logger,
-		group:        group,
-		errors:       make(chan error),
-		cancelErrors: cancel,
+	managedGrp := &managedGroupImpl{
+		logger:            logger,
+		saramaGroup:       group,
+		transferredErrors: make(chan error),
+		cancelErrors:      cancelErrors,
+		cancelConsume:     cancelConsume,
+		groupMutex:        sync.RWMutex{},
 	}
 
 	// Atomic values must be initialized with their desired type before being accessed, or a nil
@@ -55,13 +78,106 @@ func createManagedGroup(logger *zap.Logger, group sarama.ConsumerGroup, cancel f
 	managedGrp.lockedBy.Store("")   // Empty token string indicates "unlocked"
 	managedGrp.stopped.Store(false) // A managed group defaults to "started" when created
 
-	return &managedGrp
+	// Begin listening on the group's Errors() channel and write them to the managedGroup's errors channel
+	managedGrp.transferErrors(ctx)
+
+	return managedGrp
+}
+
+// stop stops the managed group (which means closing the internal ConsumerGroup and marking the managed group as stopped)
+func (m *managedGroupImpl) stop() error {
+	// 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.
+	m.createRestartChannel()
+
+	// Close the inner sarama ConsumerGroup, which will cause our consume() function to stop
+	// and wait for the managedGroup to start again.
+	if err := m.getSaramaGroup().Close(); err != nil {
+		// Don't leave the start channel open if the group.Close() call failed; that would be misleading
+		m.closeRestartChannel()
+		return err
+	}
+	return nil
+}
+
+// start creates a Sarama ConsumerGroup using the provided createGroup function and marks the
+// managedGroup as "started" by closing the restartWaitChannel
+func (m *managedGroupImpl) start(createGroup createSaramaGroupFn) error {
+	group, err := createGroup()
+	if err != nil {
+		return err
+	}
+	m.setSaramaGroup(group)
+	m.closeRestartChannel() // Closing this allows the waitForStart function to finish
+	return nil
+}
+
+// close shuts down the internal sarama ConsumerGroup, canceling the consume and/or error transfer
+// loops first.  This is distinct from "stop" which expects the consume/error loops to continue while
+// waiting for a restart.
+func (m *managedGroupImpl) close() error {
+	// 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.
+	m.closeRestartChannel()
+
+	// Stop the error transfer and consume loops of this managedGroup, if cancel functions exist
+	if m.cancelErrors == nil {
+		m.logger.Warn("The cancelErrors function of the managed group is nil")
+	} else {
+		m.cancelErrors() // This will terminate any transferError call that is waiting for a restart
+	}
+	if m.cancelConsume == nil {
+		m.logger.Warn("The cancelConsume function of the managed group is nil")
+	} else {
+		m.cancelConsume() // This will stop the factory's consume loop after the ConsumerGroup is closed
+	}
+	return m.getSaramaGroup().Close()
+}
+
+// consume calls the Consume function on the managed ConsumerGroup, supporting the stop/start functionality
+func (m *managedGroupImpl) consume(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) error {
+	for {
+		// Call the internal sarama ConsumerGroup's Consume function directly
+		err := m.getSaramaGroup().Consume(ctx, topics, handler)
+		if !m.isStopped() {
+			m.logger.Debug("Managed Consume Finished Without Stop", 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
+		m.logger.Debug("Consume is waiting for managed group restart")
+		if !m.waitForStart(ctx) {
+			// Context was canceled; abort
+			m.logger.Debug("Managed Consume Canceled")
+			return fmt.Errorf("context was canceled waiting for group to start")
+		}
+	}
+}
+
+// getErrors returns the error channel, which is relayed from the internal managed ConsumerGroup
+func (m *managedGroupImpl) errors() chan error {
+	return m.transferredErrors
+}
+
+// getSaramaGroup returns the internal Sarama ConsumerGroup, protected by the mutex
+func (m *managedGroupImpl) getSaramaGroup() sarama.ConsumerGroup {
+	m.groupMutex.RLock()
+	defer m.groupMutex.RUnlock()
+	return m.saramaGroup
+}
+
+// setSaramaGroup sets the internal Sarama ConsumerGroup, protected by the mutex
+func (m *managedGroupImpl) setSaramaGroup(group sarama.ConsumerGroup) {
+	m.groupMutex.Lock()
+	defer m.groupMutex.Unlock()
+	m.saramaGroup = group
 }
 
 // resetLock will set the "lockedBy" field to the given lockToken (need not be the same as the
 // existing one) and reset the timer to the provided timeout.  After the timer expires, the lockToken
 // will be set to an empty string (representing "unlocked")
-func (m *managedGroup) resetLock(lockToken string, timeout time.Duration) {
+func (m *managedGroupImpl) resetLock(lockToken string, timeout time.Duration) {
 
 	// Stop any existing timer (without releasing the lock) so that it won't inadvertently do an unlock later
 	if m.cancelLockTimeout != nil {
@@ -77,7 +193,7 @@ func (m *managedGroup) resetLock(lockToken string, timeout time.Duration) {
 		m.cancelLockTimeout = cancel
 		// Mark this group as "locked" by the provided token
 		m.lockedBy.Store(lockToken)
-		m.logger.Info("Managed group locked", zap.String("token", lockToken))
+		m.logger.Info("Managed group locked", zap.String("token", lockToken), zap.Duration("Timeout", timeout))
 
 		// Reset the lockedBy field to an empty string when the lockTimer expires.  We create a new routine
 		// each time (instead of calling lockTimer.Reset) because an existing timer may have expired long ago
@@ -106,13 +222,13 @@ func (m *managedGroup) resetLock(lockToken string, timeout time.Duration) {
 
 // canUnlock returns true if the provided token is sufficient to unlock the group (that is,
 // if it either matches the existing token, or if the existing token is empty)
-func (m *managedGroup) canUnlock(token string) bool {
+func (m *managedGroupImpl) canUnlock(token string) bool {
 	lockToken := m.lockedBy.Load()
 	return lockToken == "" || lockToken == token
 }
 
 // removeLock sets the lockedBy token to an empty string, meaning "unlocked"
-func (m *managedGroup) removeLock() {
+func (m *managedGroupImpl) removeLock() {
 	if m.lockedBy.Load() != "" {
 		m.logger.Debug("Managed Group lock removed")
 		m.lockedBy.Store("")
@@ -120,16 +236,51 @@ func (m *managedGroup) removeLock() {
 	}
 }
 
+// processLock handles setting and removing the managedGroup's lock status:
+// - Lock the group, if "lock" is true and cmdLock.LockBefore is true
+// - Unlock the group, if "lock" is false and cmdLock.UnlockAfter is true
+// Returns an error if the lock.Token field doesn't match the existing token (for any case even if
+// no CommandLock is provided), as this indicates that the group is locked by the other token.
+func (m *managedGroupImpl) processLock(cmdLock *commands.CommandLock, lock bool) error {
+
+	token := ""
+	if cmdLock != nil {
+		token = cmdLock.Token
+	}
+
+	if !m.canUnlock(token) {
+		m.logger.Info("Managed group access denied; already locked with a different token",
+			zap.String("Token", token))
+		return GroupLockedError // Already locked by a different command token
+	}
+
+	if cmdLock == nil {
+		return nil // If neither a lock nor unlock were requested, no need to go any further
+	}
+
+	if lock && cmdLock.LockBefore {
+		timeout := cmdLock.Timeout
+		if timeout == 0 {
+			timeout = defaultLockTimeout
+		}
+		m.resetLock(cmdLock.Token, timeout)
+	} else if !lock && cmdLock.UnlockAfter {
+		m.removeLock()
+	}
+
+	return nil // Lock succeeded
+}
+
 // 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 {
+func (m *managedGroupImpl) 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() {
+func (m *managedGroupImpl) 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{})
@@ -138,7 +289,7 @@ func (m *managedGroup) createRestartChannel() {
 }
 
 // closeRestartChannel sets the state of the managed group to "started" by closing the restartWaitChannel channel.
-func (m *managedGroup) closeRestartChannel() {
+func (m *managedGroupImpl) closeRestartChannel() {
 	// If the managed group is already started, don't try to close the restart wait channel
 	if m.isStopped() {
 		m.stopped.Store(false)
@@ -148,7 +299,7 @@ func (m *managedGroup) closeRestartChannel() {
 
 // 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 {
+func (m *managedGroupImpl) waitForStart(ctx context.Context) bool {
 	if !m.isStopped() {
 		return true // group is already started; don't try to read from closed channel
 	}
@@ -165,18 +316,18 @@ func (m *managedGroup) waitForStart(ctx context.Context) bool {
 // 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) {
+func (m *managedGroupImpl) transferErrors(ctx context.Context) {
 	go func() {
 		for {
 			m.logger.Debug("Starting managed group error transfer")
-			for groupErr := range m.group.Errors() {
-				m.errors <- groupErr
+			for groupErr := range m.getSaramaGroup().Errors() {
+				m.transferredErrors <- 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)
+				close(m.transferredErrors)
 				return
 			}
 			// Wait for the manager to restart the Consumer Group before calling m.group.Errors() again


Test patch:
diff --git a/pkg/common/consumer/consumer_manager_test.go b/pkg/common/consumer/consumer_manager_test.go
index cabbc02ac..1acf245c9 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -19,7 +19,6 @@ package consumer
 import (
 	"context"
 	"fmt"
-	"sync"
 	"testing"
 	"time"
 
@@ -132,7 +131,6 @@ func TestCloseConsumerGroup(t *testing.T) {
 		groupId   string
 		expectErr bool
 		closeErr  error
-		cancel    bool
 	}{
 		{
 			name:      "Nonexistent GroupID",
@@ -148,39 +146,24 @@ func TestCloseConsumerGroup(t *testing.T) {
 			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
+			manager, group, _, server := getManagerWithMockGroup(t, testCase.groupId, false)
 			if group != nil {
 				group.On("Close").Return(testCase.closeErr)
-				if testCase.cancel {
-					managedGrp.cancelConsume = func() { cancelConsumeCalled = true }
-					managedGrp.cancelErrors = func() { cancelErrorsCalled = true }
-				}
+				group.On("Errors").Return(make(chan error))
 			}
 			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
 	}{
 		{
@@ -188,59 +171,55 @@ func TestConsume(t *testing.T) {
 			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",
+			groupId: "test-group-id",
 		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			manager := &kafkaConsumerGroupManagerImpl{logger: logtesting.TestLogger(t).Desugar(), groups: make(groupMap)}
+			if testCase.groupId != "" {
+				mockGroup := &mockManagedGroup{}
+				mockGroup.On("consume", context.Background(), []string{"topic"}, nil).Return(nil)
+				manager.groups[testCase.groupId] = mockGroup
+			}
+			err := manager.consume(context.Background(), testCase.groupId, []string{"topic"}, nil)
+			if testCase.expectErr != "" {
+				assert.NotNil(t, err)
+				assert.Equal(t, testCase.expectErr, err.Error())
+			} else {
+				assert.Nil(t, err)
+			}
+		})
+	}
+}
+
+func TestLockUnlockWrappers(t *testing.T) {
+
+	for _, testCase := range []struct {
+		name      string
+		groupId   string
+		expectErr bool
+	}{
 		{
-			name:      "Existing GroupID, Stopped",
-			groupId:   "test-group-id",
-			stopped:   true,
-			expectErr: "kafka: tried to use a consumer group that was closed",
+			name: "Nonexistent GroupID",
 		},
 		{
-			name:      "Existing GroupID, Canceled",
+			name:      "Existing GroupID",
 			groupId:   "test-group-id",
-			stopped:   true,
-			cancel:    true,
-			expectErr: "context was canceled waiting for group 'test-group-id' to start",
+			expectErr: true,
 		},
 	} {
 		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)
+			manager := &kafkaConsumerGroupManagerImpl{logger: logtesting.TestLogger(t).Desugar(), groups: make(groupMap)}
+			if testCase.groupId != "" {
+				mockGroup := &mockManagedGroup{}
+				mockGroup.On("processLock", mock.Anything, mock.Anything).Return(fmt.Errorf("test error"))
+				manager.groups[testCase.groupId] = mockGroup
 			}
-
-			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)
+			err := manager.lockBefore(nil, testCase.groupId)
+			assert.Equal(t, testCase.expectErr, err != nil)
+			err = manager.unlockAfter(nil, testCase.groupId)
+			assert.Equal(t, testCase.expectErr, err != nil)
 		})
 	}
 }
@@ -287,6 +266,7 @@ func TestNotifications(t *testing.T) {
 		closeErr    error
 		expectClose bool
 		lockFail    bool
+		unlockFail  bool
 		lock        *commands.CommandLock
 	}{
 		{
@@ -367,17 +347,16 @@ func TestNotifications(t *testing.T) {
 			version:     1,
 			lock:        &commands.CommandLock{UnlockAfter: true, Token: "token"},
 			initialStop: true,
+			expectStop:  true, // Still stopped, because the group is locked by a different token
 			lockFail:    true,
 		},
 		{
-			name:        "Stop Group OpCode, Unlock Failure",
-			opcode:      commands.StopConsumerGroupOpCode,
-			groupId:     "test-group-id",
-			version:     1,
-			lock:        &commands.CommandLock{UnlockAfter: true, Token: "token"},
-			expectClose: true,
-			lockFail:    true,
-			expectStop:  true,
+			name:     "Stop Group OpCode, Unlock Failure",
+			opcode:   commands.StopConsumerGroupOpCode,
+			groupId:  "test-group-id",
+			version:  1,
+			lock:     &commands.CommandLock{UnlockAfter: true, Token: "token"},
+			lockFail: true,
 		},
 		{
 			name:        "Start Group OpCode, Lock Success",
@@ -396,12 +375,31 @@ func TestNotifications(t *testing.T) {
 			expectStop:  true,
 			version:     1,
 		},
+		{
+			name:        "Start Group OpCode, Lock Success, Unlock Failure",
+			opcode:      commands.StartConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			version:     1,
+			lock:        &commands.CommandLock{LockBefore: true, Token: "token"},
+			initialStop: true,
+			unlockFail:  true,
+		},
+		{
+			name:       "Stop Group OpCode, Lock Success, Unlock Failure",
+			opcode:     commands.StopConsumerGroupOpCode,
+			groupId:    "test-group-id",
+			version:    1,
+			lock:       &commands.CommandLock{LockBefore: true, Token: "token"},
+			expectStop: true,
+			unlockFail: true,
+		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
+
 			manager, group, managedGrp, serverHandler := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
 			impl := manager.(*kafkaConsumerGroupManagerImpl)
 			if testCase.initialStop {
-				impl.groups[testCase.groupId].createRestartChannel()
+				impl.groups[testCase.groupId].(*managedGroupImpl).createRestartChannel()
 			}
 			if group != nil && testCase.expectClose {
 				group.On("Close").Return(testCase.closeErr)
@@ -411,9 +409,21 @@ func TestNotifications(t *testing.T) {
 			assert.True(t, ok)
 
 			if testCase.lockFail {
-				managedGrp.lockedBy.Store("different-lock-token")
+				managedGrp.(*managedGroupImpl).lockedBy.Store("different-lock-token")
 			} else if testCase.lock != nil {
-				managedGrp.lockedBy.Store(testCase.lock.Token)
+				managedGrp.(*managedGroupImpl).lockedBy.Store(testCase.lock.Token)
+			}
+
+			mockingManagedGroup := false
+			if testCase.unlockFail {
+				// Having lock succeed but unlock fail is difficult to accomplish practically, so use a mock instead
+				mockingManagedGroup = true
+				mockGroup := &mockManagedGroup{}
+				mockGroup.On("processLock", mock.Anything, true).Return(nil)
+				mockGroup.On("stop").Return(nil)
+				mockGroup.On("start", mock.Anything).Return(nil)
+				mockGroup.On("processLock", mock.Anything, false).Return(fmt.Errorf("unlock error"))
+				impl.groups[testCase.groupId] = mockGroup
 			}
 
 			testCommand := commands.ConsumerGroupAsyncCommand{
@@ -432,7 +442,9 @@ func TestNotifications(t *testing.T) {
 			}))
 
 			if group != nil {
-				assert.Equal(t, testCase.expectStop, impl.groups[testCase.groupId].isStopped())
+				if !mockingManagedGroup {
+					assert.Equal(t, testCase.expectStop, impl.groups[testCase.groupId].(*managedGroupImpl).isStopped())
+				}
 				group.AssertExpectations(t)
 			}
 			serverHandler.AssertExpectations(t)
@@ -440,103 +452,6 @@ func TestNotifications(t *testing.T) {
 	}
 }
 
-func TestProcessLock(t *testing.T) {
-	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
-	const existingGroup = "existing-group"
-	const newToken = "new-token"
-	const shortTimeout = 60 * time.Millisecond
-
-	for _, testCase := range []struct {
-		name            string
-		groupId         string
-		lock            *commands.CommandLock
-		existingToken   string
-		expectLock      string
-		expectUnlock    string
-		expectErrBefore error
-		expectErrAfter  error
-		expectTimerStop bool
-	}{
-		{
-			name:    "Nil LockCommand",
-			groupId: existingGroup,
-		},
-		{
-			name:    "Nonexistent Group",
-			groupId: "nonexistent-group",
-			lock:    &commands.CommandLock{LockBefore: true, UnlockAfter: true},
-		},
-		{
-			name:            "Different Lock Token",
-			groupId:         existingGroup,
-			existingToken:   "existing-token",
-			expectLock:      "existing-token",
-			expectUnlock:    "existing-token",
-			lock:            &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
-			expectErrBefore: GroupLockedError,
-			expectErrAfter:  GroupLockedError,
-		},
-		{
-			name:          "Same Lock Token",
-			groupId:       existingGroup,
-			existingToken: newToken,
-			expectLock:    newToken,
-			lock:          &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
-		},
-		{
-			name:       "Zero Timeout",
-			groupId:    existingGroup,
-			lock:       &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
-			expectLock: newToken,
-		},
-		{
-			name:            "Explicit Timeout",
-			groupId:         existingGroup,
-			lock:            &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken, Timeout: shortTimeout},
-			expectLock:      newToken,
-			expectTimerStop: true,
-		},
-		{
-			name:         "LockBefore Only",
-			groupId:      existingGroup,
-			lock:         &commands.CommandLock{LockBefore: true, Token: newToken},
-			expectLock:   newToken,
-			expectUnlock: newToken,
-		},
-		{
-			name:    "UnlockAfter Only",
-			groupId: existingGroup,
-			lock:    &commands.CommandLock{UnlockAfter: true, Token: newToken},
-		},
-		{
-			name:    "No Lock Or Unlock",
-			groupId: existingGroup,
-			lock:    &commands.CommandLock{Token: newToken},
-		},
-	} {
-		t.Run(testCase.name, func(t *testing.T) {
-			time.Sleep(time.Millisecond)
-			manager, _, managedGrp, server := getManagerWithMockGroup(t, existingGroup, false)
-			if testCase.existingToken != "" {
-				managedGrp.lockedBy.Store(testCase.existingToken)
-			}
-			impl := manager.(*kafkaConsumerGroupManagerImpl)
-			errBefore := impl.processLock(testCase.lock, testCase.groupId, true)
-			assert.Equal(t, testCase.expectLock, managedGrp.lockedBy.Load())
-			if testCase.expectTimerStop {
-				time.Sleep(2 * shortTimeout)
-				assert.Equal(t, "", managedGrp.lockedBy.Load())
-			}
-			errAfter := impl.processLock(testCase.lock, testCase.groupId, false)
-			assert.Equal(t, testCase.expectUnlock, managedGrp.lockedBy.Load())
-			assert.Equal(t, testCase.expectErrBefore, errBefore)
-			assert.Equal(t, testCase.expectErrAfter, errAfter)
-
-			server.AssertExpectations(t)
-		})
-	}
-}
-
 func TestProcessAsyncGroupNotification_BadMessage(t *testing.T) {
 	cmdFunctionCalled := false
 	cmdFunction := func(lock *commands.CommandLock, groupId string) error {
@@ -551,7 +466,7 @@ func TestProcessAsyncGroupNotification_BadMessage(t *testing.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,
+	managedGroup,
 	*controltesting.MockServerHandler) {
 
 	serverHandler := getMockServerHandler()
@@ -559,17 +474,28 @@ func getManagerWithMockGroup(t *testing.T, groupId string, factoryErr bool) (Kaf
 		if factoryErr {
 			return nil, fmt.Errorf("factory error")
 		}
-		return kafkatesting.NewMockConsumerGroup(), nil
+		mockGroup := kafkatesting.NewMockConsumerGroup()
+		mockGroup.On("Errors").Return(make(chan error))
+		return mockGroup, nil
 	}
 	manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), serverHandler, []string{}, &sarama.Config{})
 	if groupId != "" {
-		mockGrp, managedGrp := createTestGroup(t)
+		mockGroup, managedGrp := createMockAndManagedGroups(t)
 		manager.(*kafkaConsumerGroupManagerImpl).groups[groupId] = managedGrp
-		return manager, mockGrp, managedGrp, serverHandler
+		return manager, mockGroup, managedGrp, serverHandler
 	}
 	return manager, nil, nil, serverHandler
 }
 
+func createMockAndManagedGroups(t *testing.T) (*kafkatesting.MockConsumerGroup, *managedGroupImpl) {
+	mockGroup := kafkatesting.NewMockConsumerGroup()
+	mockGroup.On("Errors").Return(make(chan error))
+	managedGrp := createManagedGroup(context.Background(), logtesting.TestLogger(t).Desugar(), mockGroup, func() {}, func() {})
+	// let the transferErrors function start (otherwise AssertExpectations will randomly fail because Errors() isn't called)
+	time.Sleep(5 * time.Millisecond)
+	return mockGroup, managedGrp.(*managedGroupImpl)
+}
+
 func getMockServerHandler() *controltesting.MockServerHandler {
 	server := controltesting.GetMockServerHandler()
 	server.On("AddAsyncHandler", commands.StopConsumerGroupOpCode, commands.StopConsumerGroupResultOpCode, mock.Anything, mock.Anything).Return()
@@ -581,11 +507,6 @@ func getMockServerHandler() *controltesting.MockServerHandler {
 	return server
 }
 
-func createTestGroup(t *testing.T) (*kafkatesting.MockConsumerGroup, *managedGroup) {
-	mockGroup := kafkatesting.NewMockConsumerGroup()
-	return mockGroup, createManagedGroup(logtesting.TestLogger(t).Desugar(), 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
index df8fbdada..c01312d88 100644
--- a/pkg/common/consumer/managed_group_test.go
+++ b/pkg/common/consumer/managed_group_test.go
@@ -23,9 +23,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	logtesting "knative.dev/pkg/logging/testing"
 
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
 	kafkatesting "knative.dev/eventing-kafka/pkg/common/kafka/testing"
 )
 
@@ -55,7 +58,10 @@ func TestManagedGroup(t *testing.T) {
 			// Test stop/start of a managedGroup
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
-			group := createManagedGroup(logtesting.TestLogger(t).Desugar(), nil, cancel)
+
+			mockGroup := kafkatesting.NewMockConsumerGroup()
+			mockGroup.On("Errors").Return(make(chan error))
+			group := createManagedGroup(ctx, logtesting.TestLogger(t).Desugar(), mockGroup, cancel, func() {}).(*managedGroupImpl)
 			waitGroup := sync.WaitGroup{}
 			assert.False(t, group.isStopped())
 
@@ -82,6 +88,110 @@ func TestManagedGroup(t *testing.T) {
 
 }
 
+func TestProcessLock(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
+	const newToken = "new-token"
+	const shortTimeout = 60 * time.Millisecond
+
+	for _, testCase := range []struct {
+		name            string
+		lock            *commands.CommandLock
+		existingToken   string
+		expectLock      string
+		expectUnlock    string
+		expectErrBefore error
+		expectErrAfter  error
+		expectTimerStop bool
+	}{
+		{
+			name: "Nil LockCommand",
+		},
+		{
+			name:            "Different Lock Token",
+			existingToken:   "existing-token",
+			expectLock:      "existing-token",
+			expectUnlock:    "existing-token",
+			lock:            &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
+			expectErrBefore: GroupLockedError,
+			expectErrAfter:  GroupLockedError,
+		},
+		{
+			name:            "No Lock Provided, ManagedGroup Locked",
+			existingToken:   "existing-token",
+			expectLock:      "existing-token",
+			expectUnlock:    "existing-token",
+			expectErrBefore: GroupLockedError,
+			expectErrAfter:  GroupLockedError,
+		},
+		{
+			name:            "Empty Token, ManagedGroup Locked",
+			existingToken:   "existing-token",
+			expectLock:      "existing-token",
+			expectUnlock:    "existing-token",
+			lock:            &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: ""},
+			expectErrBefore: GroupLockedError,
+			expectErrAfter:  GroupLockedError,
+		},
+		{
+			name:            "Different Lock Token, No LockBefore Or UnlockAfter Specified",
+			existingToken:   "existing-token",
+			expectLock:      "existing-token",
+			expectUnlock:    "existing-token",
+			lock:            &commands.CommandLock{Token: newToken},
+			expectErrBefore: GroupLockedError,
+			expectErrAfter:  GroupLockedError,
+		},
+		{
+			name:          "Same Lock Token",
+			existingToken: newToken,
+			expectLock:    newToken,
+			lock:          &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
+		},
+		{
+			name:       "Zero Timeout",
+			lock:       &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
+			expectLock: newToken,
+		},
+		{
+			name:            "Explicit Timeout",
+			lock:            &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken, Timeout: shortTimeout},
+			expectLock:      newToken,
+			expectTimerStop: true,
+		},
+		{
+			name:         "LockBefore Only",
+			lock:         &commands.CommandLock{LockBefore: true, Token: newToken},
+			expectLock:   newToken,
+			expectUnlock: newToken,
+		},
+		{
+			name: "UnlockAfter Only",
+			lock: &commands.CommandLock{UnlockAfter: true, Token: newToken},
+		},
+		{
+			name: "No Lock Or Unlock",
+			lock: &commands.CommandLock{Token: newToken},
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			_, managedGrp := createMockAndManagedGroups(t)
+			if testCase.existingToken != "" {
+				managedGrp.lockedBy.Store(testCase.existingToken)
+			}
+			errBefore := managedGrp.processLock(testCase.lock, true)
+			assert.Equal(t, testCase.expectLock, managedGrp.lockedBy.Load())
+			if testCase.expectTimerStop {
+				time.Sleep(2 * shortTimeout)
+				assert.Equal(t, "", managedGrp.lockedBy.Load())
+			}
+			errAfter := managedGrp.processLock(testCase.lock, false)
+			assert.Equal(t, testCase.expectUnlock, managedGrp.lockedBy.Load())
+			assert.Equal(t, testCase.expectErrBefore, errBefore)
+			assert.Equal(t, testCase.expectErrAfter, errAfter)
+		})
+	}
+}
+
 func TestResetLockTimer(t *testing.T) {
 	const shortTimeout = 60 * time.Millisecond
 
@@ -144,7 +254,7 @@ func TestResetLockTimer(t *testing.T) {
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
 			time.Sleep(time.Millisecond)
-			_, managedGrp := createTestGroup(t)
+			_, managedGrp := createMockAndManagedGroups(t)
 			if testCase.cancelTimer != nil {
 				managedGrp.cancelLockTimeout = testCase.cancelTimer
 			}
@@ -175,6 +285,165 @@ func TestResetLockTimer(t *testing.T) {
 	}
 }
 
+func TestManagedGroupConsume(t *testing.T) {
+
+	for _, testCase := range []struct {
+		name      string
+		stopped   bool
+		cancel    bool
+		expectErr string
+	}{
+		{
+			name:      "Started",
+			expectErr: "kafka: tried to use a consumer group that was closed",
+		},
+		{
+			name:      "Stopped",
+			stopped:   true,
+			expectErr: "kafka: tried to use a consumer group that was closed",
+		},
+		{
+			name:      "Canceled",
+			stopped:   true,
+			cancel:    true,
+			expectErr: "context was canceled waiting for group to start",
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			mockGroup, mgdGroup := createMockAndManagedGroups(t)
+			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 := mgdGroup.consume(ctx, []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.saramaGroup.Close()) // Stops the MockConsumerGroup's Consume() call
+			}
+			waitGroup.Wait() // Allows the goroutine with the consume call to finish
+		})
+	}
+}
+
+func TestStopStart(t *testing.T) {
+
+	for _, testCase := range []struct {
+		name        string
+		errStopping bool
+		errStarting bool
+		stopped     bool
+	}{
+		{
+			name: "Initially Started",
+		},
+		{
+			name:        "Initially Started, Error Stopping",
+			errStopping: true,
+		},
+		{
+			name:    "Initially Stopped",
+			stopped: true,
+		},
+		{
+			name:        "Initially Stopped, Error Starting",
+			stopped:     true,
+			errStarting: true,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			mockGroup, mgdGroup := createMockAndManagedGroups(t)
+			errStopped := fmt.Errorf("error stopping")
+			if !testCase.errStopping {
+				errStopped = nil
+			}
+			mockGroup.On("Close").Return(errStopped)
+
+			if testCase.stopped {
+				// Simulate a stopped group
+				mgdGroup.createRestartChannel()
+			} else {
+				mgdGroup.stopped.Store(testCase.stopped)
+			}
+
+			err := mgdGroup.start(func() (sarama.ConsumerGroup, error) {
+				startErr := fmt.Errorf("error starting")
+				if !testCase.errStarting {
+					startErr = nil
+				}
+				return mockGroup, startErr
+			})
+			assert.Equal(t, testCase.errStarting, err != nil)
+
+			// Verify that the group is not stopped (unless there was an error)
+			assert.Equal(t, testCase.errStarting, mgdGroup.stopped.Load().(bool))
+
+			err = mgdGroup.stop()
+
+			assert.Equal(t, !testCase.errStopping || testCase.stopped, mgdGroup.stopped.Load().(bool))
+			assert.Equal(t, testCase.errStopping, err != nil)
+
+			mockGroup.AssertExpectations(t)
+
+		})
+	}
+}
+
+func TestClose(t *testing.T) {
+
+	for _, testCase := range []struct {
+		name   string
+		cancel bool
+	}{
+		{
+			name:   "With Cancel Functions",
+			cancel: true,
+		},
+		{
+			name: "Without Cancel Functions",
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			mockGroup, mgdGroup := createMockAndManagedGroups(t)
+			mockGroup.On("Close").Return(nil)
+
+			cancelConsumeCalled := false
+			cancelErrorsCalled := false
+			if testCase.cancel {
+				mgdGroup.cancelConsume = func() { cancelConsumeCalled = true }
+				mgdGroup.cancelErrors = func() { cancelErrorsCalled = true }
+			} else {
+				mgdGroup.cancelConsume = nil
+				mgdGroup.cancelErrors = nil
+			}
+
+			err := mgdGroup.close()
+			assert.Nil(t, err)
+			assert.Equal(t, testCase.cancel, cancelConsumeCalled)
+			assert.Equal(t, testCase.cancel, cancelErrorsCalled)
+		})
+	}
+}
+
 func TestTransferErrors(t *testing.T) {
 	const shortTimeout = 60 * time.Millisecond
 
@@ -204,14 +473,24 @@ func TestTransferErrors(t *testing.T) {
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
 			time.Sleep(time.Millisecond)
-			mockGrp, managedGrp := createTestGroup(t)
+
+			mockGrp := kafkatesting.NewMockConsumerGroup()
+			managedGrp := managedGroupImpl{
+				logger:            logtesting.TestLogger(t).Desugar(),
+				saramaGroup:       mockGrp,
+				transferredErrors: make(chan error),
+				groupMutex:        sync.RWMutex{},
+			}
+			managedGrp.lockedBy.Store("")
+			managedGrp.stopped.Store(false)
+
 			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
+			err := <-managedGrp.errors()
 			assert.NotNil(t, err)
 			assert.Equal(t, "test-error", err.Error())
 			if testCase.stopGroup {
@@ -228,13 +507,13 @@ func TestTransferErrors(t *testing.T) {
 				// Simulate the effects of startConsumerGroup (new ConsumerGroup, same managedConsumerGroup)
 				mockGrp = kafkatesting.NewMockConsumerGroup()
 				mockGrp.On("Errors").Return(mockGrp.ErrorChan)
-				managedGrp.group = mockGrp
+				managedGrp.saramaGroup = mockGrp
 				managedGrp.closeRestartChannel()
 
 				time.Sleep(shortTimeout) // Let the waitForStart function finish
 				// Verify that errors work again after restart
 				mockGrp.ErrorChan <- fmt.Errorf("test-error-2")
-				err = <-managedGrp.errors
+				err = <-managedGrp.errors()
 				assert.NotNil(t, err)
 				assert.Equal(t, "test-error-2", err.Error())
 				close(mockGrp.ErrorChan)
@@ -243,3 +522,36 @@ func TestTransferErrors(t *testing.T) {
 		})
 	}
 }
+
+//
+// Mock managedGroup
+//
+
+// mockManagedGroup implements the managedGroup interface
+type mockManagedGroup struct {
+	mock.Mock
+}
+
+func (m *mockManagedGroup) consume(ctx context.Context, topics []string, handler sarama.ConsumerGroupHandler) error {
+	return m.Called(ctx, topics, handler).Error(0)
+}
+
+func (m *mockManagedGroup) start(createGroup createSaramaGroupFn) error {
+	return m.Called(createGroup).Error(0)
+}
+
+func (m *mockManagedGroup) stop() error {
+	return m.Called().Error(0)
+}
+
+func (m *mockManagedGroup) close() error {
+	return m.Called().Error(0)
+}
+
+func (m *mockManagedGroup) errors() chan error {
+	return m.Called().Get(0).(chan error)
+}
+
+func (m *mockManagedGroup) processLock(cmdLock *commands.CommandLock, lock bool) error {
+	return m.Called(cmdLock, lock).Error(0)
+}