Repository: knative-sandbox/eventing-kafka
Pull request #589 (2021-04-29 20:56:59)

ResetOffset API & CRD
Sub-Task of Issue #588 to create the ResetOffset API, CRD, and generated source

## Proposed Changes
- Defined the CRD
- Created the API
- Generated DeepCopy, Client, Informers, Listers, etc

**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.

**Notes**
- This is the first step in the ResetOffset implementation.  The new API / CRD are not used anywhere and should not impact any existing implementations.
- This implementation treats the Status slightly different from the FeatureTrack document.  The same approach is still in place, but the names have been altered to fit with the Knative `NewBatchConditionSet()` and `Succeeded` state.  I can update the document to match if necessary once this PR is approved.

A sample ResetOffset CRD can be created as...
```yaml
apiVersion: kafka.eventing.knative.dev/v1alpha1
kind: ResetOffset
metadata:
  name: sample-reset-offset-1
  namespace: tenant1
spec:
  offset: earliest
  ref:
    apiVersion: messaging.knative.dev/v1beta1
    kind: Subscription
    namespace: tenant1
    name: sample-kafka-subscription-1
```
...and an example status could be...
```bash
> kubectl get resetoffset -n tenant1 sample-reset-offset-1
NAME                    SUCCEEDED   REASON                      AGE
sample-reset-offset-1   False       FailedToStopConsumerGroup   15s

> kubectl get resetoffset -n tenant1 sample-reset-offset-1 -o wide
NAME                    SUCCEEDED   REASON                      REF NAMESPACE   REF NAME                      AGE
sample-reset-offset-1   False       FailedToStopConsumerGroup   tenant1         sample-kafka-subscription-1   18
```
...with detailed status of...
```bash
> kubectl get resetoffset -n tenant1 sample-reset-offset-1 -o yaml
apiVersion: kafka.eventing.knative.dev/v1alpha1
kind: ResetOffset
metadata:
  creationTimestamp: "2021-04-29T18:59:11Z"
  finalizers:
  - resetoffsets.kafka.eventing.knative.dev
  generation: 1
  managedFields:
    ...
  name: sample-reset-offset-1
  namespace: tenant1
  resourceVersion: "50428288"
  selfLink: /apis/kafka.eventing.knative.dev/v1alpha1/namespaces/tenant1/resetoffsets/sample-reset-offset-1
  uid: f2b8c647-c717-4b68-8dec-3baa8cff716e
spec:
  offset: earliest
  ref:
    apiVersion: messaging.knative.dev/v1beta1
    kind: Subscription
    name: sample-kafka-subscription-1
    namespace: tenant1
status:
  annotations:
    group: my-group
    partitions: '[{"partition":0,"oldOffset":1,"newOffset":2},{"partition":1,"oldOffset":2,"newOffset":3},{"partition":2,"oldOffset":3,"newOffset":4},{"partition":3,"oldOffset":4,"newOffset":5}]'
    topic: my-topic
  conditions:
  - lastTransitionTime: "2021-04-29T18:59:11Z"
    status: "True"
    type: 1-ResetInitiated
  - lastTransitionTime: "2021-04-29T18:59:11Z"
    status: "True"
    type: 2-OffsetsCalculated
  - lastTransitionTime: "2021-04-29T18:59:11Z"
    message: failed to stop consumergroup 123
    reason: FailedToStopConsumerGroup
    status: "False"
    type: 3-ConsumerGroupsStopped
  - lastTransitionTime: "2021-04-29T18:59:11Z"
    status: Unknown
    type: 4-OffsetsUpdated
  - lastTransitionTime: "2021-04-29T18:59:11Z"
    status: Unknown
    type: 5-ConsumerGroupsStarted
  - lastTransitionTime: "2021-04-29T18:59:11Z"
    message: failed to stop consumergroup 123
    reason: FailedToStopConsumerGroup
    status: "False"
    type: Succeeded
  observedGeneration: 1
```

/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-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/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387873589755121664/artifacts/line-cov.html#file0) | Do not exist | 0.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_annotations.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387873589755121664/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387873589755121664/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387873589755121664/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387873589755121664/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387873589755121664/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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-go-coverage | e706bbae6b2cae62465fb04e5c8f62e8d7440d69 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387873589755121664) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/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/589?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     #589      +/-   ##
==========================================
+ Coverage   71.69%   71.75%   +0.05%     
==========================================
  Files         137      143       +6     
  Lines        5645     5710      +65     
==========================================
+ Hits         4047     4097      +50     
- Misses       1371     1386      +15     
  Partials      227      227              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/apis/kafka/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVnaXN0ZXIuZ28=) | `0.00% <0.00%> (ø)` | |
