Repository: knative-sandbox/eventing-kafka
Pull request #626 (2021-05-17 14:27:04)

Resetoffset controller
Subtask of #588 to create the shared ResetOffset Controller framework.

## Proposed Changes

- Create a reusable Controller for the ResetOffset CRD which is flexible enough to support different ResetOffset.Spec.Ref types, as well as custom topic/group naming conventions within a specific Ref type.
- Implementation is incomplete in that it is not yet stopping / starting ConsumerGroups.  To manually test this with the distributed KafkaChannel implementation you can simple scale the Dispatcher replicas between 1 and 0 as needed.
- Enhanced the lifecycle states to include concepts of "initiated" and "completed" as well tracking the status of the "Ref mapping" (turning the Spec.Ref KReference into a Kafka TopicName / ConsumerGroup ID).


**Release Note**
Release Notes will be provided near the end of the ResetOffset feature development.

**Docs**
Documentation will be provided near the end of the ResetOffset feature development.

**Dev Note**
To add this new Controller to an existing deployment would look like the following (taken from dev testing against the distributed KafkaChannel)...
```
// Create A Subscription RefMapper Factory With Custom Topic/Group Naming
subscriptionRefMapperFactory := refmappers.NewSubscriptionRefMapperFactory(controllerutil.TopicNameMapper, controllerutil.GroupIdMapper)

// Create A ResetOffset ControllerConstructor Factory With Custom Subscription Ref Mapping
resetOffsetControllerConstructor := resetoffset.NewControllerFactory(subscriptionRefMapperFactory)

// Create The SharedMain Instance With The Various Controllers
sharedmain.MainWithContext(ctx, constants.ControllerComponentName, kafkachannel.NewController, resetOffsetControllerConstructor)
```
...where the above is creating a "Subscription" mapper with custom TopicName / GroupId mappers.  Both layers of abstraction are need to support the desired use cases.  If someone (kafka-broker) wanted to support Triggers instead of Subscriptions then they would implement a new "Trigger" mapper.  The "distributed" and "consolidated" KafkaChannel implementations both want to use the Subscription mapper but with different TopicName / GroupId mappers.

/hold

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **travis-minke-sap** after the PR has been reviewed.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["travis-minke-sap"]} -->
knative-prow-robot: @travis-minke-sap: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1394298439491129344) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1394298439491129344) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1394298439554043904) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1394298439491129344) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1394298439554043904) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1394298439465963520) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1394298439491129344) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1394298439554043904) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1394298439465963520) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-upgrade-tests/1394298439579209728) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1394298439491129344) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1394298439554043904) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1394298439465963520) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-upgrade-tests/1394298439579209728) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1394298439533072384) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls/1394298439491129344) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-tls`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1394298439554043904) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1394298439465963520) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-upgrade-tests/1394298439579209728) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1394298439533072384) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl/1394298439512100864) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated-sasl`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394298439654707200/artifacts/line-cov.html#file4) | 82.5% | 88.3% | 5.8
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394298439654707200/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394298439654707200/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394298439654707200/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394298439654707200/artifacts/line-cov.html#file8) | Do not exist | 64.9% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394298439654707200/artifacts/line-cov.html#file9) | Do not exist | 100.0% | 

