Repository: knative-sandbox/eventing-kafka
Pull request #725 (2021-06-18 17:53:36)

Resetoffset controller control protocol
Subtask of #588 to add control-protocol communication with dataplane to the ResetOffset Controller.

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- 🧽   Refactor "ref mapper" results into a single RefInfo
- 🎁   Provide distributed KafkaChannel mappers
- 🎁   Added new mappers to account for dataplane labelling differences
- 🎁   Implement ConnectionPool and AsyncNotificationStore communication with dataplane services.
- 🎁   Unit test everything delivered so far

**Notes**
- The ResetOffset Controller is still "inactive" and has not been added to any Channels/Sources/etc.
- The ResetOffset Controller is still not "complete" and this is a stepping stone in the path to implementing the feature.  There are still TODO's in the code.
- This PR and #724 are two halves of the control-protocol communication and contain some overlap that will require updating depending on which merges first.  This is to be expected and accounts for some of the TODOs.

**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.

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/725#" title="Author self-approved">travis-minke-sap</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/OWNERS)~~ [travis-minke-sap]

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405946822458871808/artifacts/line-cov.html#file2) | 91.7% | 87.5% | -4.2
[pkg/common/commands/resetoffset/controller/dataplane.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405946822458871808/artifacts/line-cov.html#file3) | Do not exist | 86.2% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405946822458871808/artifacts/line-cov.html#file5) | 96.7% | 97.6% | 1.0
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405946822458871808/artifacts/line-cov.html#file6) | 94.3% | 95.8% | 1.5
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405946822458871808/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/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/725?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     #725      +/-   ##
==========================================
+ Coverage   73.80%   74.21%   +0.40%     
==========================================
  Files         133      135       +2     
  Lines        5848     6003     +155     
==========================================
+ Hits         4316     4455     +139     
- Misses       1311     1321      +10     
- Partials      221      227       +6     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [.../channel/distributed/controller/util/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci91dGlsL2Rpc3BhdGNoZXIuZ28=) | `100.00% <ø> (ø)` | |
| [...ommon/commands/resetoffset/controller/dataplane.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL2RhdGFwbGFuZS5nbw==) | `78.26% <78.26%> (ø)` | |
| [...mmon/commands/resetoffset/controller/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL2NvbnRyb2xsZXIuZ28=) | `85.00% <87.50%> (-0.72%)` | :arrow_down: |
| [...on/commands/resetoffset/refmappers/subscription.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9yZWZtYXBwZXJzL3N1YnNjcmlwdGlvbi5nbw==) | `95.83% <97.87%> (+3.33%)` | :arrow_up: |
| [...hannel/distributed/controller/util/subscription.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci91dGlsL3N1YnNjcmlwdGlvbi5nbw==) | `100.00% <100.00%> (ø)` | |
| [.../common/commands/resetoffset/controller/offsets.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL29mZnNldHMuZ28=) | `95.00% <100.00%> (ø)` | |
| [...mmon/commands/resetoffset/controller/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb21tYW5kcy9yZXNldG9mZnNldC9jb250cm9sbGVyL3JlY29uY2lsZXIuZ28=) | `96.51% <100.00%> (+1.51%)` | :arrow_up: |
| [...g/common/controlprotocol/commands/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb250cm9scHJvdG9jb2wvY29tbWFuZHMvY29uc3VtZXJncm91cC5nbw==) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725?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/725?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [11214e9...ebd7149](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/725?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/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405968676661760000/artifacts/line-cov.html#file2) | 91.7% | 87.5% | -4.2
[pkg/common/commands/resetoffset/controller/dataplane.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405968676661760000/artifacts/line-cov.html#file3) | Do not exist | 86.2% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405968676661760000/artifacts/line-cov.html#file5) | 96.7% | 97.6% | 1.0
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405968676661760000/artifacts/line-cov.html#file6) | 94.3% | 95.8% | 1.5
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1405968676661760000/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1407040492905959424/artifacts/line-cov.html#file2) | 91.7% | 87.5% | -4.2
[pkg/common/commands/resetoffset/controller/dataplane.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1407040492905959424/artifacts/line-cov.html#file3) | Do not exist | 85.9% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1407040492905959424/artifacts/line-cov.html#file5) | 96.7% | 97.6% | 1.0
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1407040492905959424/artifacts/line-cov.html#file6) | 94.3% | 95.8% | 1.5
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1407040492905959424/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 

slinkydeveloper: @travis-minke-sap Am I wrong or the control protocol commands in this PR it's the same code from #724?
travis-minke-sap: > @travis-minke-sap Am I wrong or the control protocol commands in this PR it's the same code from #724?

Yes, @eric-sap and I are working on the opposite sides of this (sending / receiving these commands) and have both included them in our PR.  Whichever merges first is fine and the other person will rebase, etc.  After we get #724 and #725 in we can do some formal testing and probably will have a small alignment / cleanup PR - there are a few things in this PR that will need to be adjusted/moved/etc...
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/commands/resetoffset/controller/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1408060104103694336/artifacts/line-cov.html#file2) | 91.7% | 87.5% | -4.2
[pkg/common/commands/resetoffset/controller/dataplane.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1408060104103694336/artifacts/line-cov.html#file3) | Do not exist | 85.9% | 
[pkg/common/commands/resetoffset/controller/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1408060104103694336/artifacts/line-cov.html#file5) | 96.7% | 97.6% | 1.0
[pkg/common/commands/resetoffset/refmappers/subscription.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1408060104103694336/artifacts/line-cov.html#file6) | 94.3% | 95.8% | 1.5
[pkg/common/controlprotocol/commands/consumergroup.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/725/pull-knative-sandbox-eventing-kafka-go-coverage/1408060104103694336/artifacts/line-cov.html#file7) | Do not exist | 100.0% | 


Patch:
diff --git a/go.mod b/go.mod
index 4b203088d..c50ccd129 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,7 @@ require (
 	github.com/Azure/go-autorest/autorest v0.11.10 // indirect
 	github.com/Azure/go-autorest/autorest/to v0.4.0 // indirect
 	github.com/Azure/go-autorest/autorest/validation v0.3.0 // indirect
-	github.com/Shopify/sarama v1.29.0
+	github.com/Shopify/sarama v1.29.1
 	github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2 v2.4.1
 	github.com/cloudevents/sdk-go/v2 v2.4.1
 	github.com/davecgh/go-spew v1.1.1
diff --git a/go.sum b/go.sum
index 513025e19..084d219ac 100644
--- a/go.sum
+++ b/go.sum
@@ -112,8 +112,8 @@ github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdko
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
 github.com/Shopify/sarama v1.23.1/go.mod h1:XLH1GYJnLVE0XCr6KdJGVJRTwY30moWNJ4sERjXX6fs=
 github.com/Shopify/sarama v1.25.0/go.mod h1:y/CFFTO9eaMTNriwu/Q+W4eioLqiDMGkA1W+gmdfj8w=
-github.com/Shopify/sarama v1.29.0 h1:ARid8o8oieau9XrHI55f/L3EoRAhm9px6sonbD7yuUE=
-github.com/Shopify/sarama v1.29.0/go.mod h1:2QpgD79wpdAESqNQMxNc0KYMkycd4slxGdV3TWSVqrU=
+github.com/Shopify/sarama v1.29.1 h1:wBAacXbYVLmWieEA/0X/JagDdCZ8NVFOfS6l6+2u5S0=
+github.com/Shopify/sarama v1.29.1/go.mod h1:mdtqvCSg8JOxk8PmpTNGyo6wzd4BMm4QXSfDnTXmgkE=
 github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/VividCortex/gohistogram v1.0.0/go.mod h1:Pf5mBqqDxYaXu3hDrrU+w6nw50o/4+TcAqDqk/vUH7g=
@@ -818,9 +818,9 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201112155050-0c6587e931a9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a h1:kr2P4QFmQr29mSLA43kwrOcgcReGTfbE9N577tCTuBc=
 golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8=
+golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e h1:gsTQYXdTw2Gq7RBsWvlQ91b+aEQ6bXFUngBGuR8sPpI=
+golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -902,9 +902,9 @@ golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
-golang.org/x/net v0.0.0-20210427231257-85d9c07bbe3a/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
-golang.org/x/net v0.0.0-20210525063256-abc453219eb5 h1:wjuX4b5yYQnEQHzd+CBcrcC6OVR2J1CN6mUy0oSxIPo=
 golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20210614182718-04defd469f4e h1:XpT3nA5TvE525Ne3hInMh6+GETgn27Zfm9dxsThnX2Q=
+golang.org/x/net v0.0.0-20210614182718-04defd469f4e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -985,8 +985,9 @@ golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40 h1:JWgyZ1qgdTaF3N3oxC+MdTV7qvEEgHo3otj+HB5CM7Q=
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1 h1:SrN+KX8Art/Sf4HNj6Zcz06G7VEz+7w9tdXTPOZ7+l4=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
diff --git a/pkg/channel/distributed/controller/util/dispatcher.go b/pkg/channel/distributed/controller/util/dispatcher.go
index d0f5f4ffe..a0b09e388 100644
--- a/pkg/channel/distributed/controller/util/dispatcher.go
+++ b/pkg/channel/distributed/controller/util/dispatcher.go
@@ -22,13 +22,12 @@ import (
 	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 )
 
-// Create A DNS Safe Name For The Specified KafkaChannel Suitable For Use With K8S Services
+// DispatcherDnsSafeName creates a DNS-Safe name for the specified KafkaChannel suitable for use with K8S Services
 func DispatcherDnsSafeName(channel *kafkav1beta1.KafkaChannel) string {
 
 	// In order for the resulting name to be a valid DNS component is 63 characters.  We are appending 13 characters to
 	// separate the components and to indicate this is a Dispatcher, and adding 8 hash characters, which further reduces
-	// the available length to 42.
-	// We will allocate 26 characters to the channel and 16 to the namespace, leaving some extra buffer.
+	// the available length to 42. We then allocate 26 characters to the channel and 16 to the namespace.
 	safeChannelName := GenerateValidDnsName(channel.Name, 26, true, false)
 	safeChannelNamespace := GenerateValidDnsName(channel.Namespace, 16, false, false)
 	hash := GenerateHash(channel.Name+channel.Namespace, 8)
diff --git a/pkg/channel/distributed/controller/util/subscription.go b/pkg/channel/distributed/controller/util/subscription.go
index b7e1dd81e..f8b475604 100644
--- a/pkg/channel/distributed/controller/util/subscription.go
+++ b/pkg/channel/distributed/controller/util/subscription.go
@@ -23,7 +23,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	"knative.dev/pkg/system"
 
+	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	commonkafkautil "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/util"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 )
@@ -62,3 +64,38 @@ func GroupIdMapper(subscription *messagingv1.Subscription) (string, error) {
 	}
 	return commonkafkautil.GroupId(string(subscription.UID)), nil
 }
+
+// ConnectionPoolKeyMapper returns a string representing the control-protocol ControlPlaneConnectionPool Key for the specified Knative Subscription.
+func ConnectionPoolKeyMapper(subscription *messagingv1.Subscription) (string, error) {
+	return TopicNameMapper(subscription) // Distributed KafkaChannel is using the TopicName as ConnectionPool Key since it is 1:1 with KafkaChannel.
+}
+
+// DataPlaneNamespaceMapper returns the Kubernetes Namespace where the data-plane components
+// (e.g. Dispatcher) are created by the distributed KafkaChannel controller.
+func DataPlaneNamespaceMapper(_ *messagingv1.Subscription) (string, error) {
+	return system.Namespace(), nil
+}
+
+// DataPlaneLabelsMapper returns a map of Kubernetes Labels identifying the distributed
+// KafkaChannels' Dispatcher pods.
+func DataPlaneLabelsMapper(subscription *messagingv1.Subscription) (map[string]string, error) {
+
+	channelName := subscription.Spec.Channel.Name
+	channelNamespace := subscription.Spec.Channel.Namespace
+	if channelNamespace == "" {
+		channelNamespace = subscription.Namespace
+	}
+
+	sparseKafkaChannel := &kafkav1beta1.KafkaChannel{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      channelName,
+			Namespace: channelNamespace,
+		},
+	}
+
+	dispatcherAppLabelValue := DispatcherDnsSafeName(sparseKafkaChannel)
+
+	labels := map[string]string{constants.AppLabel: dispatcherAppLabelValue}
+
+	return labels, nil
+}
diff --git a/pkg/common/commands/resetoffset/controller/controller.go b/pkg/common/commands/resetoffset/controller/controller.go
index f3b195e2d..8c0b1c355 100644
--- a/pkg/common/commands/resetoffset/controller/controller.go
+++ b/pkg/common/commands/resetoffset/controller/controller.go
@@ -20,6 +20,9 @@ import (
 	"context"
 
 	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/types"
+	ctrlreconciler "knative.dev/control-protocol/pkg/reconciler"
+	"knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection"
@@ -33,7 +36,9 @@ import (
 )
 
 // NewControllerFactory returns a ControllerConstructor function capable of creating a "typed" ResetOffset Controller
-func NewControllerFactory(refMapperFactory refmappers.ResetOffsetRefMapperFactory) injection.ControllerConstructor {
+func NewControllerFactory(
+	refMapperFactory refmappers.ResetOffsetRefMapperFactory,
+	connectionPool ctrlreconciler.ControlPlaneConnectionPool) injection.ControllerConstructor {
 
 	// Return The New ResetOffset ControllerConstructor Function
 	return func(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
@@ -42,15 +47,25 @@ func NewControllerFactory(refMapperFactory refmappers.ResetOffsetRefMapperFactor
 		logger := logging.FromContext(ctx)
 
 		// Get The Needed Informers
+		podInformer := pod.Get(ctx)
 		resetoffsetInformer := resetoffset.Get(ctx)
 
 		// Create The RefMapper Via The Supplied Factory Using Initialized Context
 		refMapper := refMapperFactory.Create(ctx)
 
+		// Create A Control-Protocol AsyncCommandNotificationStore - No-Op Enqueue Function For ResetOffset Use Case
+		enqueueFunc := func(key types.NamespacedName) {
+			logger.Debug("Control-Protocol Enqueue Function", zap.String("Key", key.String()))
+		}
+		asyncCommandNotificationStore := ctrlreconciler.NewAsyncCommandNotificationStore(enqueueFunc)
+
 		// Create A ResetOffset Reconciler
 		reconciler := &Reconciler{
-			resetoffsetLister: resetoffsetInformer.Lister(),
-			refMapper:         refMapper,
+			podLister:                     podInformer.Lister(),
+			resetoffsetLister:             resetoffsetInformer.Lister(),
+			refMapper:                     refMapper,
+			connectionPool:                connectionPool,
+			asyncCommandNotificationStore: asyncCommandNotificationStore,
 		}
 
 		// Setup Reconciler To Watch The Kafka ConfigMap For Changes
diff --git a/pkg/common/commands/resetoffset/controller/dataplane.go b/pkg/common/commands/resetoffset/controller/dataplane.go
new file mode 100644
index 000000000..ad42ee008
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/dataplane.go
@@ -0,0 +1,213 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"fmt"
+	"hash/fnv"
+	"strings"
+	"sync"
+	"time"
+
+	"go.uber.org/multierr"
+	"go.uber.org/zap"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	ctrl "knative.dev/control-protocol/pkg"
+	ctrlmessage "knative.dev/control-protocol/pkg/message"
+	ctrlreconciler "knative.dev/control-protocol/pkg/reconciler"
+	"knative.dev/pkg/logging"
+
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+)
+
+const (
+	asyncCommandResultPollDuration    = 1 * time.Second  // AsyncCommandResult Polling Duration
+	asyncCommandResultTimeoutDuration = 10 * time.Second // AsyncCommandResult Timeout Duration
+
+	ControlProtocolServerPort = 8085 // TODO - Use common constant once eric delivers!
+)
+
+// reconcileDataPlaneServices updates the Reconciler ConnectionPool Services associated with the specified RefInfo.
+func (r *Reconciler) reconcileDataPlaneServices(ctx context.Context, resetOffset *kafkav1alpha1.ResetOffset, refInfo *refmappers.RefInfo) (map[string]ctrl.Service, error) {
+
+	// Get The Logger From Context
+	logger := logging.FromContext(ctx).Desugar().With(zap.Any("RefInfo", refInfo))
+
+	// Create A Control-Protocol PodIpGetter & Get The Pod IPs
+	podIpGetter := ctrlreconciler.PodIpGetter{Lister: r.podLister}
+	podIPs, err := podIpGetter.GetAllPodsIp(refInfo.DataPlaneNamespace, labels.Set(refInfo.DataPlaneLabels).AsSelector())
+	if err != nil {
+		logger.Error("Failed to getting data-plane Pod IPs", zap.Error(err))
+		return nil, err
+	}
+
+	// Append The Control-Protocol Server Port Number To The PodIPs If Not Already Present
+	for index, podIP := range podIPs {
+		if !strings.Contains(podIP, ":") {
+			podIPs[index] = fmt.Sprintf("%s:%d", podIP, ControlProtocolServerPort)
+		}
+	}
+	logger.Debug("Detected DataPlane Services", zap.Any("Pod IPs", podIPs))
+
+	// Define Service Callback Functions To Manage The Reconciler AsyncCommandNotificationStore
+	resetOffsetNamespacedName := types.NamespacedName{
+		Namespace: resetOffset.GetNamespace(),
+		Name:      resetOffset.GetName(),
+	}
+	newServiceCallbackFn := func(newHost string, service ctrl.Service) {
+		logger.Debug("New Control-Protocol Service Callback", zap.String("Host", newHost))
+		service.MessageHandler(r.asyncCommandNotificationStore.MessageHandler(resetOffsetNamespacedName, newHost))
+	}
+	oldServiceCallbackFn := func(oldHost string) {
+		logger.Debug("Old Control-Protocol Service Callback", zap.String("Host", oldHost))
+		r.asyncCommandNotificationStore.CleanPodNotification(resetOffsetNamespacedName, oldHost)
+	}
+
+	// Reconcile The Services/Connections For Specified Key / Pods
+	services, err := r.connectionPool.ReconcileConnections(ctx, refInfo.ConnectionPoolKey, podIPs, newServiceCallbackFn, oldServiceCallbackFn)
+	if err != nil {
+		logger.Error("Failed to reconcile connections", zap.Error(err))
+		return nil, err
+	}
+
+	// Return Success
+	return services, nil
+}
+
+// startConsumerGroups sends Start messages to the specified DataPlane services for a Topic / ConsumerGroup and
+// waits for the async responses.  A multi-error is returned if any ConsumerGroup was not started successfully.
+func (r *Reconciler) startConsumerGroups(ctx context.Context, resetOffset *kafkav1alpha1.ResetOffset, services map[string]ctrl.Service, refInfo *refmappers.RefInfo) error {
+	return r.sendConsumerGroupAsyncCommands(ctx, resetOffset, services, refInfo, commands.StartConsumerGroupOpCode)
+}
+
+// stopConsumerGroups sends Stop messages to the specified DataPlane services for a Topic / ConsumerGroup and
+// waits for the async responses.  A multi-error is returned if any ConsumerGroup was not stopped successfully.
+func (r *Reconciler) stopConsumerGroups(ctx context.Context, resetOffset *kafkav1alpha1.ResetOffset, services map[string]ctrl.Service, refInfo *refmappers.RefInfo) error {
+	return r.sendConsumerGroupAsyncCommands(ctx, resetOffset, services, refInfo, commands.StopConsumerGroupOpCode)
+}
+
+// sendConsumerGroupAsyncCommands sends ConsumerGroupAsyncCommands to the specified control-protocol Services in
+// parallel and blocks waiting for all the AsyncCommandResults.  Any errors are returned in a single multi-error.
+func (r *Reconciler) sendConsumerGroupAsyncCommands(ctx context.Context,
+	resetOffset *kafkav1alpha1.ResetOffset,
+	services map[string]ctrl.Service,
+	refInfo *refmappers.RefInfo,
+	opCode ctrl.OpCode) error {
+
+	// Send To All The ConsumerGroups In Parallel
+	waitGroup := &sync.WaitGroup{}
+	waitGroup.Add(len(services))
+	errChan := make(chan error, len(services))
+	for podIP, service := range services {
+		go func(podIP string, service ctrl.Service) {
+			defer waitGroup.Done()
+			err := r.sendConsumerGroupAsyncCommand(ctx, resetOffset, podIP, service, refInfo, opCode)
+			if err != nil {
+				errChan <- err
+			}
+		}(podIP, service)
+	}
+
+	// Wait For All The AsyncCommand Results
+	waitGroup.Wait()
+
+	// Close & Drain The Error Channel
+	close(errChan)
+	var multiErr error
+	for err := range errChan {
+		multierr.AppendInto(&multiErr, err)
+	}
+
+	// Return Any Errors
+	return multiErr
+}
+
+// sendConsumerGroupAsyncCommand sends a ConsumerGroupAsyncCommand with the specified opCode to the
+// specified pods and blocks waiting for AsyncCommandResult response which is then returned.
+func (r *Reconciler) sendConsumerGroupAsyncCommand(ctx context.Context,
+	resetOffset *kafkav1alpha1.ResetOffset,
+	podIP string,
+	service ctrl.Service,
+	refInfo *refmappers.RefInfo,
+	opCode ctrl.OpCode) error {
+
+	// Get The Logger From Context
+	logger := logging.FromContext(ctx).Desugar().With(zap.String("PodIP", podIP), zap.Int("OpCode", int(opCode)))
+
+	// Generate A CommandID For The ResetOffset
+	commandId, err := generateCommandId(resetOffset, podIP, opCode)
+	if err != nil {
+		logger.Error("Failed to generate Command ID for ResetOffset", zap.Error(err))
+		return fmt.Errorf("failed to generate Command ID for ResetOffset: %v", err)
+	}
+
+	// Create And Send A The ConsumerGroupAsyncCommand to The Control-Protocol Service
+	consumerGroupAsyncCommand := commands.NewConsumerGroupAsyncCommand(commandId, refInfo.TopicName, refInfo.GroupId)
+	err = service.SendAndWaitForAck(opCode, consumerGroupAsyncCommand)
+	if err != nil {
+		logger.Error("Failed to send ConsumerGroup AsyncCommand", zap.Int64("CommandID", commandId), zap.Error(err))
+		return fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d' : %v", commandId, err)
+	}
+
+	// Wait For The AsyncCommand Result & Return Results
+	return r.waitForAsyncCommandResult(resetOffset, podIP, consumerGroupAsyncCommand)
+}
+
+// waitForAsyncCommandResult polls the Reconciler AsyncCommandNotificationStore waiting for the
+// AsyncCommandResult corresponding to the specified AsyncCommand.  The ConsumerGroupAsyncCommands
+// are inherently asynchronous so that they can be used in other scenarios (Pause/Resume), but the
+// ResetOffset implementation treats them as Synchronous to facilitate single-pass reconciliation.
+func (r *Reconciler) waitForAsyncCommandResult(resetOffset *kafkav1alpha1.ResetOffset, podIP string, asyncCommand ctrlmessage.AsyncCommand) error {
+
+	// Create A NamespacedName For The ResetOffset
+	resetOffsetNamespacedName := types.NamespacedName{
+		Namespace: resetOffset.GetNamespace(),
+		Name:      resetOffset.GetName(),
+	}
+
+	// Poll The AsyncCommandNotificationStore For AsyncCommandResult
+	err := wait.Poll(asyncCommandResultPollDuration, asyncCommandResultTimeoutDuration, func() (done bool, err error) {
+		asyncCommandResult := r.asyncCommandNotificationStore.GetCommandResult(resetOffsetNamespacedName, podIP, asyncCommand)
+		if asyncCommandResult != nil {
+			if asyncCommandResult.IsFailed() {
+				return true, fmt.Errorf("AsyncCommand ID '%x' resulted in error: %s", asyncCommand.SerializedId(), asyncCommandResult.Error)
+			} else {
+				return true, nil // Return Success
+			}
+		} else {
+			return false, nil // Not Found - Try Again
+		}
+	})
+
+	// Return The Result
+	return err
+}
+
+// generateCommandId returns an int64 hash based on the specified ResetOffset.
+func generateCommandId(resetOffset *kafkav1alpha1.ResetOffset, podIP string, opCode ctrl.OpCode) (int64, error) {
+	hash := fnv.New32a()
+	_, err := hash.Write([]byte(fmt.Sprintf("%s-%d-%s-%d", string(resetOffset.UID), resetOffset.Generation, podIP, opCode)))
+	if err != nil {
+		return -1, err
+	}
+	return int64(hash.Sum32()), nil
+}
diff --git a/pkg/common/commands/resetoffset/controller/offsets.go b/pkg/common/commands/resetoffset/controller/offsets.go
index 956728068..ec67e78d7 100644
--- a/pkg/common/commands/resetoffset/controller/offsets.go
+++ b/pkg/common/commands/resetoffset/controller/offsets.go
@@ -27,6 +27,7 @@ import (
 	"knative.dev/pkg/logging"
 
 	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
 )
 
 // PartitionOffsetManagers is a map of Partition -> Sarama PartitionOffsetManager
@@ -51,12 +52,12 @@ var SaramaNewOffsetManagerFromClientFn SaramaNewOffsetManagerFromClientFnType =
 // offsetTime (millis since epoch) and return OffsetMappings of the old/new
 // state.  An error will be returned and the Offsets will not be committed
 // if any problems occur.
-func (r *Reconciler) reconcileOffsets(ctx context.Context, topicName string, groupId string, offsetTime int64) ([]kafkav1alpha1.OffsetMapping, error) {
+func (r *Reconciler) reconcileOffsets(ctx context.Context, refInfo *refmappers.RefInfo, offsetTime int64) ([]kafkav1alpha1.OffsetMapping, error) {
 
 	// Get The Logger From The Context & Enhance The With Parameters
 	logger := logging.FromContext(ctx).Desugar().With(
-		zap.String("Topic", topicName),
-		zap.String("Group", groupId),
+		zap.String("Topic", refInfo.TopicName),
+		zap.String("Group", refInfo.GroupId),
 		zap.Int64("Time", offsetTime))
 
 	// Initialize A New Sarama Client
@@ -74,7 +75,7 @@ func (r *Reconciler) reconcileOffsets(ctx context.Context, topicName string, gro
 	}
 
 	// Get The Partitions Of The Specified Kafka Topic
-	partitions, err := saramaClient.Partitions(topicName)
+	partitions, err := saramaClient.Partitions(refInfo.TopicName)
 	if err != nil {
 		logger.Error("Failed to determine Partitions for Topic", zap.Error(err))
 		return nil, err
@@ -82,14 +83,14 @@ func (r *Reconciler) reconcileOffsets(ctx context.Context, topicName string, gro
 	logger.Debug("Found Topic Partitions", zap.Any("Partitions", partitions))
 
 	// Create An OffsetManager For The Specified ConsumerGroup
-	offsetManager, err := SaramaNewOffsetManagerFromClientFn(groupId, saramaClient)
+	offsetManager, err := SaramaNewOffsetManagerFromClientFn(refInfo.GroupId, saramaClient)
 	if offsetManager == nil || err != nil {
 		logger.Error("Failed to create OffsetManager for ConsumerGroup", zap.Error(err))
 		return nil, err
 	}
 
 	// Create The Required PartitionOffsetManagers For The Specified Topic / Partitions
-	partitionOffsetManagers, err := createPartitionOffsetManagers(offsetManager, topicName, partitions)
+	partitionOffsetManagers, err := createPartitionOffsetManagers(offsetManager, refInfo.TopicName, partitions)
 	if err != nil {
 		logger.Error("Failed to create PartitionOffsetManagers for Topic Partitions", zap.Error(err))
 		_ = closeManagersAndDrainErrors(logger, offsetManager, partitionOffsetManagers)
@@ -97,7 +98,7 @@ func (r *Reconciler) reconcileOffsets(ctx context.Context, topicName string, gro
 	}
 
 	// Update All Topic Partitions To The Specified Offset Time
-	offsetMappings, err := updateOffsets(logger, saramaClient, offsetManager, partitionOffsetManagers, topicName, partitions, offsetTime)
+	offsetMappings, err := updateOffsets(logger, saramaClient, offsetManager, partitionOffsetManagers, refInfo.TopicName, partitions, offsetTime)
 	if err != nil {
 		logger.Error("Failed to update Offsets for Topic Partitions", zap.Error(err))
 		_ = closeManagersAndDrainErrors(logger, offsetManager, partitionOffsetManagers)
diff --git a/pkg/common/commands/resetoffset/controller/reconciler.go b/pkg/common/commands/resetoffset/controller/reconciler.go
index 1d1cdbc6d..1806e4dd7 100644
--- a/pkg/common/commands/resetoffset/controller/reconciler.go
+++ b/pkg/common/commands/resetoffset/controller/reconciler.go
@@ -24,6 +24,10 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	corev1listers "k8s.io/client-go/listers/core/v1"
+	control "knative.dev/control-protocol/pkg"
+	ctrlreconciler "knative.dev/control-protocol/pkg/reconciler"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/reconciler"
 
@@ -44,10 +48,13 @@ var (
 
 // Reconciler Implements controller.Reconciler for ResetOffset Resources
 type Reconciler struct {
-	kafkaBrokers      []string
-	saramaConfig      *sarama.Config
-	resetoffsetLister kafkalisters.ResetOffsetLister
-	refMapper         refmappers.ResetOffsetRefMapper
+	kafkaBrokers                  []string
+	saramaConfig                  *sarama.Config
+	podLister                     corev1listers.PodLister
+	resetoffsetLister             kafkalisters.ResetOffsetLister
+	refMapper                     refmappers.ResetOffsetRefMapper
+	connectionPool                ctrlreconciler.ControlPlaneConnectionPool
+	asyncCommandNotificationStore ctrlreconciler.AsyncCommandNotificationStore
 }
 
 // ReconcileKind implements the Reconciler Interface and is responsible for performing Offset repositioning.
@@ -57,7 +64,14 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, resetOffset *kafkav1alph
 	logger := logging.FromContext(ctx).Desugar()
 	logger.Debug("<==========  START RESET-OFFSET RECONCILIATION  ==========>")
 
+	//
 	// Ignore Previously Initiated ResetOffset Instances
+	//
+	// Normally this would be achieved via a cache.FilteringResourceEventHandler{} directly
+	// in the controller, but that only has a metav1.Object to deal with which does not
+	// expose the ResetOffset.Status we need to make the filtering decision, so we are left
+	// to handle it here in the Reconciler.
+	//
 	if resetOffset.Status.IsInitiated() || resetOffset.Status.IsCompleted() {
 		logger.Debug("Skipping reconciliation of previously executed ResetOffset instance")
 		return reconciler.NewEvent(corev1.EventTypeNormal, ResetOffsetSkipped.String(), "Skipped previously executed ResetOffset")
@@ -67,15 +81,15 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, resetOffset *kafkav1alph
 	resetOffset.Status.InitializeConditions()
 
 	// Map The ResetOffset's Ref To Kafka Topic Name / ConsumerGroup ID
-	topic, group, err := r.refMapper.MapRef(resetOffset)
+	refInfo, err := r.refMapper.MapRef(resetOffset)
 	if err != nil {
 		logger.Error("Failed to map ResetOffset.Spec.Ref to Kafka Topic name and ConsumerGroup ID", zap.Error(err))
 		resetOffset.Status.MarkRefMappedFailed("FailedToMapRef", "Failed to map 'ref' to Kafka Topic and Group: %v", err)
 		return fmt.Errorf("failed to map 'ref' to Kafka Topic and Group: %v", err)
 	}
-	logger.Info("Successfully mapped ResetOffset.Spec.Ref to Kafka Topic name and ConsumerGroup ID", zap.String("Topic", topic), zap.String("Group", group))
-	resetOffset.Status.SetTopic(topic)
-	resetOffset.Status.SetGroup(group)
+	logger.Info("Successfully mapped ResetOffset.Spec.Ref", zap.Any("RefInfo", refInfo))
+	resetOffset.Status.SetTopic(refInfo.TopicName)
+	resetOffset.Status.SetGroup(refInfo.GroupId)
 	resetOffset.Status.MarkRefMappedTrue()
 
 	// Parse The Sarama Offset Time From ResetOffset Spec
@@ -86,19 +100,34 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, resetOffset *kafkav1alph
 	}
 	logger.Info("Successfully parsed Sarama Offset time from ResetOffset Spec", zap.Int64("Time (millis)", offsetTime))
 
-	// TODO - Map ResetOffset.Spec.Ref to "owning" Dispatcher Deployment/Service (Consolidated = one, Distributed = many)
+	// Reconcile The DataPlane "Services" From The ConnectionPool For Specified Key
+	dataPlaneServices, err := r.reconcileDataPlaneServices(ctx, resetOffset, refInfo)
+	if err != nil {
+		logger.Error("Failed to reconcile DataPlane services from ConnectionPool", zap.Error(err))
+		// TODO - how to handle this - need another status condition and mark as failed?
+		return fmt.Errorf("failed to reconcile DataPlane Services from ConnectionPool: %v", err)
+	}
+	logger.Info("Successfully reconciled DataPlane Services", zap.Any("Services", dataPlaneServiceIPs(dataPlaneServices)))
 
 	// TODO - Send "Initiate" Message to Dispatcher Replicas to lock mutext and start process.
 	resetOffset.Status.MarkResetInitiatedTrue()
 
-	// TODO - Send "Stop" Message to Dispatcher Replicas to stop ConsumerGroups
+	// Stop The ConsumerGroup In Associated Dispatchers
+	err = r.stopConsumerGroups(ctx, resetOffset, dataPlaneServices, refInfo)
+	if err != nil {
+		logger.Error("Failed to stop ConsumerGroups", zap.Error(err))
+		resetOffset.Status.MarkConsumerGroupsStoppedFailed("FailedToStopConsumerGroups", "Failed to stop one or more ConsumerGroups: %v", err)
+		// TODO - should we re-reconcile and keep trying (stuck) or rollback here or in stopConsumerGroups() ???
+		return nil
+	}
+	logger.Info("Successfully stopped ConsumerGroups")
 	resetOffset.Status.MarkConsumerGroupsStoppedTrue()
 
 	// Update The Sarama Offsets & Update ResetOffset CRD With OffsetMappings
-	offsetMappings, err := r.reconcileOffsets(ctx, topic, group, offsetTime)
+	offsetMappings, err := r.reconcileOffsets(ctx, refInfo, offsetTime)
 	if err != nil {
 		logger.Error("Failed to update Offsets of one or more partitions", zap.Error(err))
-		resetOffset.Status.MarkOffsetsUpdatedFailed("FailedToUpdateOffsets", "Failed to update offsets of one or more partitions: %v", err)
+		resetOffset.Status.MarkOffsetsUpdatedFailed("FailedToUpdateOffsets", "Failed to update Offsets of one or more Partitions: %v", err)
 		return fmt.Errorf("failed to update offsets of one or more partitions: %v", err)
 	}
 	if offsetMappings != nil {
@@ -107,7 +136,15 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, resetOffset *kafkav1alph
 	logger.Info("Successfully updated Offsets of all partitions")
 	resetOffset.Status.MarkOffsetsUpdatedTrue()
 
-	// TODO - Send "Start" Message to Dispatcher Replicas to restart ConsumerGroups
+	// Start The ConsumerGroup In Associated Dispatchers
+	err = r.startConsumerGroups(ctx, resetOffset, dataPlaneServices, refInfo)
+	if err != nil {
+		logger.Error("Failed to start ConsumerGroups", zap.Error(err))
+		resetOffset.Status.MarkConsumerGroupsStartedFailed("FailedToStartConsumerGroups", "Failed to start one or more ConsumerGroups: %v", err)
+		// TODO - should we re-reconcile and keep trying (stuck) or rollback here or in stopConsumerGroups() ???
+		return nil // TODO - for now stopping without re-queueing
+	}
+	logger.Info("Successfully started ConsumerGroups")
 	resetOffset.Status.MarkConsumerGroupsStartedTrue()
 
 	// Return Reconciled Success Event
@@ -127,6 +164,12 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, resetOffset *kafkav1alpha
 		return fmt.Errorf("skipping finalization of in-progress ResetOffset instance")
 	}
 
+	// Clean The AsyncCommandNotificationStore
+	r.asyncCommandNotificationStore.CleanPodsNotifications(types.NamespacedName{
+		Namespace: resetOffset.Namespace,
+		Name:      resetOffset.Name,
+	})
+
 	// No-Op Finalization - Nothing To Do
 	logger.Info("No-Op Finalization Successful")
 
@@ -183,3 +226,14 @@ func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.Co
 	// Update Reconciler With New Config
 	r.saramaConfig = ekConfig.Sarama.Config
 }
+
+// dataPlaneServiceIPs returns the control-protocol Service IPs which are the keys in the specified map.
+func dataPlaneServiceIPs(services map[string]control.Service) []string {
+	i := 0
+	ips := make([]string, len(services))
+	for ip := range services {
+		ips[i] = ip
+		i++
+	}
+	return ips
+}
diff --git a/pkg/common/commands/resetoffset/refmappers/subscription.go b/pkg/common/commands/resetoffset/refmappers/subscription.go
index 87162e48e..db227fc05 100644
--- a/pkg/common/commands/resetoffset/refmappers/subscription.go
+++ b/pkg/common/commands/resetoffset/refmappers/subscription.go
@@ -40,20 +40,39 @@ var _ ResetOffsetRefMapperFactory = &SubscriptionRefMapperFactory{}
 
 // SubscriptionRefMapperFactory implements the ResetOffsetRefMapperFactory for Knative Subscriptions
 type SubscriptionRefMapperFactory struct {
-	TopicNameMapper SubscriptionTopicNameMapper
-	GroupIdMapper   SubscriptionConsumerGroupIdMapper
+	TopicNameMapper          SubscriptionTopicNameMapper
+	GroupIdMapper            SubscriptionConsumerGroupIdMapper
+	ConnectionPoolKeyMapper  SubscriptionConnectionPoolKeyMapper
+	DataPlaneNamespaceMapper SubscriptionDataPlaneNamespaceMapper
+	DataPlaneLabelsMapper    SubscriptionDataPlaneLabelsMapper
 }
 
 // NewSubscriptionRefMapperFactory returns an initialized SubscriptionRefMapperFactory
-func NewSubscriptionRefMapperFactory(topicNameMapper SubscriptionTopicNameMapper, groupIdMapper SubscriptionConsumerGroupIdMapper) *SubscriptionRefMapperFactory {
-	return &SubscriptionRefMapperFactory{TopicNameMapper: topicNameMapper, GroupIdMapper: groupIdMapper}
+func NewSubscriptionRefMapperFactory(topicNameMapper SubscriptionTopicNameMapper,
+	groupIdMapper SubscriptionConsumerGroupIdMapper,
+	connectionPoolKeyMapper SubscriptionConnectionPoolKeyMapper,
+	dataPlaneNamespaceMapper SubscriptionDataPlaneNamespaceMapper,
+	dataPlaneLabelsMapper SubscriptionDataPlaneLabelsMapper) *SubscriptionRefMapperFactory {
+
+	return &SubscriptionRefMapperFactory{
+		TopicNameMapper:          topicNameMapper,
+		GroupIdMapper:            groupIdMapper,
+		ConnectionPoolKeyMapper:  connectionPoolKeyMapper,
+		DataPlaneNamespaceMapper: dataPlaneNamespaceMapper,
+		DataPlaneLabelsMapper:    dataPlaneLabelsMapper,
+	}
 }
 
 // Create implements the ResetOffsetRefMapperFactory interface for Subscription references.  It will return
 // a new SubscriptionRefMapper instance using the specific TopicName / GroupId mappers.  It also relies on
 // the Context having injected informers (SubscriptionInformer).
 func (f *SubscriptionRefMapperFactory) Create(ctx context.Context) ResetOffsetRefMapper {
-	return NewSubscriptionRefMapper(ctx, f.TopicNameMapper, f.GroupIdMapper)
+	return NewSubscriptionRefMapper(ctx,
+		f.TopicNameMapper,
+		f.GroupIdMapper,
+		f.ConnectionPoolKeyMapper,
+		f.DataPlaneNamespaceMapper,
+		f.DataPlaneLabelsMapper)
 }
 
 //
@@ -66,21 +85,36 @@ type SubscriptionTopicNameMapper func(*messagingv1.Subscription) (string, error)
 // SubscriptionConsumerGroupIdMapper defines a function signature for mapping a Subscription to a Kafka ConsumerGroup ID.
 type SubscriptionConsumerGroupIdMapper func(*messagingv1.Subscription) (string, error)
 
+// SubscriptionConnectionPoolKeyMapper defines a function signature for mapping a Subscription to a control-protocol ControlPlaneConnectionPool Key.
+type SubscriptionConnectionPoolKeyMapper func(*messagingv1.Subscription) (string, error)
+
+// SubscriptionDataPlaneNamespaceMapper defines a function signature for mapping a Subscription to the Kubernetes namespace of the DataPlane components.
+type SubscriptionDataPlaneNamespaceMapper func(subscription *messagingv1.Subscription) (string, error)
+
+// SubscriptionDataPlaneLabelsMapper defines a function signature for mapping a Subscription to the Kubernetes labels of the DataPlane Pods.
+type SubscriptionDataPlaneLabelsMapper func(subscription *messagingv1.Subscription) (map[string]string, error)
+
 // Verify The Subscription ResetOffsetRefMapper Implements The Interface
 var _ ResetOffsetRefMapper = &SubscriptionRefMapper{}
 
 // SubscriptionRefMapper implements the ResetOffsetRefMapper for Knative Subscriptions
 type SubscriptionRefMapper struct {
-	logger             *zap.Logger
-	subscriptionLister messaginglisters.SubscriptionLister
-	topicNameMapper    SubscriptionTopicNameMapper
-	groupIdMapper      SubscriptionConsumerGroupIdMapper
+	logger                   *zap.Logger
+	subscriptionLister       messaginglisters.SubscriptionLister
+	topicNameMapper          SubscriptionTopicNameMapper
+	groupIdMapper            SubscriptionConsumerGroupIdMapper
+	connectionPoolKeyMapper  SubscriptionConnectionPoolKeyMapper
+	dataPlaneNamespaceMapper SubscriptionDataPlaneNamespaceMapper
+	dataPlaneLabelsMapper    SubscriptionDataPlaneLabelsMapper
 }
 
 // NewSubscriptionRefMapper returns an initialized ResetOffsetSubscriptionRefMapper
 func NewSubscriptionRefMapper(ctx context.Context,
 	topicNameMapper SubscriptionTopicNameMapper,
-	groupIdMapper SubscriptionConsumerGroupIdMapper) *SubscriptionRefMapper {
+	groupIdMapper SubscriptionConsumerGroupIdMapper,
+	connectionPoolKeyMapper SubscriptionConnectionPoolKeyMapper,
+	dataPlaneNamespaceMapper SubscriptionDataPlaneNamespaceMapper,
+	dataPlaneLabelsMapper SubscriptionDataPlaneLabelsMapper) *SubscriptionRefMapper {
 
 	// Get The Logger From Context
 	logger := logging.FromContext(ctx).Desugar()
@@ -90,21 +124,24 @@ func NewSubscriptionRefMapper(ctx context.Context,
 
 	// Return An Initialized ResetOffsetSubscriptionRefMapper
 	return &SubscriptionRefMapper{
-		logger:             logger,
-		subscriptionLister: subscriptionInformer.Lister(),
-		topicNameMapper:    topicNameMapper,
-		groupIdMapper:      groupIdMapper,
+		logger:                   logger,
+		subscriptionLister:       subscriptionInformer.Lister(),
+		topicNameMapper:          topicNameMapper,
+		groupIdMapper:            groupIdMapper,
+		connectionPoolKeyMapper:  connectionPoolKeyMapper,
+		dataPlaneNamespaceMapper: dataPlaneNamespaceMapper,
+		dataPlaneLabelsMapper:    dataPlaneLabelsMapper,
 	}
 }
 
 // MapRef implements the ResetOffsetRefMapper interface for Subscription references. It will return an
 // error in all cases other than successfully mapping the ResetOffset.Spec.Ref to a Kafka Topic / Group.
-func (m *SubscriptionRefMapper) MapRef(resetOffset *kafkav1alpha1.ResetOffset) (string, string, error) {
+func (m *SubscriptionRefMapper) MapRef(resetOffset *kafkav1alpha1.ResetOffset) (*RefInfo, error) {
 
 	// Validate The ResetOffset
 	if resetOffset == nil {
 		m.logger.Warn("Received nil ResetOffset argument")
-		return "", "", fmt.Errorf("unable to map nil ResetOffset")
+		return nil, fmt.Errorf("unable to map nil ResetOffset")
 	}
 
 	// Get The ResetOffset Ref From Spec & Enhance Logger
@@ -114,11 +151,11 @@ func (m *SubscriptionRefMapper) MapRef(resetOffset *kafkav1alpha1.ResetOffset) (
 	// Validate The Reference
 	if !strings.HasPrefix(ref.APIVersion, messaging.GroupName) || ref.Kind != "Subscription" {
 		m.logger.Warn("Received ResetOffset with non Subscription reference")
-		return "", "", fmt.Errorf("received ResetOffset with non Subscription reference: %v", ref)
+		return nil, fmt.Errorf("received ResetOffset with non Subscription reference: %v", ref)
 	}
 	if ref.Name == "" {
 		m.logger.Warn("Received ResetOffset with unnamed Subscription reference")
-		return "", "", fmt.Errorf("received ResetOffset with unnamed Subscription reference: %v", ref)
+		return nil, fmt.Errorf("received ResetOffset with unnamed Subscription reference: %v", ref)
 	}
 
 	// Default Optional Ref.Namespace If Not Provided
@@ -131,27 +168,57 @@ func (m *SubscriptionRefMapper) MapRef(resetOffset *kafkav1alpha1.ResetOffset) (
 	subscription, err := m.subscriptionLister.Subscriptions(refNamespace).Get(ref.Name)
 	if err != nil {
 		logger.Error("Failed to get Subscription referenced by ResetOffset", zap.Error(err))
-		return "", "", fmt.Errorf("failed to get Subscription referenced by ResetOffset.Spec.Ref '%v': %v", ref, err)
+		return nil, fmt.Errorf("failed to get Subscription referenced by ResetOffset.Spec.Ref '%v': %v", ref, err)
 	}
 	if subscription == nil {
 		logger.Info("No Subscription found for ResetOffset reference")
-		return "", "", fmt.Errorf("no Subscription found for ResetOffset.Spec.Ref %v", ref)
+		return nil, fmt.Errorf("no Subscription found for ResetOffset.Spec.Ref %v", ref)
 	}
 
 	// Map The Subscription To Kafka Topic Name Via Custom SubscriptionTopicNameMapper
 	topicName, err := m.topicNameMapper(subscription)
 	if err != nil {
 		logger.Error("Failed to map Knative Subscription to Kafka Topic name", zap.Error(err))
-		return "", "", fmt.Errorf("failed to map Knative Subscription '%v' to Kafka Topic name: %v", ref, err)
+		return nil, fmt.Errorf("failed to map Knative Subscription '%v' to Kafka Topic name: %v", ref, err)
 	}
 
 	// Map The Subscription To Kafka ConsumerGroup ID Via Custom SubscriptionGroupIdMapper
 	groupId, err := m.groupIdMapper(subscription)
 	if err != nil {
 		logger.Error("Failed to map Knative Subscription to Kafka ConsumerGroup ID", zap.Error(err))
-		return "", "", fmt.Errorf("failed to map Knative Subscription '%v' to Kafka ConsumerGroup ID: %v", ref, err)
+		return nil, fmt.Errorf("failed to map Knative Subscription '%v' to Kafka ConsumerGroup ID: %v", ref, err)
+	}
+
+	// Map The Subscription To The control-protocol ControlPlaneConnectionPool Key Via Custom SubscriptionConnectionPoolKeyMapper
+	connectionPoolKey, err := m.connectionPoolKeyMapper(subscription)
+	if err != nil {
+		logger.Error("Failed to map Knative Subscription to ControlPlaneConnectionPool Key", zap.Error(err))
+		return nil, fmt.Errorf("failed to map Knative Subscription '%v' to ConnectinPool Key: %v", ref, err)
+	}
+
+	// Map The Subscription To The Kubernetes Namespace of the DataPlane Pods.
+	dataPlaneNamespace, err := m.dataPlaneNamespaceMapper(subscription)
+	if err != nil {
+		logger.Error("Failed to map Knative Subscription to DataPlane Namespace", zap.Error(err))
+		return nil, fmt.Errorf("failed to map Knative Subscription '%v' to DataPlane Namespace: %v", ref, err)
+	}
+
+	// Map The Subscription To The Kubernetes Namespace of the DataPlane Pods.
+	dataPlaneLabels, err := m.dataPlaneLabelsMapper(subscription)
+	if err != nil {
+		logger.Error("Failed to map Knative Subscription to DataPlane Pod Labels", zap.Error(err))
+		return nil, fmt.Errorf("failed to map Knative Subscription '%v' to DataPlane Pod Labels: %v", ref, err)
+	}
+
+	// Create The RefInfo Struct
+	refInfo := &RefInfo{
+		TopicName:          topicName,
+		GroupId:            groupId,
+		ConnectionPoolKey:  connectionPoolKey,
+		DataPlaneNamespace: dataPlaneNamespace,
+		DataPlaneLabels:    dataPlaneLabels,
 	}
 
-	// Successfully Mapped The Ref To Kafka Topic / Group - Return Results
-	return topicName, groupId, nil
+	// Successfully Mapped The Ref - Return Results
+	return refInfo, nil
 }
diff --git a/pkg/common/commands/resetoffset/refmappers/types.go b/pkg/common/commands/resetoffset/refmappers/types.go
index 8d8007c1e..4131ede22 100644
--- a/pkg/common/commands/resetoffset/refmappers/types.go
+++ b/pkg/common/commands/resetoffset/refmappers/types.go
@@ -26,12 +26,25 @@ import (
 // instances based on the provided Context which comes from SharedMain and includes the
 // injected Informers.  It is necessary for delayed initialization against that Context.
 type ResetOffsetRefMapperFactory interface {
-	Create(ctx context.Context) ResetOffsetRefMapper
+	Create(context.Context) ResetOffsetRefMapper
 }
 
 // ResetOffsetRefMapper defines the interface for the capability to map ResetOffset.Spec.Ref
-// to Kafka Topic / Group. This abstraction allows for future extensibility to support
-// additional KRef types, such as Triggers, instead of just Subscriptions.
+// to Kafka Topic, Group and ConnectionPool Key. This abstraction allows for future
+// extensibility to support additional KRef types, such as Triggers, instead of just
+// Subscriptions.
 type ResetOffsetRefMapper interface {
-	MapRef(resetOffset *kafkav1alpha1.ResetOffset) (topic string, group string, err error)
+	MapRef(*kafkav1alpha1.ResetOffset) (*RefInfo, error)
+}
+
+// RefInfo contains the data necessary for ResetOffset reconciliation which is specific
+// to a particular use-case, as provided by a customized ResetOffsetRefMapper implementation.
+// This allows implementations of Kafka Channels/Brokers/etc to differ from one another
+// and still make use of the shared ResetOffset Controller.
+type RefInfo struct {
+	TopicName          string
+	GroupId            string
+	ConnectionPoolKey  string
+	DataPlaneNamespace string
+	DataPlaneLabels    map[string]string
 }
diff --git a/pkg/common/controlprotocol/commands/consumergroup.go b/pkg/common/controlprotocol/commands/consumergroup.go
new file mode 100644
index 000000000..96fd7e0da
--- /dev/null
+++ b/pkg/common/controlprotocol/commands/consumergroup.go
@@ -0,0 +1,69 @@
+/*
+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 commands
+
+import (
+	"encoding/json"
+
+	ctrl "knative.dev/control-protocol/pkg"
+	ctrlmessage "knative.dev/control-protocol/pkg/message"
+)
+
+const (
+	ConsumerGroupAsyncCommandVersion int16 = 1 // Basic AsyncCommand Compatibility Check
+
+	StopConsumerGroupOpCode        ctrl.OpCode = 10
+	StopConsumerGroupResultOpCode  ctrl.OpCode = 11
+	StartConsumerGroupOpCode       ctrl.OpCode = 12
+	StartConsumerGroupResultOpCode ctrl.OpCode = 13
+)
+
+// Verify The ConsumerGroupAsyncCommand Implements The Control-Protocol AsyncCommand Interface
+var _ ctrlmessage.AsyncCommand = (*ConsumerGroupAsyncCommand)(nil)
+
+// ConsumerGroupAsyncCommand implements an AsyncCommand for handling ConsumerGroup related operations.
+type ConsumerGroupAsyncCommand struct {
+	Version   int16  `json:"version"`
+	CommandId int64  `json:"id"`
+	TopicName string `json:"topic"`
+	GroupId   string `json:"group"`
+}
+
+// NewConsumerGroupAsyncCommand constructs and returns a new ConsumerGroupAsyncCommand.
+func NewConsumerGroupAsyncCommand(commandId int64, topicName string, groupId string) *ConsumerGroupAsyncCommand {
+	return &ConsumerGroupAsyncCommand{
+		Version:   ConsumerGroupAsyncCommandVersion, // Only One Version For Now - Validate Compatibility In Handler ; )
+		CommandId: commandId,
+		TopicName: topicName,
+		GroupId:   groupId,
+	}
+}
+
+// MarshalBinary implements the Control-Protocol AsyncCommand interface.
+func (s *ConsumerGroupAsyncCommand) MarshalBinary() (data []byte, err error) {
+	return json.Marshal(s)
+}
+
+// UnmarshalBinary implements the Control-Protocol AsyncCommand interface.
+func (s *ConsumerGroupAsyncCommand) UnmarshalBinary(data []byte) error {
+	return json.Unmarshal(data, &s)
+}
+
+// SerializedId implements the Control-Protocol AsyncCommand interface.
+func (s *ConsumerGroupAsyncCommand) SerializedId() []byte {
+	return ctrlmessage.Int64CommandId(s.CommandId)
+}
diff --git a/vendor/github.com/Shopify/sarama/acl_types.go b/vendor/github.com/Shopify/sarama/acl_types.go
index 16a21fc3f..c3ba8ddcf 100644
--- a/vendor/github.com/Shopify/sarama/acl_types.go
+++ b/vendor/github.com/Shopify/sarama/acl_types.go
@@ -141,6 +141,7 @@ const (
 	AclResourceGroup
 	AclResourceCluster
 	AclResourceTransactionalID
+	AclResourceDelegationToken
 )
 
 func (a *AclResourceType) String() string {
@@ -151,6 +152,7 @@ func (a *AclResourceType) String() string {
 		AclResourceGroup:           "Group",
 		AclResourceCluster:         "Cluster",
 		AclResourceTransactionalID: "TransactionalID",
+		AclResourceDelegationToken: "DelegationToken",
 	}
 	s, ok := mapping[*a]
 	if !ok {
@@ -174,6 +176,7 @@ func (a *AclResourceType) UnmarshalText(text []byte) error {
 		"group":           AclResourceGroup,
 		"cluster":         AclResourceCluster,
 		"transactionalid": AclResourceTransactionalID,
+		"delegationtoken": AclResourceDelegationToken,
 	}
 
 	art, ok := mapping[normalized]
diff --git a/vendor/github.com/Shopify/sarama/alter_configs_response.go b/vendor/github.com/Shopify/sarama/alter_configs_response.go
index 89065ab11..cfb6369ac 100644
--- a/vendor/github.com/Shopify/sarama/alter_configs_response.go
+++ b/vendor/github.com/Shopify/sarama/alter_configs_response.go
@@ -23,16 +23,9 @@ func (a *AlterConfigsResponse) encode(pe packetEncoder) error {
 		return err
 	}
 
-	for i := range a.Resources {
-		pe.putInt16(a.Resources[i].ErrorCode)
-		err := pe.putString(a.Resources[i].ErrorMsg)
-		if err != nil {
-			return nil
-		}
-		pe.putInt8(int8(a.Resources[i].Type))
-		err = pe.putString(a.Resources[i].Name)
-		if err != nil {
-			return nil
+	for _, v := range a.Resources {
+		if err := v.encode(pe); err != nil {
+			return err
 		}
 	}
 
@@ -56,30 +49,52 @@ func (a *AlterConfigsResponse) decode(pd packetDecoder, version int16) error {
 	for i := range a.Resources {
 		a.Resources[i] = new(AlterConfigsResourceResponse)
 
-		errCode, err := pd.getInt16()
-		if err != nil {
+		if err := a.Resources[i].decode(pd, version); err != nil {
 			return err
 		}
-		a.Resources[i].ErrorCode = errCode
+	}
 
-		e, err := pd.getString()
-		if err != nil {
-			return err
-		}
-		a.Resources[i].ErrorMsg = e
+	return nil
+}
 
-		t, err := pd.getInt8()
-		if err != nil {
-			return err
-		}
-		a.Resources[i].Type = ConfigResourceType(t)
+func (a *AlterConfigsResourceResponse) encode(pe packetEncoder) error {
+	pe.putInt16(a.ErrorCode)
+	err := pe.putString(a.ErrorMsg)
+	if err != nil {
+		return nil
+	}
+	pe.putInt8(int8(a.Type))
+	err = pe.putString(a.Name)
+	if err != nil {
+		return nil
+	}
+	return nil
+}
 
-		name, err := pd.getString()
-		if err != nil {
-			return err
-		}
-		a.Resources[i].Name = name
+func (a *AlterConfigsResourceResponse) decode(pd packetDecoder, version int16) error {
+	errCode, err := pd.getInt16()
+	if err != nil {
+		return err
+	}
+	a.ErrorCode = errCode
+
+	e, err := pd.getString()
+	if err != nil {
+		return err
+	}
+	a.ErrorMsg = e
+
+	t, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.Type = ConfigResourceType(t)
+
+	name, err := pd.getString()
+	if err != nil {
+		return err
 	}
+	a.Name = name
 
 	return nil
 }
diff --git a/vendor/github.com/Shopify/sarama/broker.go b/vendor/github.com/Shopify/sarama/broker.go
index 664b74f54..dd01e4ef1 100644
--- a/vendor/github.com/Shopify/sarama/broker.go
+++ b/vendor/github.com/Shopify/sarama/broker.go
@@ -666,6 +666,18 @@ func (b *Broker) AlterConfigs(request *AlterConfigsRequest) (*AlterConfigsRespon
 	return response, nil
 }
 
+// IncrementalAlterConfigs sends a request to incremental alter config and return a response or error
+func (b *Broker) IncrementalAlterConfigs(request *IncrementalAlterConfigsRequest) (*IncrementalAlterConfigsResponse, error) {
+	response := new(IncrementalAlterConfigsResponse)
+
+	err := b.sendAndReceive(request, response)
+	if err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}
+
 // DeleteGroups sends a request to delete groups and returns a response or error
 func (b *Broker) DeleteGroups(request *DeleteGroupsRequest) (*DeleteGroupsResponse, error) {
 	response := new(DeleteGroupsResponse)
diff --git a/vendor/github.com/Shopify/sarama/consumer.go b/vendor/github.com/Shopify/sarama/consumer.go
index 46c26134c..f9cd172b4 100644
--- a/vendor/github.com/Shopify/sarama/consumer.go
+++ b/vendor/github.com/Shopify/sarama/consumer.go
@@ -468,9 +468,7 @@ feederLoop:
 		}
 
 		for i, msg := range msgs {
-			for _, interceptor := range child.conf.Consumer.Interceptors {
-				msg.safelyApplyInterceptor(interceptor)
-			}
+			child.interceptors(msg)
 		messageSelect:
 			select {
 			case <-child.dying:
@@ -484,6 +482,7 @@ feederLoop:
 					child.broker.acks.Done()
 				remainingLoop:
 					for _, msg = range msgs[i:] {
+						child.interceptors(msg)
 						select {
 						case child.messages <- msg:
 						case <-child.dying:
@@ -715,6 +714,12 @@ func (child *partitionConsumer) parseResponse(response *FetchResponse) ([]*Consu
 	return messages, nil
 }
 
+func (child *partitionConsumer) interceptors(msg *ConsumerMessage) {
+	for _, interceptor := range child.conf.Consumer.Interceptors {
+		msg.safelyApplyInterceptor(interceptor)
+	}
+}
+
 type brokerConsumer struct {
 	consumer         *consumer
 	broker           *Broker
diff --git a/vendor/github.com/Shopify/sarama/consumer_group.go b/vendor/github.com/Shopify/sarama/consumer_group.go
index 80f625d80..2bf236ae5 100644
--- a/vendor/github.com/Shopify/sarama/consumer_group.go
+++ b/vendor/github.com/Shopify/sarama/consumer_group.go
@@ -732,6 +732,9 @@ func (s *consumerGroupSession) heartbeatLoop() {
 	pause := time.NewTicker(s.parent.config.Consumer.Group.Heartbeat.Interval)
 	defer pause.Stop()
 
+	retryBackoff := time.NewTimer(s.parent.config.Metadata.Retry.Backoff)
+	defer retryBackoff.Stop()
+
 	retries := s.parent.config.Metadata.Retry.Max
 	for {
 		coordinator, err := s.parent.client.Coordinator(s.parent.groupID)
@@ -740,11 +743,11 @@ func (s *consumerGroupSession) heartbeatLoop() {
 				s.parent.handleError(err, "", -1)
 				return
 			}
-
+			retryBackoff.Reset(s.parent.config.Metadata.Retry.Backoff)
 			select {
 			case <-s.hbDying:
 				return
-			case <-time.After(s.parent.config.Metadata.Retry.Backoff):
+			case <-retryBackoff.C:
 				retries--
 			}
 			continue
diff --git a/vendor/github.com/Shopify/sarama/go.mod b/vendor/github.com/Shopify/sarama/go.mod
index 3cc2b5b9e..ccbd8e2d1 100644
--- a/vendor/github.com/Shopify/sarama/go.mod
+++ b/vendor/github.com/Shopify/sarama/go.mod
@@ -19,8 +19,9 @@ require (
 	github.com/stretchr/testify v1.7.0
 	github.com/xdg/scram v1.0.3
 	github.com/xdg/stringprep v1.0.3 // indirect
-	golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b // indirect
-	golang.org/x/net v0.0.0-20210427231257-85d9c07bbe3a
+	golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e // indirect
+	golang.org/x/net v0.0.0-20210614182718-04defd469f4e
+	golang.org/x/text v0.3.6 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
diff --git a/vendor/github.com/Shopify/sarama/go.sum b/vendor/github.com/Shopify/sarama/go.sum
index fdd796125..a49776049 100644
--- a/vendor/github.com/Shopify/sarama/go.sum
+++ b/vendor/github.com/Shopify/sarama/go.sum
@@ -61,17 +61,20 @@ github.com/xdg/stringprep v1.0.3 h1:cmL5Enob4W83ti/ZHuZLuKD/xqJfus4fVPwE+/BDm+4=
 github.com/xdg/stringprep v1.0.3/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20201112155050-0c6587e931a9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b h1:7mWr3k41Qtv8XlltBkDkl8LoP3mpSgBW8BUoxtEdbXg=
-golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
+golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e h1:gsTQYXdTw2Gq7RBsWvlQ91b+aEQ6bXFUngBGuR8sPpI=
+golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210427231257-85d9c07bbe3a h1:njMmldwFTyDLqonHMagNXKBWptTBeDZOdblgaDsNEGQ=
 golang.org/x/net v0.0.0-20210427231257-85d9c07bbe3a/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
+golang.org/x/net v0.0.0-20210614182718-04defd469f4e h1:XpT3nA5TvE525Ne3hInMh6+GETgn27Zfm9dxsThnX2Q=
+golang.org/x/net v0.0.0-20210614182718-04defd469f4e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
diff --git a/vendor/github.com/Shopify/sarama/gssapi_kerberos.go b/vendor/github.com/Shopify/sarama/gssapi_kerberos.go
index b2c199000..ab8b70196 100644
--- a/vendor/github.com/Shopify/sarama/gssapi_kerberos.go
+++ b/vendor/github.com/Shopify/sarama/gssapi_kerberos.go
@@ -57,10 +57,10 @@ type KerberosClient interface {
 
 // writePackage appends length in big endian before the payload, and sends it to kafka
 func (krbAuth *GSSAPIKerberosAuth) writePackage(broker *Broker, payload []byte) (int, error) {
-	length := len(payload)
+	length := uint64(len(payload))
 	size := length + 4 // 4 byte length header + payload
-	if size > math.MaxUint32 {
-		return 0, errors.New("payload too large, will overflow uint32")
+	if size > math.MaxInt32 {
+		return 0, errors.New("payload too large, will overflow int32")
 	}
 	finalPackage := make([]byte, size)
 	copy(finalPackage[4:], payload)
diff --git a/vendor/github.com/Shopify/sarama/incremental_alter_configs_request.go b/vendor/github.com/Shopify/sarama/incremental_alter_configs_request.go
new file mode 100644
index 000000000..c4d05a972
--- /dev/null
+++ b/vendor/github.com/Shopify/sarama/incremental_alter_configs_request.go
@@ -0,0 +1,173 @@
+package sarama
+
+type IncrementalAlterConfigsOperation int8
+
+const (
+	IncrementalAlterConfigsOperationSet IncrementalAlterConfigsOperation = iota
+	IncrementalAlterConfigsOperationDelete
+	IncrementalAlterConfigsOperationAppend
+	IncrementalAlterConfigsOperationSubtract
+)
+
+// IncrementalAlterConfigsRequest is an incremental alter config request type
+type IncrementalAlterConfigsRequest struct {
+	Resources    []*IncrementalAlterConfigsResource
+	ValidateOnly bool
+}
+
+type IncrementalAlterConfigsResource struct {
+	Type          ConfigResourceType
+	Name          string
+	ConfigEntries map[string]IncrementalAlterConfigsEntry
+}
+
+type IncrementalAlterConfigsEntry struct {
+	Operation IncrementalAlterConfigsOperation
+	Value     *string
+}
+
+func (a *IncrementalAlterConfigsRequest) encode(pe packetEncoder) error {
+	if err := pe.putArrayLength(len(a.Resources)); err != nil {
+		return err
+	}
+
+	for _, r := range a.Resources {
+		if err := r.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	pe.putBool(a.ValidateOnly)
+	return nil
+}
+
+func (a *IncrementalAlterConfigsRequest) decode(pd packetDecoder, version int16) error {
+	resourceCount, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	a.Resources = make([]*IncrementalAlterConfigsResource, resourceCount)
+	for i := range a.Resources {
+		r := &IncrementalAlterConfigsResource{}
+		err = r.decode(pd, version)
+		if err != nil {
+			return err
+		}
+		a.Resources[i] = r
+	}
+
+	validateOnly, err := pd.getBool()
+	if err != nil {
+		return err
+	}
+
+	a.ValidateOnly = validateOnly
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsResource) encode(pe packetEncoder) error {
+	pe.putInt8(int8(a.Type))
+
+	if err := pe.putString(a.Name); err != nil {
+		return err
+	}
+
+	if err := pe.putArrayLength(len(a.ConfigEntries)); err != nil {
+		return err
+	}
+
+	for name, e := range a.ConfigEntries {
+		if err := pe.putString(name); err != nil {
+			return err
+		}
+
+		if err := e.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsResource) decode(pd packetDecoder, version int16) error {
+	t, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.Type = ConfigResourceType(t)
+
+	name, err := pd.getString()
+	if err != nil {
+		return err
+	}
+	a.Name = name
+
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	if n > 0 {
+		a.ConfigEntries = make(map[string]IncrementalAlterConfigsEntry, n)
+		for i := 0; i < n; i++ {
+			name, err := pd.getString()
+			if err != nil {
+				return err
+			}
+
+			var v IncrementalAlterConfigsEntry
+
+			if err := v.decode(pd, version); err != nil {
+				return err
+			}
+
+			a.ConfigEntries[name] = v
+		}
+	}
+	return err
+}
+
+func (a *IncrementalAlterConfigsEntry) encode(pe packetEncoder) error {
+	pe.putInt8(int8(a.Operation))
+
+	if err := pe.putNullableString(a.Value); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsEntry) decode(pd packetDecoder, version int16) error {
+	t, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.Operation = IncrementalAlterConfigsOperation(t)
+
+	s, err := pd.getNullableString()
+	if err != nil {
+		return err
+	}
+
+	a.Value = s
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsRequest) key() int16 {
+	return 44
+}
+
+func (a *IncrementalAlterConfigsRequest) version() int16 {
+	return 0
+}
+
+func (a *IncrementalAlterConfigsRequest) headerVersion() int16 {
+	return 1
+}
+
+func (a *IncrementalAlterConfigsRequest) requiredVersion() KafkaVersion {
+	return V2_3_0_0
+}
diff --git a/vendor/github.com/Shopify/sarama/incremental_alter_configs_response.go b/vendor/github.com/Shopify/sarama/incremental_alter_configs_response.go
new file mode 100644
index 000000000..3e8c4500c
--- /dev/null
+++ b/vendor/github.com/Shopify/sarama/incremental_alter_configs_response.go
@@ -0,0 +1,66 @@
+package sarama
+
+import "time"
+
+// IncrementalAlterConfigsResponse is a response type for incremental alter config
+type IncrementalAlterConfigsResponse struct {
+	ThrottleTime time.Duration
+	Resources    []*AlterConfigsResourceResponse
+}
+
+func (a *IncrementalAlterConfigsResponse) encode(pe packetEncoder) error {
+	pe.putInt32(int32(a.ThrottleTime / time.Millisecond))
+
+	if err := pe.putArrayLength(len(a.Resources)); err != nil {
+		return err
+	}
+
+	for _, v := range a.Resources {
+		if err := v.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsResponse) decode(pd packetDecoder, version int16) error {
+	throttleTime, err := pd.getInt32()
+	if err != nil {
+		return err
+	}
+	a.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
+
+	responseCount, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	a.Resources = make([]*AlterConfigsResourceResponse, responseCount)
+
+	for i := range a.Resources {
+		a.Resources[i] = new(AlterConfigsResourceResponse)
+
+		if err := a.Resources[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsResponse) key() int16 {
+	return 44
+}
+
+func (a *IncrementalAlterConfigsResponse) version() int16 {
+	return 0
+}
+
+func (a *IncrementalAlterConfigsResponse) headerVersion() int16 {
+	return 0
+}
+
+func (a *IncrementalAlterConfigsResponse) requiredVersion() KafkaVersion {
+	return V2_3_0_0
+}
diff --git a/vendor/github.com/Shopify/sarama/request.go b/vendor/github.com/Shopify/sarama/request.go
index afa64ff92..d899df534 100644
--- a/vendor/github.com/Shopify/sarama/request.go
+++ b/vendor/github.com/Shopify/sarama/request.go
@@ -182,6 +182,8 @@ func allocateBody(key, version int16) protocolBody {
 		return &CreatePartitionsRequest{}
 	case 42:
 		return &DeleteGroupsRequest{}
+	case 44:
+		return &IncrementalAlterConfigsRequest{}
 	case 45:
 		return &AlterPartitionReassignmentsRequest{}
 	case 46:
diff --git a/vendor/golang.org/x/net/http2/ascii.go b/vendor/golang.org/x/net/http2/ascii.go
index 0c58d727c..17caa2058 100644
--- a/vendor/golang.org/x/net/http2/ascii.go
+++ b/vendor/golang.org/x/net/http2/ascii.go
@@ -6,6 +6,10 @@ package http2
 
 import "strings"
 
+// The HTTP protocols are defined in terms of ASCII, not Unicode. This file
+// contains helper functions which may use Unicode-aware functions which would
+// otherwise be unsafe and could introduce vulnerabilities if used improperly.
+
 // asciiEqualFold is strings.EqualFold, ASCII only. It reports whether s and t
 // are equal, ASCII-case-insensitively.
 func asciiEqualFold(s, t string) bool {
diff --git a/vendor/golang.org/x/net/http2/server.go b/vendor/golang.org/x/net/http2/server.go
index 09bc70533..0ccbe9b4c 100644
--- a/vendor/golang.org/x/net/http2/server.go
+++ b/vendor/golang.org/x/net/http2/server.go
@@ -259,16 +259,12 @@ func ConfigureServer(s *http.Server, conf *Server) error {
 
 	s.TLSConfig.PreferServerCipherSuites = true
 
-	haveNPN := false
-	for _, p := range s.TLSConfig.NextProtos {
-		if p == NextProtoTLS {
-			haveNPN = true
-			break
-		}
-	}
-	if !haveNPN {
+	if !strSliceContains(s.TLSConfig.NextProtos, NextProtoTLS) {
 		s.TLSConfig.NextProtos = append(s.TLSConfig.NextProtos, NextProtoTLS)
 	}
+	if !strSliceContains(s.TLSConfig.NextProtos, "http/1.1") {
+		s.TLSConfig.NextProtos = append(s.TLSConfig.NextProtos, "http/1.1")
+	}
 
 	if s.TLSNextProto == nil {
 		s.TLSNextProto = map[string]func(*http.Server, *tls.Conn, http.Handler){}
diff --git a/vendor/golang.org/x/net/http2/transport.go b/vendor/golang.org/x/net/http2/transport.go
index 7bd4b9c19..b97adff7d 100644
--- a/vendor/golang.org/x/net/http2/transport.go
+++ b/vendor/golang.org/x/net/http2/transport.go
@@ -264,9 +264,8 @@ type ClientConn struct {
 	peerMaxHeaderListSize uint64
 	initialWindowSize     uint32
 
-	hbuf    bytes.Buffer // HPACK encoder writes into this
-	henc    *hpack.Encoder
-	freeBuf [][]byte
+	hbuf bytes.Buffer // HPACK encoder writes into this
+	henc *hpack.Encoder
 
 	wmu  sync.Mutex // held while writing; acquire AFTER mu if holding both
 	werr error      // first write error that has occurred
@@ -913,46 +912,6 @@ func (cc *ClientConn) closeForLostPing() error {
 	return cc.closeForError(err)
 }
 
-const maxAllocFrameSize = 512 << 10
-
-// frameBuffer returns a scratch buffer suitable for writing DATA frames.
-// They're capped at the min of the peer's max frame size or 512KB
-// (kinda arbitrarily), but definitely capped so we don't allocate 4GB
-// bufers.
-func (cc *ClientConn) frameScratchBuffer() []byte {
-	cc.mu.Lock()
-	size := cc.maxFrameSize
-	if size > maxAllocFrameSize {
-		size = maxAllocFrameSize
-	}
-	for i, buf := range cc.freeBuf {
-		if len(buf) >= int(size) {
-			cc.freeBuf[i] = nil
-			cc.mu.Unlock()
-			return buf[:size]
-		}
-	}
-	cc.mu.Unlock()
-	return make([]byte, size)
-}
-
-func (cc *ClientConn) putFrameScratchBuffer(buf []byte) {
-	cc.mu.Lock()
-	defer cc.mu.Unlock()
-	const maxBufs = 4 // arbitrary; 4 concurrent requests per conn? investigate.
-	if len(cc.freeBuf) < maxBufs {
-		cc.freeBuf = append(cc.freeBuf, buf)
-		return
-	}
-	for i, old := range cc.freeBuf {
-		if old == nil {
-			cc.freeBuf[i] = buf
-			return
-		}
-	}
-	// forget about it.
-}
-
 // errRequestCanceled is a copy of net/http's errRequestCanceled because it's not
 // exported. At least they'll be DeepEqual for h1-vs-h2 comparisons tests.
 var errRequestCanceled = errors.New("net/http: request canceled")
@@ -1295,11 +1254,35 @@ var (
 	errReqBodyTooLong = errors.New("http2: request body larger than specified content length")
 )
 
+// frameScratchBufferLen returns the length of a buffer to use for
+// outgoing request bodies to read/write to/from.
+//
+// It returns max(1, min(peer's advertised max frame size,
+// Request.ContentLength+1, 512KB)).
+func (cs *clientStream) frameScratchBufferLen(maxFrameSize int) int {
+	const max = 512 << 10
+	n := int64(maxFrameSize)
+	if n > max {
+		n = max
+	}
+	if cl := actualContentLength(cs.req); cl != -1 && cl+1 < n {
+		// Add an extra byte past the declared content-length to
+		// give the caller's Request.Body io.Reader a chance to
+		// give us more bytes than they declared, so we can catch it
+		// early.
+		n = cl + 1
+	}
+	if n < 1 {
+		return 1
+	}
+	return int(n) // doesn't truncate; max is 512K
+}
+
+var bufPool sync.Pool // of *[]byte
+
 func (cs *clientStream) writeRequestBody(body io.Reader, bodyCloser io.Closer) (err error) {
 	cc := cs.cc
 	sentEnd := false // whether we sent the final DATA frame w/ END_STREAM
-	buf := cc.frameScratchBuffer()
-	defer cc.putFrameScratchBuffer(buf)
 
 	defer func() {
 		traceWroteRequest(cs.trace, err)
@@ -1318,9 +1301,24 @@ func (cs *clientStream) writeRequestBody(body io.Reader, bodyCloser io.Closer) (
 	remainLen := actualContentLength(req)
 	hasContentLen := remainLen != -1
 
+	cc.mu.Lock()
+	maxFrameSize := int(cc.maxFrameSize)
+	cc.mu.Unlock()
+
+	// Scratch buffer for reading into & writing from.
+	scratchLen := cs.frameScratchBufferLen(maxFrameSize)
+	var buf []byte
+	if bp, ok := bufPool.Get().(*[]byte); ok && len(*bp) >= scratchLen {
+		defer bufPool.Put(bp)
+		buf = *bp
+	} else {
+		buf = make([]byte, scratchLen)
+		defer bufPool.Put(&buf)
+	}
+
 	var sawEOF bool
 	for !sawEOF {
-		n, err := body.Read(buf[:len(buf)-1])
+		n, err := body.Read(buf[:len(buf)])
 		if hasContentLen {
 			remainLen -= int64(n)
 			if remainLen == 0 && err == nil {
@@ -1331,8 +1329,9 @@ func (cs *clientStream) writeRequestBody(body io.Reader, bodyCloser io.Closer) (
 				// to send the END_STREAM bit early, double-check that we're actually
 				// at EOF. Subsequent reads should return (0, EOF) at this point.
 				// If either value is different, we return an error in one of two ways below.
+				var scratch [1]byte
 				var n1 int
-				n1, err = body.Read(buf[n:])
+				n1, err = body.Read(scratch[:])
 				remainLen -= int64(n1)
 			}
 			if remainLen < 0 {
@@ -1402,10 +1401,6 @@ func (cs *clientStream) writeRequestBody(body io.Reader, bodyCloser io.Closer) (
 		}
 	}
 
-	cc.mu.Lock()
-	maxFrameSize := int(cc.maxFrameSize)
-	cc.mu.Unlock()
-
 	cc.wmu.Lock()
 	defer cc.wmu.Unlock()
 
diff --git a/vendor/golang.org/x/sys/unix/mkerrors.sh b/vendor/golang.org/x/sys/unix/mkerrors.sh
index c4e3e1eee..3f670faba 100644
--- a/vendor/golang.org/x/sys/unix/mkerrors.sh
+++ b/vendor/golang.org/x/sys/unix/mkerrors.sh
@@ -239,6 +239,7 @@ struct ltchars {
 #include <linux/netfilter/nfnetlink.h>
 #include <linux/netlink.h>
 #include <linux/net_namespace.h>
+#include <linux/nfc.h>
 #include <linux/nsfs.h>
 #include <linux/perf_event.h>
 #include <linux/pps.h>
@@ -502,6 +503,9 @@ ccflags="$@"
 		$2 ~ /^LO_(KEY|NAME)_SIZE$/ ||
 		$2 ~ /^LOOP_(CLR|CTL|GET|SET)_/ ||
 		$2 ~ /^(AF|SOCK|SO|SOL|IPPROTO|IP|IPV6|TCP|MCAST|EVFILT|NOTE|SHUT|PROT|MAP|MFD|T?PACKET|MSG|SCM|MCL|DT|MADV|PR|LOCAL)_/ ||
+		$2 ~ /^NFC_(GENL|PROTO|COMM|RF|SE|DIRECTION|LLCP|SOCKPROTO)_/ ||
+		$2 ~ /^NFC_.*_(MAX)?SIZE$/ ||
+		$2 ~ /^RAW_PAYLOAD_/ ||
 		$2 ~ /^TP_STATUS_/ ||
 		$2 ~ /^FALLOC_/ ||
 		$2 ~ /^ICMPV?6?_(FILTER|SEC)/ ||
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux.go b/vendor/golang.org/x/sys/unix/syscall_linux.go
index 2dd7c8e34..41b91fdfb 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux.go
@@ -904,6 +904,46 @@ func (sa *SockaddrIUCV) sockaddr() (unsafe.Pointer, _Socklen, error) {
 	return unsafe.Pointer(&sa.raw), SizeofSockaddrIUCV, nil
 }
 
+type SockaddrNFC struct {
+	DeviceIdx   uint32
+	TargetIdx   uint32
+	NFCProtocol uint32
+	raw         RawSockaddrNFC
+}
+
+func (sa *SockaddrNFC) sockaddr() (unsafe.Pointer, _Socklen, error) {
+	sa.raw.Sa_family = AF_NFC
+	sa.raw.Dev_idx = sa.DeviceIdx
+	sa.raw.Target_idx = sa.TargetIdx
+	sa.raw.Nfc_protocol = sa.NFCProtocol
+	return unsafe.Pointer(&sa.raw), SizeofSockaddrNFC, nil
+}
+
+type SockaddrNFCLLCP struct {
+	DeviceIdx      uint32
+	TargetIdx      uint32
+	NFCProtocol    uint32
+	DestinationSAP uint8
+	SourceSAP      uint8
+	ServiceName    string
+	raw            RawSockaddrNFCLLCP
+}
+
+func (sa *SockaddrNFCLLCP) sockaddr() (unsafe.Pointer, _Socklen, error) {
+	sa.raw.Sa_family = AF_NFC
+	sa.raw.Dev_idx = sa.DeviceIdx
+	sa.raw.Target_idx = sa.TargetIdx
+	sa.raw.Nfc_protocol = sa.NFCProtocol
+	sa.raw.Dsap = sa.DestinationSAP
+	sa.raw.Ssap = sa.SourceSAP
+	if len(sa.ServiceName) > len(sa.raw.Service_name) {
+		return nil, 0, EINVAL
+	}
+	copy(sa.raw.Service_name[:], sa.ServiceName)
+	sa.raw.SetServiceNameLen(len(sa.ServiceName))
+	return unsafe.Pointer(&sa.raw), SizeofSockaddrNFCLLCP, nil
+}
+
 var socketProtocol = func(fd int) (int, error) {
 	return GetsockoptInt(fd, SOL_SOCKET, SO_PROTOCOL)
 }
@@ -1144,6 +1184,37 @@ func anyToSockaddr(fd int, rsa *RawSockaddrAny) (Sockaddr, error) {
 			}
 			return sa, nil
 		}
+	case AF_NFC:
+		proto, err := socketProtocol(fd)
+		if err != nil {
+			return nil, err
+		}
+		switch proto {
+		case NFC_SOCKPROTO_RAW:
+			pp := (*RawSockaddrNFC)(unsafe.Pointer(rsa))
+			sa := &SockaddrNFC{
+				DeviceIdx:   pp.Dev_idx,
+				TargetIdx:   pp.Target_idx,
+				NFCProtocol: pp.Nfc_protocol,
+			}
+			return sa, nil
+		case NFC_SOCKPROTO_LLCP:
+			pp := (*RawSockaddrNFCLLCP)(unsafe.Pointer(rsa))
+			if uint64(pp.Service_name_len) > uint64(len(pp.Service_name)) {
+				return nil, EINVAL
+			}
+			sa := &SockaddrNFCLLCP{
+				DeviceIdx:      pp.Dev_idx,
+				TargetIdx:      pp.Target_idx,
+				NFCProtocol:    pp.Nfc_protocol,
+				DestinationSAP: pp.Dsap,
+				SourceSAP:      pp.Ssap,
+				ServiceName:    string(pp.Service_name[:pp.Service_name_len]),
+			}
+			return sa, nil
+		default:
+			return nil, EINVAL
+		}
 	}
 	return nil, EAFNOSUPPORT
 }
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_386.go b/vendor/golang.org/x/sys/unix/syscall_linux_386.go
index 7b52e5d8a..b430536c8 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_386.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_386.go
@@ -378,6 +378,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint32(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint32(length)
+}
+
 //sys	poll(fds *PollFd, nfds int, timeout int) (n int, err error)
 
 func Poll(fds []PollFd, timeout int) (n int, err error) {
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_amd64.go b/vendor/golang.org/x/sys/unix/syscall_linux_amd64.go
index 28b764115..85cd97da0 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_amd64.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_amd64.go
@@ -172,6 +172,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint64(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint64(length)
+}
+
 //sys	poll(fds *PollFd, nfds int, timeout int) (n int, err error)
 
 func Poll(fds []PollFd, timeout int) (n int, err error) {
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_arm.go b/vendor/golang.org/x/sys/unix/syscall_linux_arm.go
index 68877728e..39a864d4e 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_arm.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_arm.go
@@ -256,6 +256,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint32(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint32(length)
+}
+
 //sys	poll(fds *PollFd, nfds int, timeout int) (n int, err error)
 
 func Poll(fds []PollFd, timeout int) (n int, err error) {
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_arm64.go b/vendor/golang.org/x/sys/unix/syscall_linux_arm64.go
index 7ed703476..7f27ebf2f 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_arm64.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_arm64.go
@@ -207,6 +207,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint64(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint64(length)
+}
+
 func InotifyInit() (fd int, err error) {
 	return InotifyInit1(0)
 }
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_mips64x.go b/vendor/golang.org/x/sys/unix/syscall_linux_mips64x.go
index 06dec06fa..27aee81d9 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_mips64x.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_mips64x.go
@@ -217,6 +217,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint64(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint64(length)
+}
+
 func InotifyInit() (fd int, err error) {
 	return InotifyInit1(0)
 }
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_mipsx.go b/vendor/golang.org/x/sys/unix/syscall_linux_mipsx.go
index 8f0d0a5b5..3a5621e37 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_mipsx.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_mipsx.go
@@ -229,6 +229,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint32(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint32(length)
+}
+
 //sys	poll(fds *PollFd, nfds int, timeout int) (n int, err error)
 
 func Poll(fds []PollFd, timeout int) (n int, err error) {
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_ppc.go b/vendor/golang.org/x/sys/unix/syscall_linux_ppc.go
index 7e65e088d..cf0d36f76 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_ppc.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_ppc.go
@@ -215,6 +215,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint32(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint32(length)
+}
+
 //sysnb	pipe(p *[2]_C_int) (err error)
 
 func Pipe(p []int) (err error) {
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_ppc64x.go b/vendor/golang.org/x/sys/unix/syscall_linux_ppc64x.go
index 0b1f0d6da..5259a5fea 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_ppc64x.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_ppc64x.go
@@ -100,6 +100,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint64(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint64(length)
+}
+
 //sysnb	pipe(p *[2]_C_int) (err error)
 
 func Pipe(p []int) (err error) {
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_riscv64.go b/vendor/golang.org/x/sys/unix/syscall_linux_riscv64.go
index ce9bcd317..8ef821e5d 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_riscv64.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_riscv64.go
@@ -188,6 +188,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint64(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint64(length)
+}
+
 func InotifyInit() (fd int, err error) {
 	return InotifyInit1(0)
 }
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_s390x.go b/vendor/golang.org/x/sys/unix/syscall_linux_s390x.go
index a1e45694b..a1c0574b5 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_s390x.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_s390x.go
@@ -129,6 +129,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint64(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint64(length)
+}
+
 // Linux on s390x uses the old mmap interface, which requires arguments to be passed in a struct.
 // mmap2 also requires arguments to be passed in a struct; it is currently not exposed in <asm/unistd.h>.
 func mmap(addr uintptr, length uintptr, prot int, flags int, fd int, offset int64) (xaddr uintptr, err error) {
diff --git a/vendor/golang.org/x/sys/unix/syscall_linux_sparc64.go b/vendor/golang.org/x/sys/unix/syscall_linux_sparc64.go
index 49055a3cf..de14b8898 100644
--- a/vendor/golang.org/x/sys/unix/syscall_linux_sparc64.go
+++ b/vendor/golang.org/x/sys/unix/syscall_linux_sparc64.go
@@ -116,6 +116,10 @@ func (cmsg *Cmsghdr) SetLen(length int) {
 	cmsg.Len = uint64(length)
 }
 
+func (rsa *RawSockaddrNFCLLCP) SetServiceNameLen(length int) {
+	rsa.Service_name_len = uint64(length)
+}
+
 //sysnb	pipe(p *[2]_C_int) (err error)
 
 func Pipe(p []int) (err error) {
diff --git a/vendor/golang.org/x/sys/unix/zerrors_linux.go b/vendor/golang.org/x/sys/unix/zerrors_linux.go
index 4e4583b61..c3fa22486 100644
--- a/vendor/golang.org/x/sys/unix/zerrors_linux.go
+++ b/vendor/golang.org/x/sys/unix/zerrors_linux.go
@@ -1566,6 +1566,59 @@ const (
 	NETLINK_XFRM                                = 0x6
 	NETNSA_MAX                                  = 0x5
 	NETNSA_NSID_NOT_ASSIGNED                    = -0x1
+	NFC_ATR_REQ_GB_MAXSIZE                      = 0x30
+	NFC_ATR_REQ_MAXSIZE                         = 0x40
+	NFC_ATR_RES_GB_MAXSIZE                      = 0x2f
+	NFC_ATR_RES_MAXSIZE                         = 0x40
+	NFC_COMM_ACTIVE                             = 0x0
+	NFC_COMM_PASSIVE                            = 0x1
+	NFC_DEVICE_NAME_MAXSIZE                     = 0x8
+	NFC_DIRECTION_RX                            = 0x0
+	NFC_DIRECTION_TX                            = 0x1
+	NFC_FIRMWARE_NAME_MAXSIZE                   = 0x20
+	NFC_GB_MAXSIZE                              = 0x30
+	NFC_GENL_MCAST_EVENT_NAME                   = "events"
+	NFC_GENL_NAME                               = "nfc"
+	NFC_GENL_VERSION                            = 0x1
+	NFC_HEADER_SIZE                             = 0x1
+	NFC_ISO15693_UID_MAXSIZE                    = 0x8
+	NFC_LLCP_MAX_SERVICE_NAME                   = 0x3f
+	NFC_LLCP_MIUX                               = 0x1
+	NFC_LLCP_REMOTE_LTO                         = 0x3
+	NFC_LLCP_REMOTE_MIU                         = 0x2
+	NFC_LLCP_REMOTE_RW                          = 0x4
+	NFC_LLCP_RW                                 = 0x0
+	NFC_NFCID1_MAXSIZE                          = 0xa
+	NFC_NFCID2_MAXSIZE                          = 0x8
+	NFC_NFCID3_MAXSIZE                          = 0xa
+	NFC_PROTO_FELICA                            = 0x3
+	NFC_PROTO_FELICA_MASK                       = 0x8
+	NFC_PROTO_ISO14443                          = 0x4
+	NFC_PROTO_ISO14443_B                        = 0x6
+	NFC_PROTO_ISO14443_B_MASK                   = 0x40
+	NFC_PROTO_ISO14443_MASK                     = 0x10
+	NFC_PROTO_ISO15693                          = 0x7
+	NFC_PROTO_ISO15693_MASK                     = 0x80
+	NFC_PROTO_JEWEL                             = 0x1
+	NFC_PROTO_JEWEL_MASK                        = 0x2
+	NFC_PROTO_MAX                               = 0x8
+	NFC_PROTO_MIFARE                            = 0x2
+	NFC_PROTO_MIFARE_MASK                       = 0x4
+	NFC_PROTO_NFC_DEP                           = 0x5
+	NFC_PROTO_NFC_DEP_MASK                      = 0x20
+	NFC_RAW_HEADER_SIZE                         = 0x2
+	NFC_RF_INITIATOR                            = 0x0
+	NFC_RF_NONE                                 = 0x2
+	NFC_RF_TARGET                               = 0x1
+	NFC_SENSB_RES_MAXSIZE                       = 0xc
+	NFC_SENSF_RES_MAXSIZE                       = 0x12
+	NFC_SE_DISABLED                             = 0x0
+	NFC_SE_EMBEDDED                             = 0x2
+	NFC_SE_ENABLED                              = 0x1
+	NFC_SE_UICC                                 = 0x1
+	NFC_SOCKPROTO_LLCP                          = 0x1
+	NFC_SOCKPROTO_MAX                           = 0x2
+	NFC_SOCKPROTO_RAW                           = 0x0
 	NFNETLINK_V0                                = 0x0
 	NFNLGRP_ACCT_QUOTA                          = 0x8
 	NFNLGRP_CONNTRACK_DESTROY                   = 0x3
@@ -1991,6 +2044,11 @@ const (
 	QNX4_SUPER_MAGIC                            = 0x2f
 	QNX6_SUPER_MAGIC                            = 0x68191122
 	RAMFS_MAGIC                                 = 0x858458f6
+	RAW_PAYLOAD_DIGITAL                         = 0x3
+	RAW_PAYLOAD_HCI                             = 0x2
+	RAW_PAYLOAD_LLCP                            = 0x0
+	RAW_PAYLOAD_NCI                             = 0x1
+	RAW_PAYLOAD_PROPRIETARY                     = 0x4
 	RDTGROUP_SUPER_MAGIC                        = 0x7655821
 	REISERFS_SUPER_MAGIC                        = 0x52654973
 	RENAME_EXCHANGE                             = 0x2
diff --git a/vendor/golang.org/x/sys/unix/ztypes_dragonfly_amd64.go b/vendor/golang.org/x/sys/unix/ztypes_dragonfly_amd64.go
index 1d049d7a1..d0ba8e9b8 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_dragonfly_amd64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_dragonfly_amd64.go
@@ -431,6 +431,9 @@ type Winsize struct {
 const (
 	AT_FDCWD            = 0xfffafdcd
 	AT_SYMLINK_NOFOLLOW = 0x1
+	AT_REMOVEDIR        = 0x2
+	AT_EACCESS          = 0x4
+	AT_SYMLINK_FOLLOW   = 0x8
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_freebsd_386.go b/vendor/golang.org/x/sys/unix/ztypes_freebsd_386.go
index c51bc88ff..1f99c024a 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_freebsd_386.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_freebsd_386.go
@@ -672,9 +672,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_REMOVEDIR        = 0x800
-	AT_SYMLINK_FOLLOW   = 0x400
+	AT_EACCESS          = 0x100
 	AT_SYMLINK_NOFOLLOW = 0x200
+	AT_SYMLINK_FOLLOW   = 0x400
+	AT_REMOVEDIR        = 0x800
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_freebsd_amd64.go b/vendor/golang.org/x/sys/unix/ztypes_freebsd_amd64.go
index 395b69187..ddf0305a5 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_freebsd_amd64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_freebsd_amd64.go
@@ -675,9 +675,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_REMOVEDIR        = 0x800
-	AT_SYMLINK_FOLLOW   = 0x400
+	AT_EACCESS          = 0x100
 	AT_SYMLINK_NOFOLLOW = 0x200
+	AT_SYMLINK_FOLLOW   = 0x400
+	AT_REMOVEDIR        = 0x800
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_freebsd_arm.go b/vendor/golang.org/x/sys/unix/ztypes_freebsd_arm.go
index d3f9d2541..dce0a5c80 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_freebsd_arm.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_freebsd_arm.go
@@ -656,9 +656,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_REMOVEDIR        = 0x800
-	AT_SYMLINK_FOLLOW   = 0x400
+	AT_EACCESS          = 0x100
 	AT_SYMLINK_NOFOLLOW = 0x200
+	AT_SYMLINK_FOLLOW   = 0x400
+	AT_REMOVEDIR        = 0x800
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_freebsd_arm64.go b/vendor/golang.org/x/sys/unix/ztypes_freebsd_arm64.go
index 434d6e8e8..e23244702 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_freebsd_arm64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_freebsd_arm64.go
@@ -653,9 +653,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_REMOVEDIR        = 0x800
-	AT_SYMLINK_FOLLOW   = 0x400
+	AT_EACCESS          = 0x100
 	AT_SYMLINK_NOFOLLOW = 0x200
+	AT_SYMLINK_FOLLOW   = 0x400
+	AT_REMOVEDIR        = 0x800
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux.go b/vendor/golang.org/x/sys/unix/ztypes_linux.go
index c9b2c9aae..72887abe5 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux.go
@@ -351,6 +351,13 @@ type RawSockaddrIUCV struct {
 	Name    [8]int8
 }
 
+type RawSockaddrNFC struct {
+	Sa_family    uint16
+	Dev_idx      uint32
+	Target_idx   uint32
+	Nfc_protocol uint32
+}
+
 type _Socklen uint32
 
 type Linger struct {
@@ -464,6 +471,7 @@ const (
 	SizeofSockaddrL2TPIP    = 0x10
 	SizeofSockaddrL2TPIP6   = 0x20
 	SizeofSockaddrIUCV      = 0x20
+	SizeofSockaddrNFC       = 0x10
 	SizeofLinger            = 0x8
 	SizeofIPMreq            = 0x8
 	SizeofIPMreqn           = 0xc
@@ -3828,3 +3836,72 @@ const (
 	MTD_FILE_MODE_OTP_USER    = 0x2
 	MTD_FILE_MODE_RAW         = 0x3
 )
+
+const (
+	NFC_CMD_UNSPEC                    = 0x0
+	NFC_CMD_GET_DEVICE                = 0x1
+	NFC_CMD_DEV_UP                    = 0x2
+	NFC_CMD_DEV_DOWN                  = 0x3
+	NFC_CMD_DEP_LINK_UP               = 0x4
+	NFC_CMD_DEP_LINK_DOWN             = 0x5
+	NFC_CMD_START_POLL                = 0x6
+	NFC_CMD_STOP_POLL                 = 0x7
+	NFC_CMD_GET_TARGET                = 0x8
+	NFC_EVENT_TARGETS_FOUND           = 0x9
+	NFC_EVENT_DEVICE_ADDED            = 0xa
+	NFC_EVENT_DEVICE_REMOVED          = 0xb
+	NFC_EVENT_TARGET_LOST             = 0xc
+	NFC_EVENT_TM_ACTIVATED            = 0xd
+	NFC_EVENT_TM_DEACTIVATED          = 0xe
+	NFC_CMD_LLC_GET_PARAMS            = 0xf
+	NFC_CMD_LLC_SET_PARAMS            = 0x10
+	NFC_CMD_ENABLE_SE                 = 0x11
+	NFC_CMD_DISABLE_SE                = 0x12
+	NFC_CMD_LLC_SDREQ                 = 0x13
+	NFC_EVENT_LLC_SDRES               = 0x14
+	NFC_CMD_FW_DOWNLOAD               = 0x15
+	NFC_EVENT_SE_ADDED                = 0x16
+	NFC_EVENT_SE_REMOVED              = 0x17
+	NFC_EVENT_SE_CONNECTIVITY         = 0x18
+	NFC_EVENT_SE_TRANSACTION          = 0x19
+	NFC_CMD_GET_SE                    = 0x1a
+	NFC_CMD_SE_IO                     = 0x1b
+	NFC_CMD_ACTIVATE_TARGET           = 0x1c
+	NFC_CMD_VENDOR                    = 0x1d
+	NFC_CMD_DEACTIVATE_TARGET         = 0x1e
+	NFC_ATTR_UNSPEC                   = 0x0
+	NFC_ATTR_DEVICE_INDEX             = 0x1
+	NFC_ATTR_DEVICE_NAME              = 0x2
+	NFC_ATTR_PROTOCOLS                = 0x3
+	NFC_ATTR_TARGET_INDEX             = 0x4
+	NFC_ATTR_TARGET_SENS_RES          = 0x5
+	NFC_ATTR_TARGET_SEL_RES           = 0x6
+	NFC_ATTR_TARGET_NFCID1            = 0x7
+	NFC_ATTR_TARGET_SENSB_RES         = 0x8
+	NFC_ATTR_TARGET_SENSF_RES         = 0x9
+	NFC_ATTR_COMM_MODE                = 0xa
+	NFC_ATTR_RF_MODE                  = 0xb
+	NFC_ATTR_DEVICE_POWERED           = 0xc
+	NFC_ATTR_IM_PROTOCOLS             = 0xd
+	NFC_ATTR_TM_PROTOCOLS             = 0xe
+	NFC_ATTR_LLC_PARAM_LTO            = 0xf
+	NFC_ATTR_LLC_PARAM_RW             = 0x10
+	NFC_ATTR_LLC_PARAM_MIUX           = 0x11
+	NFC_ATTR_SE                       = 0x12
+	NFC_ATTR_LLC_SDP                  = 0x13
+	NFC_ATTR_FIRMWARE_NAME            = 0x14
+	NFC_ATTR_SE_INDEX                 = 0x15
+	NFC_ATTR_SE_TYPE                  = 0x16
+	NFC_ATTR_SE_AID                   = 0x17
+	NFC_ATTR_FIRMWARE_DOWNLOAD_STATUS = 0x18
+	NFC_ATTR_SE_APDU                  = 0x19
+	NFC_ATTR_TARGET_ISO15693_DSFID    = 0x1a
+	NFC_ATTR_TARGET_ISO15693_UID      = 0x1b
+	NFC_ATTR_SE_PARAMS                = 0x1c
+	NFC_ATTR_VENDOR_ID                = 0x1d
+	NFC_ATTR_VENDOR_SUBCMD            = 0x1e
+	NFC_ATTR_VENDOR_DATA              = 0x1f
+	NFC_SDP_ATTR_UNSPEC               = 0x0
+	NFC_SDP_ATTR_URI                  = 0x1
+	NFC_SDP_ATTR_SAP                  = 0x2
+)
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_386.go b/vendor/golang.org/x/sys/unix/ztypes_linux_386.go
index 4d4d283de..235c62e46 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_386.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_386.go
@@ -128,6 +128,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint32
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -160,9 +171,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x8
-	SizeofMsghdr  = 0x1c
-	SizeofCmsghdr = 0xc
+	SizeofSockaddrNFCLLCP = 0x58
+	SizeofIovec           = 0x8
+	SizeofMsghdr          = 0x1c
+	SizeofCmsghdr         = 0xc
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_amd64.go b/vendor/golang.org/x/sys/unix/ztypes_linux_amd64.go
index 8a2eed5ec..99b1e5b6a 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_amd64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_amd64.go
@@ -130,6 +130,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -163,9 +174,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_arm.go b/vendor/golang.org/x/sys/unix/ztypes_linux_arm.go
index 94b34add6..cc8bba791 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_arm.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_arm.go
@@ -134,6 +134,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint32
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]uint8
@@ -166,9 +177,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x8
-	SizeofMsghdr  = 0x1c
-	SizeofCmsghdr = 0xc
+	SizeofSockaddrNFCLLCP = 0x58
+	SizeofIovec           = 0x8
+	SizeofMsghdr          = 0x1c
+	SizeofCmsghdr         = 0xc
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_arm64.go b/vendor/golang.org/x/sys/unix/ztypes_linux_arm64.go
index 2143de4d5..fa8fe3a75 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_arm64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_arm64.go
@@ -131,6 +131,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -164,9 +175,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_mips.go b/vendor/golang.org/x/sys/unix/ztypes_linux_mips.go
index a40216eee..e7fb8d9b7 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_mips.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_mips.go
@@ -133,6 +133,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint32
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -165,9 +176,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x8
-	SizeofMsghdr  = 0x1c
-	SizeofCmsghdr = 0xc
+	SizeofSockaddrNFCLLCP = 0x58
+	SizeofIovec           = 0x8
+	SizeofMsghdr          = 0x1c
+	SizeofCmsghdr         = 0xc
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_mips64.go b/vendor/golang.org/x/sys/unix/ztypes_linux_mips64.go
index e834b069f..2fa61d593 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_mips64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_mips64.go
@@ -131,6 +131,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -164,9 +175,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_mips64le.go b/vendor/golang.org/x/sys/unix/ztypes_linux_mips64le.go
index e31083b04..7f3639933 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_mips64le.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_mips64le.go
@@ -131,6 +131,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -164,9 +175,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_mipsle.go b/vendor/golang.org/x/sys/unix/ztypes_linux_mipsle.go
index 42811f7fb..f3c20cb86 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_mipsle.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_mipsle.go
@@ -133,6 +133,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint32
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -165,9 +176,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x8
-	SizeofMsghdr  = 0x1c
-	SizeofCmsghdr = 0xc
+	SizeofSockaddrNFCLLCP = 0x58
+	SizeofIovec           = 0x8
+	SizeofMsghdr          = 0x1c
+	SizeofCmsghdr         = 0xc
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_ppc.go b/vendor/golang.org/x/sys/unix/ztypes_linux_ppc.go
index af7a72017..885d27950 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_ppc.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_ppc.go
@@ -134,6 +134,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint32
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]uint8
@@ -166,9 +177,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x8
-	SizeofMsghdr  = 0x1c
-	SizeofCmsghdr = 0xc
+	SizeofSockaddrNFCLLCP = 0x58
+	SizeofIovec           = 0x8
+	SizeofMsghdr          = 0x1c
+	SizeofCmsghdr         = 0xc
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_ppc64.go b/vendor/golang.org/x/sys/unix/ztypes_linux_ppc64.go
index 2a3afbaef..a94eb8e18 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_ppc64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_ppc64.go
@@ -132,6 +132,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]uint8
@@ -165,9 +176,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_ppc64le.go b/vendor/golang.org/x/sys/unix/ztypes_linux_ppc64le.go
index c0de30a65..659e32ebd 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_ppc64le.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_ppc64le.go
@@ -132,6 +132,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]uint8
@@ -165,9 +176,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_riscv64.go b/vendor/golang.org/x/sys/unix/ztypes_linux_riscv64.go
index 74faf2e91..ab8ec604f 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_riscv64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_riscv64.go
@@ -131,6 +131,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]uint8
@@ -164,9 +175,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_s390x.go b/vendor/golang.org/x/sys/unix/ztypes_linux_s390x.go
index 9a8f0c2c6..3ec08237f 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_s390x.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_s390x.go
@@ -130,6 +130,17 @@ const (
 	FADV_NOREUSE  = 0x7
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -163,9 +174,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_linux_sparc64.go b/vendor/golang.org/x/sys/unix/ztypes_linux_sparc64.go
index 72cdda75b..23d474470 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_linux_sparc64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_linux_sparc64.go
@@ -134,6 +134,17 @@ const (
 	FADV_NOREUSE  = 0x5
 )
 
+type RawSockaddrNFCLLCP struct {
+	Sa_family        uint16
+	Dev_idx          uint32
+	Target_idx       uint32
+	Nfc_protocol     uint32
+	Dsap             uint8
+	Ssap             uint8
+	Service_name     [63]uint8
+	Service_name_len uint64
+}
+
 type RawSockaddr struct {
 	Family uint16
 	Data   [14]int8
@@ -167,9 +178,10 @@ type Cmsghdr struct {
 }
 
 const (
-	SizeofIovec   = 0x10
-	SizeofMsghdr  = 0x38
-	SizeofCmsghdr = 0x10
+	SizeofSockaddrNFCLLCP = 0x60
+	SizeofIovec           = 0x10
+	SizeofMsghdr          = 0x38
+	SizeofCmsghdr         = 0x10
 )
 
 const (
diff --git a/vendor/golang.org/x/sys/unix/ztypes_netbsd_386.go b/vendor/golang.org/x/sys/unix/ztypes_netbsd_386.go
index b10e73abf..2fd2060e6 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_netbsd_386.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_netbsd_386.go
@@ -445,8 +445,10 @@ type Ptmget struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x400
+	AT_EACCESS          = 0x100
 	AT_SYMLINK_NOFOLLOW = 0x200
+	AT_SYMLINK_FOLLOW   = 0x400
+	AT_REMOVEDIR        = 0x800
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_netbsd_amd64.go b/vendor/golang.org/x/sys/unix/ztypes_netbsd_amd64.go
index 28ed6d55a..6a5a1a8ae 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_netbsd_amd64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_netbsd_amd64.go
@@ -453,8 +453,10 @@ type Ptmget struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x400
+	AT_EACCESS          = 0x100
 	AT_SYMLINK_NOFOLLOW = 0x200
+	AT_SYMLINK_FOLLOW   = 0x400
+	AT_REMOVEDIR        = 0x800
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_netbsd_arm.go b/vendor/golang.org/x/sys/unix/ztypes_netbsd_arm.go
index 4ba196ebe..84cc8d01e 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_netbsd_arm.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_netbsd_arm.go
@@ -450,8 +450,10 @@ type Ptmget struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x400
+	AT_EACCESS          = 0x100
 	AT_SYMLINK_NOFOLLOW = 0x200
+	AT_SYMLINK_FOLLOW   = 0x400
+	AT_REMOVEDIR        = 0x800
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_netbsd_arm64.go b/vendor/golang.org/x/sys/unix/ztypes_netbsd_arm64.go
index dd642bd9c..c844e7096 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_netbsd_arm64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_netbsd_arm64.go
@@ -453,8 +453,10 @@ type Ptmget struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x400
+	AT_EACCESS          = 0x100
 	AT_SYMLINK_NOFOLLOW = 0x200
+	AT_SYMLINK_FOLLOW   = 0x400
+	AT_REMOVEDIR        = 0x800
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_openbsd_386.go b/vendor/golang.org/x/sys/unix/ztypes_openbsd_386.go
index 1fdb0e5fa..2a8b1e6f7 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_openbsd_386.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_openbsd_386.go
@@ -438,8 +438,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x4
+	AT_EACCESS          = 0x1
 	AT_SYMLINK_NOFOLLOW = 0x2
+	AT_SYMLINK_FOLLOW   = 0x4
+	AT_REMOVEDIR        = 0x8
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_openbsd_amd64.go b/vendor/golang.org/x/sys/unix/ztypes_openbsd_amd64.go
index e2fc93c7c..b1759cf70 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_openbsd_amd64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_openbsd_amd64.go
@@ -438,8 +438,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x4
+	AT_EACCESS          = 0x1
 	AT_SYMLINK_NOFOLLOW = 0x2
+	AT_SYMLINK_FOLLOW   = 0x4
+	AT_REMOVEDIR        = 0x8
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_openbsd_arm.go b/vendor/golang.org/x/sys/unix/ztypes_openbsd_arm.go
index 8d34b5a2f..e807de206 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_openbsd_arm.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_openbsd_arm.go
@@ -439,8 +439,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x4
+	AT_EACCESS          = 0x1
 	AT_SYMLINK_NOFOLLOW = 0x2
+	AT_SYMLINK_FOLLOW   = 0x4
+	AT_REMOVEDIR        = 0x8
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_openbsd_arm64.go b/vendor/golang.org/x/sys/unix/ztypes_openbsd_arm64.go
index ea8f1a0d9..ff3aecaee 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_openbsd_arm64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_openbsd_arm64.go
@@ -432,8 +432,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x4
+	AT_EACCESS          = 0x1
 	AT_SYMLINK_NOFOLLOW = 0x2
+	AT_SYMLINK_FOLLOW   = 0x4
+	AT_REMOVEDIR        = 0x8
 )
 
 type PollFd struct {
diff --git a/vendor/golang.org/x/sys/unix/ztypes_openbsd_mips64.go b/vendor/golang.org/x/sys/unix/ztypes_openbsd_mips64.go
index ec6e8bc3f..9ecda6917 100644
--- a/vendor/golang.org/x/sys/unix/ztypes_openbsd_mips64.go
+++ b/vendor/golang.org/x/sys/unix/ztypes_openbsd_mips64.go
@@ -432,8 +432,10 @@ type Winsize struct {
 
 const (
 	AT_FDCWD            = -0x64
-	AT_SYMLINK_FOLLOW   = 0x4
+	AT_EACCESS          = 0x1
 	AT_SYMLINK_NOFOLLOW = 0x2
+	AT_SYMLINK_FOLLOW   = 0x4
+	AT_REMOVEDIR        = 0x8
 )
 
 type PollFd struct {
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake/fake.go
new file mode 100644
index 000000000..db12d56a0
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/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"
+
+	pod "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
+	fake "knative.dev/pkg/client/injection/kube/informers/factory/fake"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = pod.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Core().V1().Pods()
+	return context.WithValue(ctx, pod.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 265d0bfec..26d395d2e 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -69,7 +69,7 @@ github.com/Azure/go-autorest/tracing
 github.com/PuerkitoBio/purell
 # github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578
 github.com/PuerkitoBio/urlesc
-# github.com/Shopify/sarama v1.29.0
+# github.com/Shopify/sarama v1.29.1
 ## explicit
 github.com/Shopify/sarama
 # github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751
@@ -467,7 +467,7 @@ go.uber.org/zap/internal/exit
 go.uber.org/zap/internal/ztest
 go.uber.org/zap/zapcore
 go.uber.org/zap/zaptest
-# golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a
+# golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e
 golang.org/x/crypto/cast5
 golang.org/x/crypto/md4
 golang.org/x/crypto/openpgp
@@ -486,7 +486,7 @@ golang.org/x/lint/golint
 # golang.org/x/mod v0.4.2
 golang.org/x/mod/module
 golang.org/x/mod/semver
-# golang.org/x/net v0.0.0-20210525063256-abc453219eb5
+# golang.org/x/net v0.0.0-20210614182718-04defd469f4e
 golang.org/x/net/context
 golang.org/x/net/context/ctxhttp
 golang.org/x/net/http/httpguts
@@ -511,7 +511,7 @@ golang.org/x/oauth2/jwt
 ## explicit
 golang.org/x/sync/errgroup
 golang.org/x/sync/semaphore
-# golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40
+# golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1
 golang.org/x/sys/execabs
 golang.org/x/sys/internal/unsafeheader
 golang.org/x/sys/plan9
@@ -1276,6 +1276,7 @@ knative.dev/pkg/client/injection/kube/informers/core/v1/endpoints
 knative.dev/pkg/client/injection/kube/informers/core/v1/namespace
 knative.dev/pkg/client/injection/kube/informers/core/v1/node
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod
+knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/service
 knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount


Test patch:
diff --git a/pkg/channel/distributed/controller/util/subscription_test.go b/pkg/channel/distributed/controller/util/subscription_test.go
index 8bb15016e..9445f16a5 100644
--- a/pkg/channel/distributed/controller/util/subscription_test.go
+++ b/pkg/channel/distributed/controller/util/subscription_test.go
@@ -18,6 +18,7 @@ package util
 
 import (
 	"fmt"
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -27,6 +28,7 @@ import (
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	logtesting "knative.dev/pkg/logging/testing"
+	"knative.dev/pkg/system"
 
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 )
@@ -193,3 +195,148 @@ func TestGroupIdMapper(t *testing.T) {
 		})
 	}
 }
+
+func TestConnectionPoolKeyMapper(t *testing.T) {
+
+	// Test Data
+	kafkaChannelGroupVersion := schema.GroupVersion{
+		Group:   messagingv1.SchemeGroupVersion.Group,
+		Version: messagingv1.SchemeGroupVersion.Version,
+	}
+
+	// Define The TestCases
+	tests := []struct {
+		name         string
+		subscription *messagingv1.Subscription
+		expected     string
+		err          bool
+	}{
+		{
+			name: "fully populated subscription",
+			subscription: &messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      subscriptionName,
+					Namespace: subscriptionNamespace,
+				},
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						Kind:       constants.KafkaChannelKind,
+						Namespace:  channelNamespace,
+						Name:       channelName,
+						APIVersion: kafkaChannelGroupVersion.String(),
+					},
+				},
+			},
+			expected: fmt.Sprintf("%s.%s", channelNamespace, channelName),
+			err:      false,
+		},
+		{
+			name: "sparsely populated subscription",
+			subscription: &messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      subscriptionName,
+					Namespace: subscriptionNamespace,
+				},
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						Kind:       constants.KafkaChannelKind,
+						Name:       channelName,
+						APIVersion: kafkaChannelGroupVersion.String(),
+					},
+				},
+			},
+			expected: fmt.Sprintf("%s.%s", subscriptionNamespace, channelName),
+			err:      false,
+		},
+		{
+			name:         "nil subscription",
+			subscription: nil,
+			expected:     "",
+			err:          true,
+		},
+	}
+
+	// Execute The Tests
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Perform The Test
+			actual, err := ConnectionPoolKeyMapper(test.subscription)
+
+			// Verify Results
+			assert.Equal(t, test.err, err != nil)
+			assert.Equal(t, test.expected, actual)
+		})
+	}
+}
+
+func TestDataPlaneNamespaceMapper(t *testing.T) {
+
+	// Test Data
+	namespace := "test-namespace"
+
+	// Setup The Environment
+	os.Clearenv()
+	assert.Nil(t, os.Setenv(system.NamespaceEnvKey, namespace))
+
+	// Perform The Test
+	result, err := DataPlaneNamespaceMapper(nil)
+
+	// Verify Results
+	assert.Nil(t, err)
+	assert.Equal(t, namespace, result)
+}
+
+func TestDataPlaneLabelsMapper(t *testing.T) {
+
+	// Define The TestCases
+	tests := []struct {
+		name         string
+		subscription *messagingv1.Subscription
+		expected     map[string]string
+	}{
+		{
+			name: "Complete Channel Reference",
+			subscription: &messagingv1.Subscription{
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						Namespace: channelNamespace,
+						Name:      channelName,
+					},
+				},
+			},
+			expected: map[string]string{
+				constants.AppLabel: "test-channel-name-test-channel-nam-59d657a5-dispatcher",
+			},
+		},
+		{
+			name: "Sparse Channel Reference",
+			subscription: &messagingv1.Subscription{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: channelNamespace,
+				},
+				Spec: messagingv1.SubscriptionSpec{
+					Channel: duckv1.KReference{
+						Name: channelName,
+					},
+				},
+			},
+			expected: map[string]string{
+				constants.AppLabel: "test-channel-name-test-channel-nam-59d657a5-dispatcher",
+			},
+		},
+	}
+
+	// Execute The Tests
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Perform The Test
+			result, err := DataPlaneLabelsMapper(test.subscription)
+
+			// Verify Results
+			assert.Nil(t, err)
+			assert.Equal(t, test.expected, result)
+		})
+	}
+}
diff --git a/pkg/common/commands/resetoffset/controller/controller_test.go b/pkg/common/commands/resetoffset/controller/controller_test.go
index 8f10aa0fc..0d3c2a421 100644
--- a/pkg/common/commands/resetoffset/controller/controller_test.go
+++ b/pkg/common/commands/resetoffset/controller/controller_test.go
@@ -28,14 +28,16 @@ import (
 	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
 
-	fakeKafkaClient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	fakekafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/kafka/v1alpha1/resetoffset/fake" // Force Fake Informer Injection
+	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
 	refmapperstesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers/testing"
 	configtesting "knative.dev/eventing-kafka/pkg/common/config/testing"
 	"knative.dev/eventing-kafka/pkg/common/configmaploader"
 	fakeConfigmapLoader "knative.dev/eventing-kafka/pkg/common/configmaploader/fake"
 	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake" // Knative Fake Informer Injection
 )
 
 // Test The NewControllerFactory() Functionality
@@ -61,7 +63,7 @@ func TestNewControllerFactory(t *testing.T) {
 	ctx = context.WithValue(ctx, configmaploader.Key{}, configmapLoader.Load)
 
 	// Add The Fake Kafka Clientset To The Context (Empty)
-	ctx, fakeKafkaClientset := fakeKafkaClient.With(ctx)
+	ctx, fakeKafkaClientset := fakekafkaclient.With(ctx)
 	assert.NotNil(t, fakeKafkaClientset)
 
 	// Create A Watcher On The Configuration Settings ConfigMap & Dynamically Update Configuration
@@ -72,8 +74,11 @@ func TestNewControllerFactory(t *testing.T) {
 	mockResetOffsetRefMapperFactory := &refmapperstesting.MockResetOffsetRefMapperFactory{}
 	mockResetOffsetRefMapperFactory.On("Create", ctx).Return(mockResetOffsetRefMapper)
 
+	// Create Mock ConnectionPool For Testing
+	mockConnectionPool := &controllertesting.MockConnectionPool{}
+
 	// Verify The ResetOffset ControllerFactory Creates A ControllerConstructor
-	controllerConstructor := NewControllerFactory(mockResetOffsetRefMapperFactory)
+	controllerConstructor := NewControllerFactory(mockResetOffsetRefMapperFactory, mockConnectionPool)
 	assert.NotNil(t, controllerConstructor)
 
 	// Verify The ResetOffset ControllerConstructor
diff --git a/pkg/common/commands/resetoffset/controller/dataplane_test.go b/pkg/common/commands/resetoffset/controller/dataplane_test.go
new file mode 100644
index 000000000..db59ec080
--- /dev/null
+++ b/pkg/common/commands/resetoffset/controller/dataplane_test.go
@@ -0,0 +1,255 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"go.uber.org/multierr"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "knative.dev/control-protocol/pkg"
+	ctrlmessage "knative.dev/control-protocol/pkg/message"
+	"knative.dev/pkg/logging"
+	logtesting "knative.dev/pkg/logging/testing"
+
+	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
+	refmapperstesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers/testing"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
+)
+
+func TestReconciler_ReconcileDataPlaneServices(t *testing.T) {
+
+	// Test Data
+	podIp1 := "1.2.3.4"
+	podIp2 := "2.3.4.5"
+	podIpPort1 := fmt.Sprintf("%s:%d", podIp1, ControlProtocolServerPort)
+	podIpPort2 := fmt.Sprintf("%s:%d", podIp2, ControlProtocolServerPort)
+	podIpPorts := []string{podIpPort1, podIpPort2}
+	pods := []*corev1.Pod{
+		{Status: corev1.PodStatus{PodIP: podIp1}},
+		{Status: corev1.PodStatus{PodIP: podIp2}},
+	}
+	mockDataPlaneService1 := &controllertesting.MockService{}
+	mockDataPlaneService2 := &controllertesting.MockService{}
+	testErr := fmt.Errorf("test-error")
+
+	// Create A Context With Test Logger
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.Background(), logger)
+
+	// Define The Test Cases
+	tests := []struct {
+		name              string
+		podListerErr      error
+		connectionPoolErr error
+		expectedServices  map[string]ctrl.Service
+		expectedErr       error
+	}{
+		{
+			name:              "Success",
+			podListerErr:      nil,
+			connectionPoolErr: nil,
+			expectedServices:  map[string]ctrl.Service{podIp1: mockDataPlaneService1, podIp2: mockDataPlaneService2},
+			expectedErr:       nil,
+		},
+		{
+			name:              "PodLister Error",
+			podListerErr:      testErr,
+			connectionPoolErr: nil,
+			expectedServices:  nil,
+			expectedErr:       testErr,
+		},
+		{
+			name:              "ConnectionPool Error",
+			podListerErr:      nil,
+			connectionPoolErr: testErr,
+			expectedServices:  nil,
+			expectedErr:       testErr,
+		},
+	}
+
+	// Execute The Test Cases
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Create A ResetOffset To Test
+			resetOffset := controllertesting.NewResetOffset()
+
+			// Create A RefInfo To Test
+			refInfo := refmapperstesting.NewRefInfo()
+
+			// Create A Mock PodLister
+			mockPodNamespaceLister := &controllertesting.MockPodNamespaceLister{}
+			mockPodNamespaceLister.On("List", labels.Set(refInfo.DataPlaneLabels).AsSelector()).Return(pods, test.podListerErr)
+			mockPodLister := &controllertesting.MockPodLister{}
+			mockPodLister.On("Pods", refInfo.DataPlaneNamespace).Return(mockPodNamespaceLister)
+
+			// Create A Mock Control-Protocol ConnectionPool
+			mockConnectionPool := &controllertesting.MockConnectionPool{}
+			if test.podListerErr == nil {
+				mockConnectionPool.On("ReconcileConnections",
+					ctx,
+					refInfo.ConnectionPoolKey,
+					podIpPorts,
+					mock.AnythingOfType("func(string, control.Service)"),
+					mock.AnythingOfType("func(string)")).
+					Return(test.expectedServices, test.connectionPoolErr)
+			}
+
+			// Create A Mock Control-Protocol AsyncCommandNotificationStore
+			mockAsyncCommandNotificationStore := &controllertesting.MockAsyncCommandNotificationStore{}
+
+			// Create A Reconciler To Test
+			reconciler := &Reconciler{
+				podLister:                     mockPodLister,
+				asyncCommandNotificationStore: mockAsyncCommandNotificationStore,
+				connectionPool:                mockConnectionPool,
+			}
+
+			// Perform The Test
+			actualServices, actualErr := reconciler.reconcileDataPlaneServices(ctx, resetOffset, refInfo)
+
+			// Verify The Results
+			assert.Equal(t, test.expectedErr, actualErr)
+			assert.Equal(t, test.expectedServices, actualServices)
+			mockPodLister.AssertExpectations(t)
+			mockAsyncCommandNotificationStore.AssertExpectations(t)
+			mockConnectionPool.AssertExpectations(t)
+		})
+	}
+}
+
+func TestReconciler_StartConsumerGroups(t *testing.T) {
+	performStartStopConsumerGroupAsyncCommandsTest(t, commands.StartConsumerGroupOpCode)
+}
+
+func TestReconciler_StopConsumerGroups(t *testing.T) {
+	performStartStopConsumerGroupAsyncCommandsTest(t, commands.StopConsumerGroupOpCode)
+}
+
+func performStartStopConsumerGroupAsyncCommandsTest(t *testing.T, opCode ctrl.OpCode) {
+
+	// Test Data
+	podIp1 := "1.2.3.4"
+	podIp2 := "2.3.4.5"
+	testErr := fmt.Errorf("test-error")
+
+	// Create A Context With Test Logger
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.Background(), logger)
+
+	// Define The Test Cases
+	tests := []struct {
+		name    string
+		sendErr error
+		result  *ctrlmessage.AsyncCommandResult
+	}{
+		{
+			name:    "Success",
+			sendErr: nil,
+			result:  &ctrlmessage.AsyncCommandResult{},
+		},
+		{
+			name:    "SendAndWaitForAck Error",
+			sendErr: testErr,
+			result:  nil,
+		},
+	}
+
+	// Execute The Test Cases
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+
+			// Create A ResetOffset To Test
+			resetOffset := controllertesting.NewResetOffset()
+			resetOffsetNamespacedName := types.NamespacedName{
+				Namespace: resetOffset.GetNamespace(),
+				Name:      resetOffset.GetName(),
+			}
+
+			// Create A RefInfo To Test
+			refInfo := refmapperstesting.NewRefInfo()
+
+			// Determine The Expected CommandIDs
+			commandId1, err := generateCommandId(resetOffset, podIp1, opCode)
+			assert.Nil(t, err)
+			commandId2, err := generateCommandId(resetOffset, podIp2, opCode)
+			assert.Nil(t, err)
+
+			// Create Test ConsumerGroupAsyncCommands For Each Pod
+			consumerGroupAsyncCommand1 := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: commandId1, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId}
+			consumerGroupAsyncCommand2 := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: commandId2, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId}
+
+			// Create A Mock Control-Protocol AsyncCommandNotificationStore & Assign To Reconciler
+			mockAsyncCommandNotificationStore := &controllertesting.MockAsyncCommandNotificationStore{}
+			if test.result != nil {
+				mockAsyncCommandNotificationStore.On("GetCommandResult", resetOffsetNamespacedName, podIp1, consumerGroupAsyncCommand1).Return(test.result)
+				mockAsyncCommandNotificationStore.On("GetCommandResult", resetOffsetNamespacedName, podIp2, consumerGroupAsyncCommand2).Return(test.result)
+			}
+
+			// Create The Mock Services To Test Against
+			mockDataPlaneService1 := &controllertesting.MockService{}
+			mockDataPlaneService2 := &controllertesting.MockService{}
+			mockDataPlaneService1.On("SendAndWaitForAck", opCode, consumerGroupAsyncCommand1).Return(test.sendErr)
+			mockDataPlaneService2.On("SendAndWaitForAck", opCode, consumerGroupAsyncCommand2).Return(test.sendErr)
+			services := map[string]ctrl.Service{podIp1: mockDataPlaneService1, podIp2: mockDataPlaneService2}
+
+			// Create A Reconciler To Test
+			reconciler := &Reconciler{
+				asyncCommandNotificationStore: mockAsyncCommandNotificationStore,
+			}
+
+			// Perform The Test Based On OpCode
+			var actualErr error
+			switch opCode {
+			case commands.StartConsumerGroupOpCode:
+				actualErr = reconciler.startConsumerGroups(ctx, resetOffset, services, refInfo)
+			case commands.StopConsumerGroupOpCode:
+				actualErr = reconciler.stopConsumerGroups(ctx, resetOffset, services, refInfo)
+			default:
+				assert.Fail(t, "Unexpected OpCode")
+			}
+
+			// Setup Expected Errors Based On Test
+			var expectedErrs []error
+			if test.sendErr != nil {
+				expectedErrs = []error{
+					fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d' : %v", commandId1, testErr),
+					fmt.Errorf("failed to send ConsumerGroup AsyncCommand '%d' : %v", commandId2, testErr),
+				}
+			}
+
+			// Verify The Results
+			if expectedErrs == nil {
+				assert.Nil(t, actualErr)
+			} else {
+				for _, expectedErr := range expectedErrs {
+					assert.Contains(t, multierr.Errors(actualErr), expectedErr)
+				}
+			}
+			mockAsyncCommandNotificationStore.AssertExpectations(t)
+			mockDataPlaneService1.AssertExpectations(t)
+			mockDataPlaneService2.AssertExpectations(t)
+		})
+	}
+}
diff --git a/pkg/common/commands/resetoffset/controller/offsets_test.go b/pkg/common/commands/resetoffset/controller/offsets_test.go
index 681382786..073923839 100644
--- a/pkg/common/commands/resetoffset/controller/offsets_test.go
+++ b/pkg/common/commands/resetoffset/controller/offsets_test.go
@@ -29,6 +29,7 @@ import (
 
 	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
 	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
 )
 
 //
@@ -310,7 +311,7 @@ func TestReconciler_ReconcileOffsets(t *testing.T) {
 				controllertesting.WithOffsetManagerMockManagePartition(topicName, partition, partitionOffsetManager, nil)(test.offsetManager)
 			}
 
-			// Create A Reconciler
+			// Create A Reconciler To Test
 			reconciler := &Reconciler{
 				kafkaBrokers:      kafkaBrokers,
 				saramaConfig:      saramaConfig,
@@ -318,8 +319,14 @@ func TestReconciler_ReconcileOffsets(t *testing.T) {
 				refMapper:         nil,
 			}
 
+			// Create The RefInfo
+			refInfo := &refmappers.RefInfo{
+				TopicName: topicName,
+				GroupId:   groupId,
+			}
+
 			// Perform The Test
-			offsetMappings, err := reconciler.reconcileOffsets(ctx, topicName, groupId, offsetTime)
+			offsetMappings, err := reconciler.reconcileOffsets(ctx, refInfo, offsetTime)
 
 			// Verify The Results
 			assert.Equal(t, test.expectedErr, err)
diff --git a/pkg/common/commands/resetoffset/controller/reconciler_test.go b/pkg/common/commands/resetoffset/controller/reconciler_test.go
index c7fb3caa6..4f21577d0 100644
--- a/pkg/common/commands/resetoffset/controller/reconciler_test.go
+++ b/pkg/common/commands/resetoffset/controller/reconciler_test.go
@@ -26,10 +26,14 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/fake"
 	clientgotesting "k8s.io/client-go/testing"
+	ctrl "knative.dev/control-protocol/pkg"
+	ctrlmessage "knative.dev/control-protocol/pkg/message"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -43,6 +47,7 @@ import (
 	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
 	refmapperstesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers/testing"
 	"knative.dev/eventing-kafka/pkg/common/constants"
+	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
 )
 
@@ -68,6 +73,11 @@ func TestReconcile(t *testing.T) {
 		{Partition: 0, OldOffset: oldOffset, NewOffset: newOffset},
 	}
 
+	podIp := "1.2.3.4"
+	pods := []*corev1.Pod{{Status: corev1.PodStatus{PodIP: podIp}}}
+	podIpPort := fmt.Sprintf("%s:%d", podIp, ControlProtocolServerPort)
+	podIpPorts := []string{podIpPort}
+
 	testErr := fmt.Errorf("test-error")
 
 	// Define The ResetOffset Reconciler Test Cases
@@ -176,6 +186,45 @@ func TestReconcile(t *testing.T) {
 				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to map 'ref' to Kafka Topic and Group: %v", testErr.Error())),
 			},
 		},
+		{
+			Name:          "Reconcile DataPlane Services Error",
+			Key:           controllertesting.ResetOffsetKey,
+			Objects:       []runtime.Object{controllertesting.NewResetOffset(controllertesting.WithFinalizer)},
+			OtherTestData: map[string]interface{}{"ReconcileConnectionsErr": testErr},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewResetOffset(
+						controllertesting.WithFinalizer,
+						controllertesting.WithStatusInitialized,
+						controllertesting.WithStatusTopic(topicName),
+						controllertesting.WithStatusGroup(groupId),
+						controllertesting.WithStatusRefMapped(true)),
+				},
+			},
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to reconcile DataPlane Services from ConnectionPool: %v", testErr.Error())),
+			},
+		},
+		{
+			Name:          "Stop ConsumerGroups Error",
+			Key:           controllertesting.ResetOffsetKey,
+			Objects:       []runtime.Object{controllertesting.NewResetOffset(controllertesting.WithFinalizer)},
+			OtherTestData: map[string]interface{}{"StopConsumerGroupsErr": testErr},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewResetOffset(
+						controllertesting.WithFinalizer,
+						controllertesting.WithStatusInitialized,
+						controllertesting.WithStatusTopic(topicName),
+						controllertesting.WithStatusGroup(groupId),
+						controllertesting.WithStatusRefMapped(true),
+						controllertesting.WithStatusResetInitiated(true),
+						controllertesting.WithStatusConsumerGroupsStopped(false, "FailedToStopConsumerGroups", fmt.Sprintf("Failed to stop one or more ConsumerGroups: failed to send ConsumerGroup AsyncCommand '3866008807' : %v", testErr.Error()))),
+				},
+			},
+			WantErr: false,
+		},
 		{
 			Name:          "Reconcile Offsets Error",
 			Key:           controllertesting.ResetOffsetKey,
@@ -191,7 +240,7 @@ func TestReconcile(t *testing.T) {
 						controllertesting.WithStatusRefMapped(true),
 						controllertesting.WithStatusResetInitiated(true),
 						controllertesting.WithStatusConsumerGroupsStopped(true),
-						controllertesting.WithStatusOffsetsUpdated(false, "FailedToUpdateOffsets", "Failed to update offsets of one or more partitions: test-error")),
+						controllertesting.WithStatusOffsetsUpdated(false, "FailedToUpdateOffsets", "Failed to update Offsets of one or more Partitions: test-error")),
 				},
 			},
 			WantErr: true,
@@ -199,6 +248,28 @@ func TestReconcile(t *testing.T) {
 				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf("failed to update offsets of one or more partitions: %v", testErr.Error())),
 			},
 		},
+		{
+			Name:          "Start ConsumerGroups Error",
+			Key:           controllertesting.ResetOffsetKey,
+			Objects:       []runtime.Object{controllertesting.NewResetOffset(controllertesting.WithFinalizer)},
+			OtherTestData: map[string]interface{}{"StartConsumerGroupsErr": testErr},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: controllertesting.NewResetOffset(
+						controllertesting.WithFinalizer,
+						controllertesting.WithStatusInitialized,
+						controllertesting.WithStatusTopic(topicName),
+						controllertesting.WithStatusGroup(groupId),
+						controllertesting.WithStatusPartitions(offsetMappings),
+						controllertesting.WithStatusRefMapped(true),
+						controllertesting.WithStatusResetInitiated(true),
+						controllertesting.WithStatusConsumerGroupsStopped(true),
+						controllertesting.WithStatusOffsetsUpdated(true),
+						controllertesting.WithStatusConsumerGroupsStarted(false, "FailedToStartConsumerGroups", fmt.Sprintf("Failed to start one or more ConsumerGroups: failed to send ConsumerGroup AsyncCommand '3899564045' : %v", testErr))),
+				},
+			},
+			WantErr: false,
+		},
 
 		//
 		// Finalize Tests
@@ -271,9 +342,80 @@ func TestReconcile(t *testing.T) {
 			}
 		}
 
+		// Check ReconcileConnectionsErr Option
+		var reconcileConnectionsErr error
+		if options != nil {
+			mapRefErrOption := options["ReconcileConnectionsErr"]
+			if err, ok := mapRefErrOption.(error); ok {
+				reconcileConnectionsErr = err
+			}
+		}
+
+		// Check StopConsumerGroupsErr Option
+		var stopConsumerGroupsErr error
+		if options != nil {
+			mapRefErrOption := options["StopConsumerGroupsErr"]
+			if err, ok := mapRefErrOption.(error); ok {
+				stopConsumerGroupsErr = err
+			}
+		}
+
+		// Check StartConsumerGroupsErr Option
+		var startConsumerGroupsErr error
+		if options != nil {
+			mapRefErrOption := options["StartConsumerGroupsErr"]
+			if err, ok := mapRefErrOption.(error); ok {
+				startConsumerGroupsErr = err
+			}
+		}
+
+		// Create A RefInfo For Testing
+		refInfo := refmapperstesting.NewRefInfo()
+
+		// Create A Mock PodLister
+		mockPodNamespaceLister := &controllertesting.MockPodNamespaceLister{}
+		mockPodNamespaceLister.On("List", labels.Set(refInfo.DataPlaneLabels).AsSelector()).Return(pods, nil)
+		mockPodLister := &controllertesting.MockPodLister{}
+		mockPodLister.On("Pods", refInfo.DataPlaneNamespace).Return(mockPodNamespaceLister)
+
+		// Create Test ConsumerGroupAsyncCommands
+		stopCommandId, err := generateCommandId(controllertesting.NewResetOffset(), podIp, commands.StopConsumerGroupOpCode)
+		assert.Nil(t, err)
+		startCommandId, err := generateCommandId(controllertesting.NewResetOffset(), podIp, commands.StartConsumerGroupOpCode)
+		assert.Nil(t, err)
+		stopConsumerGroupAsyncCommand := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: stopCommandId, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId}
+		startConsumerGroupAsyncCommand := &commands.ConsumerGroupAsyncCommand{Version: 1, CommandId: startCommandId, TopicName: refInfo.TopicName, GroupId: refInfo.GroupId}
+
+		// Create The Mock Service To Test Against
+		mockDataPlaneService := &controllertesting.MockService{}
+		mockDataPlaneService.On("SendAndWaitForAck", commands.StopConsumerGroupOpCode, stopConsumerGroupAsyncCommand).Return(stopConsumerGroupsErr)
+		mockDataPlaneService.On("SendAndWaitForAck", commands.StartConsumerGroupOpCode, startConsumerGroupAsyncCommand).Return(startConsumerGroupsErr)
+		services := map[string]ctrl.Service{podIp: mockDataPlaneService}
+
+		// Create A Mock Control-Protocol ConnectionPool
+		mockConnectionPool := &controllertesting.MockConnectionPool{}
+		mockConnectionPool.On("ReconcileConnections",
+			mock.Anything,
+			refInfo.ConnectionPoolKey,
+			podIpPorts,
+			mock.AnythingOfType("func(string, control.Service)"),
+			mock.AnythingOfType("func(string)")).
+			Return(services, reconcileConnectionsErr)
+
 		// Create A Mock ResetOffset RefMapper
 		mockResetOffsetRefMapper := &refmapperstesting.MockResetOffsetRefMapper{}
-		mockResetOffsetRefMapper.On("MapRef", mock.Anything).Return(controllertesting.TopicName, controllertesting.GroupId, mapRefErr)
+		mockResetOffsetRefMapper.On("MapRef", mock.Anything).Return(refInfo, mapRefErr)
+
+		// Create A Mock Control-Protocol AsyncCommandNotificationStore & Assign To Reconciler
+		resetOffsetNamespacedName := controllertesting.NewResetOffsetNamespacedName()
+		successResult := &ctrlmessage.AsyncCommandResult{}
+		mockAsyncCommandNotificationStore := &controllertesting.MockAsyncCommandNotificationStore{}
+		mockAsyncCommandNotificationStore.On("GetCommandResult", resetOffsetNamespacedName, podIp, stopConsumerGroupAsyncCommand).Return(successResult)
+		mockAsyncCommandNotificationStore.On("GetCommandResult", resetOffsetNamespacedName, podIp, startConsumerGroupAsyncCommand).Return(successResult)
+		mockAsyncCommandNotificationStore.On("CleanPodsNotifications", types.NamespacedName{
+			Namespace: controllertesting.ResetOffsetNamespace,
+			Name:      controllertesting.ResetOffsetName,
+		}).Return()
 
 		// Check SaramaNewClientFnErr Option
 		var saramaNewClientFnErr error
@@ -292,10 +434,13 @@ func TestReconcile(t *testing.T) {
 
 		// Create The ResetOffset Reconciler Struct
 		r := &Reconciler{
-			kafkaBrokers:      kafkaBrokers,
-			saramaConfig:      saramaConfig,
-			resetoffsetLister: listers.GetResetOffsetLister(),
-			refMapper:         mockResetOffsetRefMapper,
+			kafkaBrokers:                  kafkaBrokers,
+			saramaConfig:                  saramaConfig,
+			podLister:                     mockPodLister,
+			resetoffsetLister:             listers.GetResetOffsetLister(),
+			refMapper:                     mockResetOffsetRefMapper,
+			connectionPool:                mockConnectionPool,
+			asyncCommandNotificationStore: mockAsyncCommandNotificationStore,
 		}
 
 		// Create / Return The Full Reconciler
diff --git a/pkg/common/commands/resetoffset/controller/testing/data.go b/pkg/common/commands/resetoffset/controller/testing/data.go
index 06beb81bd..434a1e1a9 100644
--- a/pkg/common/commands/resetoffset/controller/testing/data.go
+++ b/pkg/common/commands/resetoffset/controller/testing/data.go
@@ -18,6 +18,7 @@ package testing
 
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
@@ -175,3 +176,10 @@ func WithStatusConsumerGroupsStarted(state bool, failed ...string) ResetOffsetOp
 		}
 	}
 }
+
+func NewResetOffsetNamespacedName() types.NamespacedName {
+	return types.NamespacedName{
+		Namespace: ResetOffsetNamespace,
+		Name:      ResetOffsetName,
+	}
+}
diff --git a/pkg/common/commands/resetoffset/controller/testing/mock.go b/pkg/common/commands/resetoffset/controller/testing/mock.go
index 455860c59..1eaf95286 100644
--- a/pkg/common/commands/resetoffset/controller/testing/mock.go
+++ b/pkg/common/commands/resetoffset/controller/testing/mock.go
@@ -17,15 +17,25 @@ limitations under the License.
 package testing
 
 import (
+	"context"
+	"encoding"
+
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/mock"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	listerscorev1 "k8s.io/client-go/listers/core/v1"
+	ctrl "knative.dev/control-protocol/pkg"
+	ctrlmessage "knative.dev/control-protocol/pkg/message"
+	ctrlreconciler "knative.dev/control-protocol/pkg/reconciler"
 )
 
 //
 // Mock Sarama Client
 //
 
-var _ sarama.Client = &MockClient{}
+var _ sarama.Client = (*MockClient)(nil)
 
 type MockClient struct {
 	mock.Mock
@@ -169,7 +179,7 @@ func WithClientMockClose(err error) MockClientOption {
 // Mock Sarama OffsetManager
 //
 
-var _ sarama.OffsetManager = &MockOffsetManager{}
+var _ sarama.OffsetManager = (*MockOffsetManager)(nil)
 
 type MockOffsetManager struct {
 	mock.Mock
@@ -226,7 +236,7 @@ func WithOffsetManagerMockClose(err error) MockOffsetManagerOption {
 // Mock Sarama PartitionOffsetManager
 //
 
-var _ sarama.PartitionOffsetManager = &MockPartitionOffsetManager{}
+var _ sarama.PartitionOffsetManager = (*MockPartitionOffsetManager)(nil)
 
 type MockPartitionOffsetManager struct {
 	mock.Mock
@@ -315,3 +325,143 @@ func WithPartitionOffsetManagerMockAsyncClose() MockPartitionOffsetManagerOption
 		mockPartitionOffsetManager.On("AsyncClose").Return()
 	}
 }
+
+// TODO - Move these control-protocol mocks to common/controlprotocol/testing/mocks.go once eric merges!
+
+//
+// Mock Control-Protocol ConnectionPool
+//
+
+var _ ctrlreconciler.ControlPlaneConnectionPool = (*MockConnectionPool)(nil)
+
+type MockConnectionPool struct {
+	mock.Mock
+}
+
+func (c *MockConnectionPool) GetConnectedHosts(key string) []string {
+	args := c.Called(key)
+	return args.Get(0).([]string)
+}
+
+func (c *MockConnectionPool) GetServices(key string) map[string]ctrl.Service {
+	args := c.Called(key)
+	return args.Get(0).(map[string]ctrl.Service)
+}
+
+func (c *MockConnectionPool) ResolveControlInterface(key string, host string) (string, ctrl.Service) {
+	args := c.Called(key, host)
+	return args.String(0), args.Get(1).(ctrl.Service)
+}
+
+func (c *MockConnectionPool) RemoveConnection(ctx context.Context, key string, host string) {
+	c.Called(ctx, key, host)
+}
+
+func (c *MockConnectionPool) RemoveAllConnections(ctx context.Context, key string) {
+	c.Called(ctx, key)
+}
+
+func (c *MockConnectionPool) Close(ctx context.Context) {
+	c.Called(ctx)
+}
+
+func (c *MockConnectionPool) ReconcileConnections(ctx context.Context, key string, wantConnections []string, newServiceCb func(string, ctrl.Service), oldServiceCb func(string)) (map[string]ctrl.Service, error) {
+	args := c.Called(ctx, key, wantConnections, newServiceCb, oldServiceCb)
+	return args.Get(0).(map[string]ctrl.Service), args.Error(1)
+}
+
+func (c *MockConnectionPool) DialControlService(ctx context.Context, key string, host string) (string, ctrl.Service, error) {
+	args := c.Called(ctx, key, host)
+	return args.String(0), args.Get(1).(ctrl.Service), args.Error(2)
+}
+
+//
+// Mock Control-Protocol AsyncCommandNotificationStore
+//
+
+var _ ctrlreconciler.AsyncCommandNotificationStore = (*MockAsyncCommandNotificationStore)(nil)
+
+type MockAsyncCommandNotificationStore struct {
+	mock.Mock
+}
+
+func (a *MockAsyncCommandNotificationStore) GetCommandResult(srcName types.NamespacedName, pod string, command ctrlmessage.AsyncCommand) *ctrlmessage.AsyncCommandResult {
+	args := a.Called(srcName, pod, command)
+	return args.Get(0).(*ctrlmessage.AsyncCommandResult)
+}
+
+func (a *MockAsyncCommandNotificationStore) CleanPodsNotifications(srcName types.NamespacedName) {
+	a.Called(srcName)
+}
+
+func (a *MockAsyncCommandNotificationStore) CleanPodNotification(srcName types.NamespacedName, pod string) {
+	a.Called(srcName, pod)
+}
+
+func (a *MockAsyncCommandNotificationStore) MessageHandler(srcName types.NamespacedName, pod string) ctrl.MessageHandler {
+	args := a.Called(srcName, pod)
+	return args.Get(0).(ctrl.MessageHandler)
+}
+
+//
+// Mock Control-Protocol Service
+//
+
+var _ ctrl.Service = (*MockService)(nil)
+
+type MockService struct {
+	mock.Mock
+}
+
+func (s *MockService) SendAndWaitForAck(opcode ctrl.OpCode, payload encoding.BinaryMarshaler) error {
+	args := s.Called(opcode, payload)
+	return args.Error(0)
+}
+
+func (s *MockService) MessageHandler(handler ctrl.MessageHandler) {
+	s.Called(handler)
+}
+
+func (s *MockService) ErrorHandler(handler ctrl.ErrorHandler) {
+	s.Called(handler)
+}
+
+//
+// Mock K8S PodLister
+//
+
+var _ listerscorev1.PodLister = (*MockPodLister)(nil)
+
+type MockPodLister struct {
+	mock.Mock
+}
+
+func (l *MockPodLister) List(selector labels.Selector) (ret []*corev1.Pod, err error) {
+	args := l.Called(selector)
+	return args.Get(0).([]*corev1.Pod), args.Error(1)
+}
+
+func (l *MockPodLister) Pods(namespace string) listerscorev1.PodNamespaceLister {
+	args := l.Called(namespace)
+	return args.Get(0).(listerscorev1.PodNamespaceLister)
+}
+
+//
+// Mock K8S PodNamespaceLister
+//
+
+var _ listerscorev1.PodNamespaceLister = (*MockPodNamespaceLister)(nil)
+
+type MockPodNamespaceLister struct {
+	mock.Mock
+}
+
+func (nl *MockPodNamespaceLister) List(selector labels.Selector) (ret []*corev1.Pod, err error) {
+	args := nl.Called(selector)
+	return args.Get(0).([]*corev1.Pod), args.Error(1)
+}
+
+func (nl *MockPodNamespaceLister) Get(name string) (*corev1.Pod, error) {
+	args := nl.Called(name)
+	return args.Get(0).(*corev1.Pod), args.Error(1)
+}
diff --git a/pkg/common/commands/resetoffset/refmappers/subscription_test.go b/pkg/common/commands/resetoffset/refmappers/subscription_test.go
index c477ba339..cd16f29bb 100644
--- a/pkg/common/commands/resetoffset/refmappers/subscription_test.go
+++ b/pkg/common/commands/resetoffset/refmappers/subscription_test.go
@@ -42,8 +42,18 @@ const (
 	SubscriptionNamespace = "subscription-namespace"
 	SubscriptionName      = "subscription-name"
 
-	TopicName = "TestTopicName"
-	GroupId   = "TestGroupId"
+	TopicName            = "TestTopicName"
+	GroupId              = "TestGroupId"
+	ConnectionPoolKey    = "TestConnectionPoolKey"
+	DataPlaneNamespace   = "TestDataPlaneNamespace"
+	DataPlaneLabelKey1   = "TestDataPlaneLabelKey1"
+	DataPlaneLabelValue1 = "TestDataPlaneLabelValue1"
+	DataPlaneLabelKey2   = "TestDataPlaneLabelKey2"
+	DataPlaneLabelValue2 = "TestDataPlaneLabelValue2"
+)
+
+var (
+	DataPlaneLabels = map[string]string{DataPlaneLabelKey1: DataPlaneLabelValue1, DataPlaneLabelKey2: DataPlaneLabelValue2}
 )
 
 func TestNewSubscriptionRefMapperFactory(t *testing.T) {
@@ -59,9 +69,12 @@ func TestNewSubscriptionRefMapperFactory(t *testing.T) {
 	// Create Test Mappers
 	topicNameMapper := newMockSubscriptionTopicNameMapper(t, nil, TopicName, nil)
 	groupIdMapper := newMockSubscriptionConsumerGroupIdMapper(t, nil, GroupId, nil)
+	connectionPoolKeyMapper := newMockSubscriptionConnectionPoolKeyMapper(t, nil, ConnectionPoolKey, nil)
+	dataPlaneNamespaceMapper := newMockSubscriptionDataPlaneNamespaceMapper(t, nil, DataPlaneNamespace, nil)
+	dataPlaneLabelsMapper := newMockSubscriptionDataPlaneLabelsMapper(t, nil, DataPlaneLabels, nil)
 
 	// Perform The Test - Create New Subscription RefMapper Factory
-	factory := NewSubscriptionRefMapperFactory(topicNameMapper, groupIdMapper)
+	factory := NewSubscriptionRefMapperFactory(topicNameMapper, groupIdMapper, connectionPoolKeyMapper, dataPlaneNamespaceMapper, dataPlaneLabelsMapper)
 	assert.NotNil(t, factory)
 
 	// Test The Factory Create()
@@ -82,9 +95,12 @@ func TestNewResetOffsetSubscriptionRefMapper(t *testing.T) {
 	// Create Test Mappers
 	topicNameMapper := newMockSubscriptionTopicNameMapper(t, nil, TopicName, nil)
 	groupIdMapper := newMockSubscriptionConsumerGroupIdMapper(t, nil, GroupId, nil)
+	connectionPoolKeyMapper := newMockSubscriptionConnectionPoolKeyMapper(t, nil, ConnectionPoolKey, nil)
+	dataPlaneNamespaceMapper := newMockSubscriptionDataPlaneNamespaceMapper(t, nil, DataPlaneNamespace, nil)
+	dataPlaneLabelsMapper := newMockSubscriptionDataPlaneLabelsMapper(t, nil, DataPlaneLabels, nil)
 
 	// Perform The Test - Create A New SubscriptionRefMapper
-	resetOffsetSubscriptionRefMapper := NewSubscriptionRefMapper(ctx, topicNameMapper, groupIdMapper)
+	resetOffsetSubscriptionRefMapper := NewSubscriptionRefMapper(ctx, topicNameMapper, groupIdMapper, connectionPoolKeyMapper, dataPlaneNamespaceMapper, dataPlaneLabelsMapper)
 
 	// Verify The Results
 	assert.NotNil(t, resetOffsetSubscriptionRefMapper)
@@ -92,6 +108,7 @@ func TestNewResetOffsetSubscriptionRefMapper(t *testing.T) {
 	assert.NotNil(t, resetOffsetSubscriptionRefMapper.subscriptionLister)
 	assert.NotNil(t, resetOffsetSubscriptionRefMapper.topicNameMapper) // Testify / DeepEqual Cannot Compare func Types
 	assert.NotNil(t, resetOffsetSubscriptionRefMapper.groupIdMapper)
+	assert.NotNil(t, resetOffsetSubscriptionRefMapper.connectionPoolKeyMapper)
 }
 
 func TestResetOffsetSubscriptionRefMapper_MapRef(t *testing.T) {
@@ -122,24 +139,34 @@ func TestResetOffsetSubscriptionRefMapper_MapRef(t *testing.T) {
 
 	// Define The Test Cases
 	tests := []struct {
-		name            string
-		subscription    *messagingv1.Subscription
-		subscriptionErr error
-		resetOffset     *kafkav1alpha1.ResetOffset
-		topicNameMapper SubscriptionTopicNameMapper
-		groupIdMapper   SubscriptionConsumerGroupIdMapper
-		wantTopic       string
-		wantGroup       string
-		wantErr         bool
+		name                     string
+		subscription             *messagingv1.Subscription
+		subscriptionErr          error
+		resetOffset              *kafkav1alpha1.ResetOffset
+		topicNameMapper          SubscriptionTopicNameMapper
+		groupIdMapper            SubscriptionConsumerGroupIdMapper
+		connectionPoolKeyMapper  SubscriptionConnectionPoolKeyMapper
+		dataPlaneNamespaceMapper SubscriptionDataPlaneNamespaceMapper
+		dataPlaneLabelsMapper    SubscriptionDataPlaneLabelsMapper
+		wantRefInfo              *RefInfo
+		wantErr                  bool
 	}{
 		{
-			name:            "Success",
-			subscription:    subscription,
-			resetOffset:     controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
-			topicNameMapper: newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
-			groupIdMapper:   newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
-			wantTopic:       TopicName,
-			wantGroup:       GroupId,
+			name:                     "Success",
+			subscription:             subscription,
+			resetOffset:              controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper:          newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
+			groupIdMapper:            newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
+			connectionPoolKeyMapper:  newMockSubscriptionConnectionPoolKeyMapper(t, subscription, ConnectionPoolKey, nil),
+			dataPlaneNamespaceMapper: newMockSubscriptionDataPlaneNamespaceMapper(t, subscription, DataPlaneNamespace, nil),
+			dataPlaneLabelsMapper:    newMockSubscriptionDataPlaneLabelsMapper(t, subscription, DataPlaneLabels, nil),
+			wantRefInfo: &RefInfo{
+				TopicName:          TopicName,
+				GroupId:            GroupId,
+				ConnectionPoolKey:  ConnectionPoolKey,
+				DataPlaneNamespace: DataPlaneNamespace,
+				DataPlaneLabels:    DataPlaneLabels,
+			},
 		},
 		{
 			name:        "Nil ResetOffset",
@@ -157,12 +184,34 @@ func TestResetOffsetSubscriptionRefMapper_MapRef(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "Sparse ResetOffset.Spec.Ref",
+			name:         "ResetOffset.Spec.Ref Without Namespace",
+			subscription: subscription,
 			resetOffset: controllertesting.NewResetOffset(controllertesting.WithSpecRef(&duckv1.KReference{
 				Kind:       "Subscription",
 				APIVersion: messagingv1.SchemeGroupVersion.String(),
 				Name:       SubscriptionName,
 			})),
+			topicNameMapper:          newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
+			groupIdMapper:            newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
+			connectionPoolKeyMapper:  newMockSubscriptionConnectionPoolKeyMapper(t, subscription, ConnectionPoolKey, nil),
+			dataPlaneNamespaceMapper: newMockSubscriptionDataPlaneNamespaceMapper(t, subscription, DataPlaneNamespace, nil),
+			dataPlaneLabelsMapper:    newMockSubscriptionDataPlaneLabelsMapper(t, subscription, DataPlaneLabels, nil),
+			wantRefInfo: &RefInfo{
+				TopicName:          TopicName,
+				GroupId:            GroupId,
+				ConnectionPoolKey:  ConnectionPoolKey,
+				DataPlaneNamespace: DataPlaneNamespace,
+				DataPlaneLabels:    DataPlaneLabels,
+			},
+			wantErr: false,
+		},
+		{
+			name: "ResetOffset.Spec.Ref Without Name",
+			resetOffset: controllertesting.NewResetOffset(controllertesting.WithSpecRef(&duckv1.KReference{
+				Kind:       "Subscription",
+				APIVersion: messagingv1.SchemeGroupVersion.String(),
+				Namespace:  SubscriptionNamespace,
+			})),
 			wantErr: true,
 		},
 		{
@@ -172,20 +221,59 @@ func TestResetOffsetSubscriptionRefMapper_MapRef(t *testing.T) {
 			wantErr:         true,
 		},
 		{
-			name:            "TopicName Mapper Error",
-			subscription:    subscription,
-			resetOffset:     controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
-			topicNameMapper: newMockSubscriptionTopicNameMapper(t, subscription, TopicName, testErr),
-			groupIdMapper:   newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
-			wantErr:         true,
+			name:                     "TopicName Mapper Error",
+			subscription:             subscription,
+			resetOffset:              controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper:          newMockSubscriptionTopicNameMapper(t, subscription, TopicName, testErr),
+			groupIdMapper:            newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
+			connectionPoolKeyMapper:  newMockSubscriptionConnectionPoolKeyMapper(t, subscription, ConnectionPoolKey, nil),
+			dataPlaneNamespaceMapper: newMockSubscriptionDataPlaneNamespaceMapper(t, subscription, DataPlaneNamespace, nil),
+			dataPlaneLabelsMapper:    newMockSubscriptionDataPlaneLabelsMapper(t, subscription, DataPlaneLabels, nil),
+			wantErr:                  true,
 		},
 		{
-			name:            "GroupId Mapper Error",
-			subscription:    subscription,
-			resetOffset:     controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
-			topicNameMapper: newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
-			groupIdMapper:   newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, testErr),
-			wantErr:         true,
+			name:                     "GroupId Mapper Error",
+			subscription:             subscription,
+			resetOffset:              controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper:          newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
+			groupIdMapper:            newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, testErr),
+			connectionPoolKeyMapper:  newMockSubscriptionConnectionPoolKeyMapper(t, subscription, ConnectionPoolKey, nil),
+			dataPlaneNamespaceMapper: newMockSubscriptionDataPlaneNamespaceMapper(t, subscription, DataPlaneNamespace, nil),
+			dataPlaneLabelsMapper:    newMockSubscriptionDataPlaneLabelsMapper(t, subscription, DataPlaneLabels, nil),
+			wantErr:                  true,
+		},
+		{
+			name:                     "ConnectionPoolKey Mapper Error",
+			subscription:             subscription,
+			resetOffset:              controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper:          newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
+			groupIdMapper:            newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
+			connectionPoolKeyMapper:  newMockSubscriptionConnectionPoolKeyMapper(t, subscription, ConnectionPoolKey, testErr),
+			dataPlaneNamespaceMapper: newMockSubscriptionDataPlaneNamespaceMapper(t, subscription, DataPlaneNamespace, nil),
+			dataPlaneLabelsMapper:    newMockSubscriptionDataPlaneLabelsMapper(t, subscription, DataPlaneLabels, nil),
+			wantErr:                  true,
+		},
+		{
+			name:                     "DataPlaneNamespace Mapper Error",
+			subscription:             subscription,
+			resetOffset:              controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper:          newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
+			groupIdMapper:            newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
+			connectionPoolKeyMapper:  newMockSubscriptionConnectionPoolKeyMapper(t, subscription, ConnectionPoolKey, nil),
+			dataPlaneNamespaceMapper: newMockSubscriptionDataPlaneNamespaceMapper(t, subscription, DataPlaneNamespace, testErr),
+			dataPlaneLabelsMapper:    newMockSubscriptionDataPlaneLabelsMapper(t, subscription, DataPlaneLabels, nil),
+			wantErr:                  true,
+		},
+		{
+			name:                     "DataPlaneLabels Mapper Error",
+			subscription:             subscription,
+			resetOffset:              controllertesting.NewResetOffset(controllertesting.WithSpecRef(subscriptionRef)),
+			topicNameMapper:          newMockSubscriptionTopicNameMapper(t, subscription, TopicName, nil),
+			groupIdMapper:            newMockSubscriptionConsumerGroupIdMapper(t, subscription, GroupId, nil),
+			connectionPoolKeyMapper:  newMockSubscriptionConnectionPoolKeyMapper(t, subscription, ConnectionPoolKey, nil),
+			dataPlaneNamespaceMapper: newMockSubscriptionDataPlaneNamespaceMapper(t, subscription, DataPlaneNamespace, nil),
+			dataPlaneLabelsMapper:    newMockSubscriptionDataPlaneLabelsMapper(t, subscription, DataPlaneLabels, testErr),
+			wantErr:                  true,
 		},
 	}
 
@@ -204,19 +292,21 @@ func TestResetOffsetSubscriptionRefMapper_MapRef(t *testing.T) {
 
 			// Create A New SubscriptionRefMapper To Test
 			subscriptionRefMapper := &SubscriptionRefMapper{
-				logger:             logger,
-				subscriptionLister: mockSubscriptionLister,
-				topicNameMapper:    test.topicNameMapper,
-				groupIdMapper:      test.groupIdMapper,
+				logger:                   logger,
+				subscriptionLister:       mockSubscriptionLister,
+				topicNameMapper:          test.topicNameMapper,
+				groupIdMapper:            test.groupIdMapper,
+				connectionPoolKeyMapper:  test.connectionPoolKeyMapper,
+				dataPlaneNamespaceMapper: test.dataPlaneNamespaceMapper,
+				dataPlaneLabelsMapper:    test.dataPlaneLabelsMapper,
 			}
 
 			// Perform The Test - Map A Subscription To Kafka Topic Name & ConsumerGroup ID
-			topicName, groupId, err := subscriptionRefMapper.MapRef(test.resetOffset)
+			refInfo, err := subscriptionRefMapper.MapRef(test.resetOffset)
 
 			// Validate The Results
 			assert.Equal(t, test.wantErr, err != nil)
-			assert.Equal(t, test.wantTopic, topicName)
-			assert.Equal(t, test.wantGroup, groupId)
+			assert.Equal(t, test.wantRefInfo, refInfo)
 		})
 	}
 }
@@ -278,3 +368,24 @@ func newMockSubscriptionConsumerGroupIdMapper(t *testing.T, expectedSubscription
 		return groupId, err
 	}
 }
+
+func newMockSubscriptionConnectionPoolKeyMapper(t *testing.T, expectedSubscription *messagingv1.Subscription, connectionPoolKey string, err error) SubscriptionConnectionPoolKeyMapper {
+	return func(subscription *messagingv1.Subscription) (string, error) {
+		assert.Equal(t, expectedSubscription, subscription)
+		return connectionPoolKey, err
+	}
+}
+
+func newMockSubscriptionDataPlaneNamespaceMapper(t *testing.T, expectedSubscription *messagingv1.Subscription, dataPlaneNamespace string, err error) SubscriptionDataPlaneNamespaceMapper {
+	return func(subscription *messagingv1.Subscription) (string, error) {
+		assert.Equal(t, expectedSubscription, subscription)
+		return dataPlaneNamespace, err
+	}
+}
+
+func newMockSubscriptionDataPlaneLabelsMapper(t *testing.T, expectedSubscription *messagingv1.Subscription, dataPlaneLabels map[string]string, err error) SubscriptionDataPlaneLabelsMapper {
+	return func(subscription *messagingv1.Subscription) (map[string]string, error) {
+		assert.Equal(t, expectedSubscription, subscription)
+		return dataPlaneLabels, err
+	}
+}
diff --git a/pkg/common/commands/resetoffset/refmappers/testing/data.go b/pkg/common/commands/resetoffset/refmappers/testing/data.go
new file mode 100644
index 000000000..d730c599f
--- /dev/null
+++ b/pkg/common/commands/resetoffset/refmappers/testing/data.go
@@ -0,0 +1,59 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package testing
+
+import (
+	controllertesting "knative.dev/eventing-kafka/pkg/common/commands/resetoffset/controller/testing"
+	"knative.dev/eventing-kafka/pkg/common/commands/resetoffset/refmappers"
+)
+
+const (
+	ConnectionPoolKey    = "TestConnectionPoolKey"
+	DataPlaneNamespace   = "TestDataPlaneNamespace"
+	DataPlaneLabelKey1   = "TestDataPlaneLabelKey1"
+	DataPlaneLabelValue1 = "TestDataPlaneLabelValue1"
+	DataPlaneLabelKey2   = "TestDataPlaneLabelKey2"
+	DataPlaneLabelValue2 = "TestDataPlaneLabelValue2"
+)
+
+//
+// RefInfo Resources
+//
+
+// RefInfoOption allow for customizing a RefInfo
+type RefInfoOption func(refInfo *refmappers.RefInfo)
+
+// NewRefInfo creates a custom NewRefInfo
+func NewRefInfo(options ...RefInfoOption) *refmappers.RefInfo {
+
+	// Create The Default Test RefInfo
+	refInfo := &refmappers.RefInfo{
+		TopicName:          controllertesting.TopicName,
+		GroupId:            controllertesting.GroupId,
+		ConnectionPoolKey:  ConnectionPoolKey,
+		DataPlaneNamespace: DataPlaneNamespace,
+		DataPlaneLabels:    map[string]string{DataPlaneLabelKey1: DataPlaneLabelValue1, DataPlaneLabelKey2: DataPlaneLabelValue2},
+	}
+
+	// Apply The Specified Customizations
+	for _, option := range options {
+		option(refInfo)
+	}
+
+	// Return The Custom RefInfo
+	return refInfo
+}
diff --git a/pkg/common/commands/resetoffset/refmappers/testing/mocks.go b/pkg/common/commands/resetoffset/refmappers/testing/mocks.go
index 79f625d0b..efd61e479 100644
--- a/pkg/common/commands/resetoffset/refmappers/testing/mocks.go
+++ b/pkg/common/commands/resetoffset/refmappers/testing/mocks.go
@@ -50,7 +50,7 @@ type MockResetOffsetRefMapper struct {
 	mock.Mock
 }
 
-func (m *MockResetOffsetRefMapper) MapRef(resetOffset *kafkav1alpha1.ResetOffset) (string, string, error) {
+func (m *MockResetOffsetRefMapper) MapRef(resetOffset *kafkav1alpha1.ResetOffset) (*refmappers.RefInfo, error) {
 	args := m.Called(resetOffset)
-	return args.String(0), args.String(1), args.Error(2)
+	return args.Get(0).(*refmappers.RefInfo), args.Error(1)
 }
diff --git a/pkg/common/controlprotocol/commands/consumergroup_test.go b/pkg/common/controlprotocol/commands/consumergroup_test.go
new file mode 100644
index 000000000..463e8c58a
--- /dev/null
+++ b/pkg/common/controlprotocol/commands/consumergroup_test.go
@@ -0,0 +1,72 @@
+/*
+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 commands
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewConsumerGroupAsyncCommand(t *testing.T) {
+
+	// Test Data
+	commandId := int64(1234)
+	topicName := "TestTopicName"
+	groupId := "TestGroupId"
+
+	// Perform The Test
+	consumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(commandId, topicName, groupId)
+
+	// Verify The Results
+	assert.NotNil(t, consumerGroupAsyncCommand)
+	assert.Equal(t, ConsumerGroupAsyncCommandVersion, consumerGroupAsyncCommand.Version)
+	assert.Equal(t, commandId, consumerGroupAsyncCommand.CommandId)
+	assert.Equal(t, topicName, consumerGroupAsyncCommand.TopicName)
+	assert.Equal(t, groupId, consumerGroupAsyncCommand.GroupId)
+}
+
+func TestConsumerGroupAsyncCommand_MarshalUnmarshal(t *testing.T) {
+
+	// Create A ConsumerGroupAsyncCommand To Test
+	origConsumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(int64(1234), "TestTopicName", "TestGroupId")
+
+	// Perform The Test (Marshal & Unmarshal Round Trip)
+	binaryData, err := origConsumerGroupAsyncCommand.MarshalBinary()
+	assert.Nil(t, err)
+	newConsumerGroupAsyncCommand := &ConsumerGroupAsyncCommand{}
+	err = newConsumerGroupAsyncCommand.UnmarshalBinary(binaryData)
+	assert.Nil(t, err)
+
+	// Verify The Results
+	assert.Equal(t, ConsumerGroupAsyncCommandVersion, newConsumerGroupAsyncCommand.Version)
+	assert.Equal(t, origConsumerGroupAsyncCommand.CommandId, newConsumerGroupAsyncCommand.CommandId)
+	assert.Equal(t, origConsumerGroupAsyncCommand.TopicName, newConsumerGroupAsyncCommand.TopicName)
+	assert.Equal(t, origConsumerGroupAsyncCommand.GroupId, newConsumerGroupAsyncCommand.GroupId)
+}
+
+func TestConsumerGroupAsyncCommand_SerializedId(t *testing.T) {
+
+	// Create A ConsumerGroupAsyncCommand To Test
+	consumerGroupAsyncCommand := NewConsumerGroupAsyncCommand(int64(1234), "TestTopicName", "TestGroupId1")
+
+	// Perform The Test
+	serializedId := consumerGroupAsyncCommand.SerializedId()
+
+	// Verify The Results (1234 serialized)
+	assert.Equal(t, []byte{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x4, 0xd2}, serializedId)
+}
diff --git a/test/rekt/features/kafkasource/readiness.go b/test/rekt/features/kafkasource/readiness.go
index a2f2a6279..1b2f27c45 100644
--- a/test/rekt/features/kafkasource/readiness.go
+++ b/test/rekt/features/kafkasource/readiness.go
@@ -22,12 +22,18 @@ import (
 	"knative.dev/reconciler-test/pkg/feature"
 
 	"knative.dev/eventing-kafka/test/rekt/resources/kafkasource"
+	"knative.dev/eventing-kafka/test/rekt/resources/kafkatopic"
 )
 
 // KafkaSourceGoesReady returns a feature testing if a KafkaSourceGoesReady becomes ready.
 func KafkaSourceGoesReady(name string, cfg ...kafkasource.CfgFn) *feature.Feature {
 	f := feature.NewFeatureNamed("KafkaSource goes ready.")
 
+	topicName := feature.MakeRandomK8sName("topic") // A k8s name is also a valid topic name.
+	f.Setup("install kafka topic", kafkatopic.Install(topicName))
+
+	cfg = append(cfg, kafkasource.WithTopics([]string{topicName}))
+
 	sink := feature.MakeRandomK8sName("sink")
 	f.Setup("install a service", svc.Install(sink, "app", "rekt"))
 
@@ -38,7 +44,9 @@ func KafkaSourceGoesReady(name string, cfg ...kafkasource.CfgFn) *feature.Featur
 	}, ""))
 
 	f.Setup("install a kafkasource", kafkasource.Install(name, cfg...))
-	f.Requirement("be ready", kafkasource.IsReady(name))
+
+	f.Requirement("topic is ready", kafkatopic.IsReady(topicName))
+	f.Requirement("kafkasource is ready", kafkasource.IsReady(name))
 
 	return f
 }
diff --git a/test/rekt/kafka_source_test.go b/test/rekt/kafka_source_test.go
index cb41647fc..0fb732407 100644
--- a/test/rekt/kafka_source_test.go
+++ b/test/rekt/kafka_source_test.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"knative.dev/eventing-kafka/test/rekt/features/kafkasource"
+	ks "knative.dev/eventing-kafka/test/rekt/resources/kafkasource"
 	"knative.dev/eventing/pkg/utils"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/system"
@@ -33,11 +34,12 @@ import (
 )
 
 const (
-	kafkaSASLSecret = "strimzi-sasl-secret"
-	kafkaTLSSecret  = "strimzi-tls-secret"
+	kafkaSASLSecret        = "strimzi-sasl-secret"
+	kafkaTLSSecret         = "strimzi-tls-secret"
+	kafkaBootstrapUrlPlain = "my-cluster-kafka-bootstrap.kafka.svc:9092"
 )
 
-var test_mt_soure = os.Getenv("TEST_MT_SOURCE")
+var test_mt_source = os.Getenv("TEST_MT_SOURCE")
 
 func TestKafkaSource(t *testing.T) {
 	t.Parallel()
@@ -62,10 +64,13 @@ func TestKafkaSource(t *testing.T) {
 	}
 
 	testFeatures := kafkasource.DataPlaneDelivery()
-	if test_mt_soure == "1" {
+	if test_mt_source == "1" {
 		testFeatures = append(testFeatures, kafkasource.MtDataPlaneDelivery())
 	}
 
+	testFeatures = append(testFeatures,
+		kafkasource.KafkaSourceGoesReady("readysource", ks.WithBootstrapServers([]string{kafkaBootstrapUrlPlain})))
+
 	for _, f := range testFeatures {
 		env.Test(ctx, t, f)
 	}