| [...kg/apis/kafka/v1alpha1/reset\_offset\_annotations.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X2Fubm90YXRpb25zLmdv) | `100.00% <100.00%> (ø)` | |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X2RlZmF1bHRzLmdv) | `100.00% <100.00%> (ø)` | |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/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/apis/kafka/v1alpha1/reset\_offset\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X3R5cGVzLmdv) | `100.00% <100.00%> (ø)` | |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X3ZhbGlkYXRpb24uZ28=) | `100.00% <100.00%> (ø)` | |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/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% <0.00%> (-6.20%)` | :arrow_down: |
| ... and [3 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589/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/589?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/589?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [9621529...e706bba](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/589?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387892634504138752/artifacts/line-cov.html#file0) | Do not exist | 0.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_annotations.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387892634504138752/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387892634504138752/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387892634504138752/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387892634504138752/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387892634504138752/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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-go-coverage | 68a2d5f18f4eddad4d136e248cd377c04f12a7b0 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1387892634504138752) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/589). [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 -->
google-cla[bot]: We found a Contributor License Agreement for you (the sender of this pull request), but were unable to find agreements for all the commit author(s) or Co-authors.  If you authored these, maybe you used a different email address in the git commits than was used to sign the CLA ([login here](https://cla.developers.google.com/) to double check)?  If these were authored by someone else, then they will need to sign a CLA as well, and confirm that they're okay with these being contributed to Google.
In order to pass this check, please resolve this problem and then comment `@googlebot I fixed it.`. If the bot doesn't comment, it means it doesn't think anything has changed.

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

<!-- need_author_cla -->
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/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1388158220404527104/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_annotations.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1388158220404527104/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1388158220404527104/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1388158220404527104/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1388158220404527104/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1388158220404527104/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

travis-minke-sap: CLA broke because I used the github "commit suggestions" feature - anyone know how to recover from that ?
google-cla[bot]: We found a Contributor License Agreement for you (the sender of this pull request), but were unable to find agreements for all the commit author(s) or Co-authors.  If you authored these, maybe you used a different email address in the git commits than was used to sign the CLA ([login here](https://cla.developers.google.com/) to double check)?  If these were authored by someone else, then they will need to sign a CLA as well, and confirm that they're okay with these being contributed to Google.
In order to pass this check, please resolve this problem and then comment `@googlebot I fixed it.`. If the bot doesn't comment, it means it doesn't think anything has changed.

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

<!-- need_author_cla -->
slinkydeveloper: > CLA broke because I used the github "commit suggestions" feature - anyone know how to recover from that ?

Force push changing the commit message author and adding the sign off string is the easiest way :) 
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/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389243734230568960/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_annotations.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389243734230568960/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389243734230568960/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389243734230568960/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389243734230568960/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389243734230568960/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389283407023837184/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_annotations.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389283407023837184/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389283407023837184/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389283407023837184/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389283407023837184/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389283407023837184/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389598085528686592/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_annotations.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389598085528686592/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389598085528686592/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389598085528686592/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389598085528686592/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389598085528686592/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389604078421544960/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_annotations.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389604078421544960/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389604078421544960/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389604078421544960/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389604078421544960/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389604078421544960/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389960597705068544/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_annotations.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389960597705068544/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389960597705068544/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389960597705068544/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389960597705068544/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1389960597705068544/artifacts/line-cov.html#file5) | Do not exist | 100.0% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390044042888220672/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390044042888220672/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390044042888220672/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390044042888220672/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390044042888220672/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 

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/589#pullrequestreview-653244884" title="Approved">slinkydeveloper</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)~~ [slinkydeveloper]

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":[]} -->
matzew: /unhold
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 | b9d7d57249dc13019c9efd42d66270e6254ffcb7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1390272418592854016) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/589). [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: /unhold

Thanks for all the reviews @slinkydeveloper and @matzew and for bearing with me on the annotations discussion/confusion ; )
knative-prow-robot: New changes are detected. LGTM label has been removed.
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/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390298261775978496/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390298261775978496/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390298261775978496/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390298261775978496/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390298261775978496/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 

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 | 53f586b47be709e300a31c6a4e5c253ba5decf9b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1390298261515931648) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/589). [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 | 53f586b47be709e300a31c6a4e5c253ba5decf9b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1390298261515931648) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 53f586b47be709e300a31c6a4e5c253ba5decf9b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1390298261654343680) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/589). [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 tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-integration-test-mt-source | 53f586b47be709e300a31c6a4e5c253ba5decf9b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-integration-test-mt-source/1390298261654343680) | `/test pull-knative-sandbox-eventing-kafka-integration-test-mt-source`
pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated | 53f586b47be709e300a31c6a4e5c253ba5decf9b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1390306160359247872) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/589). [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-mt-source
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 | 53f586b47be709e300a31c6a4e5c253ba5decf9b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated/1390359555602386944) | `/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/589). [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/apis/kafka/v1alpha1/register.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390373529853104128/artifacts/line-cov.html#file0) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390373529853104128/artifacts/line-cov.html#file1) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390373529853104128/artifacts/line-cov.html#file2) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_types.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390373529853104128/artifacts/line-cov.html#file3) | Do not exist | 100.0% | 
[pkg/apis/kafka/v1alpha1/reset_offset_validation.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/589/pull-knative-sandbox-eventing-kafka-go-coverage/1390373529853104128/artifacts/line-cov.html#file4) | Do not exist | 100.0% | 

travis-minke-sap: /test pull-knative-sandbox-eventing-kafka-upgrade-tests

Patch:
diff --git a/config/command/resources/resetoffset-crd.yaml b/config/command/resources/resetoffset-crd.yaml
new file mode 100644
index 000000000..c6337355b
--- /dev/null
+++ b/config/command/resources/resetoffset-crd.yaml
@@ -0,0 +1,181 @@
+# 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.
+
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: resetoffsets.kafka.eventing.knative.dev
+  labels:
+    kafka.eventing.knative.dev/release: devel
+    knative.dev/crd-install: "true"
+spec:
+  group: kafka.eventing.knative.dev
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    subresources:
+      status: { }
+    schema:
+      openAPIV3Schema:
+        description: 'ResetOffset is a one-time "command" or "request" indicating the desire to
+            reposition the Kafka Topic/Partition Offsets of a specific ConsumerGroup as referenced
+            by a Subscription / Trigger, etc.  The ResetOffset status tracks the progress of the
+            steps involved in this repositioning. Once the Succeeded status is set (true / false),
+            the ResetOffset is "finished" and cannot be re-executed.  Another ResetOffset will be
+            required in order to make another attempt once the underlying issue has been resolved.'
+        type: object
+        properties:
+          spec:
+            description: 'Specifies the desired "time" to reposition the Offsets, as well as
+                a reference object which is used to identify the specific ConsumerGroups to be
+                repositioned.'
+            type: object
+            properties:
+              offset:
+                description: 'Wrapper containing various options for specifying the desired Offset.
+                Currently only the "time" option is provided, but this will allow for additional
+                configuration (e.g. explicit partition/offset values) to be added in the future.'
+                type: object
+                properties:
+                  time:
+                    description: 'String defining the time to which the Kafka Topic / Partition
+                    Offsets will be reset. Supported values include "earliest", "latest", or a
+                    valid date / time string in the RFC3339 format (e.g. "2021-05-04T05:04:01Z").
+                    The "earliest" and "latest" values indicate the beginning and end, respectively,
+                    of the persistence window of the Topic. There is no guarantee of precision, and
+                    the exact time/offset will depend on the state of the persistence window when
+                    the ResetOffset command is executed. There is no default value, and invalid
+                    values will result in the ResetOffset operation being rejected as failed.'
+                    type: string
+              ref:
+                description: 'Reference to a Kafka resource which can be mapped to a specific
+                    ConsumerGroup, such as a Subscription or Trigger. This open type allows various
+                    implementations (Channels, Brokers, etc) to support the ResetOffset CRD without
+                    changing the schema. Each such implementation is responsible for validating and
+                    rejecting unsupported referenced types.  For example, the KafkaChannel Controller
+                    will only support Subscriptions and will reject any other reference type, while
+                    the KafkaBroker implementation might do the same for Triggers.  Check your
+                    specific use case to determine what resource types are supported! There is no
+                    default value, and invalid values will result in the ResetOffset operation being
+                    rejected as failed.'
+                type: object
+                properties:
+                  apiVersion:
+                    description: 'API version of the referent.'
+                    type: string
+                  kind:
+                    description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                    type: string
+                  name:
+                    description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                    type: string
+                  namespace:
+                    description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/'
+                    type: string
+          status:
+            description: "Status (computed) for a ResetOffset"
+            type: object
+            properties:
+              topic:
+                description: 'The Kafka Topic name associated with the specified Spec.Ref instance.'
+                type: string
+              group:
+                description: 'The Kafka ConsumerGroup ID associated with the specified Spec.Ref instance.'
+                type: string
+              partitions:
+                description: 'The Offset information for each Kafka Partition associated with the specified Spec.Ref instance.'
+                type: array
+                items:
+                  type: object
+                  properties:
+                    partition:
+                      description: 'The Partition number for the associated Topic / ConsumerGroup.'
+                      type: integer
+                    oldOffset:
+                      description: 'The current Offset of the Kafka Partition prior to being reset.'
+                      type: integer
+                    newOffset:
+                      description: 'The new Offset to which the Kafka Partition will be reset.'
+                      type: integer
+              annotations:
+                description: 'Annotations is additional Status fields for the Resource to save some
+                    additional State as well as convey more information to the user. This is roughly
+                    akin to Annotations on any k8s resource, just the reconciler conveying richer
+                    information outwards.'
+                type: object
+                x-kubernetes-preserve-unknown-fields: true
+              conditions:
+                description: 'Conditions is the latest available observations of a resource''s current state.'
+                type: array
+                items:
+                  type: object
+                  properties:
+                    lastTransitionTime:
+                      description: 'LastTransitionTime is the last time the condition transitioned
+                          from one status to another. We use VolatileTime in place of metav1.Time
+                          to exclude this from creating equality.Semantic differences (all other
+                          things held constant).'
+                      type: string
+                    message:
+                      description: 'A human readable message indicating details about the transition.'
+                      type: string
+                    reason:
+                      description: 'The reason for the condition''s last transition.'
+                      type: string
+                    severity:
+                      description: 'Severity with which to treat failures of this type of condition.
+                          When this is not specified, it defaults to Error.'
+                      type: string
+                    status:
+                      description: 'Status of the condition, one of True, False, Unknown.'
+                      type: string
+                    type:
+                      description: 'Type of condition.'
+                      type: string
+              observedGeneration:
+                description: 'ObservedGeneration is the ''Generation'' of the Service that was last
+                    processed by the controller.'
+                type: integer
+                format: int64
+    additionalPrinterColumns:
+    - name: Succeeded
+      type: string
+      jsonPath: ".status.conditions[?(@.type==\"Succeeded\")].status"
+    - name: Reason
+      type: string
+      jsonPath: ".status.conditions[?(@.type==\"Succeeded\")].reason"
+    - name: Ref Namespace
+      type: string
+      jsonPath: ".spec.ref.namespace"
+      priority: 1
+    - name: Ref Name
+      type: string
+      jsonPath: ".spec.ref.name"
+      priority: 1
+    - name: Age
+      type: date
+      jsonPath: .metadata.creationTimestamp
+  names:
+    kind: ResetOffset
+    plural: resetoffsets
+    singular: resetoffset
+    categories:
+    - all
+    - knative
+    - eventing
+    - kafka
+    shortNames:
+    - ro
+  scope: Namespaced
diff --git a/hack/update-codegen.sh b/hack/update-codegen.sh
index b81854c8d..b07728618 100755
--- a/hack/update-codegen.sh
+++ b/hack/update-codegen.sh
@@ -33,7 +33,7 @@ group "Kubernetes Codegen"
 #                  instead of the $GOPATH directly. For normal projects this can be dropped.
 ${CODEGEN_PKG}/generate-groups.sh "deepcopy,client,informer,lister" \
 "knative.dev/eventing-kafka/pkg/client" "knative.dev/eventing-kafka/pkg/apis" \
-"sources:v1alpha1 sources:v1beta1 bindings:v1alpha1 bindings:v1beta1 messaging:v1beta1" \
+"sources:v1alpha1 sources:v1beta1 bindings:v1alpha1 bindings:v1beta1 messaging:v1beta1 kafka:v1alpha1" \
 --go-header-file ${REPO_ROOT_DIR}/hack/boilerplate.go.txt
 
 # Only deepcopy the Duck types, as they are not real resources.
@@ -48,7 +48,7 @@ group "Knative Codegen"
 # Knative Injection
 ${KNATIVE_CODEGEN_PKG}/hack/generate-knative.sh "injection" \
 "knative.dev/eventing-kafka/pkg/client" "knative.dev/eventing-kafka/pkg/apis" \
-"sources:v1alpha1 sources:v1beta1 bindings:v1alpha1 bindings:v1beta1 messaging:v1beta1" \
+"sources:v1alpha1 sources:v1beta1 bindings:v1alpha1 bindings:v1beta1 messaging:v1beta1 kafka:v1alpha1" \
 --go-header-file ${REPO_ROOT_DIR}/hack/boilerplate.go.txt
 
 group "Deepcopy Gen"
diff --git a/pkg/apis/kafka/register.go b/pkg/apis/kafka/register.go
new file mode 100644
index 000000000..5aabce5b6
--- /dev/null
+++ b/pkg/apis/kafka/register.go
@@ -0,0 +1,31 @@
+/*
+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 kafka
+
+import "k8s.io/apimachinery/pkg/runtime/schema"
+
+const (
+	GroupName = "kafka.eventing.knative.dev"
+)
+
+var (
+	// ResetOffsetsResource represents a ResetOffset
+	ResetOffsetsResource = schema.GroupResource{
+		Group:    GroupName,
+		Resource: "resetoffsets",
+	}
+)
diff --git a/pkg/apis/kafka/v1alpha1/doc.go b/pkg/apis/kafka/v1alpha1/doc.go
new file mode 100644
index 000000000..2ae154121
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/doc.go
@@ -0,0 +1,20 @@
+/*
+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 v1alpha1 is the v1alpha1 version of the API.
+// +k8s:deepcopy-gen=package
+// +groupName=kafka.eventing.knative.dev
+package v1alpha1
diff --git a/pkg/apis/kafka/v1alpha1/register.go b/pkg/apis/kafka/v1alpha1/register.go
new file mode 100644
index 000000000..c40ebfc8d
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/register.go
@@ -0,0 +1,53 @@
+/*
+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 v1alpha1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"knative.dev/eventing-kafka/pkg/apis/kafka"
+)
+
+// SchemeGroupVersion is group version used to register these objects
+var SchemeGroupVersion = schema.GroupVersion{Group: kafka.GroupName, Version: "v1alpha1"}
+
+// Kind takes an unqualified kind and returns back a Group qualified GroupKind
+func Kind(kind string) schema.GroupKind {
+	return SchemeGroupVersion.WithKind(kind).GroupKind()
+}
+
+// Resource takes an unqualified resource and returns a Group qualified GroupResource
+func Resource(resource string) schema.GroupResource {
+	return SchemeGroupVersion.WithResource(resource).GroupResource()
+}
+
+var (
+	SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)
+	AddToScheme   = SchemeBuilder.AddToScheme
+)
+
+// Adds the list of known types to Scheme.
+func addKnownTypes(scheme *runtime.Scheme) error {
+	scheme.AddKnownTypes(SchemeGroupVersion,
+		&ResetOffset{},
+		&ResetOffsetList{},
+	)
+	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
+	return nil
+}
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_defaults.go b/pkg/apis/kafka/v1alpha1/reset_offset_defaults.go
new file mode 100644
index 000000000..11eca5c84
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_defaults.go
@@ -0,0 +1,29 @@
+/*
+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 v1alpha1
+
+import (
+	"context"
+)
+
+func (ro *ResetOffset) SetDefaults(ctx context.Context) {
+	ro.Spec.SetDefaults(ctx)
+}
+
+func (ros *ResetOffsetSpec) SetDefaults(_ context.Context) {
+	// Currently no fields can be defaulted.
+}
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go
new file mode 100644
index 000000000..4455965ad
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle.go
@@ -0,0 +1,145 @@
+/*
+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 v1alpha1
+
+import (
+	"sync"
+
+	"knative.dev/pkg/apis"
+)
+
+var cs = apis.NewBatchConditionSet(
+	ResetOffsetConditionResetInitiated,
+	ResetOffsetConditionConsumerGroupsStopped,
+	ResetOffsetConditionOffsetsUpdated,
+	ResetOffsetConditionConsumerGroupsStarted)
+
+var condSetLock = sync.RWMutex{}
+
+const (
+	// ResetOffsetConditionSucceeded has status True when all sub-conditions below have been set to True.
+	ResetOffsetConditionSucceeded = apis.ConditionSucceeded
+
+	// 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.
+	ResetOffsetConditionResetInitiated apis.ConditionType = "ResetInitiated"
+
+	// ResetOffsetConditionConsumerGroupsStopped has status True when all of the ConsumerGroups
+	// associated with the referenced object (Subscription, Trigger, etc.) have been stopped.
+	ResetOffsetConditionConsumerGroupsStopped apis.ConditionType = "ConsumerGroupsStopped"
+
+	// ResetOffsetConditionOffsetsUpdated has status True when all of the individual offsets
+	// of each Partition in the Topic have been updated to their new values.
+	ResetOffsetConditionOffsetsUpdated apis.ConditionType = "OffsetsUpdated"
+
+	// ResetOffsetConditionConsumerGroupsStarted has status True when all of the ConsumerGroups
+	// associated with the referenced object (Subscription, Trigger, etc.) have been restarted.
+	ResetOffsetConditionConsumerGroupsStarted apis.ConditionType = "ConsumerGroupsStarted"
+)
+
+// RegisterAlternateResetOffsetConditionSet register a different apis.ConditionSet.
+func RegisterAlternateResetOffsetConditionSet(conditionSet apis.ConditionSet) {
+	condSetLock.Lock()
+	defer condSetLock.Unlock()
+	cs = conditionSet
+}
+
+// GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
+func (*ResetOffset) GetConditionSet() apis.ConditionSet {
+	condSetLock.RLock()
+	defer condSetLock.RUnlock()
+	return cs
+}
+
+// GetConditionSet retrieves the condition set for this resource.
+func (*ResetOffsetStatus) GetConditionSet() apis.ConditionSet {
+	condSetLock.RLock()
+	defer condSetLock.RUnlock()
+	return cs
+}
+
+// GetCondition returns the condition currently associated with the given type, or nil.
+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.
+func (ros *ResetOffsetStatus) IsCompleted() bool {
+	return ros.GetConditionSet().Manage(ros).IsHappy()
+}
+
+// InitializeConditions sets relevant unset conditions to Unknown state.
+func (ros *ResetOffsetStatus) InitializeConditions() {
+	ros.GetConditionSet().Manage(ros).InitializeConditions()
+}
+
+func (ros *ResetOffsetStatus) MarkResetInitiatedFailed(reason, messageFormat string, messageA ...interface{}) {
+	ros.GetConditionSet().Manage(ros).MarkFalse(ResetOffsetConditionResetInitiated, reason, messageFormat, messageA...)
+}
+
+func (ros *ResetOffsetStatus) MarkResetInitiatedTrue() {
+	ros.GetConditionSet().Manage(ros).MarkTrue(ResetOffsetConditionResetInitiated)
+}
+
+func (ros *ResetOffsetStatus) MarkConsumerGroupsStoppedFailed(reason, messageFormat string, messageA ...interface{}) {
+	ros.GetConditionSet().Manage(ros).MarkFalse(ResetOffsetConditionConsumerGroupsStopped, reason, messageFormat, messageA...)
+}
+
+func (ros *ResetOffsetStatus) MarkConsumerGroupsStoppedTrue() {
+	ros.GetConditionSet().Manage(ros).MarkTrue(ResetOffsetConditionConsumerGroupsStopped)
+}
+
+func (ros *ResetOffsetStatus) MarkOffsetsUpdatedFailed(reason, messageFormat string, messageA ...interface{}) {
+	ros.GetConditionSet().Manage(ros).MarkFalse(ResetOffsetConditionOffsetsUpdated, reason, messageFormat, messageA...)
+}
+
+func (ros *ResetOffsetStatus) MarkOffsetsUpdatedTrue() {
+	ros.GetConditionSet().Manage(ros).MarkTrue(ResetOffsetConditionOffsetsUpdated)
+}
+
+func (ros *ResetOffsetStatus) MarkConsumerGroupsStartedFailed(reason, messageFormat string, messageA ...interface{}) {
+	ros.GetConditionSet().Manage(ros).MarkFalse(ResetOffsetConditionConsumerGroupsStarted, reason, messageFormat, messageA...)
+}
+
+func (ros *ResetOffsetStatus) MarkConsumerGroupsStartedTrue() {
+	ros.GetConditionSet().Manage(ros).MarkTrue(ResetOffsetConditionConsumerGroupsStarted)
+}
+
+func (ros *ResetOffsetStatus) GetTopic() string {
+	return ros.Topic
+}
+
+func (ros *ResetOffsetStatus) SetTopic(topic string) {
+	ros.Topic = topic
+}
+
+func (ros *ResetOffsetStatus) GetGroup() string {
+	return ros.Group
+}
+
+func (ros *ResetOffsetStatus) SetGroup(group string) {
+	ros.Group = group
+}
+
+func (ros *ResetOffsetStatus) GetPartitions() []OffsetMapping {
+	return ros.Partitions
+}
+
+func (ros *ResetOffsetStatus) SetPartitions(offsetMappings []OffsetMapping) {
+	ros.Partitions = offsetMappings
+}
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_types.go b/pkg/apis/kafka/v1alpha1/reset_offset_types.go
new file mode 100644
index 000000000..b74ede068
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_types.go
@@ -0,0 +1,182 @@
+/*
+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 v1alpha1
+
+import (
+	"time"
+
+	"github.com/Shopify/sarama"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
+)
+
+const (
+	OffsetEarliest = "earliest"
+	OffsetLatest   = "latest"
+)
+
+// +genclient
+// +genreconciler
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// ResetOffset is a resource representing a "command" for re-positioning the
+// offsets of a specific Kafka resource (Subscription, Trigger, etc.)
+type ResetOffset struct {
+	metav1.TypeMeta `json:",inline"`
+	// +optional
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	// Spec defines the desired state of the ResetOffset.
+	Spec ResetOffsetSpec `json:"spec,omitempty"`
+
+	// Status represents the current state of the ResetOffset.
+	// This data may be out of date.
+	// +optional
+	Status ResetOffsetStatus `json:"status,omitempty"`
+}
+
+var (
+	// Check that this resource can be validated and defaulted.
+	_ apis.Validatable = (*ResetOffset)(nil)
+	_ apis.Defaultable = (*ResetOffset)(nil)
+
+	_ runtime.Object = (*ResetOffset)(nil)
+
+	// Check that we can create OwnerReferences to an this resource.
+	_ kmeta.OwnerRefable = (*ResetOffset)(nil)
+
+	// Check that the type conforms to the duck Knative Resource shape.
+	_ duckv1.KRShaped = (*ResetOffset)(nil)
+)
+
+// ResetOffsetSpec defines the specification for a ResetOffset.
+type ResetOffsetSpec struct {
+
+	// Offset is an object representing the desired offset position to which all partitions
+	// will be reset. It provides for future extensibility in supporting various types of
+	// offset information (time based, explicit offset numbers, etc.)
+	Offset OffsetSpec `json:"offset"`
+
+	// Ref is a KReference specifying the Knative resource, related to a Kafka ConsumerGroup,
+	// whose partitions offsets will be reset (e.g. Subscription, Trigger, etc.)  The referenced
+	// object MUST be related to a Kafka Topic in such a way that it's specific partitions
+	// can be identified.  Thus, even though the KReference is a wide-open type, it is up
+	// to the user to provide an appropriate value as supported by the Controller in question
+	// (KafkaChannel vs KafkaBroker, etc).  Failure to provide a valid value will result in
+	// the ResetOffset operation being rejected as failed.
+	Ref duckv1.KReference `json:"ref"`
+}
+
+// OffsetSpec defines the intended values to move the offsets to.
+// Note: This simple wrapper might seem unnecessary, but is provided to allow future extension
+//       in order to support specifying explicit offset (int64) values for each Partition.
+type OffsetSpec struct {
+
+	// Time is an string representing the desired offset position to which all partitions
+	// will be reset.  Supported values include "earliest", "latest", or a valid date / time
+	// string in the time.RFC3339 format. The "earliest" and "latest" values indicate the
+	// beginning and end, respectively, of the persistence window of the Topic.  There is no
+	// default value, and invalid values will result in the ResetOffset operation being
+	// rejected as failed.
+	Time string `json:"time"`
+}
+
+// IsOffsetEarliest returns True if the Offset value is "earliest"
+func (ros *ResetOffsetSpec) IsOffsetEarliest() bool {
+	return ros.Offset.Time == OffsetEarliest
+}
+
+// IsOffsetLatest returns True if the Offset value is "latest"
+func (ros *ResetOffsetSpec) IsOffsetLatest() bool {
+	return ros.Offset.Time == OffsetLatest
+}
+
+// ParseOffsetTime returns the parsed Offset Time if valid (RFC3339 format) or an error for invalid content.
+func (ros *ResetOffsetSpec) ParseOffsetTime() (time.Time, error) {
+	return time.Parse(time.RFC3339, ros.Offset.Time)
+}
+
+// ParseSaramaOffsetTime returns the Sarama Offset Time (millis since epoch) int64 as parsed from the specified ResetOffset.Spec
+func (ros *ResetOffsetSpec) ParseSaramaOffsetTime() (int64, error) {
+	var saramaOffsetTime int64
+	if ros.IsOffsetEarliest() {
+		saramaOffsetTime = sarama.OffsetOldest
+	} else if ros.IsOffsetLatest() {
+		saramaOffsetTime = sarama.OffsetNewest
+	} else {
+		offsetTime, err := ros.ParseOffsetTime()
+		if err != nil {
+			return 0, err
+		}
+		saramaOffsetTime = offsetTime.UnixNano() / 1000000 // Convert Nanos To Millis For Sarama
+	}
+	return saramaOffsetTime, nil
+}
+
+// ResetOffsetStatus represents the current state of a ResetOffset.
+type ResetOffsetStatus struct {
+
+	// Topic is a string representing the Kafka Topic name associated with the ResetOffsetSpec.Ref
+	// +optional
+	Topic string `json:"topic,omitempty"`
+
+	// Group is a string representing the Kafka ConsumerGroup ID associated with the ResetOffsetSpec.Ref
+	// +optional
+	Group string `json:"group,omitempty"`
+
+	// Partitions is an array of OffsetMapping structs which represent the Offsets (old / new) of
+	// all Kafka Partitions associated with the ResetOffsetSpec.Ref
+	// +optional
+	Partitions []OffsetMapping `json:"partitions,omitempty"`
+
+	// inherits duck/v1 Status, which currently provides:
+	// * ObservedGeneration - the 'Generation' of the Service that was last processed by the controller.
+	// * Conditions - the latest available observations of a resource's current state.
+	// * Annotations - optional status information to be conveyed to users.
+	duckv1.Status `json:",inline"`
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// ResetOffsetList is a collection of ResetOffsets.
+type ResetOffsetList struct {
+	metav1.TypeMeta `json:",inline"`
+	// +optional
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []ResetOffset `json:"items"`
+}
+
+// GetGroupVersionKind returns GroupVersionKind for ResetOffset
+func (ro *ResetOffset) GetGroupVersionKind() schema.GroupVersionKind {
+	return SchemeGroupVersion.WithKind("ResetOffset")
+}
+
+// GetStatus retrieves the duck status for this resource. Implements the KRShaped interface.
+func (ro *ResetOffset) GetStatus() *duckv1.Status {
+	return &ro.Status.Status
+}
+
+// OffsetMapping represents a single Kafka Partition's Offset values before and after repositioning.
+type OffsetMapping struct {
+	Partition int32 `json:"partition"`
+	OldOffset int64 `json:"oldOffset"`
+	NewOffset int64 `json:"newOffset"`
+}
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_validation.go b/pkg/apis/kafka/v1alpha1/reset_offset_validation.go
new file mode 100644
index 000000000..f269163e9
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_validation.go
@@ -0,0 +1,48 @@
+/*
+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 v1alpha1
+
+import (
+	"context"
+	"time"
+
+	"knative.dev/pkg/apis"
+)
+
+// Validate verifies the ResetOffset and returns errors for any invalid fields.
+func (ro *ResetOffset) Validate(ctx context.Context) *apis.FieldError {
+	return ro.Spec.Validate(ctx).ViaField("spec")
+}
+
+// Validate verifies the ResetOffsetSpec and returns errors for an invalid fields.
+func (ros *ResetOffsetSpec) Validate(ctx context.Context) *apis.FieldError {
+
+	var errs *apis.FieldError
+
+	// Validate The Offset String ("earliest", "latest", or valid date string)
+	if !ros.IsOffsetEarliest() && !ros.IsOffsetLatest() {
+		offsetTime, err := ros.ParseOffsetTime()
+		if err != nil || offsetTime.After(time.Now()) {
+			errs = errs.Also(apis.ErrInvalidValue(ros.Offset.Time, "offset"))
+		}
+	}
+
+	// Validate The Ref KReference Basics (Kafka Topic relation which is expected to be done in Controllers!)
+	errs = errs.Also(ros.Ref.Validate(ctx))
+
+	return errs
+}
diff --git a/pkg/apis/kafka/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/kafka/v1alpha1/zz_generated.deepcopy.go
new file mode 100644
index 000000000..3b9dfc800
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/zz_generated.deepcopy.go
@@ -0,0 +1,158 @@
+// +build !ignore_autogenerated
+
+/*
+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 deepcopy-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	runtime "k8s.io/apimachinery/pkg/runtime"
+)
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *OffsetMapping) DeepCopyInto(out *OffsetMapping) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OffsetMapping.
+func (in *OffsetMapping) DeepCopy() *OffsetMapping {
+	if in == nil {
+		return nil
+	}
+	out := new(OffsetMapping)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *OffsetSpec) DeepCopyInto(out *OffsetSpec) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OffsetSpec.
+func (in *OffsetSpec) DeepCopy() *OffsetSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(OffsetSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ResetOffset) DeepCopyInto(out *ResetOffset) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	out.Spec = in.Spec
+	in.Status.DeepCopyInto(&out.Status)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResetOffset.
+func (in *ResetOffset) DeepCopy() *ResetOffset {
+	if in == nil {
+		return nil
+	}
+	out := new(ResetOffset)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *ResetOffset) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ResetOffsetList) DeepCopyInto(out *ResetOffsetList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]ResetOffset, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResetOffsetList.
+func (in *ResetOffsetList) DeepCopy() *ResetOffsetList {
+	if in == nil {
+		return nil
+	}
+	out := new(ResetOffsetList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *ResetOffsetList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ResetOffsetSpec) DeepCopyInto(out *ResetOffsetSpec) {
+	*out = *in
+	out.Offset = in.Offset
+	out.Ref = in.Ref
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResetOffsetSpec.
+func (in *ResetOffsetSpec) DeepCopy() *ResetOffsetSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(ResetOffsetSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ResetOffsetStatus) DeepCopyInto(out *ResetOffsetStatus) {
+	*out = *in
+	if in.Partitions != nil {
+		in, out := &in.Partitions, &out.Partitions
+		*out = make([]OffsetMapping, len(*in))
+		copy(*out, *in)
+	}
+	in.Status.DeepCopyInto(&out.Status)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ResetOffsetStatus.
+func (in *ResetOffsetStatus) DeepCopy() *ResetOffsetStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(ResetOffsetStatus)
+	in.DeepCopyInto(out)
+	return out
+}
diff --git a/pkg/client/clientset/versioned/clientset.go b/pkg/client/clientset/versioned/clientset.go
index aed42177e..77d85f60a 100644
--- a/pkg/client/clientset/versioned/clientset.go
+++ b/pkg/client/clientset/versioned/clientset.go
@@ -26,6 +26,7 @@ import (
 	flowcontrol "k8s.io/client-go/util/flowcontrol"
 	bindingsv1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1alpha1"
 	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1"
 	sourcesv1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1alpha1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1"
@@ -35,6 +36,7 @@ type Interface interface {
 	Discovery() discovery.DiscoveryInterface
 	BindingsV1alpha1() bindingsv1alpha1.BindingsV1alpha1Interface
 	BindingsV1beta1() bindingsv1beta1.BindingsV1beta1Interface
+	KafkaV1alpha1() kafkav1alpha1.KafkaV1alpha1Interface
 	MessagingV1beta1() messagingv1beta1.MessagingV1beta1Interface
 	SourcesV1alpha1() sourcesv1alpha1.SourcesV1alpha1Interface
 	SourcesV1beta1() sourcesv1beta1.SourcesV1beta1Interface
@@ -46,6 +48,7 @@ type Clientset struct {
 	*discovery.DiscoveryClient
 	bindingsV1alpha1 *bindingsv1alpha1.BindingsV1alpha1Client
 	bindingsV1beta1  *bindingsv1beta1.BindingsV1beta1Client
+	kafkaV1alpha1    *kafkav1alpha1.KafkaV1alpha1Client
 	messagingV1beta1 *messagingv1beta1.MessagingV1beta1Client
 	sourcesV1alpha1  *sourcesv1alpha1.SourcesV1alpha1Client
 	sourcesV1beta1   *sourcesv1beta1.SourcesV1beta1Client
@@ -61,6 +64,11 @@ func (c *Clientset) BindingsV1beta1() bindingsv1beta1.BindingsV1beta1Interface {
 	return c.bindingsV1beta1
 }
 
+// KafkaV1alpha1 retrieves the KafkaV1alpha1Client
+func (c *Clientset) KafkaV1alpha1() kafkav1alpha1.KafkaV1alpha1Interface {
+	return c.kafkaV1alpha1
+}
+
 // MessagingV1beta1 retrieves the MessagingV1beta1Client
 func (c *Clientset) MessagingV1beta1() messagingv1beta1.MessagingV1beta1Interface {
 	return c.messagingV1beta1
@@ -105,6 +113,10 @@ func NewForConfig(c *rest.Config) (*Clientset, error) {
 	if err != nil {
 		return nil, err
 	}
+	cs.kafkaV1alpha1, err = kafkav1alpha1.NewForConfig(&configShallowCopy)
+	if err != nil {
+		return nil, err
+	}
 	cs.messagingV1beta1, err = messagingv1beta1.NewForConfig(&configShallowCopy)
 	if err != nil {
 		return nil, err
@@ -131,6 +143,7 @@ func NewForConfigOrDie(c *rest.Config) *Clientset {
 	var cs Clientset
 	cs.bindingsV1alpha1 = bindingsv1alpha1.NewForConfigOrDie(c)
 	cs.bindingsV1beta1 = bindingsv1beta1.NewForConfigOrDie(c)
+	cs.kafkaV1alpha1 = kafkav1alpha1.NewForConfigOrDie(c)
 	cs.messagingV1beta1 = messagingv1beta1.NewForConfigOrDie(c)
 	cs.sourcesV1alpha1 = sourcesv1alpha1.NewForConfigOrDie(c)
 	cs.sourcesV1beta1 = sourcesv1beta1.NewForConfigOrDie(c)
@@ -144,6 +157,7 @@ func New(c rest.Interface) *Clientset {
 	var cs Clientset
 	cs.bindingsV1alpha1 = bindingsv1alpha1.New(c)
 	cs.bindingsV1beta1 = bindingsv1beta1.New(c)
+	cs.kafkaV1alpha1 = kafkav1alpha1.New(c)
 	cs.messagingV1beta1 = messagingv1beta1.New(c)
 	cs.sourcesV1alpha1 = sourcesv1alpha1.New(c)
 	cs.sourcesV1beta1 = sourcesv1beta1.New(c)
diff --git a/pkg/client/clientset/versioned/fake/clientset_generated.go b/pkg/client/clientset/versioned/fake/clientset_generated.go
index 1072a3839..ae31ee9dc 100644
--- a/pkg/client/clientset/versioned/fake/clientset_generated.go
+++ b/pkg/client/clientset/versioned/fake/clientset_generated.go
@@ -29,6 +29,8 @@ import (
 	fakebindingsv1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1alpha1/fake"
 	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1"
 	fakebindingsv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1"
+	fakekafkav1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1"
 	fakemessagingv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake"
 	sourcesv1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1alpha1"
@@ -94,6 +96,11 @@ func (c *Clientset) BindingsV1beta1() bindingsv1beta1.BindingsV1beta1Interface {
 	return &fakebindingsv1beta1.FakeBindingsV1beta1{Fake: &c.Fake}
 }
 
+// KafkaV1alpha1 retrieves the KafkaV1alpha1Client
+func (c *Clientset) KafkaV1alpha1() kafkav1alpha1.KafkaV1alpha1Interface {
+	return &fakekafkav1alpha1.FakeKafkaV1alpha1{Fake: &c.Fake}
+}
+
 // MessagingV1beta1 retrieves the MessagingV1beta1Client
 func (c *Clientset) MessagingV1beta1() messagingv1beta1.MessagingV1beta1Interface {
 	return &fakemessagingv1beta1.FakeMessagingV1beta1{Fake: &c.Fake}
diff --git a/pkg/client/clientset/versioned/fake/register.go b/pkg/client/clientset/versioned/fake/register.go
index 7d289c44d..95d7bff1d 100644
--- a/pkg/client/clientset/versioned/fake/register.go
+++ b/pkg/client/clientset/versioned/fake/register.go
@@ -26,6 +26,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	bindingsv1alpha1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1alpha1"
 	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	sourcesv1alpha1 "knative.dev/eventing-kafka/pkg/apis/sources/v1alpha1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
@@ -37,6 +38,7 @@ var codecs = serializer.NewCodecFactory(scheme)
 var localSchemeBuilder = runtime.SchemeBuilder{
 	bindingsv1alpha1.AddToScheme,
 	bindingsv1beta1.AddToScheme,
+	kafkav1alpha1.AddToScheme,
 	messagingv1beta1.AddToScheme,
 	sourcesv1alpha1.AddToScheme,
 	sourcesv1beta1.AddToScheme,
diff --git a/pkg/client/clientset/versioned/scheme/register.go b/pkg/client/clientset/versioned/scheme/register.go
index af3f01a4c..a3628c477 100644
--- a/pkg/client/clientset/versioned/scheme/register.go
+++ b/pkg/client/clientset/versioned/scheme/register.go
@@ -26,6 +26,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	bindingsv1alpha1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1alpha1"
 	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	sourcesv1alpha1 "knative.dev/eventing-kafka/pkg/apis/sources/v1alpha1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
@@ -37,6 +38,7 @@ var ParameterCodec = runtime.NewParameterCodec(Scheme)
 var localSchemeBuilder = runtime.SchemeBuilder{
 	bindingsv1alpha1.AddToScheme,
 	bindingsv1beta1.AddToScheme,
+	kafkav1alpha1.AddToScheme,
 	messagingv1beta1.AddToScheme,
 	sourcesv1alpha1.AddToScheme,
 	sourcesv1beta1.AddToScheme,
diff --git a/pkg/client/clientset/versioned/typed/kafka/v1alpha1/doc.go b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/doc.go
new file mode 100644
index 000000000..f905d3a21
--- /dev/null
+++ b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/doc.go
@@ -0,0 +1,20 @@
+/*
+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 client-gen. DO NOT EDIT.
+
+// This package has the automatically generated typed clients.
+package v1alpha1
diff --git a/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+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 client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.go b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.go
new file mode 100644
index 000000000..53377b6b9
--- /dev/null
+++ b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.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 client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1"
+)
+
+type FakeKafkaV1alpha1 struct {
+	*testing.Fake
+}
+
+func (c *FakeKafkaV1alpha1) ResetOffsets(namespace string) v1alpha1.ResetOffsetInterface {
+	return &FakeResetOffsets{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeKafkaV1alpha1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go
new file mode 100644
index 000000000..af76f8c0b
--- /dev/null
+++ b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go
@@ -0,0 +1,142 @@
+/*
+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 client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+// FakeResetOffsets implements ResetOffsetInterface
+type FakeResetOffsets struct {
+	Fake *FakeKafkaV1alpha1
+	ns   string
+}
+
+var resetoffsetsResource = schema.GroupVersionResource{Group: "kafka.eventing.knative.dev", Version: "v1alpha1", Resource: "resetoffsets"}
+
+var resetoffsetsKind = schema.GroupVersionKind{Group: "kafka.eventing.knative.dev", Version: "v1alpha1", Kind: "ResetOffset"}
+
+// Get takes name of the resetOffset, and returns the corresponding resetOffset object, and an error if there is any.
+func (c *FakeResetOffsets) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(resetoffsetsResource, c.ns, name), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// List takes label and field selectors, and returns the list of ResetOffsets that match those selectors.
+func (c *FakeResetOffsets) List(ctx context.Context, opts v1.ListOptions) (result *v1alpha1.ResetOffsetList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(resetoffsetsResource, resetoffsetsKind, c.ns, opts), &v1alpha1.ResetOffsetList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1alpha1.ResetOffsetList{ListMeta: obj.(*v1alpha1.ResetOffsetList).ListMeta}
+	for _, item := range obj.(*v1alpha1.ResetOffsetList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested resetOffsets.
+func (c *FakeResetOffsets) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(resetoffsetsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a resetOffset and creates it.  Returns the server's representation of the resetOffset, and an error, if there is any.
+func (c *FakeResetOffsets) Create(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.CreateOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(resetoffsetsResource, c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// Update takes the representation of a resetOffset and updates it. Returns the server's representation of the resetOffset, and an error, if there is any.
+func (c *FakeResetOffsets) Update(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(resetoffsetsResource, c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeResetOffsets) UpdateStatus(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (*v1alpha1.ResetOffset, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(resetoffsetsResource, "status", c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// Delete takes name of the resetOffset and deletes it. Returns an error if one occurs.
+func (c *FakeResetOffsets) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(resetoffsetsResource, c.ns, name), &v1alpha1.ResetOffset{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeResetOffsets) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(resetoffsetsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1alpha1.ResetOffsetList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched resetOffset.
+func (c *FakeResetOffsets) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(resetoffsetsResource, c.ns, name, pt, data, subresources...), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
diff --git a/pkg/client/clientset/versioned/typed/kafka/v1alpha1/generated_expansion.go b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/generated_expansion.go
new file mode 100644
index 000000000..7431bce90
--- /dev/null
+++ b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/generated_expansion.go
@@ -0,0 +1,21 @@
+/*
+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 client-gen. DO NOT EDIT.
+
+package v1alpha1
+
+type ResetOffsetExpansion interface{}
diff --git a/pkg/client/clientset/versioned/typed/kafka/v1alpha1/kafka_client.go b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/kafka_client.go
new file mode 100644
index 000000000..9cd6948ba
--- /dev/null
+++ b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/kafka_client.go
@@ -0,0 +1,89 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	rest "k8s.io/client-go/rest"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
+)
+
+type KafkaV1alpha1Interface interface {
+	RESTClient() rest.Interface
+	ResetOffsetsGetter
+}
+
+// KafkaV1alpha1Client is used to interact with features provided by the kafka.eventing.knative.dev group.
+type KafkaV1alpha1Client struct {
+	restClient rest.Interface
+}
+
+func (c *KafkaV1alpha1Client) ResetOffsets(namespace string) ResetOffsetInterface {
+	return newResetOffsets(c, namespace)
+}
+
+// NewForConfig creates a new KafkaV1alpha1Client for the given config.
+func NewForConfig(c *rest.Config) (*KafkaV1alpha1Client, error) {
+	config := *c
+	if err := setConfigDefaults(&config); err != nil {
+		return nil, err
+	}
+	client, err := rest.RESTClientFor(&config)
+	if err != nil {
+		return nil, err
+	}
+	return &KafkaV1alpha1Client{client}, nil
+}
+
+// NewForConfigOrDie creates a new KafkaV1alpha1Client for the given config and
+// panics if there is an error in the config.
+func NewForConfigOrDie(c *rest.Config) *KafkaV1alpha1Client {
+	client, err := NewForConfig(c)
+	if err != nil {
+		panic(err)
+	}
+	return client
+}
+
+// New creates a new KafkaV1alpha1Client for the given RESTClient.
+func New(c rest.Interface) *KafkaV1alpha1Client {
+	return &KafkaV1alpha1Client{c}
+}
+
+func setConfigDefaults(config *rest.Config) error {
+	gv := v1alpha1.SchemeGroupVersion
+	config.GroupVersion = &gv
+	config.APIPath = "/apis"
+	config.NegotiatedSerializer = scheme.Codecs.WithoutConversion()
+
+	if config.UserAgent == "" {
+		config.UserAgent = rest.DefaultKubernetesUserAgent()
+	}
+
+	return nil
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *KafkaV1alpha1Client) RESTClient() rest.Interface {
+	if c == nil {
+		return nil
+	}
+	return c.restClient
+}
diff --git a/pkg/client/clientset/versioned/typed/kafka/v1alpha1/resetoffset.go b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/resetoffset.go
new file mode 100644
index 000000000..c4072de43
--- /dev/null
+++ b/pkg/client/clientset/versioned/typed/kafka/v1alpha1/resetoffset.go
@@ -0,0 +1,195 @@
+/*
+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 client-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"context"
+	"time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	rest "k8s.io/client-go/rest"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	scheme "knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
+)
+
+// ResetOffsetsGetter has a method to return a ResetOffsetInterface.
+// A group's client should implement this interface.
+type ResetOffsetsGetter interface {
+	ResetOffsets(namespace string) ResetOffsetInterface
+}
+
+// ResetOffsetInterface has methods to work with ResetOffset resources.
+type ResetOffsetInterface interface {
+	Create(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.CreateOptions) (*v1alpha1.ResetOffset, error)
+	Update(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (*v1alpha1.ResetOffset, error)
+	UpdateStatus(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (*v1alpha1.ResetOffset, error)
+	Delete(ctx context.Context, name string, opts v1.DeleteOptions) error
+	DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error
+	Get(ctx context.Context, name string, opts v1.GetOptions) (*v1alpha1.ResetOffset, error)
+	List(ctx context.Context, opts v1.ListOptions) (*v1alpha1.ResetOffsetList, error)
+	Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error)
+	Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1alpha1.ResetOffset, err error)
+	ResetOffsetExpansion
+}
+
+// resetOffsets implements ResetOffsetInterface
+type resetOffsets struct {
+	client rest.Interface
+	ns     string
+}
+
+// newResetOffsets returns a ResetOffsets
+func newResetOffsets(c *KafkaV1alpha1Client, namespace string) *resetOffsets {
+	return &resetOffsets{
+		client: c.RESTClient(),
+		ns:     namespace,
+	}
+}
+
+// Get takes name of the resetOffset, and returns the corresponding resetOffset object, and an error if there is any.
+func (c *resetOffsets) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1alpha1.ResetOffset, err error) {
+	result = &v1alpha1.ResetOffset{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		Name(name).
+		VersionedParams(&options, scheme.ParameterCodec).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// List takes label and field selectors, and returns the list of ResetOffsets that match those selectors.
+func (c *resetOffsets) List(ctx context.Context, opts v1.ListOptions) (result *v1alpha1.ResetOffsetList, err error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	result = &v1alpha1.ResetOffsetList{}
+	err = c.client.Get().
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Watch returns a watch.Interface that watches the requested resetOffsets.
+func (c *resetOffsets) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	var timeout time.Duration
+	if opts.TimeoutSeconds != nil {
+		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
+	}
+	opts.Watch = true
+	return c.client.Get().
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Watch(ctx)
+}
+
+// Create takes the representation of a resetOffset and creates it.  Returns the server's representation of the resetOffset, and an error, if there is any.
+func (c *resetOffsets) Create(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.CreateOptions) (result *v1alpha1.ResetOffset, err error) {
+	result = &v1alpha1.ResetOffset{}
+	err = c.client.Post().
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(resetOffset).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Update takes the representation of a resetOffset and updates it. Returns the server's representation of the resetOffset, and an error, if there is any.
+func (c *resetOffsets) Update(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (result *v1alpha1.ResetOffset, err error) {
+	result = &v1alpha1.ResetOffset{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		Name(resetOffset.Name).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(resetOffset).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *resetOffsets) UpdateStatus(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (result *v1alpha1.ResetOffset, err error) {
+	result = &v1alpha1.ResetOffset{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		Name(resetOffset.Name).
+		SubResource("status").
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(resetOffset).
+		Do(ctx).
+		Into(result)
+	return
+}
+
+// Delete takes name of the resetOffset and deletes it. Returns an error if one occurs.
+func (c *resetOffsets) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		Name(name).
+		Body(&opts).
+		Do(ctx).
+		Error()
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *resetOffsets) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	var timeout time.Duration
+	if listOpts.TimeoutSeconds != nil {
+		timeout = time.Duration(*listOpts.TimeoutSeconds) * time.Second
+	}
+	return c.client.Delete().
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		VersionedParams(&listOpts, scheme.ParameterCodec).
+		Timeout(timeout).
+		Body(&opts).
+		Do(ctx).
+		Error()
+}
+
+// Patch applies the patch and returns the patched resetOffset.
+func (c *resetOffsets) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1alpha1.ResetOffset, err error) {
+	result = &v1alpha1.ResetOffset{}
+	err = c.client.Patch(pt).
+		Namespace(c.ns).
+		Resource("resetoffsets").
+		Name(name).
+		SubResource(subresources...).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Body(data).
+		Do(ctx).
+		Into(result)
+	return
+}
diff --git a/pkg/client/informers/externalversions/factory.go b/pkg/client/informers/externalversions/factory.go
index 98e20f1a4..5634640fb 100644
--- a/pkg/client/informers/externalversions/factory.go
+++ b/pkg/client/informers/externalversions/factory.go
@@ -30,6 +30,7 @@ import (
 	versioned "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	bindings "knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings"
 	internalinterfaces "knative.dev/eventing-kafka/pkg/client/informers/externalversions/internalinterfaces"
+	kafka "knative.dev/eventing-kafka/pkg/client/informers/externalversions/kafka"
 	messaging "knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging"
 	sources "knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources"
 )
@@ -175,6 +176,7 @@ type SharedInformerFactory interface {
 	WaitForCacheSync(stopCh <-chan struct{}) map[reflect.Type]bool
 
 	Bindings() bindings.Interface
+	Kafka() kafka.Interface
 	Messaging() messaging.Interface
 	Sources() sources.Interface
 }
@@ -183,6 +185,10 @@ func (f *sharedInformerFactory) Bindings() bindings.Interface {
 	return bindings.New(f, f.namespace, f.tweakListOptions)
 }
 
+func (f *sharedInformerFactory) Kafka() kafka.Interface {
+	return kafka.New(f, f.namespace, f.tweakListOptions)
+}
+
 func (f *sharedInformerFactory) Messaging() messaging.Interface {
 	return messaging.New(f, f.namespace, f.tweakListOptions)
 }
diff --git a/pkg/client/informers/externalversions/generic.go b/pkg/client/informers/externalversions/generic.go
index 661cdf708..05db38e8d 100644
--- a/pkg/client/informers/externalversions/generic.go
+++ b/pkg/client/informers/externalversions/generic.go
@@ -25,6 +25,7 @@ import (
 	cache "k8s.io/client-go/tools/cache"
 	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1alpha1"
 	v1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	sourcesv1alpha1 "knative.dev/eventing-kafka/pkg/apis/sources/v1alpha1"
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
@@ -64,6 +65,10 @@ func (f *sharedInformerFactory) ForResource(resource schema.GroupVersionResource
 	case v1beta1.SchemeGroupVersion.WithResource("kafkabindings"):
 		return &genericInformer{resource: resource.GroupResource(), informer: f.Bindings().V1beta1().KafkaBindings().Informer()}, nil
 
+		// Group=kafka.eventing.knative.dev, Version=v1alpha1
+	case kafkav1alpha1.SchemeGroupVersion.WithResource("resetoffsets"):
+		return &genericInformer{resource: resource.GroupResource(), informer: f.Kafka().V1alpha1().ResetOffsets().Informer()}, nil
+
 		// Group=messaging.knative.dev, Version=v1beta1
 	case messagingv1beta1.SchemeGroupVersion.WithResource("kafkachannels"):
 		return &genericInformer{resource: resource.GroupResource(), informer: f.Messaging().V1beta1().KafkaChannels().Informer()}, nil
diff --git a/pkg/client/informers/externalversions/kafka/interface.go b/pkg/client/informers/externalversions/kafka/interface.go
new file mode 100644
index 000000000..11b43cf9c
--- /dev/null
+++ b/pkg/client/informers/externalversions/kafka/interface.go
@@ -0,0 +1,46 @@
+/*
+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 informer-gen. DO NOT EDIT.
+
+package kafka
+
+import (
+	internalinterfaces "knative.dev/eventing-kafka/pkg/client/informers/externalversions/internalinterfaces"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/kafka/v1alpha1"
+)
+
+// Interface provides access to each of this group's versions.
+type Interface interface {
+	// V1alpha1 provides access to shared informers for resources in V1alpha1.
+	V1alpha1() v1alpha1.Interface
+}
+
+type group struct {
+	factory          internalinterfaces.SharedInformerFactory
+	namespace        string
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+}
+
+// New returns a new Interface.
+func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) Interface {
+	return &group{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
+}
+
+// V1alpha1 returns a new v1alpha1.Interface.
+func (g *group) V1alpha1() v1alpha1.Interface {
+	return v1alpha1.New(g.factory, g.namespace, g.tweakListOptions)
+}
diff --git a/pkg/client/informers/externalversions/kafka/v1alpha1/interface.go b/pkg/client/informers/externalversions/kafka/v1alpha1/interface.go
new file mode 100644
index 000000000..2b422c1a2
--- /dev/null
+++ b/pkg/client/informers/externalversions/kafka/v1alpha1/interface.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 informer-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	internalinterfaces "knative.dev/eventing-kafka/pkg/client/informers/externalversions/internalinterfaces"
+)
+
+// Interface provides access to all the informers in this group version.
+type Interface interface {
+	// ResetOffsets returns a ResetOffsetInformer.
+	ResetOffsets() ResetOffsetInformer
+}
+
+type version struct {
+	factory          internalinterfaces.SharedInformerFactory
+	namespace        string
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+}
+
+// New returns a new Interface.
+func New(f internalinterfaces.SharedInformerFactory, namespace string, tweakListOptions internalinterfaces.TweakListOptionsFunc) Interface {
+	return &version{factory: f, namespace: namespace, tweakListOptions: tweakListOptions}
+}
+
+// ResetOffsets returns a ResetOffsetInformer.
+func (v *version) ResetOffsets() ResetOffsetInformer {
+	return &resetOffsetInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
+}
diff --git a/pkg/client/informers/externalversions/kafka/v1alpha1/resetoffset.go b/pkg/client/informers/externalversions/kafka/v1alpha1/resetoffset.go
new file mode 100644
index 000000000..c9a3f45c1
--- /dev/null
+++ b/pkg/client/informers/externalversions/kafka/v1alpha1/resetoffset.go
@@ -0,0 +1,90 @@
+/*
+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 informer-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"context"
+	time "time"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	cache "k8s.io/client-go/tools/cache"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	versioned "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	internalinterfaces "knative.dev/eventing-kafka/pkg/client/informers/externalversions/internalinterfaces"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/listers/kafka/v1alpha1"
+)
+
+// ResetOffsetInformer provides access to a shared informer and lister for
+// ResetOffsets.
+type ResetOffsetInformer interface {
+	Informer() cache.SharedIndexInformer
+	Lister() v1alpha1.ResetOffsetLister
+}
+
+type resetOffsetInformer struct {
+	factory          internalinterfaces.SharedInformerFactory
+	tweakListOptions internalinterfaces.TweakListOptionsFunc
+	namespace        string
+}
+
+// NewResetOffsetInformer constructs a new informer for ResetOffset type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewResetOffsetInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
+	return NewFilteredResetOffsetInformer(client, namespace, resyncPeriod, indexers, nil)
+}
+
+// NewFilteredResetOffsetInformer constructs a new informer for ResetOffset type.
+// Always prefer using an informer factory to get a shared informer instead of getting an independent
+// one. This reduces memory footprint and number of connections to the server.
+func NewFilteredResetOffsetInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KafkaV1alpha1().ResetOffsets(namespace).List(context.TODO(), options)
+			},
+			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
+				if tweakListOptions != nil {
+					tweakListOptions(&options)
+				}
+				return client.KafkaV1alpha1().ResetOffsets(namespace).Watch(context.TODO(), options)
+			},
+		},
+		&kafkav1alpha1.ResetOffset{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+func (f *resetOffsetInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
+	return NewFilteredResetOffsetInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions)
+}
+
+func (f *resetOffsetInformer) Informer() cache.SharedIndexInformer {
+	return f.factory.InformerFor(&kafkav1alpha1.ResetOffset{}, f.defaultInformer)
+}
+
+func (f *resetOffsetInformer) Lister() v1alpha1.ResetOffsetLister {
+	return v1alpha1.NewResetOffsetLister(f.Informer().GetIndexer())
+}
diff --git a/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/fake/fake.go b/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/fake/fake.go
new file mode 100644
index 000000000..c296b6b89
--- /dev/null
+++ b/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/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-kafka/pkg/client/injection/informers/factory/fake"
+	resetoffset "knative.dev/eventing-kafka/pkg/client/injection/informers/kafka/v1alpha1/resetoffset"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = resetoffset.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Kafka().V1alpha1().ResetOffsets()
+	return context.WithValue(ctx, resetoffset.Key{}, inf), inf.Informer()
+}
diff --git a/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/filtered/fake/fake.go b/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/filtered/fake/fake.go
new file mode 100644
index 000000000..ee5a8c500
--- /dev/null
+++ b/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/filtered/fake/fake.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 fake
+
+import (
+	context "context"
+
+	factoryfiltered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+var Get = filtered.Get
+
+func init() {
+	injection.Fake.RegisterFilteredInformers(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(factoryfiltered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := factoryfiltered.Get(ctx, selector)
+		inf := f.Kafka().V1alpha1().ResetOffsets()
+		ctx = context.WithValue(ctx, filtered.Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
diff --git a/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/filtered/resetoffset.go b/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/filtered/resetoffset.go
new file mode 100644
index 000000000..26cecd642
--- /dev/null
+++ b/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/filtered/resetoffset.go
@@ -0,0 +1,65 @@
+/*
+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 filtered
+
+import (
+	context "context"
+
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/kafka/v1alpha1"
+	filtered "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/filtered"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterFilteredInformers(withInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct {
+	Selector string
+}
+
+func withInformer(ctx context.Context) (context.Context, []controller.Informer) {
+	untyped := ctx.Value(filtered.LabelKey{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch labelkey from context.")
+	}
+	labelSelectors := untyped.([]string)
+	infs := []controller.Informer{}
+	for _, selector := range labelSelectors {
+		f := filtered.Get(ctx, selector)
+		inf := f.Kafka().V1alpha1().ResetOffsets()
+		ctx = context.WithValue(ctx, Key{Selector: selector}, inf)
+		infs = append(infs, inf.Informer())
+	}
+	return ctx, infs
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context, selector string) v1alpha1.ResetOffsetInformer {
+	untyped := ctx.Value(Key{Selector: selector})
+	if untyped == nil {
+		logging.FromContext(ctx).Panicf(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/kafka/v1alpha1.ResetOffsetInformer with selector %s from context.", selector)
+	}
+	return untyped.(v1alpha1.ResetOffsetInformer)
+}
diff --git a/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/resetoffset.go b/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/resetoffset.go
new file mode 100644
index 000000000..e3d6b65ed
--- /dev/null
+++ b/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/resetoffset.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 resetoffset
+
+import (
+	context "context"
+
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/kafka/v1alpha1"
+	factory "knative.dev/eventing-kafka/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.Kafka().V1alpha1().ResetOffsets()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1alpha1.ResetOffsetInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/kafka/v1alpha1.ResetOffsetInformer from context.")
+	}
+	return untyped.(v1alpha1.ResetOffsetInformer)
+}
diff --git a/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/controller.go b/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/controller.go
new file mode 100644
index 000000000..2dd3c6916
--- /dev/null
+++ b/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/controller.go
@@ -0,0 +1,153 @@
+/*
+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 resetoffset
+
+import (
+	context "context"
+	fmt "fmt"
+	reflect "reflect"
+	strings "strings"
+
+	zap "go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	scheme "k8s.io/client-go/kubernetes/scheme"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	record "k8s.io/client-go/tools/record"
+	versionedscheme "knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
+	client "knative.dev/eventing-kafka/pkg/client/injection/client"
+	resetoffset "knative.dev/eventing-kafka/pkg/client/injection/informers/kafka/v1alpha1/resetoffset"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	controller "knative.dev/pkg/controller"
+	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+const (
+	defaultControllerAgentName = "resetoffset-controller"
+	defaultFinalizerName       = "resetoffsets.kafka.eventing.knative.dev"
+)
+
+// NewImpl returns a controller.Impl that handles queuing and feeding work from
+// the queue through an implementation of controller.Reconciler, delegating to
+// the provided Interface and optional Finalizer methods. OptionsFn is used to return
+// controller.Options to be used by the internal reconciler.
+func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsFn) *controller.Impl {
+	logger := logging.FromContext(ctx)
+
+	// Check the options function input. It should be 0 or 1.
+	if len(optionsFns) > 1 {
+		logger.Fatal("Up to one options function is supported, found: ", len(optionsFns))
+	}
+
+	resetoffsetInformer := resetoffset.Get(ctx)
+
+	lister := resetoffsetInformer.Lister()
+
+	rec := &reconcilerImpl{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				all, err := lister.List(labels.Everything())
+				if err != nil {
+					return err
+				}
+				for _, elt := range all {
+					// TODO: Consider letting users specify a filter in options.
+					enq(bkt, types.NamespacedName{
+						Namespace: elt.GetNamespace(),
+						Name:      elt.GetName(),
+					})
+				}
+				return nil
+			},
+		},
+		Client:        client.Get(ctx),
+		Lister:        lister,
+		reconciler:    r,
+		finalizerName: defaultFinalizerName,
+	}
+
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
+
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "kafka.eventing.knative.dev.ResetOffset"),
+	)
+
+	impl := controller.NewImpl(rec, logger, ctrTypeName)
+	agentName := defaultControllerAgentName
+
+	// Pass impl to the options. Save any optional results.
+	for _, fn := range optionsFns {
+		opts := fn(impl)
+		if opts.ConfigStore != nil {
+			rec.configStore = opts.ConfigStore
+		}
+		if opts.FinalizerName != "" {
+			rec.finalizerName = opts.FinalizerName
+		}
+		if opts.AgentName != "" {
+			agentName = opts.AgentName
+		}
+		if opts.SkipStatusUpdates {
+			rec.skipStatusUpdates = true
+		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
+	}
+
+	rec.Recorder = createRecorder(ctx, agentName)
+
+	return impl
+}
+
+func createRecorder(ctx context.Context, agentName string) record.EventRecorder {
+	logger := logging.FromContext(ctx)
+
+	recorder := controller.GetEventRecorder(ctx)
+	if recorder == nil {
+		// Create event broadcaster
+		logger.Debug("Creating event broadcaster")
+		eventBroadcaster := record.NewBroadcaster()
+		watches := []watch.Interface{
+			eventBroadcaster.StartLogging(logger.Named("event-broadcaster").Infof),
+			eventBroadcaster.StartRecordingToSink(
+				&v1.EventSinkImpl{Interface: kubeclient.Get(ctx).CoreV1().Events("")}),
+		}
+		recorder = eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: agentName})
+		go func() {
+			<-ctx.Done()
+			for _, w := range watches {
+				w.Stop()
+			}
+		}()
+	}
+
+	return recorder
+}
+
+func init() {
+	versionedscheme.AddToScheme(scheme.Scheme)
+}
diff --git a/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/reconciler.go b/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/reconciler.go
new file mode 100644
index 000000000..30b02c3a8
--- /dev/null
+++ b/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/reconciler.go
@@ -0,0 +1,455 @@
+/*
+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 resetoffset
+
+import (
+	context "context"
+	json "encoding/json"
+	fmt "fmt"
+
+	zap "go.uber.org/zap"
+	v1 "k8s.io/api/core/v1"
+	equality "k8s.io/apimachinery/pkg/api/equality"
+	errors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	sets "k8s.io/apimachinery/pkg/util/sets"
+	record "k8s.io/client-go/tools/record"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	versioned "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/client/listers/kafka/v1alpha1"
+	controller "knative.dev/pkg/controller"
+	kmp "knative.dev/pkg/kmp"
+	logging "knative.dev/pkg/logging"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+// Interface defines the strongly typed interfaces to be implemented by a
+// controller reconciling v1alpha1.ResetOffset.
+type Interface interface {
+	// ReconcileKind implements custom logic to reconcile v1alpha1.ResetOffset. Any changes
+	// to the objects .Status or .Finalizers will be propagated to the stored
+	// object. It is recommended that implementors do not call any update calls
+	// for the Kind inside of ReconcileKind, it is the responsibility of the calling
+	// controller to propagate those properties. The resource passed to ReconcileKind
+	// will always have an empty deletion timestamp.
+	ReconcileKind(ctx context.Context, o *v1alpha1.ResetOffset) reconciler.Event
+}
+
+// Finalizer defines the strongly typed interfaces to be implemented by a
+// controller finalizing v1alpha1.ResetOffset.
+type Finalizer interface {
+	// FinalizeKind implements custom logic to finalize v1alpha1.ResetOffset. Any changes
+	// to the objects .Status or .Finalizers will be ignored. Returning a nil or
+	// Normal type reconciler.Event will allow the finalizer to be deleted on
+	// the resource. The resource passed to FinalizeKind will always have a set
+	// deletion timestamp.
+	FinalizeKind(ctx context.Context, o *v1alpha1.ResetOffset) reconciler.Event
+}
+
+// ReadOnlyInterface defines the strongly typed interfaces to be implemented by a
+// controller reconciling v1alpha1.ResetOffset if they want to process resources for which
+// they are not the leader.
+type ReadOnlyInterface interface {
+	// ObserveKind implements logic to observe v1alpha1.ResetOffset.
+	// This method should not write to the API.
+	ObserveKind(ctx context.Context, o *v1alpha1.ResetOffset) reconciler.Event
+}
+
+// ReadOnlyFinalizer defines the strongly typed interfaces to be implemented by a
+// controller finalizing v1alpha1.ResetOffset if they want to process tombstoned resources
+// even when they are not the leader.  Due to the nature of how finalizers are handled
+// there are no guarantees that this will be called.
+type ReadOnlyFinalizer interface {
+	// ObserveFinalizeKind implements custom logic to observe the final state of v1alpha1.ResetOffset.
+	// This method should not write to the API.
+	ObserveFinalizeKind(ctx context.Context, o *v1alpha1.ResetOffset) reconciler.Event
+}
+
+type doReconcile func(ctx context.Context, o *v1alpha1.ResetOffset) reconciler.Event
+
+// reconcilerImpl implements controller.Reconciler for v1alpha1.ResetOffset resources.
+type reconcilerImpl struct {
+	// LeaderAwareFuncs is inlined to help us implement reconciler.LeaderAware.
+	reconciler.LeaderAwareFuncs
+
+	// Client is used to write back status updates.
+	Client versioned.Interface
+
+	// Listers index properties about resources.
+	Lister kafkav1alpha1.ResetOffsetLister
+
+	// Recorder is an event recorder for recording Event resources to the
+	// Kubernetes API.
+	Recorder record.EventRecorder
+
+	// configStore allows for decorating a context with config maps.
+	// +optional
+	configStore reconciler.ConfigStore
+
+	// reconciler is the implementation of the business logic of the resource.
+	reconciler Interface
+
+	// finalizerName is the name of the finalizer to reconcile.
+	finalizerName string
+
+	// skipStatusUpdates configures whether or not this reconciler automatically updates
+	// the status of the reconciled resource.
+	skipStatusUpdates bool
+}
+
+// Check that our Reconciler implements controller.Reconciler.
+var _ controller.Reconciler = (*reconcilerImpl)(nil)
+
+// Check that our generated Reconciler is always LeaderAware.
+var _ reconciler.LeaderAware = (*reconcilerImpl)(nil)
+
+func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versioned.Interface, lister kafkav1alpha1.ResetOffsetLister, recorder record.EventRecorder, r Interface, options ...controller.Options) controller.Reconciler {
+	// Check the options function input. It should be 0 or 1.
+	if len(options) > 1 {
+		logger.Fatal("Up to one options struct is supported, found: ", len(options))
+	}
+
+	// Fail fast when users inadvertently implement the other LeaderAware interface.
+	// For the typed reconcilers, Promote shouldn't take any arguments.
+	if _, ok := r.(reconciler.LeaderAware); ok {
+		logger.Fatalf("%T implements the incorrect LeaderAware interface. Promote() should not take an argument as genreconciler handles the enqueuing automatically.", r)
+	}
+	// TODO: Consider validating when folks implement ReadOnlyFinalizer, but not Finalizer.
+
+	rec := &reconcilerImpl{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				all, err := lister.List(labels.Everything())
+				if err != nil {
+					return err
+				}
+				for _, elt := range all {
+					// TODO: Consider letting users specify a filter in options.
+					enq(bkt, types.NamespacedName{
+						Namespace: elt.GetNamespace(),
+						Name:      elt.GetName(),
+					})
+				}
+				return nil
+			},
+		},
+		Client:        client,
+		Lister:        lister,
+		Recorder:      recorder,
+		reconciler:    r,
+		finalizerName: defaultFinalizerName,
+	}
+
+	for _, opts := range options {
+		if opts.ConfigStore != nil {
+			rec.configStore = opts.ConfigStore
+		}
+		if opts.FinalizerName != "" {
+			rec.finalizerName = opts.FinalizerName
+		}
+		if opts.SkipStatusUpdates {
+			rec.skipStatusUpdates = true
+		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
+	}
+
+	return rec
+}
+
+// Reconcile implements controller.Reconciler
+func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
+	logger := logging.FromContext(ctx)
+
+	// Initialize the reconciler state. This will convert the namespace/name
+	// string into a distinct namespace and name, determine if this instance of
+	// the reconciler is the leader, and any additional interfaces implemented
+	// by the reconciler. Returns an error is the resource key is invalid.
+	s, err := newState(key, r)
+	if err != nil {
+		logger.Error("Invalid resource key: ", key)
+		return nil
+	}
+
+	// If we are not the leader, and we don't implement either ReadOnly
+	// observer interfaces, then take a fast-path out.
+	if s.isNotLeaderNorObserver() {
+		return controller.NewSkipKey(key)
+	}
+
+	// If configStore is set, attach the frozen configuration to the context.
+	if r.configStore != nil {
+		ctx = r.configStore.ToContext(ctx)
+	}
+
+	// Add the recorder to context.
+	ctx = controller.WithEventRecorder(ctx, r.Recorder)
+
+	// Get the resource with this namespace/name.
+
+	getter := r.Lister.ResetOffsets(s.namespace)
+
+	original, err := getter.Get(s.name)
+
+	if errors.IsNotFound(err) {
+		// The resource may no longer exist, in which case we stop processing and call
+		// the ObserveDeletion handler if appropriate.
+		logger.Debugf("Resource %q no longer exists", key)
+		if del, ok := r.reconciler.(reconciler.OnDeletionInterface); ok {
+			return del.ObserveDeletion(ctx, types.NamespacedName{
+				Namespace: s.namespace,
+				Name:      s.name,
+			})
+		}
+		return nil
+	} else if err != nil {
+		return err
+	}
+
+	// Don't modify the informers copy.
+	resource := original.DeepCopy()
+
+	var reconcileEvent reconciler.Event
+
+	name, do := s.reconcileMethodFor(resource)
+	// Append the target method to the logger.
+	logger = logger.With(zap.String("targetMethod", name))
+	switch name {
+	case reconciler.DoReconcileKind:
+		// Set and update the finalizer on resource if r.reconciler
+		// implements Finalizer.
+		if resource, err = r.setFinalizerIfFinalizer(ctx, resource); err != nil {
+			return fmt.Errorf("failed to set finalizers: %w", err)
+		}
+
+		if !r.skipStatusUpdates {
+			reconciler.PreProcessReconcile(ctx, resource)
+		}
+
+		// Reconcile this copy of the resource and then write back any status
+		// updates regardless of whether the reconciliation errored out.
+		reconcileEvent = do(ctx, resource)
+
+		if !r.skipStatusUpdates {
+			reconciler.PostProcessReconcile(ctx, resource, original)
+		}
+
+	case reconciler.DoFinalizeKind:
+		// For finalizing reconcilers, if this resource being marked for deletion
+		// and reconciled cleanly (nil or normal event), remove the finalizer.
+		reconcileEvent = do(ctx, resource)
+
+		if resource, err = r.clearFinalizer(ctx, resource, reconcileEvent); err != nil {
+			return fmt.Errorf("failed to clear finalizers: %w", err)
+		}
+
+	case reconciler.DoObserveKind, reconciler.DoObserveFinalizeKind:
+		// Observe any changes to this resource, since we are not the leader.
+		reconcileEvent = do(ctx, resource)
+
+	}
+
+	// Synchronize the status.
+	switch {
+	case r.skipStatusUpdates:
+		// This reconciler implementation is configured to skip resource updates.
+		// This may mean this reconciler does not observe spec, but reconciles external changes.
+	case equality.Semantic.DeepEqual(original.Status, resource.Status):
+		// If we didn't change anything then don't call updateStatus.
+		// This is important because the copy we loaded from the injectionInformer's
+		// cache may be stale and we don't want to overwrite a prior update
+		// to status with this stale state.
+	case !s.isLeader:
+		// High-availability reconcilers may have many replicas watching the resource, but only
+		// the elected leader is expected to write modifications.
+		logger.Warn("Saw status changes when we aren't the leader!")
+	default:
+		if err = r.updateStatus(ctx, original, resource); err != nil {
+			logger.Warnw("Failed to update resource status", zap.Error(err))
+			r.Recorder.Eventf(resource, v1.EventTypeWarning, "UpdateFailed",
+				"Failed to update status for %q: %v", resource.Name, err)
+			return err
+		}
+	}
+
+	// Report the reconciler event, if any.
+	if reconcileEvent != nil {
+		var event *reconciler.ReconcilerEvent
+		if reconciler.EventAs(reconcileEvent, &event) {
+			logger.Infow("Returned an event", zap.Any("event", reconcileEvent))
+			r.Recorder.Event(resource, event.EventType, event.Reason, event.Error())
+
+			// the event was wrapped inside an error, consider the reconciliation as failed
+			if _, isEvent := reconcileEvent.(*reconciler.ReconcilerEvent); !isEvent {
+				return reconcileEvent
+			}
+			return nil
+		}
+
+		logger.Errorw("Returned an error", zap.Error(reconcileEvent))
+		r.Recorder.Event(resource, v1.EventTypeWarning, "InternalError", reconcileEvent.Error())
+		return reconcileEvent
+	}
+
+	return nil
+}
+
+func (r *reconcilerImpl) updateStatus(ctx context.Context, existing *v1alpha1.ResetOffset, desired *v1alpha1.ResetOffset) error {
+	existing = existing.DeepCopy()
+	return reconciler.RetryUpdateConflicts(func(attempts int) (err error) {
+		// The first iteration tries to use the injectionInformer's state, subsequent attempts fetch the latest state via API.
+		if attempts > 0 {
+
+			getter := r.Client.KafkaV1alpha1().ResetOffsets(desired.Namespace)
+
+			existing, err = getter.Get(ctx, desired.Name, metav1.GetOptions{})
+			if err != nil {
+				return err
+			}
+		}
+
+		// If there's nothing to update, just return.
+		if equality.Semantic.DeepEqual(existing.Status, desired.Status) {
+			return nil
+		}
+
+		if diff, err := kmp.SafeDiff(existing.Status, desired.Status); err == nil && diff != "" {
+			logging.FromContext(ctx).Debug("Updating status with: ", diff)
+		}
+
+		existing.Status = desired.Status
+
+		updater := r.Client.KafkaV1alpha1().ResetOffsets(existing.Namespace)
+
+		_, err = updater.UpdateStatus(ctx, existing, metav1.UpdateOptions{})
+		return err
+	})
+}
+
+// updateFinalizersFiltered will update the Finalizers of the resource.
+// TODO: this method could be generic and sync all finalizers. For now it only
+// updates defaultFinalizerName or its override.
+func (r *reconcilerImpl) updateFinalizersFiltered(ctx context.Context, resource *v1alpha1.ResetOffset) (*v1alpha1.ResetOffset, error) {
+
+	getter := r.Lister.ResetOffsets(resource.Namespace)
+
+	actual, err := getter.Get(resource.Name)
+	if err != nil {
+		return resource, err
+	}
+
+	// Don't modify the informers copy.
+	existing := actual.DeepCopy()
+
+	var finalizers []string
+
+	// If there's nothing to update, just return.
+	existingFinalizers := sets.NewString(existing.Finalizers...)
+	desiredFinalizers := sets.NewString(resource.Finalizers...)
+
+	if desiredFinalizers.Has(r.finalizerName) {
+		if existingFinalizers.Has(r.finalizerName) {
+			// Nothing to do.
+			return resource, nil
+		}
+		// Add the finalizer.
+		finalizers = append(existing.Finalizers, r.finalizerName)
+	} else {
+		if !existingFinalizers.Has(r.finalizerName) {
+			// Nothing to do.
+			return resource, nil
+		}
+		// Remove the finalizer.
+		existingFinalizers.Delete(r.finalizerName)
+		finalizers = existingFinalizers.List()
+	}
+
+	mergePatch := map[string]interface{}{
+		"metadata": map[string]interface{}{
+			"finalizers":      finalizers,
+			"resourceVersion": existing.ResourceVersion,
+		},
+	}
+
+	patch, err := json.Marshal(mergePatch)
+	if err != nil {
+		return resource, err
+	}
+
+	patcher := r.Client.KafkaV1alpha1().ResetOffsets(resource.Namespace)
+
+	resourceName := resource.Name
+	updated, err := patcher.Patch(ctx, resourceName, types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		r.Recorder.Eventf(existing, v1.EventTypeWarning, "FinalizerUpdateFailed",
+			"Failed to update finalizers for %q: %v", resourceName, err)
+	} else {
+		r.Recorder.Eventf(updated, v1.EventTypeNormal, "FinalizerUpdate",
+			"Updated %q finalizers", resource.GetName())
+	}
+	return updated, err
+}
+
+func (r *reconcilerImpl) setFinalizerIfFinalizer(ctx context.Context, resource *v1alpha1.ResetOffset) (*v1alpha1.ResetOffset, error) {
+	if _, ok := r.reconciler.(Finalizer); !ok {
+		return resource, nil
+	}
+
+	finalizers := sets.NewString(resource.Finalizers...)
+
+	// If this resource is not being deleted, mark the finalizer.
+	if resource.GetDeletionTimestamp().IsZero() {
+		finalizers.Insert(r.finalizerName)
+	}
+
+	resource.Finalizers = finalizers.List()
+
+	// Synchronize the finalizers filtered by r.finalizerName.
+	return r.updateFinalizersFiltered(ctx, resource)
+}
+
+func (r *reconcilerImpl) clearFinalizer(ctx context.Context, resource *v1alpha1.ResetOffset, reconcileEvent reconciler.Event) (*v1alpha1.ResetOffset, error) {
+	if _, ok := r.reconciler.(Finalizer); !ok {
+		return resource, nil
+	}
+	if resource.GetDeletionTimestamp().IsZero() {
+		return resource, nil
+	}
+
+	finalizers := sets.NewString(resource.Finalizers...)
+
+	if reconcileEvent != nil {
+		var event *reconciler.ReconcilerEvent
+		if reconciler.EventAs(reconcileEvent, &event) {
+			if event.EventType == v1.EventTypeNormal {
+				finalizers.Delete(r.finalizerName)
+			}
+		}
+	} else {
+		finalizers.Delete(r.finalizerName)
+	}
+
+	resource.Finalizers = finalizers.List()
+
+	// Synchronize the finalizers filtered by r.finalizerName.
+	return r.updateFinalizersFiltered(ctx, resource)
+}
diff --git a/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/state.go b/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/state.go
new file mode 100644
index 000000000..12a0e2bd5
--- /dev/null
+++ b/pkg/client/injection/reconciler/kafka/v1alpha1/resetoffset/state.go
@@ -0,0 +1,106 @@
+/*
+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 resetoffset
+
+import (
+	fmt "fmt"
+
+	types "k8s.io/apimachinery/pkg/types"
+	cache "k8s.io/client-go/tools/cache"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+// state is used to track the state of a reconciler in a single run.
+type state struct {
+	// key is the original reconciliation key from the queue.
+	key string
+	// namespace is the namespace split from the reconciliation key.
+	namespace string
+	// name is the name split from the reconciliation key.
+	name string
+	// reconciler is the reconciler.
+	reconciler Interface
+	// roi is the read only interface cast of the reconciler.
+	roi ReadOnlyInterface
+	// isROI (Read Only Interface) the reconciler only observes reconciliation.
+	isROI bool
+	// rof is the read only finalizer cast of the reconciler.
+	rof ReadOnlyFinalizer
+	// isROF (Read Only Finalizer) the reconciler only observes finalize.
+	isROF bool
+	// isLeader the instance of the reconciler is the elected leader.
+	isLeader bool
+}
+
+func newState(key string, r *reconcilerImpl) (*state, error) {
+	// Convert the namespace/name string into a distinct namespace and name.
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return nil, fmt.Errorf("invalid resource key: %s", key)
+	}
+
+	roi, isROI := r.reconciler.(ReadOnlyInterface)
+	rof, isROF := r.reconciler.(ReadOnlyFinalizer)
+
+	isLeader := r.IsLeaderFor(types.NamespacedName{
+		Namespace: namespace,
+		Name:      name,
+	})
+
+	return &state{
+		key:        key,
+		namespace:  namespace,
+		name:       name,
+		reconciler: r.reconciler,
+		roi:        roi,
+		isROI:      isROI,
+		rof:        rof,
+		isROF:      isROF,
+		isLeader:   isLeader,
+	}, nil
+}
+
+// isNotLeaderNorObserver checks to see if this reconciler with the current
+// state is enabled to do any work or not.
+// isNotLeaderNorObserver returns true when there is no work possible for the
+// reconciler.
+func (s *state) isNotLeaderNorObserver() bool {
+	if !s.isLeader && !s.isROI && !s.isROF {
+		// If we are not the leader, and we don't implement either ReadOnly
+		// interface, then take a fast-path out.
+		return true
+	}
+	return false
+}
+
+func (s *state) reconcileMethodFor(o *v1alpha1.ResetOffset) (string, doReconcile) {
+	if o.GetDeletionTimestamp().IsZero() {
+		if s.isLeader {
+			return reconciler.DoReconcileKind, s.reconciler.ReconcileKind
+		} else if s.isROI {
+			return reconciler.DoObserveKind, s.roi.ObserveKind
+		}
+	} else if fin, ok := s.reconciler.(Finalizer); s.isLeader && ok {
+		return reconciler.DoFinalizeKind, fin.FinalizeKind
+	} else if !s.isLeader && s.isROF {
+		return reconciler.DoObserveFinalizeKind, s.rof.ObserveFinalizeKind
+	}
+	return "unknown", nil
+}
diff --git a/pkg/client/listers/kafka/v1alpha1/expansion_generated.go b/pkg/client/listers/kafka/v1alpha1/expansion_generated.go
new file mode 100644
index 000000000..7af06f2a9
--- /dev/null
+++ b/pkg/client/listers/kafka/v1alpha1/expansion_generated.go
@@ -0,0 +1,27 @@
+/*
+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 lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+// ResetOffsetListerExpansion allows custom methods to be added to
+// ResetOffsetLister.
+type ResetOffsetListerExpansion interface{}
+
+// ResetOffsetNamespaceListerExpansion allows custom methods to be added to
+// ResetOffsetNamespaceLister.
+type ResetOffsetNamespaceListerExpansion interface{}
diff --git a/pkg/client/listers/kafka/v1alpha1/resetoffset.go b/pkg/client/listers/kafka/v1alpha1/resetoffset.go
new file mode 100644
index 000000000..81eada831
--- /dev/null
+++ b/pkg/client/listers/kafka/v1alpha1/resetoffset.go
@@ -0,0 +1,99 @@
+/*
+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 lister-gen. DO NOT EDIT.
+
+package v1alpha1
+
+import (
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+// ResetOffsetLister helps list ResetOffsets.
+// All objects returned here must be treated as read-only.
+type ResetOffsetLister interface {
+	// List lists all ResetOffsets in the indexer.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.ResetOffset, err error)
+	// ResetOffsets returns an object that can list and get ResetOffsets.
+	ResetOffsets(namespace string) ResetOffsetNamespaceLister
+	ResetOffsetListerExpansion
+}
+
+// resetOffsetLister implements the ResetOffsetLister interface.
+type resetOffsetLister struct {
+	indexer cache.Indexer
+}
+
+// NewResetOffsetLister returns a new ResetOffsetLister.
+func NewResetOffsetLister(indexer cache.Indexer) ResetOffsetLister {
+	return &resetOffsetLister{indexer: indexer}
+}
+
+// List lists all ResetOffsets in the indexer.
+func (s *resetOffsetLister) List(selector labels.Selector) (ret []*v1alpha1.ResetOffset, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.ResetOffset))
+	})
+	return ret, err
+}
+
+// ResetOffsets returns an object that can list and get ResetOffsets.
+func (s *resetOffsetLister) ResetOffsets(namespace string) ResetOffsetNamespaceLister {
+	return resetOffsetNamespaceLister{indexer: s.indexer, namespace: namespace}
+}
+
+// ResetOffsetNamespaceLister helps list and get ResetOffsets.
+// All objects returned here must be treated as read-only.
+type ResetOffsetNamespaceLister interface {
+	// List lists all ResetOffsets in the indexer for a given namespace.
+	// Objects returned here must be treated as read-only.
+	List(selector labels.Selector) (ret []*v1alpha1.ResetOffset, err error)
+	// Get retrieves the ResetOffset from the indexer for a given namespace and name.
+	// Objects returned here must be treated as read-only.
+	Get(name string) (*v1alpha1.ResetOffset, error)
+	ResetOffsetNamespaceListerExpansion
+}
+
+// resetOffsetNamespaceLister implements the ResetOffsetNamespaceLister
+// interface.
+type resetOffsetNamespaceLister struct {
+	indexer   cache.Indexer
+	namespace string
+}
+
+// List lists all ResetOffsets in the indexer for a given namespace.
+func (s resetOffsetNamespaceLister) List(selector labels.Selector) (ret []*v1alpha1.ResetOffset, err error) {
+	err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) {
+		ret = append(ret, m.(*v1alpha1.ResetOffset))
+	})
+	return ret, err
+}
+
+// Get retrieves the ResetOffset from the indexer for a given namespace and name.
+func (s resetOffsetNamespaceLister) Get(name string) (*v1alpha1.ResetOffset, error) {
+	obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(v1alpha1.Resource("resetoffset"), name)
+	}
+	return obj.(*v1alpha1.ResetOffset), nil
+}


Test patch:
diff --git a/pkg/apis/kafka/v1alpha1/register_test.go b/pkg/apis/kafka/v1alpha1/register_test.go
new file mode 100644
index 000000000..f095b1b78
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/register_test.go
@@ -0,0 +1,49 @@
+/*
+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 v1alpha1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"knative.dev/eventing-kafka/pkg/apis/kafka"
+)
+
+func TestKind(t *testing.T) {
+	kind := Kind("test-kind")
+	assert.Equal(t, schema.GroupKind{Group: "kafka.eventing.knative.dev", Kind: "test-kind"}, kind)
+}
+
+func TestResource(t *testing.T) {
+	resource := Resource("test-resource")
+	assert.Equal(t, schema.GroupResource{Group: kafka.GroupName, Resource: "test-resource"}, resource)
+}
+
+func TestAddKnownTypes(t *testing.T) {
+	scheme := runtime.NewScheme()
+	err := addKnownTypes(scheme)
+	assert.Nil(t, err)
+	types := scheme.KnownTypes(schema.GroupVersion{Group: kafka.GroupName, Version: "v1alpha1"})
+	assert.NotNil(t, types)
+	roType := types["ResetOffset"]
+	assert.NotNil(t, roType)
+	roListType := types["ResetOffsetList"]
+	assert.NotNil(t, roListType)
+}
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_defaults_test.go b/pkg/apis/kafka/v1alpha1/reset_offset_defaults_test.go
new file mode 100644
index 000000000..80c61a479
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_defaults_test.go
@@ -0,0 +1,31 @@
+/*
+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 v1alpha1
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestResetOffset_SetDefaults(t *testing.T) {
+	initialResetOffset := ResetOffset{}
+	resultResetOffset := initialResetOffset.DeepCopy()
+	resultResetOffset.SetDefaults(context.TODO())
+	assert.Equal(t, initialResetOffset, *resultResetOffset)
+}
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go
new file mode 100644
index 000000000..5eeda3c42
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_lifecycle_test.go
@@ -0,0 +1,331 @@
+/*
+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 v1alpha1
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+var (
+	condSucceeded             = apis.Condition{Type: ResetOffsetConditionSucceeded, 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}
+	condConsumerGroupsStarted = apis.Condition{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionTrue}
+)
+
+func TestResetOffset_GetConditionSet(t *testing.T) {
+	ro := &ResetOffset{}
+	if got, want := ro.GetConditionSet().GetTopLevelConditionType(), apis.ConditionSucceeded; got != want {
+		t.Errorf("GetTopLevelCondition=%v, want=%v", got, want)
+	}
+}
+
+func TestResetOffsetStatus_GetCondition(t *testing.T) {
+	tests := []struct {
+		name          string
+		status        *ResetOffsetStatus
+		conditionType apis.ConditionType
+		wantCondition *apis.Condition
+	}{
+		{
+			name: "ResetOffsetConditionSucceeded",
+			status: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{condSucceeded},
+				},
+			},
+			conditionType: ResetOffsetConditionSucceeded,
+			wantCondition: &condSucceeded,
+		},
+		{
+			name: "ResetOffsetConditionResetInitiated",
+			status: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{condResetInitiated},
+				},
+			},
+			conditionType: ResetOffsetConditionResetInitiated,
+			wantCondition: &condResetInitiated,
+		},
+		{
+			name: "ResetOffsetConditionConsumerGroupsStopped",
+			status: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{condConsumerGroupsStopped},
+				},
+			},
+			conditionType: ResetOffsetConditionConsumerGroupsStopped,
+			wantCondition: &condConsumerGroupsStopped,
+		},
+		{
+			name: "ResetOffsetConditionOffsetsUpdated",
+			status: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{condOffsetsUpdated},
+				},
+			},
+			conditionType: ResetOffsetConditionOffsetsUpdated,
+			wantCondition: &condOffsetsUpdated,
+		},
+		{
+			name: "ResetOffsetConditionConsumerGroupsStarted",
+			status: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{condConsumerGroupsStarted},
+				},
+			},
+			conditionType: ResetOffsetConditionConsumerGroupsStarted,
+			wantCondition: &condConsumerGroupsStarted,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			gotCondition := test.status.GetCondition(test.conditionType)
+			if diff := cmp.Diff(test.wantCondition, gotCondition); diff != "" {
+				t.Errorf("unexpected condition (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
+
+func TestResetOffsetStatus_InitializeConditions(t *testing.T) {
+	tests := []struct {
+		name          string
+		initialStatus *ResetOffsetStatus
+		wantStatus    *ResetOffsetStatus
+	}{
+		{
+			name:          "empty",
+			initialStatus: &ResetOffsetStatus{},
+			wantStatus: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
+					},
+				},
+			},
+		},
+		{
+			name: "one false",
+			initialStatus: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionFalse},
+					},
+				},
+			},
+			wantStatus: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionFalse},
+						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
+					},
+				},
+			},
+		},
+		{
+			name: "one true",
+			initialStatus: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue},
+					},
+				},
+			},
+			wantStatus: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue},
+						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionUnknown},
+						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionUnknown},
+					},
+				},
+			},
+		},
+		{
+			name: "happy true",
+			initialStatus: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionTrue},
+					},
+				},
+			},
+			wantStatus: &ResetOffsetStatus{
+				Status: duckv1.Status{
+					Conditions: []apis.Condition{
+						{Type: ResetOffsetConditionConsumerGroupsStarted, Status: corev1.ConditionTrue},
+						{Type: ResetOffsetConditionConsumerGroupsStopped, Status: corev1.ConditionTrue},
+						{Type: ResetOffsetConditionOffsetsUpdated, Status: corev1.ConditionTrue},
+						{Type: ResetOffsetConditionResetInitiated, Status: corev1.ConditionTrue},
+						{Type: ResetOffsetConditionSucceeded, Status: corev1.ConditionTrue},
+					},
+				},
+			},
+		},
+	}
+
+	ignoreAllButTypeAndStatus := cmpopts.IgnoreFields(apis.Condition{}, "LastTransitionTime", "Message", "Reason", "Severity")
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			test.initialStatus.InitializeConditions()
+			if diff := cmp.Diff(test.wantStatus, test.initialStatus, ignoreAllButTypeAndStatus); diff != "" { // Note Order Specific Comparison ; )
+				t.Errorf("unexpected conditions (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
+
+func TestResetOffsetStatus_IsCompleted(t *testing.T) {
+	tests := []struct {
+		name                      string
+		markResetInitiated        bool
+		markConsumerGroupsStopped bool
+		markOffsetsUpdated        bool
+		markConsumerGroupsStarted bool
+		wantSucceeded             bool
+	}{
+		{
+			name:                      "Happy",
+			markResetInitiated:        true,
+			markConsumerGroupsStopped: true,
+			markOffsetsUpdated:        true,
+			markConsumerGroupsStarted: true,
+			wantSucceeded:             true,
+		},
+		{
+			name:                      "ResetInitiated Failed",
+			markResetInitiated:        false,
+			markConsumerGroupsStopped: true,
+			markOffsetsUpdated:        true,
+			markConsumerGroupsStarted: true,
+			wantSucceeded:             false,
+		},
+		{
+			name:                      "ConsumerGroupsStopped Failed",
+			markResetInitiated:        true,
+			markConsumerGroupsStopped: false,
+			markOffsetsUpdated:        true,
+			markConsumerGroupsStarted: true,
+			wantSucceeded:             false,
+		},
+		{
+			name:                      "OffsetsUpdated Failed",
+			markResetInitiated:        true,
+			markConsumerGroupsStopped: true,
+			markOffsetsUpdated:        false,
+			markConsumerGroupsStarted: true,
+			wantSucceeded:             false,
+		},
+		{
+			name:                      "ConsumerGroupsStarted Failed",
+			markResetInitiated:        true,
+			markConsumerGroupsStopped: true,
+			markOffsetsUpdated:        true,
+			markConsumerGroupsStarted: false,
+			wantSucceeded:             false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resetOffsetStatus := &ResetOffsetStatus{}
+			resetOffsetStatus.InitializeConditions()
+			if test.markResetInitiated {
+				resetOffsetStatus.MarkResetInitiatedTrue()
+			} else {
+				resetOffsetStatus.MarkResetInitiatedFailed("TestingOffsetsResetInitiatedStatus", "TestMessage")
+			}
+			if test.markConsumerGroupsStopped {
+				resetOffsetStatus.MarkConsumerGroupsStoppedTrue()
+			} else {
+				resetOffsetStatus.MarkConsumerGroupsStoppedFailed("TestingConsumerGroupsStoppedStatus", "TestMessage")
+			}
+			if test.markOffsetsUpdated {
+				resetOffsetStatus.MarkOffsetsUpdatedTrue()
+			} else {
+				resetOffsetStatus.MarkOffsetsUpdatedFailed("TestingOffsetsUpdatedFailedStatus", "TestMessage")
+			}
+			if test.markConsumerGroupsStarted {
+				resetOffsetStatus.MarkConsumerGroupsStartedTrue()
+			} else {
+				resetOffsetStatus.MarkConsumerGroupsStartedFailed("TestingConsumerGroupsStartedStatus", "TestMessage")
+			}
+			gotSucceeded := resetOffsetStatus.IsCompleted()
+			if test.wantSucceeded != gotSucceeded {
+				t.Errorf("unexpected succeeded status: want %v, got %v", test.wantSucceeded, gotSucceeded)
+			}
+		})
+	}
+}
+
+func TestRegisterAlternateResetOffsetConditionSet(t *testing.T) {
+	conditionSet := apis.NewLivingConditionSet(apis.ConditionReady, "test")
+	RegisterAlternateResetOffsetConditionSet(conditionSet)
+	resetOffset := ResetOffset{}
+	assert.Equal(t, conditionSet, resetOffset.GetConditionSet())
+	assert.Equal(t, conditionSet, resetOffset.Status.GetConditionSet())
+}
+
+func TestResetOffsetStatus_Topic(t *testing.T) {
+	topic := "test-topic-name"
+	resetOffset := ResetOffset{}
+	assert.Equal(t, "", resetOffset.Status.GetTopic())
+	assert.Equal(t, resetOffset.Status.Topic, resetOffset.Status.GetTopic())
+	resetOffset.Status.SetTopic(topic)
+	assert.Equal(t, topic, resetOffset.Status.GetTopic())
+	assert.Equal(t, resetOffset.Status.Topic, resetOffset.Status.GetTopic())
+}
+
+func TestResetOffsetStatus_Group(t *testing.T) {
+	group := "test-group-id"
+	resetOffset := ResetOffset{}
+	assert.Equal(t, "", resetOffset.Status.GetGroup())
+	assert.Equal(t, resetOffset.Status.Group, resetOffset.Status.GetGroup())
+	resetOffset.Status.SetGroup(group)
+	assert.Equal(t, group, resetOffset.Status.GetGroup())
+	assert.Equal(t, resetOffset.Status.Group, resetOffset.Status.GetGroup())
+}
+
+func TestResetOffsetStatus_Partitions(t *testing.T) {
+	partitions := []OffsetMapping{{Partition: 0, OldOffset: 1, NewOffset: 2}, {Partition: 1, OldOffset: 2, NewOffset: 3}}
+	resetOffset := ResetOffset{}
+	assert.Nil(t, resetOffset.Status.GetPartitions())
+	assert.Equal(t, resetOffset.Status.Partitions, resetOffset.Status.GetPartitions())
+	resetOffset.Status.SetPartitions(partitions)
+	assert.Equal(t, partitions, resetOffset.Status.GetPartitions())
+	assert.Equal(t, resetOffset.Status.Partitions, resetOffset.Status.GetPartitions())
+}
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_types_test.go b/pkg/apis/kafka/v1alpha1/reset_offset_types_test.go
new file mode 100644
index 000000000..ea77c0de5
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_types_test.go
@@ -0,0 +1,214 @@
+/*
+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 v1alpha1
+
+import (
+	"testing"
+	"time"
+
+	"github.com/Shopify/sarama"
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+func TestResetOffset_GetGroupVersionKind(t *testing.T) {
+	resetOffset := ResetOffset{}
+	gvk := resetOffset.GetGroupVersionKind()
+	if gvk.Kind != "ResetOffset" {
+		t.Errorf("Should be 'ResetOffset'.")
+	}
+}
+
+func TestResetOffset_GetStatus(t *testing.T) {
+	status := &duckv1.Status{}
+	resetOffset := ResetOffset{
+		Status: ResetOffsetStatus{
+			Status: *status,
+		},
+	}
+	if !cmp.Equal(resetOffset.GetStatus(), status) {
+		t.Errorf("GetStatus did not retrieve status. Got=%v Want=%v", resetOffset.GetStatus(), status)
+	}
+}
+
+func TestResetOffsetSpec_IsOffsetEarliest(t *testing.T) {
+
+	tests := []struct {
+		name   string
+		offset string
+		want   bool
+	}{
+		{
+			name:   "earliest",
+			offset: OffsetEarliest,
+			want:   true,
+		},
+		{
+			name:   "latest",
+			offset: OffsetLatest,
+			want:   false,
+		},
+		{
+			name:   "time",
+			offset: "2021-05-04T05:04:01Z",
+			want:   false,
+		},
+		{
+			name:   "invalid",
+			offset: "foo",
+			want:   false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resetOffsetSpec := &ResetOffsetSpec{Offset: OffsetSpec{Time: test.offset}}
+			got := resetOffsetSpec.IsOffsetEarliest()
+			if diff := cmp.Diff(test.want, got); diff != "" {
+				t.Errorf("unexpected conditions (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
+
+func TestResetOffsetSpec_IsOffsetLatest(t *testing.T) {
+
+	tests := []struct {
+		name   string
+		offset string
+		want   bool
+	}{
+		{
+			name:   "earliest",
+			offset: OffsetEarliest,
+			want:   false,
+		},
+		{
+			name:   "latest",
+			offset: OffsetLatest,
+			want:   true,
+		},
+		{
+			name:   "time",
+			offset: "2021-05-04T05:04:01Z",
+			want:   false,
+		},
+		{
+			name:   "invalid",
+			offset: "foo",
+			want:   false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resetOffsetSpec := &ResetOffsetSpec{Offset: OffsetSpec{Time: test.offset}}
+			got := resetOffsetSpec.IsOffsetLatest()
+			if diff := cmp.Diff(test.want, got); diff != "" {
+				t.Errorf("unexpected conditions (-want, +got) = %v", diff)
+			}
+		})
+	}
+}
+
+func TestResetOffsetSpec_ParseOffsetTime(t *testing.T) {
+
+	offsetRFC3339 := time.Now().UTC().Add(-1 * time.Hour).Format(time.RFC3339)
+	timeRFC3339, _ := time.Parse(time.RFC3339, offsetRFC3339)
+
+	tests := []struct {
+		name       string
+		offset     string
+		expectTime time.Time
+		expectErr  bool
+	}{
+		{
+			name:       "valid",
+			offset:     offsetRFC3339,
+			expectTime: timeRFC3339,
+			expectErr:  false,
+		},
+		{
+			name:       "invalid",
+			offset:     "foo",
+			expectTime: time.Time{},
+			expectErr:  true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resetOffsetSpec := &ResetOffsetSpec{Offset: OffsetSpec{Time: test.offset}}
+			offsetTime, err := resetOffsetSpec.ParseOffsetTime()
+			if test.expectErr {
+				assert.NotNil(t, err)
+				assert.Equal(t, time.Time{}, offsetTime)
+			} else {
+				assert.Nil(t, err)
+				assert.Equal(t, test.expectTime, offsetTime)
+			}
+		})
+	}
+}
+
+func TestResetOffsetSpec_ParseSaramaOffsetTime(t *testing.T) {
+
+	offsetRFC3339 := time.Now().UTC().Add(-1 * time.Hour).Format(time.RFC3339)
+	timeRFC3339, _ := time.Parse(time.RFC3339, offsetRFC3339)
+
+	tests := []struct {
+		name       string
+		offset     string
+		expectTime int64
+		expectErr  bool
+	}{
+		{
+			name:       "valid earliest",
+			offset:     OffsetEarliest,
+			expectTime: sarama.OffsetOldest,
+			expectErr:  false,
+		},
+		{
+			name:       "valid latest",
+			offset:     OffsetLatest,
+			expectTime: sarama.OffsetNewest,
+			expectErr:  false,
+		},
+		{
+			name:       "valid time",
+			offset:     offsetRFC3339,
+			expectTime: timeRFC3339.UnixNano() / 1000000,
+			expectErr:  false,
+		},
+		{
+			name:       "invalid time",
+			offset:     "foo",
+			expectTime: 0,
+			expectErr:  true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resetOffsetSpec := &ResetOffsetSpec{Offset: OffsetSpec{Time: test.offset}}
+			saramaOffsetTime, err := resetOffsetSpec.ParseSaramaOffsetTime()
+			assert.Equal(t, test.expectErr, err != nil)
+			assert.Equal(t, test.expectTime, saramaOffsetTime)
+		})
+	}
+}
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_validation_test.go b/pkg/apis/kafka/v1alpha1/reset_offset_validation_test.go
new file mode 100644
index 000000000..99924fe12
--- /dev/null
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_validation_test.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 v1alpha1
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/webhook/resourcesemantics"
+)
+
+func TestResetOffset_Validate(t *testing.T) {
+
+	refAPIVersion := "messaging.knative.dev/v1beta1"
+	refKind := "Subscription"
+	refNamespace := "ref-namespace"
+	refName := "ref-name"
+
+	futureTime := time.Now().Add(1 * time.Hour).Format(time.RFC3339)
+	pastTime := time.Now().Add(-1 * time.Hour).Format(time.RFC3339)
+
+	reference := duckv1.KReference{
+		APIVersion: refAPIVersion,
+		Kind:       refKind,
+		Namespace:  refNamespace,
+		Name:       refName,
+	}
+
+	tests := []struct {
+		name string
+		cr   resourcesemantics.GenericCRD
+		want *apis.FieldError
+	}{
+		{
+			name: "valid offset earliest",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{Offset: OffsetSpec{Time: OffsetEarliest}, Ref: reference},
+			},
+		},
+		{
+			name: "valid offset latest",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{Offset: OffsetSpec{Time: OffsetLatest}, Ref: reference},
+			},
+		},
+		{
+			name: "valid offset time",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{Offset: OffsetSpec{Time: pastTime}, Ref: reference},
+			},
+		},
+		{
+			name: "invalid offset time",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{Offset: OffsetSpec{Time: futureTime}, Ref: reference},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrInvalidValue(futureTime, "spec.offset")
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		{
+			name: "invalid offset string",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{Offset: OffsetSpec{Time: "foo"}, Ref: reference},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrInvalidValue("foo", "spec.offset")
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		{
+			name: "invalid ref nil",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{Offset: OffsetSpec{Time: OffsetEarliest}},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrMissingField("spec.apiVersion", "spec.kind", "spec.name")
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		{
+			name: "invalid ref missing APIVersion",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{
+					Offset: OffsetSpec{Time: OffsetEarliest},
+					Ref: duckv1.KReference{
+						Kind:      refKind,
+						Namespace: refNamespace,
+						Name:      refName,
+					},
+				},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrMissingField("spec.apiVersion")
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		{
+			name: "invalid ref missing Kind",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{
+					Offset: OffsetSpec{Time: OffsetEarliest},
+					Ref: duckv1.KReference{
+						APIVersion: refAPIVersion,
+						Namespace:  refNamespace,
+						Name:       refName,
+					},
+				},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrMissingField("spec.kind")
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+		{
+			name: "invalid ref missing Name",
+			cr: &ResetOffset{
+				Spec: ResetOffsetSpec{
+					Offset: OffsetSpec{Time: OffsetEarliest},
+					Ref: duckv1.KReference{
+						APIVersion: refAPIVersion,
+						Kind:       refKind,
+						Namespace:  refNamespace,
+					},
+				},
+			},
+			want: func() *apis.FieldError {
+				var errs *apis.FieldError
+				fe := apis.ErrMissingField("spec.name")
+				errs = errs.Also(fe)
+				return errs
+			}(),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := test.cr.Validate(context.Background())
+			if test.want == nil {
+				if diff := cmp.Diff(test.want, got); diff != "" {
+					t.Errorf("validate (-want, +got) = %v", diff)
+				}
+			} else if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
+				t.Errorf("validate (-want, +got) = %v", diff)
+			}
+		})
+	}
+}