slinkydeveloper: /assign
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#626](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (5a54673) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/2196472699f3b63a543e0bf41e8414260b0dc630?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2196472) will **increase** coverage by `0.50%`.
> The diff coverage is `86.01%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/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/626?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     #626      +/-   ##
==========================================
+ Coverage   72.84%   73.35%   +0.50%     
==========================================
  Files         140      145       +5     
  Lines        5834     6069     +235     
==========================================
+ Hits         4250     4452     +202     
- Misses       1358     1389      +31     
- Partials      226      228       +2     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...mmon/commands/resetoffset/controller/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL3JlY29uY2lsZXIuZ28=) | `60.25% <60.25%> (ø)` | |
| [...mmon/commands/resetoffset/controller/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL2NvbnRyb2xsZXIuZ28=) | `85.71% <85.71%> (ø)` | |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X2xpZmVjeWNsZS5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/channel/distributed/common/kafka/util/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2thZmthL3V0aWwvdXRpbC5nbw==) | `100.00% <100.00%> (ø)` | |
| [...hannel/distributed/controller/util/subscription.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci91dGlsL3N1YnNjcmlwdGlvbi5nbw==) | `100.00% <100.00%> (ø)` | |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `74.33% <100.00%> (ø)` | |
| [...kg/common/commands/resetoffset/controller/event.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL2V2ZW50Lmdv) | `100.00% <100.00%> (ø)` | |
| [.../common/commands/resetoffset/controller/offsets.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL29mZnNldHMuZ28=) | `100.00% <100.00%> (ø)` | |
| [...on/commands/resetoffset/refmappers/subscription.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9yZWZtYXBwZXJzL3N1YnNjcmlwdGlvbi5nbw==) | `100.00% <100.00%> (ø)` | |
| ... and [4 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626/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/626?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/626?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [2196472...5a54673](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/626?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: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **slinkydeveloper** after the PR has been reviewed.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["slinkydeveloper"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394378847989796864/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394378847989796864/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394378847989796864/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394378847989796864/artifacts/line-cov.html#file8) | Do not exist | 64.9% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394378847989796864/artifacts/line-cov.html#file9) | Do not exist | 100.0% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1394298439554043904) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1394298439465963520) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-upgrade-tests | 5a54673541143058bf38192226b12157759ec24c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-upgrade-tests/1394298439579209728) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | 71ad66677e19d6673fd2e172e7cd9704417ea397 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1394378847893327872) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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 -->
slinkydeveloper: /assign
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/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394644676543254528/artifacts/line-cov.html#file4) | 82.5% | 88.3% | 5.8
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394644676543254528/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394644676543254528/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394644676543254528/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394644676543254528/artifacts/line-cov.html#file8) | Do not exist | 64.9% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394644676543254528/artifacts/line-cov.html#file9) | 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/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394725649045786624/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394725649045786624/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394725649045786624/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394725649045786624/artifacts/line-cov.html#file8) | Do not exist | 61.2% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394725649045786624/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394738195253432320/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394738195253432320/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394738195253432320/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394738195253432320/artifacts/line-cov.html#file8) | Do not exist | 61.2% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394738195253432320/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394764952308289536/artifacts/line-cov.html#file4) | 82.5% | 88.3% | 5.8
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394764952308289536/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394764952308289536/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394764952308289536/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394764952308289536/artifacts/line-cov.html#file8) | Do not exist | 61.2% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1394764952308289536/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395117492061343744/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395117492061343744/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395117492061343744/artifacts/line-cov.html#file7) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395117492061343744/artifacts/line-cov.html#file8) | Do not exist | 61.2% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395117492061343744/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395375007151951872/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395375007151951872/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395375007151951872/artifacts/line-cov.html#file7) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395375007151951872/artifacts/line-cov.html#file8) | Do not exist | 61.2% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395375007151951872/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

knative-prow-robot: @travis-minke-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 | 2b61d71ff424c22e423087caf131ba28128d257d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-unit-tests/1395834575917158400) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395834575938129920/artifacts/line-cov.html#file3) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395834575938129920/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395834575938129920/artifacts/line-cov.html#file5) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395834575938129920/artifacts/line-cov.html#file6) | Do not exist | 61.2% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395834575938129920/artifacts/line-cov.html#file7) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395836017910484992/artifacts/line-cov.html#file3) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395836017910484992/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395836017910484992/artifacts/line-cov.html#file5) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395836017910484992/artifacts/line-cov.html#file6) | Do not exist | 60.3% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395836017910484992/artifacts/line-cov.html#file7) | Do not exist | 94.3% | 

knative-prow-robot: @travis-minke-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 | 2b61d71ff424c22e423087caf131ba28128d257d | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-unit-tests/1395834575917158400) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-build-tests | dcd30b7fc7734fcc9cf2b7d6b9851754ade9bf8c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-build-tests/1395836017721741312) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

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

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395843311566065664/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395843311566065664/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395843311566065664/artifacts/line-cov.html#file7) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395843311566065664/artifacts/line-cov.html#file8) | Do not exist | 60.3% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395843311566065664/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

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

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-sap: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-build-tests | e4e4bda3c703cfaba02c36c7b6fe79e14390d698 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-build-tests/1395843309888344064) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | e4e4bda3c703cfaba02c36c7b6fe79e14390d698 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1395843309431164928) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395860983410331648/artifacts/line-cov.html#file4) | 82.5% | 88.3% | 5.8
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395860983410331648/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395860983410331648/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395860983410331648/artifacts/line-cov.html#file7) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395860983410331648/artifacts/line-cov.html#file8) | Do not exist | 60.3% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395860983410331648/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395863012543303680/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395863012543303680/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395863012543303680/artifacts/line-cov.html#file7) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395863012543303680/artifacts/line-cov.html#file8) | Do not exist | 60.3% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395863012543303680/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395864837472391168/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395864837472391168/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395864837472391168/artifacts/line-cov.html#file7) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395864837472391168/artifacts/line-cov.html#file8) | Do not exist | 60.3% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1395864837472391168/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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/626#" title="Author self-approved">travis-minke-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)~~ [travis-minke-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: @travis-minke-sap: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 1b8a5de61dcbc705d0bdd58fcec06e55048a75ed | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1397267612940898304) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-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 | 1b8a5de61dcbc705d0bdd58fcec06e55048a75ed | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-upgrade-tests/1397267612970258432) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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 -->
travis-minke-sap: /test pull-knative-sandbox-eventing-kafka-upgrade-tests
matzew: I find the `reconcileOffsets` a bit hard to read... but generally OK w/ this PR.

Perhaps any comments from @aliok ?
aliok: I finally have time to review this PR. Give me some time please.
aliok: I think the PR is good. I didn't get the `drainPartitionOffsetManagerErrors` part but I saw the discussion above and I don't have something to help.

LGTM
knative-prow-robot: @travis-minke-sap: PR needs rebase.

<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.
</details>
knative-prow-robot: @travis-minke-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 | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-unit-tests/1397617562937724928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-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 | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-unit-tests/1397617562937724928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397617562958696448) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-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 | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-unit-tests/1397617562937724928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397617562958696448) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-build-tests | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-build-tests/1397617562912559104) | `/test pull-knative-sandbox-eventing-kafka-build-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-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 | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-unit-tests/1397617562937724928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397617562958696448) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-build-tests | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-build-tests/1397617562912559104) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1397617562837061632) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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-prow-robot: @travis-minke-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 | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-unit-tests/1397617562937724928) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`
pull-knative-sandbox-eventing-kafka-go-coverage | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397617562958696448) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`
pull-knative-sandbox-eventing-kafka-build-tests | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-build-tests/1397617562912559104) | `/test pull-knative-sandbox-eventing-kafka-build-tests`
pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed/1397617562837061632) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-distributed`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | b8933dc9bbe97f80342d1340731f1bafbee92618 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1397617562753175552) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397624386567016448/artifacts/line-cov.html#file4) | 88.3% | 82.5% | -5.8
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397624386567016448/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397624386567016448/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397624386567016448/artifacts/line-cov.html#file7) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397624386567016448/artifacts/line-cov.html#file8) | Do not exist | 65.1% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397624386567016448/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 6d2e4577379e9356eb55e4b962d9bdaac4e21f05 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1397624386340524032) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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 -->
travis-minke-sap: /test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated
knative-prow-robot: @travis-minke-sap: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 6d2e4577379e9356eb55e4b962d9bdaac4e21f05 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1397630471675842560) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/626). [Your PR dashboard](https://gubernator.knative.dev/pr/travis-minke-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/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397645436189151232/artifacts/line-cov.html#file4) | 88.3% | 82.5% | -5.8
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397645436189151232/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397645436189151232/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397645436189151232/artifacts/line-cov.html#file7) | Do not exist | 98.7% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397645436189151232/artifacts/line-cov.html#file8) | Do not exist | 65.1% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397645436189151232/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397678893124554752/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397678893124554752/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397678893124554752/artifacts/line-cov.html#file7) | Do not exist | 96.8% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397678893124554752/artifacts/line-cov.html#file8) | Do not exist | 65.6% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397678893124554752/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

travis-minke-sap: > I find the `reconcileOffsets` a bit hard to read... but generally OK w/ this PR.

No one liked the implementation in `offsets.go` so I made another attempt at it.  I think it is better (decomposed a bit) but do keep in mind that "some" of the prior and current complexity is inherently due to the Sarama library itself.  Specifically we need to do things in the following order which just complicates the implementation...

1. Create Sarama Client / OffsetManager / PartitionOffsetManagers
2. Calculate Old/New Offsets
3. MarkOffset() or ResetOffset()
4. Commit
5. AsyncClose PartitionOffsetManagers
6. Close OffsetManager
7. Drain PartitionOffsetManagers.Errors() channel
8. Close the Sarama Client

appreciate any reviews!    You can (should) focus on just the `offsets.go` file and it's probably easier to just open and review that file by itself instead of looking at the Github Diff ; )



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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397689904082718720/artifacts/line-cov.html#file5) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397689904082718720/artifacts/line-cov.html#file6) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397689904082718720/artifacts/line-cov.html#file7) | Do not exist | 96.8% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397689904082718720/artifacts/line-cov.html#file8) | Do not exist | 65.6% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1397689904082718720/artifacts/line-cov.html#file9) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398020629201948672/artifacts/line-cov.html#file6) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398020629201948672/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398020629201948672/artifacts/line-cov.html#file8) | Do not exist | 96.8% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398020629201948672/artifacts/line-cov.html#file9) | Do not exist | 96.9% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398020629201948672/artifacts/line-cov.html#file10) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398027580279361536/artifacts/line-cov.html#file6) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398027580279361536/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398027580279361536/artifacts/line-cov.html#file8) | Do not exist | 96.8% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398027580279361536/artifacts/line-cov.html#file9) | Do not exist | 96.7% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398027580279361536/artifacts/line-cov.html#file10) | Do not exist | 94.3% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398278773995999232/artifacts/line-cov.html#file6) | Do not exist | 91.7% | 
[pkg/common/commands/resetoffset/controller/event.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398278773995999232/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 
[pkg/common/commands/resetoffset/controller/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398278773995999232/artifacts/line-cov.html#file8) | Do not exist | 96.8% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398278773995999232/artifacts/line-cov.html#file9) | Do not exist | 96.7% | 
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/626/pull-knative-sandbox-eventing-kafka-go-coverage/1398278773995999232/artifacts/line-cov.html#file10) | Do not exist | 94.3% | 

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/626#issuecomment-850470416" title="Approved">eric-sap</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/626#" title="Author self-approved">travis-minke-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)~~ [travis-minke-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":[]} -->

Patch:
diff --git a/go.mod b/go.mod
index eaadb80a3..bfaac2598 100644
--- a/go.mod
+++ b/go.mod
@@ -25,6 +25,7 @@ require (
 	github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c
 	go.opencensus.io v0.23.0
 	go.uber.org/atomic v1.7.0
+	go.uber.org/multierr v1.6.0
 	go.uber.org/zap v1.16.0
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go
index 4455965ad..08754a7d8 100644
--- a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go
@@ -19,10 +19,12 @@ package v1alpha1
 import (
 	"sync"
 
+	corev1 "k8s.io/api/core/v1"
 	"knative.dev/pkg/apis"
 )
 
 var cs = apis.NewBatchConditionSet(
+	ResetOffsetConditionRefMapped,
 	ResetOffsetConditionResetInitiated,
 	ResetOffsetConditionConsumerGroupsStopped,
 	ResetOffsetConditionOffsetsUpdated,
@@ -34,6 +36,11 @@ const (
 	// ResetOffsetConditionSucceeded has status True when all sub-conditions below have been set to True.
 	ResetOffsetConditionSucceeded = apis.ConditionSucceeded
 
+	// ResetOffsetConditionRefMapped has status True when the ResetOffset.Spec.Ref has been
+	// successfully mapped to the corresponding Kafka Topic name and ConsumerGroup ID.  These
+	// values will then be populated in the the ResetOffsetStatus.
+	ResetOffsetConditionRefMapped apis.ConditionType = "RefMapped"
+
 	// ResetOffsetConditionResetInitiated has status True when the ResetOffset is being processed
 	// Until that time it is either "Unknown" when first encountered, or "Blocked" if another
 	// ResetOffset is being processed for the same "Ref" resource.
@@ -78,8 +85,18 @@ func (ros *ResetOffsetStatus) GetCondition(t apis.ConditionType) *apis.Condition
 	return ros.GetConditionSet().Manage(ros).GetCondition(t)
 }
 
-// IsCompleted returns true if the resource is ready overall.
+// IsInitiated returns true if the ResetOffsetConditionResetInitiated status is true.
+func (ros *ResetOffsetStatus) IsInitiated() bool {
+	return ros.GetConditionSet().Manage(ros).GetCondition(ResetOffsetConditionResetInitiated).Status == corev1.ConditionTrue
+}
+
+// IsCompleted returns true if the ResetOffsetConditionSucceeded status is not unknown.
 func (ros *ResetOffsetStatus) IsCompleted() bool {
+	return ros.GetConditionSet().Manage(ros).GetCondition(ResetOffsetConditionSucceeded).Status != corev1.ConditionUnknown
+}
+
+// IsSucceeded returns true if the ResetOffsetConditionSucceeded status is true.
+func (ros *ResetOffsetStatus) IsSucceeded() bool {
 	return ros.GetConditionSet().Manage(ros).IsHappy()
 }
 
@@ -88,6 +105,14 @@ func (ros *ResetOffsetStatus) InitializeConditions() {
 	ros.GetConditionSet().Manage(ros).InitializeConditions()
 }
 
+func (ros *ResetOffsetStatus) MarkRefMappedFailed(reason, messageFormat string, messageA ...interface{}) {
+	ros.GetConditionSet().Manage(ros).MarkFalse(ResetOffsetConditionRefMapped, reason, messageFormat, messageA...)
+}
+
+func (ros *ResetOffsetStatus) MarkRefMappedTrue() {
+	ros.GetConditionSet().Manage(ros).MarkTrue(ResetOffsetConditionRefMapped)
+}
+
 func (ros *ResetOffsetStatus) MarkResetInitiatedFailed(reason, messageFormat string, messageA ...interface{}) {
 	ros.GetConditionSet().Manage(ros).MarkFalse(ResetOffsetConditionResetInitiated, reason, messageFormat, messageA...)
 }
diff --git a/pkg/channel/distributed/common/kafka/util/util.go b/pkg/channel/distributed/common/kafka/util/util.go
index 9fc10b707..06497aad4 100644
--- a/pkg/channel/distributed/common/kafka/util/util.go
+++ b/pkg/channel/distributed/common/kafka/util/util.go
@@ -23,17 +23,22 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 )
 
-// Get The Formatted Kafka Topic Name From The Specified Components
+// TopicName returns a formatted string representing the Kafka Topic name.
 func TopicName(namespace string, name string) string {
 	return fmt.Sprintf("%s.%s", namespace, name)
 }
 
-// Append The KafkaChannel Service Name Suffix To The Specified String
+// GroupId returns a formatted string representing the Kafka ConsumerGroup ID.
+func GroupId(uid string) string {
+	return fmt.Sprintf("kafka.%s", uid)
+}
+
+// AppendKafkaChannelServiceNameSuffix appends the KafkaChannel Service name suffix to the specified string.
 func AppendKafkaChannelServiceNameSuffix(channelName string) string {
 	return fmt.Sprintf("%s-%s", channelName, constants.KafkaChannelServiceNameSuffix)
 }
 
-// Remove The KafkaChannel Service Name Suffix From The Specified String
+// TrimKafkaChannelServiceNameSuffix removes the KafkaChannel Service name suffix from the specified string.
 func TrimKafkaChannelServiceNameSuffix(serviceName string) string {
 	return strings.TrimSuffix(serviceName, "-"+constants.KafkaChannelServiceNameSuffix)
 }
diff --git a/pkg/channel/distributed/controller/util/subscription.go b/pkg/channel/distributed/controller/util/subscription.go
index 3adf58f87..b7e1dd81e 100644
--- a/pkg/channel/distributed/controller/util/subscription.go
+++ b/pkg/channel/distributed/controller/util/subscription.go
@@ -22,16 +22,18 @@ import (
 	"go.uber.org/zap"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+
+	commonkafkautil "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/util"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 )
 
-// Get A Logger With Subscription Info
+// SubscriptionLogger returns a Logger with Subscription info.
 func SubscriptionLogger(logger *zap.Logger, subscription *messagingv1.Subscription) *zap.Logger {
 	return logger.With(zap.String("Subscription", fmt.Sprintf("%s/%s", subscription.Namespace, subscription.Name)))
 }
 
-// Create A New ControllerReference Model For The Specified Subscription
+// NewSubscriptionControllerRef creates a new Controller OwnerReference for the specified Knative Subscription.
 func NewSubscriptionControllerRef(subscription *messagingv1.Subscription) metav1.OwnerReference {
 	return *metav1.NewControllerRef(subscription, schema.GroupVersionKind{
 		Group:   messagingv1.SchemeGroupVersion.Group,
@@ -39,3 +41,24 @@ func NewSubscriptionControllerRef(subscription *messagingv1.Subscription) metav1
 		Kind:    constants.KnativeSubscriptionKind,
 	})
 }
+
+// TopicNameMapper returns a string representing the Kafka Topic name for the specified Knative Subscription.
+func TopicNameMapper(subscription *messagingv1.Subscription) (string, error) {
+	if subscription == nil {
+		return "", fmt.Errorf("unable to format topic name for nil Subscription")
+	}
+	channelName := subscription.Spec.Channel.Name
+	channelNamespace := subscription.Spec.Channel.Namespace
+	if len(channelNamespace) <= 0 {
+		channelNamespace = subscription.Namespace
+	}
+	return commonkafkautil.TopicName(channelNamespace, channelName), nil
+}
+
+// GroupIdMapper returns a string representing the Kafka ConsumerGroup ID for the specified Knative Subscription.
+func GroupIdMapper(subscription *messagingv1.Subscription) (string, error) {
+	if subscription == nil {
+		return "", fmt.Errorf("unable to format group id for nil Subscription")
+	}
+	return commonkafkautil.GroupId(string(subscription.UID)), nil
+}
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 691b8d4f9..6e8a1821d 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -18,26 +18,24 @@ package dispatcher
 
 import (
 	"context"
-	"fmt"
 	"strings"
 	"sync"
 	"time"
 
-	dispatcherconstants "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/constants"
-
-	gometrics "github.com/rcrowley/go-metrics"
-
-	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
-
 	"github.com/Shopify/sarama"
+	gometrics "github.com/rcrowley/go-metrics"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/pkg/channel"
+
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer"
+	commonkafkautil "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/util"
+	dispatcherconstants "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/constants"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
-	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
-	"knative.dev/eventing/pkg/channel"
 )
 
 // DispatcherConfig Defines A Dispatcher Config Struct To Hold Configuration
@@ -143,7 +141,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 		if _, ok := d.subscribers[subscriberSpec.UID]; !ok {
 
 			// Format The GroupId For The Specified Subscriber
-			groupId := fmt.Sprintf("kafka.%s", subscriberSpec.UID)
+			groupId := commonkafkautil.GroupId(string(subscriberSpec.UID))
 
 			// Create A ConsumerGroup Logger
 			logger := d.Logger.With(zap.String("GroupId", groupId))
diff --git a/pkg/common/commands/resetoffset/controller/controller.go b/pkg/common/commands/resetoffset/controller/controller.go
new file mode 100644
index 000000000..f3b195e2d
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/controller.go
@@ -0,0 +1,77 @@
+/*
+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 controller
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/injection"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/system"
+
+	"knative.dev/eventing-kafka/pkg/client/injection/informers/kafka/v1alpha1/resetoffset"
+	resetoffsetreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
+	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
+)
+
+// NewControllerFactory returns a ControllerConstructor function capable of creating a "typed" ResetOffset Controller
+func NewControllerFactory(refMapperFactory refmappers.ResetOffsetRefMapperFactory) injection.ControllerConstructor {
+
+	// Return The New ResetOffset ControllerConstructor Function
+	return func(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+
+		// Get A Logger
+		logger := logging.FromContext(ctx)
+
+		// Get The Needed Informers
+		resetoffsetInformer := resetoffset.Get(ctx)
+
+		// Create The RefMapper Via The Supplied Factory Using Initialized Context
+		refMapper := refMapperFactory.Create(ctx)
+
+		// Create A ResetOffset Reconciler
+		reconciler := &Reconciler{
+			resetoffsetLister: resetoffsetInformer.Lister(),
+			refMapper:         refMapper,
+		}
+
+		// Setup Reconciler To Watch The Kafka ConfigMap For Changes
+		err := commonconfig.InitializeKafkaConfigMapWatcher(ctx, cmw, logger, reconciler.updateKafkaConfig, system.Namespace())
+		if err != nil {
+			logger.Fatal("Failed To Initialize ConfigMap Watcher", zap.Error(err))
+		}
+
+		// Create A New ResetOffset Controller Impl With The Reconciler
+		controllerImpl := resetoffsetreconciler.NewImpl(ctx, reconciler)
+
+		// Configure The Informers' EventHandlers
+		logger.Info("Setting Up EventHandlers")
+		resetoffsetInformer.Informer().AddEventHandler(controller.HandleAll(controllerImpl.Enqueue))
+
+		// Return The ResetOffset Controller
+		return controllerImpl
+	}
+}
+
+// Shutdown performs clean tear-down of resources.
+func Shutdown() {
+	// Currently nothing to do
+}
diff --git a/pkg/common/commands/resetoffset/controller/event.go b/pkg/common/commands/resetoffset/controller/event.go
new file mode 100644
index 000000000..d6bdb24ba
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/event.go
@@ -0,0 +1,47 @@
+/*
+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 controller
+
+// CoreV1EventType "Enum" Type
+type CoreV1EventType int
+
+// CoreV1 EventType "Enum" Values For ResetOffset
+const (
+	ResetOffsetReconciled CoreV1EventType = iota
+	ResetOffsetFinalized
+	ResetOffsetSkipped
+)
+
+// CoreV1 EventType String Value
+func (et CoreV1EventType) String() string {
+
+	// Default The EventType String Value
+	eventTypeString := "Unknown Event Type"
+
+	// Map EventTypes To Their String Values
+	switch et {
+	case ResetOffsetReconciled:
+		eventTypeString = "ResetOffsetReconciled"
+	case ResetOffsetFinalized:
+		eventTypeString = "ResetOffsetFinalized"
+	case ResetOffsetSkipped:
+		eventTypeString = "ResetOffsetSkipped"
+	}
+
+	// Return The EventType String Value
+	return eventTypeString
+}
diff --git a/pkg/common/commands/resetoffset/controller/offsets.go b/pkg/common/commands/resetoffset/controller/offsets.go
new file mode 100644
index 000000000..956728068
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/offsets.go
@@ -0,0 +1,295 @@
+/*
+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 controller
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/Shopify/sarama"
+	"go.uber.org/multierr"
+	"go.uber.org/zap"
+	"knative.dev/pkg/logging"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+// PartitionOffsetManagers is a map of Partition -> Sarama PartitionOffsetManager
+type PartitionOffsetManagers map[int32]sarama.PartitionOffsetManager
+
+// SaramaNewClientFnType defines the Sarama NewClient() function signature.
+type SaramaNewClientFnType func([]string, *sarama.Config) (sarama.Client, error)
+
+// SaramaNewClientFn is a reference to the Sarama NewClient() function used
+// when reconciling offsets which facilitates stubbing in unit tests.
+var SaramaNewClientFn SaramaNewClientFnType = sarama.NewClient
+
+// SaramaNewOffsetManagerFromClientFnType defines the Sarama NewOffsetManagerFromClient() function signature.
+type SaramaNewOffsetManagerFromClientFnType func(group string, client sarama.Client) (sarama.OffsetManager, error)
+
+// SaramaNewOffsetManagerFromClientFn is a reference to the Sarama NewOffsetManagerFromClient()
+// function used when reconciling offsets which facilitates stubbing in unit tests.
+var SaramaNewOffsetManagerFromClientFn SaramaNewOffsetManagerFromClientFnType = sarama.NewOffsetManagerFromClient
+
+// reconcileOffsets updates the Offsets of all Partitions for the specified
+// Topic / ConsumerGroup to the Offset value corresponding to the specified
+// offsetTime (millis since epoch) and return OffsetMappings of the old/new
+// state.  An error will be returned and the Offsets will not be committed
+// if any problems occur.
+func (r *Reconciler) reconcileOffsets(ctx context.Context, topicName string, groupId string, offsetTime int64) ([]kafkav1alpha1.OffsetMapping, error) {
+
+	// Get The Logger From The Context & Enhance The With Parameters
+	logger := logging.FromContext(ctx).Desugar().With(
+		zap.String("Topic", topicName),
+		zap.String("Group", groupId),
+		zap.Int64("Time", offsetTime))
+
+	// Initialize A New Sarama Client
+	//
+	// ResetOffset is an infrequently used feature so there is no need for
+	// reuse, and there are also "broken-pipe" failures (non-recoverable)
+	// after periods of inactivity to deal with...
+	//   https://github.com/Shopify/sarama/issues/1162
+	//   https://github.com/Shopify/sarama/issues/866
+	saramaClient, err := SaramaNewClientFn(r.kafkaBrokers, r.saramaConfig)
+	defer safeCloseSaramaClient(logger, saramaClient)
+	if saramaClient == nil || err != nil {
+		logger.Error("Failed to create a new Sarama Client", zap.Error(err))
+		return nil, err
+	}
+
+	// Get The Partitions Of The Specified Kafka Topic
+	partitions, err := saramaClient.Partitions(topicName)
+	if err != nil {
+		logger.Error("Failed to determine Partitions for Topic", zap.Error(err))
+		return nil, err
+	}
+	logger.Debug("Found Topic Partitions", zap.Any("Partitions", partitions))
+
+	// Create An OffsetManager For The Specified ConsumerGroup
+	offsetManager, err := SaramaNewOffsetManagerFromClientFn(groupId, saramaClient)
+	if offsetManager == nil || err != nil {
+		logger.Error("Failed to create OffsetManager for ConsumerGroup", zap.Error(err))
+		return nil, err
+	}
+
+	// Create The Required PartitionOffsetManagers For The Specified Topic / Partitions
+	partitionOffsetManagers, err := createPartitionOffsetManagers(offsetManager, topicName, partitions)
+	if err != nil {
+		logger.Error("Failed to create PartitionOffsetManagers for Topic Partitions", zap.Error(err))
+		_ = closeManagersAndDrainErrors(logger, offsetManager, partitionOffsetManagers)
+		return nil, err
+	}
+
+	// Update All Topic Partitions To The Specified Offset Time
+	offsetMappings, err := updateOffsets(logger, saramaClient, offsetManager, partitionOffsetManagers, topicName, partitions, offsetTime)
+	if err != nil {
+		logger.Error("Failed to update Offsets for Topic Partitions", zap.Error(err))
+		_ = closeManagersAndDrainErrors(logger, offsetManager, partitionOffsetManagers)
+		return nil, err
+	}
+
+	// Close The Sarama Managers And Get Any Accumulated Errors
+	err = closeManagersAndDrainErrors(logger, offsetManager, partitionOffsetManagers)
+	if err != nil {
+		logger.Error("PartitionOffsetManager Errors encountered", zap.Error(err))
+		return nil, err
+	}
+
+	// Return Success
+	return offsetMappings, nil
+}
+
+// updateOffsets attempts to update all of the specified Topic's Partitions
+// and performs the final Commit() if all were successfully updated.  The
+// old/new Offset values are returned if successful.  Per the Sarama library
+// implementation, Errors directly related to Offset management are available
+// on the respective PartitionOffsetManager's Error channel.  Such errors are
+// not returned here as they should be drained after closing the Managers.
+func updateOffsets(logger *zap.Logger,
+	saramaClient sarama.Client,
+	offsetManager sarama.OffsetManager,
+	partitionOffsetManagers PartitionOffsetManagers,
+	topicName string,
+	partitions []int32,
+	offsetTime int64) ([]kafkav1alpha1.OffsetMapping, error) {
+
+	// The OffsetMappings To Be Returned For ResetOffset Status
+	offsetMappings := make([]kafkav1alpha1.OffsetMapping, len(partitions))
+
+	// Loop Over The Partitions - Updating Offsets & Tracking Results
+	for index, partition := range partitions {
+
+		// Enhance The Logger With Partition
+		logger = logger.With(zap.Int32("Partition", partition))
+
+		// Get The PartitionOffsetManager For The Current Partition
+		partitionOffsetManager := partitionOffsetManagers[partition]
+		if partitionOffsetManager == nil {
+			logger.Error("Missing PartitionOffsetManager - unable to update Offset")
+			return nil, fmt.Errorf("missing PartitionOffsetManager - unable to update Offset")
+		}
+
+		// Update The Individual Offset To Specified Time
+		offsetMapping, updateErr := updateOffset(logger, saramaClient, partitionOffsetManager, topicName, partition, offsetTime)
+		if updateErr != nil {
+			logger.Error("Failed to update Offset - skipping Commit", zap.Error(updateErr))
+			return nil, updateErr
+		}
+		offsetMappings[index] = *offsetMapping
+	}
+
+	// All Partitions Updated Successfully - Commit The New Offsets!
+	logger.Info("All Offsets updated successfully - performing Commit")
+	offsetManager.Commit() // No Errors Returned - Will be in PartitionOffsetManager.Errors() Channel Post-Close!
+
+	// Return Success!
+	return offsetMappings, nil
+}
+
+// updateOffset calculates and performs an update of a single Partition's Offset
+// and returns an OffsetMapping representing the old/new state.  No Offset changes
+// are committed to allow for atomic commit/fail decision for all Offsets.
+func updateOffset(logger *zap.Logger,
+	saramaClient sarama.Client,
+	partitionOffsetManager sarama.PartitionOffsetManager,
+	topic string,
+	partition int32,
+	offsetTime int64) (*kafkav1alpha1.OffsetMapping, error) {
+
+	// Get The New Offset Of Partition For Specified Time
+	newOffset, err := saramaClient.GetOffset(topic, partition, offsetTime)
+	if err != nil {
+		logger.Error("Failed to get Partition Offset for Time", zap.Int64("Time", offsetTime), zap.Error(err))
+		return nil, err
+	}
+
+	// Get The Current Offset Of Partition (Accuracy Depends On ConsumerGroup Having Been Stopped)
+	currentOffset, _ := partitionOffsetManager.NextOffset()
+
+	// Update The Partition's Offset Forward/Back As Needed
+	offsetMetaData := formatOffsetMetaData(offsetTime)
+	if newOffset > currentOffset {
+		partitionOffsetManager.MarkOffset(newOffset, offsetMetaData) // No Errors Returned - On PartitionOffsetManager.Errors() Channel Instead
+	} else if newOffset < currentOffset {
+		partitionOffsetManager.ResetOffset(newOffset, offsetMetaData) // No Errors Returned - On PartitionOffsetManager.Errors() Channel Instead
+	}
+
+	// Create An OffsetMapping For The Partition
+	offsetMapping := &kafkav1alpha1.OffsetMapping{
+		Partition: partition,
+		OldOffset: currentOffset,
+		NewOffset: newOffset,
+	}
+
+	// Return The OffsetMapping Success
+	return offsetMapping, nil
+}
+
+// formatOffsetMetaData returns a "metadata" string, suitable for use with MarkOffset/ResetOffset, for the specified time.
+func formatOffsetMetaData(time int64) string {
+	return fmt.Sprintf("resetoffset.%d", time)
+}
+
+// safeCloseSaramaClient will attempt to close the specified Sarama Client
+func safeCloseSaramaClient(logger *zap.Logger, client sarama.Client) {
+	if client != nil && !client.Closed() {
+		err := client.Close()
+		if err != nil {
+			logger.Warn("Failed to close Sarama Client", zap.Error(err))
+		}
+	}
+}
+
+// createPartitionOffsetManagers initializes the PartitionOffsetManagers for the specified topic / partitions.
+func createPartitionOffsetManagers(offsetManager sarama.OffsetManager, topicName string, partitions []int32) (PartitionOffsetManagers, error) {
+	partitionOffsetManagers := make(PartitionOffsetManagers, len(partitions))
+	for _, partition := range partitions {
+		partitionOffsetManager, err := offsetManager.ManagePartition(topicName, partition)
+		partitionOffsetManagers[partition] = partitionOffsetManager
+		if err != nil {
+			return partitionOffsetManagers, err
+		}
+	}
+	return partitionOffsetManagers, nil
+}
+
+// closePartitionOffsetManagers performs an AsyncClose on the PartitionOffsetManagers
+func closePartitionOffsetManagers(partitionOffsetManagers PartitionOffsetManagers) {
+	for _, partitionOffsetManager := range partitionOffsetManagers {
+		if partitionOffsetManager != nil {
+			partitionOffsetManager.AsyncClose() // Fast - No Errors Returned - Works Without AutoCommit ; )
+		}
+	}
+}
+
+// closeManagersAndDrainErrors handles the tear-down of the Sarama OffsetManager and
+// PartitionOffsetManagers in the necessary sequence, and returns any errors that have
+// accumulated.  This function should be called AFTER the OffsetManager.Commit() operation
+// has been performed, and the errors could be related to prior MarkOffset / ResetOffset
+// / Commit operations.  These Sarama "managers" are intertwined and Sarama is very
+// proscriptive about the order in which they should be closed and drained.
+func closeManagersAndDrainErrors(logger *zap.Logger, offsetManager sarama.OffsetManager, partitionOffsetManagers PartitionOffsetManagers) error {
+
+	// Close The PartitionOffsetManagers (Must Be Called Before Closing OffsetManager)
+	closePartitionOffsetManagers(partitionOffsetManagers)
+
+	// Close The OffsetManager (Must Be Called After Closing PartitionOffsetManagers)
+	if offsetManager != nil {
+		closeErr := offsetManager.Close()
+		if closeErr != nil {
+			logger.Error("Failed to close Sarama OffsetManager", zap.Error(closeErr)) // Doesn't appear to be possible but log it and continue
+		}
+	}
+
+	// Drain The PartitionOffsetManagers Error Channels (Must Be Called After Close)
+	pomErr := drainPartitionOffsetManagerErrors(partitionOffsetManagers)
+	if pomErr != nil {
+		logger.Error("Errors encountered during Offset update", zap.Errors("Sarama PartitionOffsetManager Errors", multierr.Errors(pomErr)))
+	}
+
+	// Return Any PartitionOffsetManager Errors
+	return pomErr
+}
+
+// drainPartitionOffsetManagerErrors drains the PartitionOffsetManager's Error channels and
+// returns any ConsumerErrors as a Zap multierr, and must be called after Commit() / Close().
+// This async error channel not ideal but is simply the way the Sarama library operates.
+func drainPartitionOffsetManagerErrors(partitionOffsetManagers PartitionOffsetManagers) error {
+	var multiErr error
+	for _, partitionOffsetManager := range partitionOffsetManagers {
+		if partitionOffsetManager != nil {
+			select {
+			case consumerErr, ok := <-partitionOffsetManager.Errors():
+				if consumerErr != nil {
+					if multiErr == nil {
+						multiErr = consumerErr.Unwrap()
+					} else {
+						multierr.AppendInto(&multiErr, consumerErr.Unwrap())
+					}
+				}
+				if !ok {
+					break // Error Channel Closed - Stop Draining
+				}
+			case <-time.After(5 * time.Second):
+				break // Error Channel Drain Timeout - Stop Waiting For Channel Close
+			}
+		}
+	}
+	return multiErr
+}
diff --git a/pkg/common/commands/resetoffset/controller/reconciler.go b/pkg/common/commands/resetoffset/controller/reconciler.go
new file mode 100644
index 000000000..1d1cdbc6d
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/reconciler.go
@@ -0,0 +1,185 @@
+/*
+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 controller
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/Shopify/sarama"
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/reconciler"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset"
+	kafkalisters "knative.dev/eventing-kafka/pkg/client/listers/kafka/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
+	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
+)
+
+// Component For Sarama Config
+const Component = "reset-offset-controller"
+
+var (
+	_ resetoffset.Interface = (*Reconciler)(nil) // Verify Reconciler Implements Interface
+	_ resetoffset.Finalizer = (*Reconciler)(nil) // Verify Reconciler Implements Finalizer
+)
+
+// Reconciler Implements controller.Reconciler for ResetOffset Resources
+type Reconciler struct {
+	kafkaBrokers      []string
+	saramaConfig      *sarama.Config
+	resetoffsetLister kafkalisters.ResetOffsetLister
+	refMapper         refmappers.ResetOffsetRefMapper
+}
+
+// ReconcileKind implements the Reconciler Interface and is responsible for performing Offset repositioning.
+func (r *Reconciler) ReconcileKind(ctx context.Context, resetOffset *kafkav1alpha1.ResetOffset) reconciler.Event {
+
+	// Get The Logger From Context
+	logger := logging.FromContext(ctx).Desugar()
+	logger.Debug("<==========  START RESET-OFFSET RECONCILIATION  ==========>")
+
+	// Ignore Previously Initiated ResetOffset Instances
+	if resetOffset.Status.IsInitiated() || resetOffset.Status.IsCompleted() {
+		logger.Debug("Skipping reconciliation of previously executed ResetOffset instance")
+		return reconciler.NewEvent(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously executed ResetOffset")
+	}
+
+	// Reset The ResetOffset's Status Conditions To Unknown
+	resetOffset.Status.InitializeConditions()
+
+	// Map The ResetOffset's Ref To Kafka Topic Name / ConsumerGroup ID
+	topic, group, err := r.refMapper.MapRef(resetOffset)
+	if err != nil {
+		logger.Error("Failed to map ResetOffset.Spec.Ref to Kafka Topic name and ConsumerGroup ID", zap.Error(err))
+		resetOffset.Status.MarkRefMappedFailed("FailedToMapRef", "Failed to map 'ref' to Kafka Topic and Group: %v", err)
+		return fmt.Errorf("failed to map 'ref' to Kafka Topic and Group: %v", err)
+	}
+	logger.Info("Successfully mapped ResetOffset.Spec.Ref to Kafka Topic name and ConsumerGroup ID", zap.String("Topic", topic), zap.String("Group", group))
+	resetOffset.Status.SetTopic(topic)
+	resetOffset.Status.SetGroup(group)
+	resetOffset.Status.MarkRefMappedTrue()
+
+	// Parse The Sarama Offset Time From ResetOffset Spec
+	offsetTime, err := resetOffset.Spec.ParseSaramaOffsetTime()
+	if err != nil {
+		logger.Error("Failed to parse Sarama Offset Time from ResetOffset Spec", zap.Error(err))
+		return err // Should never happen assuming Validation is in place
+	}
+	logger.Info("Successfully parsed Sarama Offset time from ResetOffset Spec", zap.Int64("Time (millis)", offsetTime))
+
+	// TODO - Map ResetOffset.Spec.Ref to "owning" Dispatcher Deployment/Service (Consolidated = one, Distributed = many)
+
+	// TODO - Send "Initiate" Message to Dispatcher Replicas to lock mutext and start process.
+	resetOffset.Status.MarkResetInitiatedTrue()
+
+	// TODO - Send "Stop" Message to Dispatcher Replicas to stop ConsumerGroups
+	resetOffset.Status.MarkConsumerGroupsStoppedTrue()
+
+	// Update The Sarama Offsets & Update ResetOffset CRD With OffsetMappings
+	offsetMappings, err := r.reconcileOffsets(ctx, topic, group, offsetTime)
+	if err != nil {
+		logger.Error("Failed to update Offsets of one or more partitions", zap.Error(err))
+		resetOffset.Status.MarkOffsetsUpdatedFailed("FailedToUpdateOffsets", "Failed to update offsets of one or more partitions: %v", err)
+		return fmt.Errorf("failed to update offsets of one or more partitions: %v", err)
+	}
+	if offsetMappings != nil {
+		resetOffset.Status.SetPartitions(offsetMappings)
+	}
+	logger.Info("Successfully updated Offsets of all partitions")
+	resetOffset.Status.MarkOffsetsUpdatedTrue()
+
+	// TODO - Send "Start" Message to Dispatcher Replicas to restart ConsumerGroups
+	resetOffset.Status.MarkConsumerGroupsStartedTrue()
+
+	// Return Reconciled Success Event
+	return reconciler.NewEvent(corev1.EventTypeNormal, ResetOffsetReconciled.String(), "Reconciled successfully")
+}
+
+// FinalizeKind implements the Finalizer Interface and is responsible for performing any necessary cleanup.
+func (r *Reconciler) FinalizeKind(ctx context.Context, resetOffset *kafkav1alpha1.ResetOffset) reconciler.Event {
+
+	// Get The Logger From Context
+	logger := logging.FromContext(ctx)
+	logger.Debug("<==========  START RESET-OFFSET FINALIZATION  ==========>")
+
+	// Do NOT Finalize ResetOffset Instances That Are "Executing"
+	if resetOffset.Status.IsInitiated() && !resetOffset.Status.IsCompleted() {
+		logger.Debug("Skipping finalization of in-progress ResetOffset instance")
+		return fmt.Errorf("skipping finalization of in-progress ResetOffset instance")
+	}
+
+	// No-Op Finalization - Nothing To Do
+	logger.Info("No-Op Finalization Successful")
+
+	// Return Finalized Success Event
+	return reconciler.NewEvent(corev1.EventTypeNormal, ResetOffsetFinalized.String(), "Finalized successfully")
+}
+
+// updateKafkaConfig is the callback function that handles changes to the ConfigMap
+func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.ConfigMap) {
+
+	// Get The Logger From Context
+	logger := logging.FromContext(ctx).Desugar()
+
+	// Validate Reconciler Reference (Sometimes nil on startup and can be ignored)
+	if r == nil {
+		return
+	}
+
+	// Validate ConfigMap Reference
+	if configMap == nil {
+		logger.Warn("Ignoring nil ConfigMap")
+		return
+	}
+
+	// Enhance Logger With ConfigMap
+	logger = logger.With(zap.String("ConfigMap", fmt.Sprintf("%s/%s", configMap.Namespace, configMap.Name)))
+
+	// Validate ConfigMap.Data
+	if configMap.Data == nil {
+		logger.Warn("Ignoring nil ConfigMap.Data")
+		return
+	}
+
+	// Load The EventingKafkaConfig From ConfigMap.Data
+	ekConfig, err := kafkasarama.LoadSettings(ctx, Component, configMap.Data, kafkasarama.LoadAuthConfig)
+	if err != nil || ekConfig == nil {
+		logger.Error("Failed to extract EventingKafkaConfig from ConfigMap", zap.Any("ConfigMap", configMap), zap.Error(err))
+		return
+	}
+
+	// Update The KafkaBrokers On Reconciler
+	r.kafkaBrokers = strings.Split(ekConfig.Kafka.Brokers, ",")
+
+	// Enable/Disable Sarama Logging Based On ConfigMap Setting
+	kafkasarama.EnableSaramaLogging(ekConfig.Sarama.EnableLogging)
+	logger.Debug("Set Sarama logging", zap.Bool("Enabled", ekConfig.Sarama.EnableLogging))
+
+	// Force Enable Consumer Error Handling
+	ekConfig.Sarama.Config.Consumer.Return.Errors = true
+
+	// Force Disable Manual Commits
+	ekConfig.Sarama.Config.Consumer.Offsets.AutoCommit.Enable = false
+
+	// Update Reconciler With New Config
+	r.saramaConfig = ekConfig.Sarama.Config
+}
diff --git a/pkg/common/commands/resetoffset/refmappers/subscription.go b/pkg/common/commands/resetoffset/refmappers/subscription.go
new file mode 100644
index 000000000..87162e48e
--- /dev/null
+++ b/pkg/common/commands/resetoffset/refmappers/subscription.go
@@ -0,0 +1,157 @@
+/*
+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 refmappers
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+	"knative.dev/eventing/pkg/apis/messaging"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	subscriptioninformers "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
+	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
+	"knative.dev/pkg/logging"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+//
+// SubscriptionRefMapperFactory
+//
+
+// Verify The Subscription ResetOffsetRefMapperFactory Implements The Interface
+var _ ResetOffsetRefMapperFactory = &SubscriptionRefMapperFactory{}
+
+// SubscriptionRefMapperFactory implements the ResetOffsetRefMapperFactory for Knative Subscriptions
+type SubscriptionRefMapperFactory struct {
+	TopicNameMapper SubscriptionTopicNameMapper
+	GroupIdMapper   SubscriptionConsumerGroupIdMapper
+}
+
+// NewSubscriptionRefMapperFactory returns an initialized SubscriptionRefMapperFactory
+func NewSubscriptionRefMapperFactory(topicNameMapper SubscriptionTopicNameMapper, groupIdMapper SubscriptionConsumerGroupIdMapper) *SubscriptionRefMapperFactory {
+	return &SubscriptionRefMapperFactory{TopicNameMapper: topicNameMapper, GroupIdMapper: groupIdMapper}
+}
+
+// Create implements the ResetOffsetRefMapperFactory interface for Subscription references.  It will return
+// a new SubscriptionRefMapper instance using the specific TopicName / GroupId mappers.  It also relies on
+// the Context having injected informers (SubscriptionInformer).
+func (f *SubscriptionRefMapperFactory) Create(ctx context.Context) ResetOffsetRefMapper {
+	return NewSubscriptionRefMapper(ctx, f.TopicNameMapper, f.GroupIdMapper)
+}
+
+//
+// SubscriptionRefMapper
+//
+
+// SubscriptionTopicNameMapper defines a function signature for mapping a Subscription to a Kafka Topic name.
+type SubscriptionTopicNameMapper func(*messagingv1.Subscription) (string, error)
+
+// SubscriptionConsumerGroupIdMapper defines a function signature for mapping a Subscription to a Kafka ConsumerGroup ID.
+type SubscriptionConsumerGroupIdMapper func(*messagingv1.Subscription) (string, error)
+
+// Verify The Subscription ResetOffsetRefMapper Implements The Interface
+var _ ResetOffsetRefMapper = &SubscriptionRefMapper{}
+
+// SubscriptionRefMapper implements the ResetOffsetRefMapper for Knative Subscriptions
+type SubscriptionRefMapper struct {
+	logger             *zap.Logger
+	subscriptionLister messaginglisters.SubscriptionLister
+	topicNameMapper    SubscriptionTopicNameMapper
+	groupIdMapper      SubscriptionConsumerGroupIdMapper
+}
+
+// NewSubscriptionRefMapper returns an initialized ResetOffsetSubscriptionRefMapper
+func NewSubscriptionRefMapper(ctx context.Context,
+	topicNameMapper SubscriptionTopicNameMapper,
+	groupIdMapper SubscriptionConsumerGroupIdMapper) *SubscriptionRefMapper {
+
+	// Get The Logger From Context
+	logger := logging.FromContext(ctx).Desugar()
+
+	// Get The Subscription Informer From Context (Context Must Have Injected Informers From SharedMain())
+	subscriptionInformer := subscriptioninformers.Get(ctx)
+
+	// Return An Initialized ResetOffsetSubscriptionRefMapper
+	return &SubscriptionRefMapper{
+		logger:             logger,
+		subscriptionLister: subscriptionInformer.Lister(),
+		topicNameMapper:    topicNameMapper,
+		groupIdMapper:      groupIdMapper,
+	}
+}
+
+// MapRef implements the ResetOffsetRefMapper interface for Subscription references. It will return an
+// error in all cases other than successfully mapping the ResetOffset.Spec.Ref to a Kafka Topic / Group.
+func (m *SubscriptionRefMapper) MapRef(resetOffset *kafkav1alpha1.ResetOffset) (string, string, error) {
+
+	// Validate The ResetOffset
+	if resetOffset == nil {
+		m.logger.Warn("Received nil ResetOffset argument")
+		return "", "", fmt.Errorf("unable to map nil ResetOffset")
+	}
+
+	// Get The ResetOffset Ref From Spec & Enhance Logger
+	ref := resetOffset.Spec.Ref
+	logger := m.logger.With(zap.Any("Ref", ref))
+
+	// Validate The Reference
+	if !strings.HasPrefix(ref.APIVersion, messaging.GroupName) || ref.Kind != "Subscription" {
+		m.logger.Warn("Received ResetOffset with non Subscription reference")
+		return "", "", fmt.Errorf("received ResetOffset with non Subscription reference: %v", ref)
+	}
+	if ref.Name == "" {
+		m.logger.Warn("Received ResetOffset with unnamed Subscription reference")
+		return "", "", fmt.Errorf("received ResetOffset with unnamed Subscription reference: %v", ref)
+	}
+
+	// Default Optional Ref.Namespace If Not Provided
+	refNamespace := ref.Namespace
+	if refNamespace == "" {
+		refNamespace = resetOffset.Namespace
+	}
+
+	// Attempt To Get The Specified Subscription
+	subscription, err := m.subscriptionLister.Subscriptions(refNamespace).Get(ref.Name)
+	if err != nil {
+		logger.Error("Failed to get Subscription referenced by ResetOffset", zap.Error(err))
+		return "", "", fmt.Errorf("failed to get Subscription referenced by ResetOffset.Spec.Ref '%v': %v", ref, err)
+	}
+	if subscription == nil {
+		logger.Info("No Subscription found for ResetOffset reference")
+		return "", "", fmt.Errorf("no Subscription found for ResetOffset.Spec.Ref %v", ref)
+	}
+
+	// Map The Subscription To Kafka Topic Name Via Custom SubscriptionTopicNameMapper
+	topicName, err := m.topicNameMapper(subscription)
+	if err != nil {
+		logger.Error("Failed to map Knative Subscription to Kafka Topic name", zap.Error(err))
+		return "", "", fmt.Errorf("failed to map Knative Subscription '%v' to Kafka Topic name: %v", ref, err)
+	}
+
+	// Map The Subscription To Kafka ConsumerGroup ID Via Custom SubscriptionGroupIdMapper
+	groupId, err := m.groupIdMapper(subscription)
+	if err != nil {
+		logger.Error("Failed to map Knative Subscription to Kafka ConsumerGroup ID", zap.Error(err))
+		return "", "", fmt.Errorf("failed to map Knative Subscription '%v' to Kafka ConsumerGroup ID: %v", ref, err)
+	}
+
+	// Successfully Mapped The Ref To Kafka Topic / Group - Return Results
+	return topicName, groupId, nil
+}
diff --git a/pkg/common/commands/resetoffset/refmappers/types.go b/pkg/common/commands/resetoffset/refmappers/types.go
new file mode 100644
index 000000000..8d8007c1e
--- /dev/null
+++ b/pkg/common/commands/resetoffset/refmappers/types.go
@@ -0,0 +1,37 @@
+/*
+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 refmappers
+
+import (
+	"context"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+// ResetOffsetRefMapperFactory defines the interface for creating ResetOffsetRefMapper
+// instances based on the provided Context which comes from SharedMain and includes the
+// injected Informers.  It is necessary for delayed initialization against that Context.
+type ResetOffsetRefMapperFactory interface {
+	Create(ctx context.Context) ResetOffsetRefMapper
+}
+
+// ResetOffsetRefMapper defines the interface for the capability to map ResetOffset.Spec.Ref
+// to Kafka Topic / Group. This abstraction allows for future extensibility to support
+// additional KRef types, such as Triggers, instead of just Subscriptions.
+type ResetOffsetRefMapper interface {
+	MapRef(resetOffset *kafkav1alpha1.ResetOffset) (topic string, group string, err error)
+}
diff --git a/third_party/VENDOR-LICENSE/github.com/stretchr/objx/LICENSE b/third_party/VENDOR-LICENSE/github.com/stretchr/objx/LICENSE
new file mode 100644
index 000000000..44d4d9d5a
--- /dev/null
+++ b/third_party/VENDOR-LICENSE/github.com/stretchr/objx/LICENSE
@@ -0,0 +1,22 @@
+The MIT License
+
+Copyright (c) 2014 Stretchr, Inc.
+Copyright (c) 2017-2018 objx contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/factory/fake/fake.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/factory/fake/fake.go
new file mode 100644
index 000000000..45898a891
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/factory/fake/fake.go
@@ -0,0 +1,45 @@
+/*
+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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	externalversions "knative.dev/eventing/pkg/client/informers/externalversions"
+	fake "knative.dev/eventing/pkg/client/injection/client/fake"
+	factory "knative.dev/eventing/pkg/client/injection/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = factory.Get
+
+func init() {
+	injection.Fake.RegisterInformerFactory(withInformerFactory)
+}
+
+func withInformerFactory(ctx context.Context) context.Context {
+	c := fake.Get(ctx)
+	opts := make([]externalversions.SharedInformerOption, 0, 1)
+	if injection.HasNamespaceScope(ctx) {
+		opts = append(opts, externalversions.WithNamespace(injection.GetNamespaceScope(ctx)))
+	}
+	return context.WithValue(ctx, factory.Key{},
+		externalversions.NewSharedInformerFactoryWithOptions(c, controller.GetResyncPeriod(ctx), opts...))
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
new file mode 100644
index 000000000..eb18a1fac
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	fake "knative.dev/eventing/pkg/client/injection/informers/factory/fake"
+	subscription "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = subscription.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Messaging().V1().Subscriptions()
+	return context.WithValue(ctx, subscription.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
new file mode 100644
index 000000000..af7107671
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
@@ -0,0 +1,52 @@
+/*
+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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package subscription
+
+import (
+	context "context"
+
+	v1 "knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1"
+	factory "knative.dev/eventing/pkg/client/injection/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Messaging().V1().Subscriptions()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.SubscriptionInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1.SubscriptionInformer from context.")
+	}
+	return untyped.(v1.SubscriptionInformer)
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index da4152bd9..e0a9d23c4 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -452,6 +452,7 @@ go.uber.org/automaxprocs/internal/cgroups
 go.uber.org/automaxprocs/internal/runtime
 go.uber.org/automaxprocs/maxprocs
 # go.uber.org/multierr v1.6.0
+## explicit
 go.uber.org/multierr
 # go.uber.org/zap v1.16.0
 ## explicit
@@ -1161,6 +1162,9 @@ knative.dev/eventing/pkg/client/injection/client
 knative.dev/eventing/pkg/client/injection/client/fake
 knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker
 knative.dev/eventing/pkg/client/injection/informers/factory
+knative.dev/eventing/pkg/client/injection/informers/factory/fake
+knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription
+knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker
 knative.dev/eventing/pkg/client/listers/eventing/v1
 knative.dev/eventing/pkg/client/listers/eventing/v1beta1


Test patch:
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go
index 5eeda3c42..a02d9e8e1 100644
--- a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go
@@ -29,6 +29,7 @@ import (
 
 var (
 	condSucceeded             = apis.Condition{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionTrue}
+	condRefMapped             = apis.Condition{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionTrue}
 	condResetInitiated        = apis.Condition{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionTrue}
 	condConsumerGroupsStopped = apis.Condition{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionTrue}
 	condOffsetsUpdated        = apis.Condition{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue}
@@ -59,6 +60,16 @@ func TestResetOffsetStatus_GetCondition(t *testing.T) {
 			conditionType: ResetOffsetConditionSucceeded,
 			wantCondition: &condSucceeded,
 		},
+		{
+			name: "ResetOffsetConditionRefMapped",
+			status: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{condRefMapped},
+				},
+			},
+			conditionType: ResetOffsetConditionRefMapped,
+			wantCondition: &condRefMapped,
+		},
 		{
 			name: "ResetOffsetConditionResetInitiated",
 			status: &ResetOffsetStatus{
@@ -126,6 +137,7 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
 					},
@@ -147,6 +159,7 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionFalse},
 						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
 					},
@@ -168,6 +181,7 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue},
+						{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
 						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
 					},
@@ -189,6 +203,7 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue},
+						{Type: ResetOffsetConditionRefMapped, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionTrue},
 						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionTrue},
 					},
@@ -209,9 +224,106 @@ func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
 	}
 }
 
+func TestResetOffsetStatus_IsInitiated(t *testing.T) {
+
+	tests := []struct {
+		name          string
+		markInitiated corev1.ConditionStatus
+		wantInitiated bool
+	}{
+		{
+			name:          "Initiated True",
+			markInitiated: corev1.ConditionTrue,
+			wantInitiated: true,
+		},
+		{
+			name:          "Initiated False",
+			markInitiated: corev1.ConditionFalse,
+			wantInitiated: false,
+		},
+		{
+			name:          "Initiated Unknown",
+			markInitiated: corev1.ConditionUnknown,
+			wantInitiated: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resetOffsetStatus := &ResetOffsetStatus{}
+			resetOffsetStatus.InitializeConditions()
+
+			switch test.markInitiated {
+			case corev1.ConditionTrue:
+				resetOffsetStatus.MarkResetInitiatedTrue()
+			case corev1.ConditionFalse:
+				resetOffsetStatus.MarkResetInitiatedFailed("TestingOffsetsResetInitiatedStatus", "TestMessage")
+			case corev1.ConditionUnknown:
+				for _, condition := range resetOffsetStatus.Conditions {
+					assert.True(t, condition.IsUnknown()) // Verify Initiated Is "Unknown" From InitializeConditions()
+				}
+			}
+			assert.Equal(t, test.wantInitiated, resetOffsetStatus.IsInitiated())
+		})
+	}
+}
+
 func TestResetOffsetStatus_IsCompleted(t *testing.T) {
+
+	tests := []struct {
+		name          string
+		markSucceeded corev1.ConditionStatus
+		wantCompleted bool
+	}{
+		{
+			name:          "Succeeded True",
+			markSucceeded: corev1.ConditionTrue,
+			wantCompleted: true,
+		},
+		{
+			name:          "Succeeded False",
+			markSucceeded: corev1.ConditionFalse,
+			wantCompleted: true,
+		},
+		{
+			name:          "Succeeded Unknown",
+			markSucceeded: corev1.ConditionUnknown,
+			wantCompleted: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resetOffsetStatus := &ResetOffsetStatus{}
+			resetOffsetStatus.InitializeConditions()
+
+			switch test.markSucceeded {
+			case corev1.ConditionTrue:
+				resetOffsetStatus.MarkRefMappedTrue()
+				resetOffsetStatus.MarkResetInitiatedTrue()
+				resetOffsetStatus.MarkConsumerGroupsStoppedTrue()
+				resetOffsetStatus.MarkOffsetsUpdatedTrue()
+				resetOffsetStatus.MarkConsumerGroupsStartedTrue()
+			case corev1.ConditionFalse:
+				resetOffsetStatus.MarkRefMappedFailed("TestingOffsetsRefMappedStatus", "TestMessage")
+				resetOffsetStatus.MarkResetInitiatedFailed("TestingOffsetsResetInitiatedStatus", "TestMessage")
+				resetOffsetStatus.MarkConsumerGroupsStoppedFailed("TestingConsumerGroupsStoppedStatus", "TestMessage")
+				resetOffsetStatus.MarkOffsetsUpdatedFailed("TestingOffsetsUpdatedFailedStatus", "TestMessage")
+				resetOffsetStatus.MarkConsumerGroupsStartedFailed("TestingConsumerGroupsStartedStatus", "TestMessage")
+			case corev1.ConditionUnknown:
+				for _, condition := range resetOffsetStatus.Conditions {
+					assert.True(t, condition.IsUnknown()) // Verify Succeeded Is "Unknown" From InitializeConditions()
+				}
+			}
+			assert.Equal(t, test.wantCompleted, resetOffsetStatus.IsCompleted())
+		})
+	}
+}
+
+func TestResetOffsetStatus_IsSucceeded(t *testing.T) {
 	tests := []struct {
 		name                      string
+		markRefMapped             bool
 		markResetInitiated        bool
 		markConsumerGroupsStopped bool
 		markOffsetsUpdated        bool
@@ -220,14 +332,25 @@ func TestResetOffsetStatus_IsCompleted(t *testing.T) {
 	}{
 		{
 			name:                      "Happy",
+			markRefMapped:             true,
 			markResetInitiated:        true,
 			markConsumerGroupsStopped: true,
 			markOffsetsUpdated:        true,
 			markConsumerGroupsStarted: true,
 			wantSucceeded:             true,
 		},
+		{
+			name:                      "RefMapped Failed",
+			markRefMapped:             false,
+			markResetInitiated:        true,
+			markConsumerGroupsStopped: true,
+			markOffsetsUpdated:        true,
+			markConsumerGroupsStarted: true,
+			wantSucceeded:             false,
+		},
 		{
 			name:                      "ResetInitiated Failed",
+			markRefMapped:             true,
 			markResetInitiated:        false,
 			markConsumerGroupsStopped: true,
 			markOffsetsUpdated:        true,
@@ -236,6 +359,7 @@ func TestResetOffsetStatus_IsCompleted(t *testing.T) {
 		},
 		{
 			name:                      "ConsumerGroupsStopped Failed",
+			markRefMapped:             true,
 			markResetInitiated:        true,
 			markConsumerGroupsStopped: false,
 			markOffsetsUpdated:        true,
@@ -244,6 +368,7 @@ func TestResetOffsetStatus_IsCompleted(t *testing.T) {
 		},
 		{
 			name:                      "OffsetsUpdated Failed",
+			markRefMapped:             true,
 			markResetInitiated:        true,
 			markConsumerGroupsStopped: true,
 			markOffsetsUpdated:        false,
@@ -252,6 +377,7 @@ func TestResetOffsetStatus_IsCompleted(t *testing.T) {
 		},
 		{
 			name:                      "ConsumerGroupsStarted Failed",
+			markRefMapped:             true,
 			markResetInitiated:        true,
 			markConsumerGroupsStopped: true,
 			markOffsetsUpdated:        true,
@@ -264,6 +390,11 @@ func TestResetOffsetStatus_IsCompleted(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			resetOffsetStatus := &ResetOffsetStatus{}
 			resetOffsetStatus.InitializeConditions()
+			if test.markRefMapped {
+				resetOffsetStatus.MarkRefMappedTrue()
+			} else {
+				resetOffsetStatus.MarkRefMappedFailed("TestingOffsetsRefMappedStatus", "TestMessage")
+			}
 			if test.markResetInitiated {
 				resetOffsetStatus.MarkResetInitiatedTrue()
 			} else {
@@ -284,10 +415,7 @@ func TestResetOffsetStatus_IsCompleted(t *testing.T) {
 			} else {
 				resetOffsetStatus.MarkConsumerGroupsStartedFailed("TestingConsumerGroupsStartedStatus", "TestMessage")
 			}
-			gotSucceeded := resetOffsetStatus.IsCompleted()
-			if test.wantSucceeded != gotSucceeded {
-				t.Errorf("unexpected succeeded status: want %v, got %v", test.wantSucceeded, gotSucceeded)
-			}
+			assert.Equal(t, test.wantSucceeded, resetOffsetStatus.IsSucceeded())
 		})
 	}
 }
diff --git a/pkg/channel/distributed/common/kafka/util/util_test.go b/pkg/channel/distributed/common/kafka/util/util_test.go
index 6223a7678..7d3fa0449 100644
--- a/pkg/channel/distributed/common/kafka/util/util_test.go
+++ b/pkg/channel/distributed/common/kafka/util/util_test.go
@@ -39,6 +39,20 @@ func TestTopicName(t *testing.T) {
 	assert.Equal(t, expectedTopicName, actualTopicName)
 }
 
+// Test The GroupId() Functionality
+func TestGroupId(t *testing.T) {
+
+	// Test Data
+	uid := "TestUID"
+
+	// Perform The Test
+	actualGroupId := GroupId(uid)
+
+	// Verify The Results
+	expectedGroupId := "kafka." + uid
+	assert.Equal(t, expectedGroupId, actualGroupId)
+}
+
 // Test The AppendChannelServiceNameSuffix() Functionality
 func TestAppendChannelServiceNameSuffix(t *testing.T) {
 
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
index c624efce8..2caff21da 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
@@ -24,44 +24,37 @@ import (
 	"sync"
 	"testing"
 
-	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
-
-	"github.com/Shopify/sarama"
-	"k8s.io/client-go/kubernetes/fake"
-	"knative.dev/eventing-kafka/pkg/common/constants"
-	"knative.dev/pkg/system"
-
-	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
-
-	"knative.dev/eventing-kafka/pkg/common/client"
-
+	"github.com/stretchr/testify/assert"
 	appsv1 "k8s.io/api/apps/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	kafkautil "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/util"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/util"
-
 	corev1 "k8s.io/api/core/v1"
-	apitypes "k8s.io/apimachinery/pkg/types"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
-	"knative.dev/pkg/apis/duck"
-
-	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	apitypes "k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	clientgotesting "k8s.io/client-go/testing"
+	"knative.dev/pkg/apis/duck"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	logtesting "knative.dev/pkg/logging/testing"
+	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/system"
+
 	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkaadmintesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin/testing"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/admin/types"
+	kafkautil "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/util"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
 	controllertesting "knative.dev/eventing-kafka/pkg/channel/distributed/controller/testing"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/util"
 	fakekafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
+	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+	kafkasarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/configmap"
-	"knative.dev/pkg/controller"
-	logtesting "knative.dev/pkg/logging/testing"
-	. "knative.dev/pkg/reconciler/testing"
 )
 
 // Initialization - Add types to scheme
@@ -1087,15 +1080,12 @@ func TestReconciler_updateKafkaConfig(t *testing.T) {
   authSecretNamespace: `+system.Namespace(), 1)
 
 	tests := []struct {
-		name         string
-		config       *commonconfig.EventingKafkaConfig
-		saramaConfig *sarama.Config
-		authConfig   *client.KafkaAuthConfig
-		configMap    *corev1.ConfigMap
-		hash         string
-		user         string
-		secretErr    bool
-		expectErr    string
+		name      string
+		configMap *corev1.ConfigMap
+		hash      string
+		user      string
+		secretErr bool
+		expectErr string
 	}{
 		{
 			name:      "Nil ConfigMap",
@@ -1171,7 +1161,6 @@ func TestReconciler_updateKafkaConfig(t *testing.T) {
 			}
 			r := &Reconciler{
 				kubeClientset:      fakeK8sClient,
-				config:             tt.config,
 				kafkaConfigMapHash: tt.hash,
 			}
 			err := r.updateKafkaConfig(ctx, tt.configMap)
@@ -1179,13 +1168,13 @@ func TestReconciler_updateKafkaConfig(t *testing.T) {
 				assert.Nil(t, err)
 				// If no error was expected, verify that the settings in the reconciler were changed to new values
 				assert.NotEqual(t, tt.hash, r.kafkaConfigMapHash)
-				assert.NotEqual(t, tt.config, r.config)
+				assert.NotNil(t, r.config)
 			} else {
 				assert.NotNil(t, err)
 				assert.Regexp(t, tt.expectErr, err.Error())
 				// If an error occurred, verify that the settings in the reconciler do not change from what they were
 				assert.Equal(t, tt.hash, r.kafkaConfigMapHash)
-				assert.Equal(t, tt.config, r.config)
+				assert.Nil(t, r.config)
 			}
 		})
 	}
diff --git a/pkg/channel/distributed/controller/util/channel_test.go b/pkg/channel/distributed/controller/util/channel_test.go
index b6b536cd6..6b1908a67 100644
--- a/pkg/channel/distributed/controller/util/channel_test.go
+++ b/pkg/channel/distributed/controller/util/channel_test.go
@@ -32,8 +32,8 @@ import (
 // Test Data
 const (
 	testPrefix       = "testkafkaprefix"
-	channelName      = "testname"
-	channelNamespace = "testnamespace"
+	channelName      = "test-channel-name"
+	channelNamespace = "test-channel-namespace"
 )
 
 // Test The ChannelLogger() Functionality
diff --git a/pkg/channel/distributed/controller/util/subscription_test.go b/pkg/channel/distributed/controller/util/subscription_test.go
index 56d67a319..8bb15016e 100644
--- a/pkg/channel/distributed/controller/util/subscription_test.go
+++ b/pkg/channel/distributed/controller/util/subscription_test.go
@@ -17,13 +17,24 @@ limitations under the License.
 package util
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	logtesting "knative.dev/pkg/logging/testing"
+
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
+)
+
+// Test Data
+const (
+	subscriptionName      = "test-subscription-name"
+	subscriptionNamespace = "test-subscription-namespace"
 )
 
 // Test The SubscriptionLogger Functionality
@@ -62,3 +73,123 @@ func TestNewSubscriptionControllerRef(t *testing.T) {
 	assert.Equal(t, subscription.ObjectMeta.Name, controllerRef.Name)
 	assert.True(t, *controllerRef.Controller)
 }
+
+// Test The TopicNameMapper Functionality
+func TestTopicNameMapper(t *testing.T) {
+
+	// Test Data
+	kafkaChannelGroupVersion := schema.GroupVersion{
+		Group:   messagingv1.SchemeGroupVersion.Group,
+		Version: messagingv1.SchemeGroupVersion.Version,
+	}
+
+	// Define The TestCases
+	tests := []struct {
+		name         string
+		subscription *messagingv1.Subscription
+		expected     string
+		err          bool
+	}{
+		{
+			name: "fully populated subscription",
+			subscription: &messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      subscriptionName,
+					Namespace: subscriptionNamespace,
+				},
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						Kind:       constants.KafkaChannelKind,
+						Namespace:  channelNamespace,
+						Name:       channelName,
+						APIVersion: kafkaChannelGroupVersion.String(),
+					},
+				},
+			},
+			expected: fmt.Sprintf("%s.%s", channelNamespace, channelName),
+			err:      false,
+		},
+		{
+			name: "sparsely populated subscription",
+			subscription: &messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      subscriptionName,
+					Namespace: subscriptionNamespace,
+				},
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						Kind:       constants.KafkaChannelKind,
+						Name:       channelName,
+						APIVersion: kafkaChannelGroupVersion.String(),
+					},
+				},
+			},
+			expected: fmt.Sprintf("%s.%s", subscriptionNamespace, channelName),
+			err:      false,
+		},
+		{
+			name:         "nil subscription",
+			subscription: nil,
+			expected:     "",
+			err:          true,
+		},
+	}
+
+	// Execute The Tests
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Perform The Test
+			actual, err := TopicNameMapper(test.subscription)
+
+			// Verify Results
+			assert.Equal(t, test.err, err != nil)
+			assert.Equal(t, test.expected, actual)
+		})
+	}
+}
+
+// Test The GroupIdMapper Functionality
+func TestGroupIdMapper(t *testing.T) {
+
+	// Test Data
+	subscriptionUID := "TestSubscriptionUID"
+
+	// Define The TestCases
+	tests := []struct {
+		name         string
+		subscription *messagingv1.Subscription
+		expected     string
+		err          bool
+	}{
+		{
+			name: "valid subscription",
+			subscription: &messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					UID: types.UID(subscriptionUID),
+				},
+			},
+			expected: fmt.Sprintf("kafka.%s", subscriptionUID),
+			err:      false,
+		},
+		{
+			name:         "nil subscription",
+			subscription: nil,
+			expected:     "",
+			err:          true,
+		},
+	}
+
+	// Execute The Tests
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Perform The Test
+			actual, err := GroupIdMapper(test.subscription)
+
+			// Verify Results
+			assert.Equal(t, test.err, err != nil)
+			assert.Equal(t, test.expected, actual)
+		})
+	}
+}
diff --git a/pkg/common/commands/resetoffset/controller/controller_test.go b/pkg/common/commands/resetoffset/controller/controller_test.go
new file mode 100644
index 000000000..8f10aa0fc
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/controller_test.go
@@ -0,0 +1,91 @@
+/*
+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 controller
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/client-go/rest"
+	"knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/injection"
+	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/pkg/logging"
+	logtesting "knative.dev/pkg/logging/testing"
+
+	fakeKafkaClient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/fake" // Force Fake Informer Injection
+	refmapperstesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers/testing"
+	configtesting "knative.dev/eventing-kafka/pkg/common/config/testing"
+	"knative.dev/eventing-kafka/pkg/common/configmaploader"
+	fakeConfigmapLoader "knative.dev/eventing-kafka/pkg/common/configmaploader/fake"
+	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
+	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
+)
+
+// Test The NewControllerFactory() Functionality
+func TestNewControllerFactory(t *testing.T) {
+
+	// Create A Context With Test Logger
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.Background(), logger)
+
+	// Register Fake Informers (See Injection "_" Imports Above!)
+	ctx, fakeInformers := injection.Fake.SetupInformers(ctx, &rest.Config{})
+	assert.NotNil(t, fakeInformers)
+
+	// Add The Fake K8S Clientset To The Context (Populated With ConfigMap)
+	configMap := commontesting.GetTestSaramaConfigMap(commonconstants.CurrentConfigVersion, commontesting.OldSaramaConfig, commontesting.TestEKConfig)
+	secret := configtesting.NewKafkaSecret()
+	ctx, fakeClientset := fake.With(ctx, configMap, secret)
+	assert.NotNil(t, fakeClientset)
+
+	// Add The Fake ConfigMap Loader To The Context
+	configmapLoader := fakeConfigmapLoader.NewFakeConfigmapLoader()
+	configmapLoader.Register(commonconstants.SettingsConfigMapMountPath, configMap.Data)
+	ctx = context.WithValue(ctx, configmaploader.Key{}, configmapLoader.Load)
+
+	// Add The Fake Kafka Clientset To The Context (Empty)
+	ctx, fakeKafkaClientset := fakeKafkaClient.With(ctx)
+	assert.NotNil(t, fakeKafkaClientset)
+
+	// Create A Watcher On The Configuration Settings ConfigMap & Dynamically Update Configuration
+	cmw := sharedmain.SetupConfigMapWatchOrDie(ctx, logger)
+
+	// Create Mock ResetOffset Ref Mapper For Testing
+	mockResetOffsetRefMapper := &refmapperstesting.MockResetOffsetRefMapper{}
+	mockResetOffsetRefMapperFactory := &refmapperstesting.MockResetOffsetRefMapperFactory{}
+	mockResetOffsetRefMapperFactory.On("Create", ctx).Return(mockResetOffsetRefMapper)
+
+	// Verify The ResetOffset ControllerFactory Creates A ControllerConstructor
+	controllerConstructor := NewControllerFactory(mockResetOffsetRefMapperFactory)
+	assert.NotNil(t, controllerConstructor)
+
+	// Verify The ResetOffset ControllerConstructor
+	controllerImpl := controllerConstructor(ctx, cmw)
+	assert.NotNil(t, controllerImpl)
+	assert.True(t, len(controllerImpl.Name) > 0)
+	assert.NotNil(t, controllerImpl.Reconciler)
+	mockResetOffsetRefMapperFactory.AssertExpectations(t)
+	mockResetOffsetRefMapper.AssertExpectations(t)
+}
+
+// Test The Shutdown() Functionality
+func TestShutdown(t *testing.T) {
+	Shutdown() // Currently nothing to test
+}
diff --git a/pkg/common/commands/resetoffset/controller/event_test.go b/pkg/common/commands/resetoffset/controller/event_test.go
new file mode 100644
index 000000000..6964bce09
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/event_test.go
@@ -0,0 +1,44 @@
+/*
+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 controller
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// Test The CoreV1 EventType "Enum" String Values
+func TestEventTypes(t *testing.T) {
+
+	tests := []struct {
+		name      string
+		eventType CoreV1EventType
+		expect    string
+	}{
+		{name: "ResetOffsetReconciled", eventType: ResetOffsetReconciled, expect: "ResetOffsetReconciled"},
+		{name: "ResetOffsetFinalized", eventType: ResetOffsetFinalized, expect: "ResetOffsetFinalized"},
+		{name: "ResetOffsetSkipped", eventType: ResetOffsetSkipped, expect: "ResetOffsetSkipped"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			actual := test.eventType.String()
+			assert.Equal(t, test.expect, actual)
+		})
+	}
+}
diff --git a/pkg/common/commands/resetoffset/controller/offsets_test.go b/pkg/common/commands/resetoffset/controller/offsets_test.go
new file mode 100644
index 000000000..681382786
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/offsets_test.go
@@ -0,0 +1,374 @@
+/*
+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 controller
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/multierr"
+	"knative.dev/pkg/logging"
+	logtesting "knative.dev/pkg/logging/testing"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
+)
+
+//
+// Test The Kafka Offset Reconciliation
+//
+// The Knative Eventing TableTest framework traditionally used to test Reconciliation is exclusively
+// focused on verifying the pre/post state of Kubernetes resources, and does not allow for easy
+// verification of other external interactions such as the Kafka Offset management happening here.
+// Therefore we will verify the Offset Reconciliation independently from the larger Reconciler testing.
+//
+func TestReconciler_ReconcileOffsets(t *testing.T) {
+
+	// Test Data
+	kafkaBrokers := []string{controllertesting.Brokers}
+	saramaConfig := sarama.NewConfig()
+	topicName := controllertesting.TopicName
+	groupId := controllertesting.GroupId
+	testErr := fmt.Errorf("test-error")
+
+	partition1 := int32(0)
+	oldOffset1 := int64(100)
+	newPastOffset1 := oldOffset1 - 50
+	newFutureOffset1 := oldOffset1 + 50
+
+	partition2 := int32(1)
+	oldOffset2 := int64(200)
+	newPastOffset2 := oldOffset2 - 50
+	newFutureOffset2 := oldOffset2 + 50
+
+	offsetTime := int64(123456789)
+	metadata := formatOffsetMetaData(offsetTime)
+
+	// Define The Test Cases
+	tests := []struct {
+		name                    string
+		client                  *controllertesting.MockClient
+		offsetManager           *controllertesting.MockOffsetManager
+		partitionOffsetManagers map[int32]*controllertesting.MockPartitionOffsetManager
+		expectedOffsetMappings  []kafkav1alpha1.OffsetMapping
+		expectedErr             error
+	}{
+		//
+		// Success Tests
+		//
+
+		{
+			name: "Successful MarkOffset",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, []int32{partition1, partition2}, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition1, offsetTime, newFutureOffset1, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition2, offsetTime, newFutureOffset2, nil),
+				controllertesting.WithClientMockClosed(false),
+				controllertesting.WithClientMockClose(nil)),
+			offsetManager: controllertesting.NewMockOffsetManager(
+				controllertesting.WithOffsetManagerMockCommit(),
+				controllertesting.WithOffsetManagerMockClose(nil)),
+			partitionOffsetManagers: map[int32]*controllertesting.MockPartitionOffsetManager{
+				partition1: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset1, ""),
+					controllertesting.WithPartitionOffsetManagerMockMarkOffset(newFutureOffset1, metadata),
+					controllertesting.WithPartitionOffsetManagerMockErrors(),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+				partition2: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset2, ""),
+					controllertesting.WithPartitionOffsetManagerMockMarkOffset(newFutureOffset2, metadata),
+					controllertesting.WithPartitionOffsetManagerMockErrors(),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+			},
+			expectedOffsetMappings: []kafkav1alpha1.OffsetMapping{
+				{Partition: partition1, OldOffset: oldOffset1, NewOffset: newFutureOffset1},
+				{Partition: partition2, OldOffset: oldOffset2, NewOffset: newFutureOffset2},
+			},
+			expectedErr: nil,
+		},
+		{
+			name: "Successful ResetOffset",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, []int32{partition1, partition2}, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition1, offsetTime, newPastOffset1, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition2, offsetTime, newPastOffset2, nil),
+				controllertesting.WithClientMockClosed(false),
+				controllertesting.WithClientMockClose(nil)),
+			offsetManager: controllertesting.NewMockOffsetManager(
+				controllertesting.WithOffsetManagerMockCommit(),
+				controllertesting.WithOffsetManagerMockClose(nil)),
+			partitionOffsetManagers: map[int32]*controllertesting.MockPartitionOffsetManager{
+				partition1: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset1, ""),
+					controllertesting.WithPartitionOffsetManagerMockResetOffset(newPastOffset1, metadata),
+					controllertesting.WithPartitionOffsetManagerMockErrors(),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+				partition2: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset2, ""),
+					controllertesting.WithPartitionOffsetManagerMockResetOffset(newPastOffset2, metadata),
+					controllertesting.WithPartitionOffsetManagerMockErrors(),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+			},
+			expectedOffsetMappings: []kafkav1alpha1.OffsetMapping{
+				{Partition: partition1, OldOffset: oldOffset1, NewOffset: newPastOffset1},
+				{Partition: partition2, OldOffset: oldOffset2, NewOffset: newPastOffset2},
+			},
+			expectedErr: nil,
+		},
+
+		//
+		// Sarama Error Tests
+		//
+
+		{
+			name:                   "SaramaNewClientFn() Error",
+			client:                 nil,
+			expectedOffsetMappings: nil,
+			expectedErr:            testErr,
+		},
+		{
+			name: "SaramaNewOffsetManagerFromClientFn() Error",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, []int32{partition1, partition2}, nil),
+				controllertesting.WithClientMockClosed(false),
+				controllertesting.WithClientMockClose(nil)),
+			offsetManager:          nil,
+			expectedOffsetMappings: nil,
+			expectedErr:            testErr,
+		},
+
+		//
+		// Client Error Tests
+		//
+
+		{
+			name: "Client.Partitions() Error",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, nil, testErr),
+				controllertesting.WithClientMockClosed(false),
+				controllertesting.WithClientMockClose(nil)),
+			expectedOffsetMappings: nil,
+			expectedErr:            testErr,
+		},
+		{
+			name: "Client.GetOffset() Error",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, []int32{partition1}, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition1, offsetTime, 0, testErr),
+				controllertesting.WithClientMockClosed(true)),
+			offsetManager: controllertesting.NewMockOffsetManager(
+				controllertesting.WithOffsetManagerMockClose(nil)),
+			partitionOffsetManagers: map[int32]*controllertesting.MockPartitionOffsetManager{
+				partition1: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockErrors(),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+			},
+			expectedOffsetMappings: nil,
+			expectedErr:            testErr,
+		},
+		{
+			name: "Client.Close() Error",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, []int32{partition1}, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition1, offsetTime, newPastOffset1, nil),
+				controllertesting.WithClientMockClosed(false),
+				controllertesting.WithClientMockClose(testErr)),
+			offsetManager: controllertesting.NewMockOffsetManager(
+				controllertesting.WithOffsetManagerMockCommit(),
+				controllertesting.WithOffsetManagerMockClose(nil)),
+			partitionOffsetManagers: map[int32]*controllertesting.MockPartitionOffsetManager{
+				partition1: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset1, ""),
+					controllertesting.WithPartitionOffsetManagerMockResetOffset(newPastOffset1, metadata),
+					controllertesting.WithPartitionOffsetManagerMockErrors(),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+			},
+			expectedOffsetMappings: []kafkav1alpha1.OffsetMapping{{Partition: partition1, OldOffset: oldOffset1, NewOffset: newPastOffset1}},
+			expectedErr:            nil,
+		},
+
+		//
+		// OffsetManager Error Tests
+		//
+
+		{
+			name: "OffsetManager.ManagePartition() Error",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, []int32{partition1}, nil),
+				controllertesting.WithClientMockClosed(true)),
+			offsetManager: controllertesting.NewMockOffsetManager(
+				controllertesting.WithOffsetManagerMockManagePartition(topicName, partition1,
+					controllertesting.NewMockPartitionOffsetManager(
+						controllertesting.WithPartitionOffsetManagerMockAsyncClose(),
+						controllertesting.WithPartitionOffsetManagerMockErrors()), testErr),
+				controllertesting.WithOffsetManagerMockClose(nil)),
+			expectedOffsetMappings: nil,
+			expectedErr:            testErr,
+		},
+		{
+			name: "OffsetManager.Close() Error",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, []int32{partition1}, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition1, offsetTime, newPastOffset1, nil),
+				controllertesting.WithClientMockClosed(true)),
+			offsetManager: controllertesting.NewMockOffsetManager(
+				controllertesting.WithOffsetManagerMockCommit(),
+				controllertesting.WithOffsetManagerMockClose(testErr)),
+			partitionOffsetManagers: map[int32]*controllertesting.MockPartitionOffsetManager{
+				partition1: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset1, ""),
+					controllertesting.WithPartitionOffsetManagerMockResetOffset(newPastOffset1, metadata),
+					controllertesting.WithPartitionOffsetManagerMockErrors(),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+			},
+			expectedOffsetMappings: []kafkav1alpha1.OffsetMapping{{Partition: partition1, OldOffset: oldOffset1, NewOffset: newPastOffset1}},
+			expectedErr:            nil,
+		},
+
+		//
+		// PartitionsOffsetManager Error Tests
+		//
+
+		{
+			name: "PartitionsOffsetManager.Errors()",
+			client: controllertesting.NewMockClient(
+				controllertesting.WithClientMockPartitions(topicName, []int32{partition1, partition2}, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition1, offsetTime, newPastOffset1, nil),
+				controllertesting.WithClientMockGetOffset(topicName, partition2, offsetTime, newPastOffset2, nil),
+				controllertesting.WithClientMockClosed(true)),
+			offsetManager: controllertesting.NewMockOffsetManager(
+				controllertesting.WithOffsetManagerMockCommit(),
+				controllertesting.WithOffsetManagerMockClose(nil)),
+			partitionOffsetManagers: map[int32]*controllertesting.MockPartitionOffsetManager{
+				partition1: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset1, ""),
+					controllertesting.WithPartitionOffsetManagerMockResetOffset(newPastOffset1, metadata),
+					controllertesting.WithPartitionOffsetManagerMockErrors(&sarama.ConsumerError{
+						Topic:     topicName,
+						Partition: partition1,
+						Err:       testErr,
+					}),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+				partition2: controllertesting.NewMockPartitionOffsetManager(
+					controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset2, ""),
+					controllertesting.WithPartitionOffsetManagerMockResetOffset(newPastOffset2, metadata),
+					controllertesting.WithPartitionOffsetManagerMockErrors(&sarama.ConsumerError{
+						Topic:     topicName,
+						Partition: partition2,
+						Err:       testErr,
+					}),
+					controllertesting.WithPartitionOffsetManagerMockAsyncClose()),
+			},
+			expectedOffsetMappings: nil,
+			expectedErr:            multierr.Combine(testErr, testErr),
+		},
+	}
+
+	// Execute The Test Cases
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Create A Context With Test Logger
+			logger := logtesting.TestLogger(t)
+			ctx := logging.WithLogger(context.Background(), logger)
+
+			// Stub The Sarama NewClient() Implementation To Return Mock Sarama Client
+			if test.client == nil {
+				stubSaramaNewClientFn(t, kafkaBrokers, saramaConfig, nil, testErr)
+			} else {
+				stubSaramaNewClientFn(t, kafkaBrokers, saramaConfig, test.client, nil)
+			}
+			defer restoreSaramaNewClientFn()
+
+			// Stub The Sarama NewOffsetManagerFromClient() Implementation To Return Mock Sarama OffsetManager
+			if test.offsetManager == nil {
+				stubSaramaNewOffsetManagerFromClientFn(t, groupId, test.client, nil, testErr)
+			} else {
+				stubSaramaNewOffsetManagerFromClientFn(t, groupId, test.client, test.offsetManager, nil)
+			}
+			defer restoreSaramaNewOffsetManagerFromClientFn()
+
+			// Configure The Test OffsetManager With Partitions
+			for partition, partitionOffsetManager := range test.partitionOffsetManagers {
+				controllertesting.WithOffsetManagerMockManagePartition(topicName, partition, partitionOffsetManager, nil)(test.offsetManager)
+			}
+
+			// Create A Reconciler
+			reconciler := &Reconciler{
+				kafkaBrokers:      kafkaBrokers,
+				saramaConfig:      saramaConfig,
+				resetoffsetLister: nil,
+				refMapper:         nil,
+			}
+
+			// Perform The Test
+			offsetMappings, err := reconciler.reconcileOffsets(ctx, topicName, groupId, offsetTime)
+
+			// Verify The Results
+			assert.Equal(t, test.expectedErr, err)
+			assert.Equal(t, test.expectedOffsetMappings, offsetMappings)
+			if test.client != nil {
+				test.client.AssertExpectations(t)
+			}
+			if test.offsetManager != nil {
+				test.offsetManager.AssertExpectations(t)
+			}
+			if len(test.partitionOffsetManagers) > 0 {
+				for _, partitionOffsetManager := range test.partitionOffsetManagers {
+					partitionOffsetManager.AssertExpectations(t)
+				}
+			}
+		})
+	}
+}
+
+//
+// Stubbing Utilities
+//
+
+// stubSaramaNewClientFn replaces the Sarama NewClient function with a test instance which performs
+// validation and returns the specified parameters.
+func stubSaramaNewClientFn(t *testing.T, expectedBrokers []string, expectedConfig *sarama.Config, saramaClient sarama.Client, err error) {
+	SaramaNewClientFn = func(brokers []string, config *sarama.Config) (sarama.Client, error) {
+		assert.Equal(t, expectedBrokers, brokers)
+		assert.Equal(t, expectedConfig, config)
+		return saramaClient, err
+	}
+}
+
+// restoreSaramaNewClient restores the default/official Sarama NewClient function.
+func restoreSaramaNewClientFn() {
+	SaramaNewClientFn = sarama.NewClient
+}
+
+// stubSaramaNewOffsetManagerFromClientFn replaces the Sarama NewOffsetManagerFromClient function
+// with a test instance which performs validation and returns the specified parameters.
+func stubSaramaNewOffsetManagerFromClientFn(t *testing.T, expectedGroupId string, expectedClient sarama.Client, offsetManager sarama.OffsetManager, err error) {
+	SaramaNewOffsetManagerFromClientFn = func(groupId string, client sarama.Client) (sarama.OffsetManager, error) {
+		assert.Equal(t, expectedGroupId, groupId)
+		assert.Equal(t, expectedClient, client)
+		return offsetManager, err
+	}
+}
+
+// restoreSaramaNewOffsetManagerFromClientFn restores the default/official Sarama NewOffsetManagerFromClient function.
+func restoreSaramaNewOffsetManagerFromClientFn() {
+	SaramaNewOffsetManagerFromClientFn = sarama.NewOffsetManagerFromClient
+}
diff --git a/pkg/common/commands/resetoffset/controller/reconciler_test.go b/pkg/common/commands/resetoffset/controller/reconciler_test.go
new file mode 100644
index 000000000..c7fb3caa6
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/reconciler_test.go
@@ -0,0 +1,448 @@
+/*
+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 controller
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/fake"
+	clientgotesting "k8s.io/client-go/testing"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	logtesting "knative.dev/pkg/logging/testing"
+	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/system"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	fakekafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	resetoffsetreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset"
+	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
+	refmapperstesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers/testing"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
+)
+
+// Test The Reconcile Functionality
+func TestReconcile(t *testing.T) {
+
+	// Test Data
+	logger := logtesting.TestLogger(t)
+
+	kafkaBrokers := []string{controllertesting.Brokers}
+	saramaConfig := sarama.NewConfig()
+	topicName := controllertesting.TopicName
+	groupId := controllertesting.GroupId
+
+	partition := int32(0)
+	oldOffset := int64(100)
+	newOffset := oldOffset - 50
+
+	offsetTime := sarama.OffsetOldest
+	metadata := formatOffsetMetaData(offsetTime)
+
+	offsetMappings := []kafkav1alpha1.OffsetMapping{
+		{Partition: 0, OldOffset: oldOffset, NewOffset: newOffset},
+	}
+
+	testErr := fmt.Errorf("test-error")
+
+	// Define The ResetOffset Reconciler Test Cases
+	commontesting.SetTestEnvironment(t)
+	tableTest := TableTest{
+
+		//
+		// Key Tests
+		//
+
+		{
+			Name: "Bad Key",
+			Key:  "too/many/parts",
+		},
+		{
+			Name: "Key Not Found",
+			Key:  "foo/not-found",
+		},
+
+		//
+		// Success Tests
+		//
+
+		{
+			Name:    "Full Reconciliation Success",
+			Key:     controllertesting.ResetOffsetKey,
+			Objects: []runtime.Object{controllertesting.NewResetOffset(controllertesting.WithFinalizer)},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewResetOffset(
+						controllertesting.WithFinalizer,
+						controllertesting.WithStatusTopic(topicName),
+						controllertesting.WithStatusGroup(groupId),
+						controllertesting.WithStatusPartitions(offsetMappings),
+						controllertesting.WithStatusRefMapped(true),
+						controllertesting.WithStatusResetInitiated(true),
+						controllertesting.WithStatusConsumerGroupsStopped(true),
+						controllertesting.WithStatusOffsetsUpdated(true),
+						controllertesting.WithStatusConsumerGroupsStarted(true)),
+				},
+			},
+			WantEvents: []string{
+				Eventf(corev1.EventTypeNormal, ResetOffsetReconciled.String(), "Reconciled successfully"),
+			},
+		},
+
+		//
+		// "Skipping" Tests
+		//
+
+		{
+			Name: "Skipping Previously Initiated",
+			Key:  controllertesting.ResetOffsetKey,
+			Objects: []runtime.Object{
+				controllertesting.NewResetOffset(controllertesting.WithFinalizer,
+					controllertesting.WithStatusResetInitiated(true)),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewResetOffset(
+						controllertesting.WithFinalizer,
+						controllertesting.WithStatusInitialized,
+						controllertesting.WithStatusResetInitiated(true)),
+				},
+			},
+			WantEvents: []string{
+				Eventf(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously executed ResetOffset"),
+			},
+		},
+		{
+			Name: "Skipping Previously Completed",
+			Key:  controllertesting.ResetOffsetKey,
+			Objects: []runtime.Object{
+				controllertesting.NewResetOffset(
+					controllertesting.WithFinalizer,
+					controllertesting.WithStatusRefMapped(true),
+					controllertesting.WithStatusResetInitiated(true),
+					controllertesting.WithStatusConsumerGroupsStopped(true),
+					controllertesting.WithStatusOffsetsUpdated(true),
+					controllertesting.WithStatusConsumerGroupsStarted(true)),
+			},
+			WantEvents: []string{
+				Eventf(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously executed ResetOffset"),
+			},
+		},
+
+		//
+		// Error Tests
+		//
+
+		{
+			Name:          "MapRef Error",
+			Key:           controllertesting.ResetOffsetKey,
+			Objects:       []runtime.Object{controllertesting.NewResetOffset(controllertesting.WithFinalizer)},
+			OtherTestData: map[string]interface{}{"MapRefErr": testErr},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewResetOffset(
+						controllertesting.WithFinalizer,
+						controllertesting.WithStatusInitialized,
+						controllertesting.WithStatusRefMapped(false, "FailedToMapRef", "Failed to map 'ref' to Kafka Topic and Group: test-error")),
+				},
+			},
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to map 'ref' to Kafka Topic and Group: %v", testErr.Error())),
+			},
+		},
+		{
+			Name:          "Reconcile Offsets Error",
+			Key:           controllertesting.ResetOffsetKey,
+			Objects:       []runtime.Object{controllertesting.NewResetOffset(controllertesting.WithFinalizer)},
+			OtherTestData: map[string]interface{}{"SaramaNewClientFnErr": testErr},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewResetOffset(
+						controllertesting.WithFinalizer,
+						controllertesting.WithStatusInitialized,
+						controllertesting.WithStatusTopic(topicName),
+						controllertesting.WithStatusGroup(groupId),
+						controllertesting.WithStatusRefMapped(true),
+						controllertesting.WithStatusResetInitiated(true),
+						controllertesting.WithStatusConsumerGroupsStopped(true),
+						controllertesting.WithStatusOffsetsUpdated(false, "FailedToUpdateOffsets", "Failed to update offsets of one or more partitions: test-error")),
+				},
+			},
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to update offsets of one or more partitions: %v", testErr.Error())),
+			},
+		},
+
+		//
+		// Finalize Tests
+		//
+
+		{
+			Name: "Full Finalization Success",
+			Key:  controllertesting.ResetOffsetKey,
+			Objects: []runtime.Object{controllertesting.NewResetOffset(
+				controllertesting.WithFinalizer,
+				controllertesting.WithDeletionTimestamp,
+				controllertesting.WithStatusTopic(topicName),
+				controllertesting.WithStatusGroup(groupId),
+				controllertesting.WithStatusPartitions(offsetMappings),
+				controllertesting.WithStatusRefMapped(true),
+				controllertesting.WithStatusResetInitiated(true),
+				controllertesting.WithStatusConsumerGroupsStopped(true),
+				controllertesting.WithStatusOffsetsUpdated(true),
+				controllertesting.WithStatusConsumerGroupsStarted(true))},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Namespace:   controllertesting.ResetOffsetNamespace,
+						Verb:        "patch",
+						Resource:    schema.GroupVersionResource{Group: kafkav1alpha1.SchemeGroupVersion.Group, Version: kafkav1alpha1.SchemeGroupVersion.Version, Resource: "resetoffset"},
+						Subresource: "",
+					},
+					Name:      controllertesting.ResetOffsetName,
+					PatchType: "application/merge-patch+json",
+					Patch:     []byte(`{"metadata":{"finalizers":[],"resourceVersion":""}}`),
+				},
+			},
+			WantEvents: []string{
+				Eventf(corev1.EventTypeNormal, "FinalizerUpdate", "Updated \"resetoffset-name\" finalizers"),
+				Eventf(corev1.EventTypeNormal, ResetOffsetFinalized.String(), "Finalized successfully"),
+			},
+		},
+		{
+			Name: "Finalization Executing Instance",
+			Key:  controllertesting.ResetOffsetKey,
+			Objects: []runtime.Object{controllertesting.NewResetOffset(
+				controllertesting.WithFinalizer,
+				controllertesting.WithDeletionTimestamp,
+				controllertesting.WithStatusTopic(topicName),
+				controllertesting.WithStatusGroup(groupId),
+				controllertesting.WithStatusPartitions(offsetMappings),
+				controllertesting.WithStatusRefMapped(true),
+				controllertesting.WithStatusResetInitiated(true),
+				controllertesting.WithStatusConsumerGroupsStopped(true))},
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, "InternalError", "skipping finalization of in-progress ResetOffset instance"),
+			},
+		},
+	}
+
+	// Restore Sarama Client / OffsetManager Stubs After Test Completion
+	defer restoreSaramaNewClientFn()
+	defer restoreSaramaNewOffsetManagerFromClientFn()
+
+	// Run The TableTest Using The ResetOffset Reconciler Provided By The Factory
+	tableTest.Test(t, controllertesting.MakeFactory(func(ctx context.Context, listers *controllertesting.Listers, cmw configmap.Watcher, options map[string]interface{}) controller.Reconciler {
+
+		// Check MapRefErr Option
+		var mapRefErr error
+		if options != nil {
+			mapRefErrOption := options["MapRefErr"]
+			if err, ok := mapRefErrOption.(error); ok {
+				mapRefErr = err
+			}
+		}
+
+		// Create A Mock ResetOffset RefMapper
+		mockResetOffsetRefMapper := &refmapperstesting.MockResetOffsetRefMapper{}
+		mockResetOffsetRefMapper.On("MapRef", mock.Anything).Return(controllertesting.TopicName, controllertesting.GroupId, mapRefErr)
+
+		// Check SaramaNewClientFnErr Option
+		var saramaNewClientFnErr error
+		if options != nil {
+			saramaNewClientFnErrOption := options["SaramaNewClientFnErr"]
+			if err, ok := saramaNewClientFnErrOption.(error); ok {
+				saramaNewClientFnErr = err
+			}
+		}
+
+		// Mock & Stub "success" Sarama Client / OffsetManager
+		mockClient := newSuccessSaramaClient(topicName, partition, offsetTime, newOffset)
+		stubSaramaNewClientFn(t, kafkaBrokers, saramaConfig, mockClient, saramaNewClientFnErr)
+		mockOffsetManager := newSuccessSaramaOffsetManager(topicName, partition, oldOffset, newOffset, metadata)
+		stubSaramaNewOffsetManagerFromClientFn(t, groupId, mockClient, mockOffsetManager, nil)
+
+		// Create The ResetOffset Reconciler Struct
+		r := &Reconciler{
+			kafkaBrokers:      kafkaBrokers,
+			saramaConfig:      saramaConfig,
+			resetoffsetLister: listers.GetResetOffsetLister(),
+			refMapper:         mockResetOffsetRefMapper,
+		}
+
+		// Create / Return The Full Reconciler
+		return resetoffsetreconciler.NewReconciler(ctx, logger, fakekafkaclient.Get(ctx), listers.GetResetOffsetLister(), controller.GetEventRecorder(ctx), r)
+
+	}, logger.Desugar()))
+}
+
+func TestReconciler_updateKafkaConfig(t *testing.T) {
+
+	// Define EKConfig String For Use In Test (Note - Preserve Indentation!)
+	ekConfigString := strings.Replace(commontesting.TestEKConfig, "kafka:", `kafka:
+  authSecretName: `+commontesting.SecretName+`
+  authSecretNamespace: `+system.Namespace(), 1)
+
+	// Define Brokers
+	defaultKafkaBrokers := []string{commontesting.BrokerString}
+
+	// Create A Sarama Config To Match commontesting.OldSaramaConfig
+	oldSaramaConfig := sarama.NewConfig()
+	oldSaramaConfig.ClientID = Component
+	oldSaramaConfig.Net.TLS.Enable = true
+	oldSaramaConfig.Net.SASL.Version = 1
+	oldSaramaConfig.Net.SASL.Enable = true
+	oldSaramaConfig.Net.SASL.Mechanism = sarama.SASLTypePlaintext
+	oldSaramaConfig.Net.SASL.User = commontesting.OldAuthUsername
+	oldSaramaConfig.Net.SASL.Password = commontesting.OldAuthPassword
+	oldSaramaConfig.Producer.Return.Successes = true
+	oldSaramaConfig.Consumer.Return.Errors = true
+	oldSaramaConfig.Consumer.Offsets.AutoCommit.Enable = false
+	oldSaramaConfig.Metadata.RefreshFrequency = 300000000000
+
+	// Define The Test Cases
+	tests := []struct {
+		name                 string
+		configMap            *corev1.ConfigMap
+		initialKafkaBrokers  []string
+		expectedKafkaBrokers []string
+		initialSaramaConfig  *sarama.Config
+		expectedSaramaConfig *sarama.Config
+	}{
+		{
+			name:                 "Nil ConfigMap",
+			configMap:            nil,
+			initialKafkaBrokers:  defaultKafkaBrokers,
+			expectedKafkaBrokers: defaultKafkaBrokers,
+			initialSaramaConfig:  oldSaramaConfig,
+			expectedSaramaConfig: oldSaramaConfig,
+		},
+		{
+			name:                 "Nil ConfigMap Data",
+			configMap:            &corev1.ConfigMap{},
+			initialKafkaBrokers:  defaultKafkaBrokers,
+			expectedKafkaBrokers: defaultKafkaBrokers,
+			initialSaramaConfig:  oldSaramaConfig,
+			expectedSaramaConfig: oldSaramaConfig,
+		},
+		{
+			name:                 "Load Settings Error",
+			configMap:            &corev1.ConfigMap{Data: map[string]string{constants.EventingKafkaSettingsConfigKey: "\tInvalid"}},
+			initialKafkaBrokers:  defaultKafkaBrokers,
+			expectedKafkaBrokers: defaultKafkaBrokers,
+			initialSaramaConfig:  oldSaramaConfig,
+			expectedSaramaConfig: oldSaramaConfig,
+		},
+
+		{
+			name: "Success",
+			configMap: &corev1.ConfigMap{Data: map[string]string{
+				constants.VersionConfigKey:               constants.CurrentConfigVersion,
+				constants.EventingKafkaSettingsConfigKey: ekConfigString,
+				constants.SaramaSettingsConfigKey:        commontesting.OldSaramaConfig}},
+			initialKafkaBrokers:  nil,
+			expectedKafkaBrokers: defaultKafkaBrokers,
+			initialSaramaConfig:  nil,
+			expectedSaramaConfig: oldSaramaConfig,
+		},
+	}
+
+	// Create A Context With Fake K8S Client To Return Kafka Secret
+	secret := commontesting.GetTestSaramaSecret(
+		commontesting.SecretName,
+		commontesting.OldAuthUsername,
+		commontesting.OldAuthPassword,
+		commontesting.OldAuthNamespace,
+		commontesting.OldAuthSaslType)
+	fakeK8sClient := fake.NewSimpleClientset(secret)
+	ctx := context.WithValue(context.TODO(), kubeclient.Key{}, fakeK8sClient)
+
+	// Execute The Test Cases
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Create The Reconciler To Test
+			r := &Reconciler{
+				kafkaBrokers: test.initialKafkaBrokers,
+				saramaConfig: test.initialSaramaConfig,
+			}
+
+			// Perform The Test
+			r.updateKafkaConfig(ctx, test.configMap)
+
+			// Verify Reconciler Updated As Expected (Can't Compare Full Sarama.Config Instances Directly Due To Functions Fields)
+			assert.Equal(t, test.expectedKafkaBrokers, r.kafkaBrokers)
+			assert.Equal(t, Component, test.expectedSaramaConfig.ClientID)
+			assert.Equal(t, test.expectedSaramaConfig.Net, r.saramaConfig.Net)
+			assert.Equal(t, test.expectedSaramaConfig.Consumer, r.saramaConfig.Consumer)
+
+			// Verify Fixed Configuration
+			if r.saramaConfig != nil {
+				assert.True(t, r.saramaConfig.Consumer.Return.Errors)
+				assert.False(t, r.saramaConfig.Consumer.Offsets.AutoCommit.Enable)
+			}
+		})
+	}
+
+	// Verify that a nil reconciler doesn't panic
+	var nilReconciler *Reconciler
+	//goland:noinspection GoNilness
+	nilReconciler.updateKafkaConfig(context.TODO(), nil)
+	assert.Nil(t, nilReconciler)
+}
+
+//
+// Sarama Mock Utilities
+//
+
+// newSuccessSaramaClient returns a "success" mock Sarama Client for the specified values.
+func newSuccessSaramaClient(topicName string, partition int32, offsetTime int64, newOffset int64) sarama.Client {
+	return controllertesting.NewMockClient(
+		controllertesting.WithClientMockPartitions(topicName, []int32{partition}, nil),
+		controllertesting.WithClientMockGetOffset(topicName, partition, offsetTime, newOffset, nil),
+		controllertesting.WithClientMockClosed(false),
+		controllertesting.WithClientMockClose(nil))
+}
+
+// newSuccessSaramaOffsetManager returns a "success" mock Sarama OffsetManager for the specified values.
+func newSuccessSaramaOffsetManager(topicName string, partition int32, oldOffset int64, newOffset int64, metadata string) sarama.OffsetManager {
+
+	mockPartitionOffsetManager := controllertesting.NewMockPartitionOffsetManager(
+		controllertesting.WithPartitionOffsetManagerMockNextOffset(oldOffset, ""),
+		controllertesting.WithPartitionOffsetManagerMockResetOffset(newOffset, metadata),
+		controllertesting.WithPartitionOffsetManagerMockErrors(),
+		controllertesting.WithPartitionOffsetManagerMockAsyncClose())
+
+	return controllertesting.NewMockOffsetManager(
+		controllertesting.WithOffsetManagerMockManagePartition(topicName, partition, mockPartitionOffsetManager, nil),
+		controllertesting.WithOffsetManagerMockCommit(),
+		controllertesting.WithOffsetManagerMockClose(nil))
+}
diff --git a/pkg/common/commands/resetoffset/controller/testing/data.go b/pkg/common/commands/resetoffset/controller/testing/data.go
new file mode 100644
index 000000000..06beb81bd
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/testing/data.go
@@ -0,0 +1,177 @@
+/*
+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 (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+const (
+	ResetOffsetNamespace = "resetoffset-namespace"
+	ResetOffsetName      = "resetoffset-name"
+	ResetOffsetKey       = ResetOffsetNamespace + "/" + ResetOffsetName
+	ResetOffsetFinalizer = "resetoffsets.kafka.eventing.knative.dev"
+
+	RefAPIVersion = "ref-apiversion"
+	RefKind       = "ref-kind"
+	RefNamespace  = "ref-namespace"
+	RefName       = "ref-name"
+
+	Brokers = "TestKafkaBrokers"
+
+	TopicName = "TestTopicName"
+	GroupId   = "TestGroupId"
+)
+
+var DeletionTimestamp = metav1.Now()
+
+//
+// ResetOffset Resources
+//
+
+// ResetOffsetOption allow for customizing a ResetOffset
+type ResetOffsetOption func(resetOffset *kafkav1alpha1.ResetOffset)
+
+// NewResetOffset creates a custom ResetOffset
+func NewResetOffset(options ...ResetOffsetOption) *kafkav1alpha1.ResetOffset {
+
+	// Create The Base ResetOffset
+	resetOffset := &kafkav1alpha1.ResetOffset{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "ResetOffset",
+			APIVersion: kafkav1alpha1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ResetOffsetNamespace,
+			Name:      ResetOffsetName,
+		},
+		Spec: kafkav1alpha1.ResetOffsetSpec{
+			Offset: kafkav1alpha1.OffsetSpec{
+				Time: kafkav1alpha1.OffsetEarliest,
+			},
+			Ref: duckv1.KReference{
+				APIVersion: RefAPIVersion,
+				Kind:       RefKind,
+				Namespace:  RefNamespace,
+				Name:       RefName,
+			},
+		},
+	}
+
+	// Apply The Specified Customizations
+	for _, option := range options {
+		option(resetOffset)
+	}
+
+	// Return The Custom ResetOffset
+	return resetOffset
+}
+
+func WithSpecOffsetTime(time string) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		resetOffset.Spec.Offset.Time = time
+	}
+}
+
+func WithSpecRef(ref *duckv1.KReference) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		resetOffset.Spec.Ref = *ref
+	}
+}
+
+func WithDeletionTimestamp(resetOffset *kafkav1alpha1.ResetOffset) {
+	resetOffset.ObjectMeta.SetDeletionTimestamp(&DeletionTimestamp)
+}
+
+func WithFinalizer(resetOffset *kafkav1alpha1.ResetOffset) {
+	resetOffset.ObjectMeta.Finalizers = []string{ResetOffsetFinalizer}
+}
+
+func WithStatusTopic(topic string) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		resetOffset.Status.Topic = topic
+	}
+}
+
+func WithStatusGroup(group string) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		resetOffset.Status.Group = group
+	}
+}
+
+func WithStatusPartitions(partitions []kafkav1alpha1.OffsetMapping) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		resetOffset.Status.Partitions = partitions
+	}
+}
+
+func WithStatusInitialized(resetOffset *kafkav1alpha1.ResetOffset) {
+	resetOffset.Status.InitializeConditions()
+}
+
+func WithStatusRefMapped(state bool, failed ...string) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		if state {
+			resetOffset.Status.MarkRefMappedTrue()
+		} else {
+			resetOffset.Status.MarkRefMappedFailed(failed[0], failed[1])
+		}
+	}
+}
+
+func WithStatusResetInitiated(state bool, failed ...string) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		if state {
+			resetOffset.Status.MarkResetInitiatedTrue()
+		} else {
+			resetOffset.Status.MarkResetInitiatedFailed(failed[0], failed[1])
+		}
+	}
+}
+
+func WithStatusConsumerGroupsStopped(state bool, failed ...string) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		if state {
+			resetOffset.Status.MarkConsumerGroupsStoppedTrue()
+		} else {
+			resetOffset.Status.MarkConsumerGroupsStoppedFailed(failed[0], failed[1])
+		}
+	}
+}
+
+func WithStatusOffsetsUpdated(state bool, failed ...string) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		if state {
+			resetOffset.Status.MarkOffsetsUpdatedTrue()
+		} else {
+			resetOffset.Status.MarkOffsetsUpdatedFailed(failed[0], failed[1])
+		}
+	}
+}
+
+func WithStatusConsumerGroupsStarted(state bool, failed ...string) ResetOffsetOption {
+	return func(resetOffset *kafkav1alpha1.ResetOffset) {
+		if state {
+			resetOffset.Status.MarkConsumerGroupsStartedTrue()
+		} else {
+			resetOffset.Status.MarkConsumerGroupsStartedFailed(failed[0], failed[1])
+		}
+	}
+}
diff --git a/pkg/common/commands/resetoffset/controller/testing/factory.go b/pkg/common/commands/resetoffset/controller/testing/factory.go
new file mode 100644
index 000000000..7e2d69ad8
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/testing/factory.go
@@ -0,0 +1,100 @@
+/*
+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"
+	"testing"
+
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/record"
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	fakedynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/reconciler"
+	. "knative.dev/pkg/reconciler/testing"
+
+	fakekafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
+)
+
+const (
+	// maxEventBufferSize is the estimated max number of event notifications that
+	// can be buffered during reconciliation.
+	maxEventBufferSize = 10
+)
+
+// Ctor functions create a k8s controller with given params.
+type Ctor func(context.Context, *Listers, configmap.Watcher, map[string]interface{}) controller.Reconciler
+
+// MakeFactory creates a reconciler factory with fake clients and controller created by `ctor`.
+func MakeFactory(ctor Ctor, logger *zap.Logger) Factory {
+	return func(t *testing.T, r *TableRow) (controller.Reconciler, ActionRecorderList, EventList) {
+
+		ls := NewListers(r.Objects)
+
+		ctx := context.Background()
+		ctx = logging.WithLogger(ctx, logger.Sugar())
+
+		ctx, kubeClient := fakekubeclient.With(ctx, ls.GetKubeObjects()...)
+		ctx, eventingClient := fakeeventingclient.With(ctx, ls.GetEventingObjects()...)
+		ctx, client := fakekafkaclient.With(ctx, ls.GetResetOffsetObjects()...)
+
+		dynamicScheme := runtime.NewScheme()
+		for _, addTo := range clientSetSchemes {
+			_ = addTo(dynamicScheme)
+		}
+
+		ctx, dynamicClient := fakedynamicclient.With(ctx, dynamicScheme, ls.GetAllObjects()...)
+
+		eventRecorder := record.NewFakeRecorder(maxEventBufferSize)
+		ctx = controller.WithEventRecorder(ctx, eventRecorder)
+
+		// Set up our Controller from the fakes.
+		c := ctor(ctx, &ls, configmap.NewStaticWatcher(), r.OtherTestData)
+
+		// The Reconciler won't do any work until it becomes the leader.
+		if la, ok := c.(reconciler.LeaderAware); ok {
+			_ = la.Promote(reconciler.UniversalBucket(), func(reconciler.Bucket, types.NamespacedName) {})
+		}
+
+		for _, reactor := range r.WithReactors {
+			kubeClient.PrependReactor("*", "*", reactor)
+			client.PrependReactor("*", "*", reactor)
+			dynamicClient.PrependReactor("*", "*", reactor)
+			eventingClient.PrependReactor("*", "*", reactor)
+		}
+
+		// Validate all Create operations through the eventing client.
+		client.PrependReactor("create", "*", func(action clientgotesting.Action) (handled bool, ret runtime.Object, err error) {
+			return ValidateCreates(context.Background(), action)
+		})
+		client.PrependReactor("update", "*", func(action clientgotesting.Action) (handled bool, ret runtime.Object, err error) {
+			return ValidateUpdates(context.Background(), action)
+		})
+
+		actionRecorderList := ActionRecorderList{dynamicClient, client, kubeClient}
+		eventList := EventList{Recorder: eventRecorder}
+
+		return c, actionRecorderList, eventList
+	}
+}
diff --git a/pkg/common/commands/resetoffset/controller/testing/lister.go b/pkg/common/commands/resetoffset/controller/testing/lister.go
new file mode 100644
index 000000000..5e5147ee2
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/testing/lister.go
@@ -0,0 +1,82 @@
+/*
+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 (
+	"k8s.io/apimachinery/pkg/runtime"
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
+	fakeeventingclientset "knative.dev/eventing/pkg/client/clientset/versioned/fake"
+	"knative.dev/pkg/reconciler/testing"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	fakekafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
+	resetoffsetlisters "knative.dev/eventing-kafka/pkg/client/listers/kafka/v1alpha1"
+)
+
+var clientSetSchemes = []func(*runtime.Scheme) error{
+	fakekubeclientset.AddToScheme,
+	fakekafkaclientset.AddToScheme,
+	fakeeventingclientset.AddToScheme,
+}
+
+type Listers struct {
+	sorter testing.ObjectSorter
+}
+
+func NewListers(objs []runtime.Object) Listers {
+
+	scheme := runtime.NewScheme()
+
+	for _, addTo := range clientSetSchemes {
+		addTo(scheme)
+	}
+
+	ls := Listers{
+		sorter: testing.NewObjectSorter(scheme),
+	}
+
+	ls.sorter.AddObjects(objs...)
+
+	return ls
+}
+
+func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
+	return l.sorter.IndexerForObjectType(obj)
+}
+
+func (l *Listers) GetKubeObjects() []runtime.Object {
+	return l.sorter.ObjectsForSchemeFunc(fakekubeclientset.AddToScheme)
+}
+
+func (l *Listers) GetEventingObjects() []runtime.Object {
+	return l.sorter.ObjectsForSchemeFunc(fakeeventingclientset.AddToScheme)
+}
+
+func (l *Listers) GetResetOffsetObjects() []runtime.Object {
+	return l.sorter.ObjectsForSchemeFunc(fakekafkaclientset.AddToScheme)
+}
+
+func (l *Listers) GetAllObjects() []runtime.Object {
+	all := l.GetResetOffsetObjects()
+	all = append(all, l.GetKubeObjects()...)
+	return all
+}
+
+func (l *Listers) GetResetOffsetLister() resetoffsetlisters.ResetOffsetLister {
+	return resetoffsetlisters.NewResetOffsetLister(l.indexerFor(&kafkav1alpha1.ResetOffset{}))
+}
diff --git a/pkg/common/commands/resetoffset/controller/testing/mock.go b/pkg/common/commands/resetoffset/controller/testing/mock.go
new file mode 100644
index 000000000..455860c59
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/testing/mock.go
@@ -0,0 +1,317 @@
+/*
+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 (
+	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/mock"
+)
+
+//
+// Mock Sarama Client
+//
+
+var _ sarama.Client = &MockClient{}
+
+type MockClient struct {
+	mock.Mock
+}
+
+func (c *MockClient) Config() *sarama.Config {
+	args := c.Called()
+	return args.Get(0).(*sarama.Config)
+}
+
+func (c *MockClient) Controller() (*sarama.Broker, error) {
+	args := c.Called()
+	return args.Get(0).(*sarama.Broker), args.Error(1)
+}
+
+func (c *MockClient) RefreshController() (*sarama.Broker, error) {
+	args := c.Called()
+	return args.Get(0).(*sarama.Broker), args.Error(1)
+}
+
+func (c *MockClient) Brokers() []*sarama.Broker {
+	args := c.Called()
+	return args.Get(0).([]*sarama.Broker)
+}
+
+func (c *MockClient) Broker(brokerID int32) (*sarama.Broker, error) {
+	args := c.Called(brokerID)
+	return args.Get(0).(*sarama.Broker), args.Error(1)
+}
+
+func (c *MockClient) Topics() ([]string, error) {
+	args := c.Called()
+	return args.Get(0).([]string), args.Error(1)
+}
+
+func (c *MockClient) Partitions(topic string) ([]int32, error) {
+	args := c.Called(topic)
+	return args.Get(0).([]int32), args.Error(1)
+}
+
+func (c *MockClient) WritablePartitions(topic string) ([]int32, error) {
+	args := c.Called(topic)
+	return args.Get(0).([]int32), args.Error(1)
+}
+
+func (c *MockClient) Leader(topic string, partitionID int32) (*sarama.Broker, error) {
+	args := c.Called(topic, partitionID)
+	return args.Get(0).(*sarama.Broker), args.Error(1)
+}
+
+func (c *MockClient) Replicas(topic string, partitionID int32) ([]int32, error) {
+	args := c.Called(topic, partitionID)
+	return args.Get(0).([]int32), args.Error(1)
+}
+
+func (c *MockClient) InSyncReplicas(topic string, partitionID int32) ([]int32, error) {
+	args := c.Called(topic, partitionID)
+	return args.Get(0).([]int32), args.Error(1)
+}
+
+func (c *MockClient) OfflineReplicas(topic string, partitionID int32) ([]int32, error) {
+	args := c.Called(topic, partitionID)
+	return args.Get(0).([]int32), args.Error(1)
+}
+
+func (c *MockClient) RefreshBrokers(addrs []string) error {
+	args := c.Called(addrs)
+	return args.Error(0)
+}
+
+func (c *MockClient) RefreshMetadata(topics ...string) error {
+	args := c.Called(topics)
+	return args.Error(0)
+}
+
+func (c *MockClient) GetOffset(topic string, partitionID int32, time int64) (int64, error) {
+	args := c.Called(topic, partitionID, time)
+	return args.Get(0).(int64), args.Error(1)
+}
+
+func (c *MockClient) Coordinator(consumerGroup string) (*sarama.Broker, error) {
+	args := c.Called(consumerGroup)
+	return args.Get(0).(*sarama.Broker), args.Error(1)
+}
+
+func (c *MockClient) RefreshCoordinator(consumerGroup string) error {
+	args := c.Called(consumerGroup)
+	return args.Error(0)
+}
+
+func (c *MockClient) InitProducerID() (*sarama.InitProducerIDResponse, error) {
+	args := c.Called()
+	return args.Get(0).(*sarama.InitProducerIDResponse), args.Error(1)
+}
+
+func (c *MockClient) Close() error {
+	args := c.Called()
+	return args.Error(0)
+}
+
+func (c *MockClient) Closed() bool {
+	args := c.Called()
+	return args.Bool(0)
+}
+
+type MockClientOption = func(*MockClient)
+
+func NewMockClient(options ...MockClientOption) *MockClient {
+	mockClient := &MockClient{}
+	for _, option := range options {
+		option(mockClient)
+	}
+	return mockClient
+}
+
+func WithClientMockPartitions(topic string, partitions []int32, err error) MockClientOption {
+	return func(mockSaramaClient *MockClient) {
+		mockSaramaClient.On("Partitions", topic).Return(partitions, err)
+	}
+}
+
+func WithClientMockGetOffset(topic string, partition int32, offsetTime int64, offset int64, err error) MockClientOption {
+	return func(mockSaramaClient *MockClient) {
+		mockSaramaClient.On("GetOffset", topic, partition, offsetTime).Return(offset, err)
+	}
+}
+
+func WithClientMockClosed(closed bool) MockClientOption {
+	return func(mockSaramaClient *MockClient) {
+		mockSaramaClient.On("Closed").Return(closed)
+	}
+}
+
+func WithClientMockClose(err error) MockClientOption {
+	return func(mockSaramaClient *MockClient) {
+		mockSaramaClient.On("Close").Return(err)
+	}
+}
+
+//
+// Mock Sarama OffsetManager
+//
+
+var _ sarama.OffsetManager = &MockOffsetManager{}
+
+type MockOffsetManager struct {
+	mock.Mock
+}
+
+func (o *MockOffsetManager) ManagePartition(topic string, partition int32) (sarama.PartitionOffsetManager, error) {
+	args := o.Called(topic, partition)
+	arg0 := args.Get(0)
+	var partitionOffsetManager sarama.PartitionOffsetManager
+	if arg0 != nil {
+		partitionOffsetManager = arg0.(sarama.PartitionOffsetManager)
+	}
+	return partitionOffsetManager, args.Error(1)
+}
+
+func (o *MockOffsetManager) Close() error {
+	args := o.Called()
+	return args.Error(0)
+}
+
+func (o *MockOffsetManager) Commit() {
+	o.Called()
+}
+
+type MockOffsetManagerOption = func(*MockOffsetManager)
+
+func NewMockOffsetManager(options ...MockOffsetManagerOption) *MockOffsetManager {
+	mockOffsetManager := &MockOffsetManager{}
+	for _, option := range options {
+		option(mockOffsetManager)
+	}
+	return mockOffsetManager
+}
+
+func WithOffsetManagerMockManagePartition(topic string, partition int32, partitionOffsetManager sarama.PartitionOffsetManager, err error) MockOffsetManagerOption {
+	return func(mockOffsetManager *MockOffsetManager) {
+		mockOffsetManager.On("ManagePartition", topic, partition).Return(partitionOffsetManager, err)
+	}
+}
+
+func WithOffsetManagerMockCommit() MockOffsetManagerOption {
+	return func(mockOffsetManager *MockOffsetManager) {
+		mockOffsetManager.On("Commit").Return()
+	}
+}
+
+func WithOffsetManagerMockClose(err error) MockOffsetManagerOption {
+	return func(mockOffsetManager *MockOffsetManager) {
+		mockOffsetManager.On("Close").Return(err)
+	}
+}
+
+//
+// Mock Sarama PartitionOffsetManager
+//
+
+var _ sarama.PartitionOffsetManager = &MockPartitionOffsetManager{}
+
+type MockPartitionOffsetManager struct {
+	mock.Mock
+}
+
+func (p *MockPartitionOffsetManager) NextOffset() (int64, string) {
+	args := p.Called()
+	return args.Get(0).(int64), args.String(1)
+}
+
+func (p *MockPartitionOffsetManager) MarkOffset(offset int64, metadata string) {
+	p.Called(offset, metadata)
+}
+
+func (p *MockPartitionOffsetManager) ResetOffset(offset int64, metadata string) {
+	p.Called(offset, metadata)
+}
+
+func (p *MockPartitionOffsetManager) Errors() <-chan *sarama.ConsumerError {
+	args := p.Called()
+	return args.Get(0).(chan *sarama.ConsumerError)
+}
+
+func (p *MockPartitionOffsetManager) AsyncClose() {
+	p.Called()
+}
+
+func (p *MockPartitionOffsetManager) Close() error {
+	args := p.Called()
+	return args.Error(0)
+}
+
+type MockPartitionOffsetManagerOption = func(*MockPartitionOffsetManager)
+
+func NewMockPartitionOffsetManager(options ...MockPartitionOffsetManagerOption) *MockPartitionOffsetManager {
+	mockPartitionOffsetManager := &MockPartitionOffsetManager{}
+	for _, option := range options {
+		option(mockPartitionOffsetManager)
+	}
+	return mockPartitionOffsetManager
+}
+
+func WithPartitionOffsetManagerMockNextOffset(offset int64, metadata string) MockPartitionOffsetManagerOption {
+	return func(mockPartitionOffsetManager *MockPartitionOffsetManager) {
+		mockPartitionOffsetManager.On("NextOffset").Return(offset, metadata)
+	}
+}
+
+func WithPartitionOffsetManagerMockMarkOffset(offset int64, metadata string) MockPartitionOffsetManagerOption {
+	return func(mockPartitionOffsetManager *MockPartitionOffsetManager) {
+		mockPartitionOffsetManager.On("MarkOffset", offset, metadata).Return()
+	}
+}
+
+func WithPartitionOffsetManagerMockResetOffset(offset int64, metadata string) MockPartitionOffsetManagerOption {
+	return func(mockPartitionOffsetManager *MockPartitionOffsetManager) {
+		mockPartitionOffsetManager.On("ResetOffset", offset, metadata).Return()
+	}
+}
+
+func WithPartitionOffsetManagerMockErrors(errors ...*sarama.ConsumerError) MockPartitionOffsetManagerOption {
+	return func(mockPartitionOffsetManager *MockPartitionOffsetManager) {
+		errChan := make(chan *sarama.ConsumerError)
+		if len(errors) > 0 {
+			go func() {
+				for _, err := range errors {
+					errChan <- err
+				}
+				close(errChan)
+			}()
+		} else {
+			close(errChan)
+		}
+		mockPartitionOffsetManager.On("Errors").Return(errChan)
+	}
+}
+
+func WithPartitionOffsetManagerMockClose(err error) MockPartitionOffsetManagerOption {
+	return func(mockPartitionOffsetManager *MockPartitionOffsetManager) {
+		mockPartitionOffsetManager.On("Close").Return(err)
+	}
+}
+
+func WithPartitionOffsetManagerMockAsyncClose() MockPartitionOffsetManagerOption {
+	return func(mockPartitionOffsetManager *MockPartitionOffsetManager) {
+		mockPartitionOffsetManager.On("AsyncClose").Return()
+	}
+}
diff --git a/pkg/common/commands/resetoffset/refmappers/subscription_test.go b/pkg/common/commands/resetoffset/refmappers/subscription_test.go
new file mode 100644
index 000000000..c477ba339
--- /dev/null
+++ b/pkg/common/commands/resetoffset/refmappers/subscription_test.go
@@ -0,0 +1,280 @@
+/*
+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 refmappers
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/rest"
+	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake" // Knative Fake Informer Injection
+	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/injection"
+	"knative.dev/pkg/logging"
+	logtesting "knative.dev/pkg/logging/testing"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
+)
+
+const (
+	SubscriptionNamespace = "subscription-namespace"
+	SubscriptionName      = "subscription-name"
+
+	TopicName = "TestTopicName"
+	GroupId   = "TestGroupId"
+)
+
+func TestNewSubscriptionRefMapperFactory(t *testing.T) {
+
+	// Create A Context With Test Logger
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.Background(), logger)
+
+	// Register Fake Informers (See Injection "_" Imports Above!)
+	ctx, fakeInformers := injection.Fake.SetupInformers(ctx, &rest.Config{})
+	assert.NotNil(t, fakeInformers)
+
+	// Create Test Mappers
+	topicNameMapper := newMockSubscriptionTopicNameMapper(t, nil, TopicName, nil)
+	groupIdMapper := newMockSubscriptionConsumerGroupIdMapper(t, nil, GroupId, nil)
+
+	// Perform The Test - Create New Subscription RefMapper Factory
+	factory := NewSubscriptionRefMapperFactory(topicNameMapper, groupIdMapper)
+	assert.NotNil(t, factory)
+
+	// Test The Factory Create()
+	refMapper := factory.Create(ctx)
+	assert.NotNil(t, refMapper)
+}
+
+func TestNewResetOffsetSubscriptionRefMapper(t *testing.T) {
+
+	// Create A Context With Test Logger
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.Background(), logger)
+
+	// Register Fake Informers (See Injection "_" Imports Above!)
+	ctx, fakeInformers := injection.Fake.SetupInformers(ctx, &rest.Config{})
+	assert.NotNil(t, fakeInformers)
+
+	// Create Test Mappers
+	topicNameMapper := newMockSubscriptionTopicNameMapper(t, nil, TopicName, nil)
+	groupIdMapper := newMockSubscriptionConsumerGroupIdMapper(t, nil, GroupId, nil)
+
+	// Perform The Test - Create A New SubscriptionRefMapper
+	resetOffsetSubscriptionRefMapper := NewSubscriptionRefMapper(ctx, topicNameMapper, groupIdMapper)
+
+	// Verify The Results
+	assert.NotNil(t, resetOffsetSubscriptionRefMapper)
+	assert.Equal(t, logger.Desugar(), resetOffsetSubscriptionRefMapper.logger)
+	assert.NotNil(t, resetOffsetSubscriptionRefMapper.subscriptionLister)
+	assert.NotNil(t, resetOffsetSubscriptionRefMapper.topicNameMapper) // Testify / DeepEqual Cannot Compare func Types
+	assert.NotNil(t, resetOffsetSubscriptionRefMapper.groupIdMapper)
+}
+
+func TestResetOffsetSubscriptionRefMapper_MapRef(t *testing.T) {
+
+	// Test Data
+	logger := logtesting.TestLogger(t).Desugar()
+	testErr := fmt.Errorf("test-error")
+
+	// Create A Test Subscription
+	subscription := &messagingv1.Subscription{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Subscription",
+			APIVersion: messagingv1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: SubscriptionNamespace,
+			Name:      SubscriptionName,
+		},
+	}
+
+	//
+	subscriptionRef := &duckv1.KReference{
+		Kind:       "Subscription",
+		APIVersion: messagingv1.SchemeGroupVersion.String(),
+		Namespace:  SubscriptionNamespace,
+		Name:       SubscriptionName,
+	}
+
+	// Define The Test Cases
+	tests := []struct {
+		name            string
+		subscription    *messagingv1.Subscription
+		subscriptionErr error
+		resetOffset     *kafkav1alpha1.ResetOffset
+		topicNameMapper SubscriptionTopicNameMapper
+		groupIdMapper   SubscriptionConsumerGroupIdMapper
+		wantTopic       string
+		wantGroup       string
+		wantErr         bool
+	}{
+		{
+			name:            "Success",
+			subscription:    subscription,
+			resetOffset:     controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper: newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
+			groupIdMapper:   newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
+			wantTopic:       TopicName,
+			wantGroup:       GroupId,
+		},
+		{
+			name:        "Nil ResetOffset",
+			resetOffset: nil,
+			wantErr:     true,
+		},
+		{
+			name: "Invalid ResetOffset.Spec.Ref",
+			resetOffset: controllertesting.NewResetOffset(controllertesting.WithSpecRef(&duckv1.KReference{
+				Kind:       "foo",
+				Namespace:  "bar",
+				Name:       "baz",
+				APIVersion: "bing",
+			})),
+			wantErr: true,
+		},
+		{
+			name: "Sparse ResetOffset.Spec.Ref",
+			resetOffset: controllertesting.NewResetOffset(controllertesting.WithSpecRef(&duckv1.KReference{
+				Kind:       "Subscription",
+				APIVersion: messagingv1.SchemeGroupVersion.String(),
+				Name:       SubscriptionName,
+			})),
+			wantErr: true,
+		},
+		{
+			name:            "Subscription Get Error",
+			subscriptionErr: testErr,
+			resetOffset:     controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			wantErr:         true,
+		},
+		{
+			name:            "TopicName Mapper Error",
+			subscription:    subscription,
+			resetOffset:     controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper: newMockSubscriptionTopicNameMapper(t, subscription, TopicName, testErr),
+			groupIdMapper:   newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
+			wantErr:         true,
+		},
+		{
+			name:            "GroupId Mapper Error",
+			subscription:    subscription,
+			resetOffset:     controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper: newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
+			groupIdMapper:   newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, testErr),
+			wantErr:         true,
+		},
+	}
+
+	// Execute The Test Cases
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Create A Mock SubscriptionLister To Return The Test Subscription
+			mockSubscriptionNamespaceLister := &MockSubscriptionNamespaceLister{}
+			mockSubscriptionNamespaceLister.On("Get", SubscriptionName).Return(test.subscription, test.subscriptionErr)
+			mockSubscriptionLister := &MockSubscriptionLister{}
+			mockSubscriptionLister.On("Subscriptions", SubscriptionNamespace).Return(mockSubscriptionNamespaceLister)
+			if test.resetOffset != nil {
+				mockSubscriptionLister.On("Subscriptions", test.resetOffset.Namespace).Return(mockSubscriptionNamespaceLister)
+			}
+
+			// Create A New SubscriptionRefMapper To Test
+			subscriptionRefMapper := &SubscriptionRefMapper{
+				logger:             logger,
+				subscriptionLister: mockSubscriptionLister,
+				topicNameMapper:    test.topicNameMapper,
+				groupIdMapper:      test.groupIdMapper,
+			}
+
+			// Perform The Test - Map A Subscription To Kafka Topic Name & ConsumerGroup ID
+			topicName, groupId, err := subscriptionRefMapper.MapRef(test.resetOffset)
+
+			// Validate The Results
+			assert.Equal(t, test.wantErr, err != nil)
+			assert.Equal(t, test.wantTopic, topicName)
+			assert.Equal(t, test.wantGroup, groupId)
+		})
+	}
+}
+
+//
+// Mock SubscriptionLister
+//
+
+var _ messaginglisters.SubscriptionLister = &MockSubscriptionLister{}
+
+type MockSubscriptionLister struct {
+	mock.Mock
+}
+
+func (l *MockSubscriptionLister) List(selector labels.Selector) (ret []*messagingv1.Subscription, err error) {
+	args := l.Called(selector)
+	return args.Get(0).([]*messagingv1.Subscription), args.Error(1)
+}
+
+func (l *MockSubscriptionLister) Subscriptions(namespace string) messaginglisters.SubscriptionNamespaceLister {
+	args := l.Called(namespace)
+	return args.Get(0).(messaginglisters.SubscriptionNamespaceLister)
+}
+
+//
+// Mock SubscriptionNamespaceLister
+//
+
+var _ messaginglisters.SubscriptionNamespaceLister = &MockSubscriptionNamespaceLister{}
+
+type MockSubscriptionNamespaceLister struct {
+	mock.Mock
+}
+
+func (nl *MockSubscriptionNamespaceLister) List(selector labels.Selector) (ret []*messagingv1.Subscription, err error) {
+	args := nl.Called(selector)
+	return args.Get(0).([]*messagingv1.Subscription), args.Error(1)
+}
+
+func (nl *MockSubscriptionNamespaceLister) Get(name string) (*messagingv1.Subscription, error) {
+	args := nl.Called(name)
+	return args.Get(0).(*messagingv1.Subscription), args.Error(1)
+}
+
+//
+// Mock Mappers
+//
+
+func newMockSubscriptionTopicNameMapper(t *testing.T, expectedSubscription *messagingv1.Subscription, topicName string, err error) SubscriptionTopicNameMapper {
+	return func(subscription *messagingv1.Subscription) (string, error) {
+		assert.Equal(t, expectedSubscription, subscription)
+		return topicName, err
+	}
+}
+
+func newMockSubscriptionConsumerGroupIdMapper(t *testing.T, expectedSubscription *messagingv1.Subscription, groupId string, err error) SubscriptionConsumerGroupIdMapper {
+	return func(subscription *messagingv1.Subscription) (string, error) {
+		assert.Equal(t, expectedSubscription, subscription)
+		return groupId, err
+	}
+}
diff --git a/pkg/common/commands/resetoffset/refmappers/testing/mocks.go b/pkg/common/commands/resetoffset/refmappers/testing/mocks.go
new file mode 100644
index 000000000..79f625d0b
--- /dev/null
+++ b/pkg/common/commands/resetoffset/refmappers/testing/mocks.go
@@ -0,0 +1,56 @@
+/*
+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"
+
+	"github.com/stretchr/testify/mock"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
+)
+
+//
+// Mock ResetOffsetRefMapperFactory
+//
+
+var _ refmappers.ResetOffsetRefMapperFactory = &MockResetOffsetRefMapperFactory{}
+
+type MockResetOffsetRefMapperFactory struct {
+	mock.Mock
+}
+
+func (f *MockResetOffsetRefMapperFactory) Create(ctx context.Context) refmappers.ResetOffsetRefMapper {
+	args := f.Called(ctx)
+	return args.Get(0).(refmappers.ResetOffsetRefMapper)
+}
+
+//
+// Mock ResetOffsetRefMapper
+//
+
+var _ refmappers.ResetOffsetRefMapper = &MockResetOffsetRefMapper{}
+
+type MockResetOffsetRefMapper struct {
+	mock.Mock
+}
+
+func (m *MockResetOffsetRefMapper) MapRef(resetOffset *kafkav1alpha1.ResetOffset) (string, string, error) {
+	args := m.Called(resetOffset)
+	return args.String(0), args.String(1), args.Error(2)
+}