Repository: knative-sandbox/eventing-kafka-broker
Pull request #1927 (2022-02-17 20:02:21)

[data-plane]: Bump micrometer-bom from 1.8.2 to 1.8.3 in /data-plane
Bumps [micrometer-bom](https://github.com/micrometer-metrics/micrometer) from 1.8.2 to 1.8.3.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/micrometer-metrics/micrometer/releases">micrometer-bom's releases</a>.</em></p>
<blockquote>
<h2>1.8.3</h2>
<h2>:lady_beetle: Bug Fixes</h2>
<ul>
<li>Dynatrace v2 ingestion reporting &quot;inconsistent gauge fields&quot; <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3007">#3007</a></li>
<li>Failed to record timer metrics in native image with percentiles configured <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3002">#3002</a></li>
<li>Correct PostgreSQL metrics for dead tuples, with integration tests and refactoring <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/pull/2474">#2474</a></li>
</ul>
<h2>:hammer: Dependency Upgrades</h2>
<ul>
<li>Upgrade Mockito to 4.3.1 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3029">#3029</a></li>
<li>Upgrade AWS SDK v1 to 1.12.158 and v2 to 2.17.129 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3028">#3028</a></li>
<li>Upgrade Netty to 4.1.74.Final <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3026">#3026</a></li>
<li>Upgrade Javaformat to 0.0.31 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3025">#3025</a></li>
<li>Upgrade JUnit to 5.8.2 and Testcontainers to 1.16.3 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3024">#3024</a></li>
<li>Upgrade Dropwizard Metrics to 4.1.30 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3023">#3023</a></li>
<li>Upgrade Slf4J to 1.7.36 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3022">#3022</a></li>
<li>Upgrade to Gradle Wrapper 7.4 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/pull/3001">#3001</a></li>
<li>Upgrade build jdk to 17.0.2, 8.0.322 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/2988">#2988</a></li>
</ul>
<h2>:memo: Tasks</h2>
<ul>
<li>Pin spectator-api to 1.0.+ for Micrometer 1.8.x <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3027">#3027</a></li>
<li>Store the Meter.Id instead of the Meter in KafkaMetrics <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/2991">#2991</a></li>
<li>Bump up Gradle plugin dependencies <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/pull/2979">#2979</a></li>
<li>Use branch name in dependencies cache key for CircleCI <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/2975">#2975</a></li>
<li>Upgrade to Gradle Shadow plugin 7.1.2 <a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/2973">#2973</a></li>
</ul>
<h2>:heart: Contributors</h2>
<p>We'd like to thank all the contributors who worked on this release!</p>
<ul>
<li><a href="https://github.com/izeye"><code>@​izeye</code></a></li>
<li><a href="https://github.com/jprinet"><code>@​jprinet</code></a></li>
<li><a href="https://github.com/easimon"><code>@​easimon</code></a></li>
<li><a href="https://github.com/pirgeo"><code>@​pirgeo</code></a></li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/eb7ba780b756113d13a0fd345d97b9750a84fb35"><code>eb7ba78</code></a> Drop summary and timer lines with a count of 0 (<a href="https://github-redirect.dependabot.com/micrometer-metrics/micrometer/issues/3030">#3030</a>)</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/fba079f540fa282d576d0ce688cd89da42df3455"><code>fba079f</code></a> Upgrade Mockito to 4.3.1</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/609982acb018c74c85be68b07f2491e5488e974d"><code>609982a</code></a> Upgrade AWS SDK v1 to 1.12.158 and v2 to 2.17.129</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/b8072e8dd311eac9df9412791a662a2b83aa2ae3"><code>b8072e8</code></a> Pin spectator-api to 1.0.+ for Micrometer 1.8.x</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/ade7cccacb841200bead0f7ba1b7c37d6f559b9e"><code>ade7ccc</code></a> Upgrade Netty to 4.1.74.Final</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/36c11318cb46273b33eddb7384191fa87c4ad7d7"><code>36c1131</code></a> Merge branch '1.7.x' into 1.8.x</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/ea8bfcbbd92889ae22752f7c1d66f4dd63e54937"><code>ea8bfcb</code></a> Upgrade Javaformat to 0.0.31</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/64d3093f6cb671ba05099a7601caeb8f331812ad"><code>64d3093</code></a> Upgrade JUnit to 5.8.2 and Testcontainers to 1.16.3</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/a226d24bdf57be1f3a05640cf2c74ac077587b84"><code>a226d24</code></a> Upgrade Dropwizard Metrics to 4.1.30</li>
<li><a href="https://github.com/micrometer-metrics/micrometer/commit/652185bf622f7a2e1428c6242706ac2af3ecfa68"><code>652185b</code></a> Upgrade Slf4J to 1.7.36</li>
<li>Additional commits viewable in <a href="https://github.com/micrometer-metrics/micrometer/compare/v1.8.2...v1.8.3">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=io.micrometer:micrometer-bom&package-manager=maven&previous-version=1.8.2&new-version=1.8.3)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1927#" title="Author self-approved">dependabot[bot]</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **julz** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @julz` in a comment when ready.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["julz"]} -->
knative-prow-robot: Hi @dependabot[bot]. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1927#" title="Author self-approved">dependabot[bot]</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** after the PR has been reviewed.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["pierdipi"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1927?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1927](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1927?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (b6a620f) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/4b9e5e2bdc861af699f0766ffd5f92bfaecdaa40?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4b9e5e2) will **increase** coverage by `0.04%`.
> The diff coverage is `n/a`.

> :exclamation: Current head b6a620f differs from pull request most recent head c03a99e. Consider uploading reports for the commit c03a99e to get more accurate results

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1927/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&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-broker/pull/1927?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    #1927      +/-   ##
============================================
+ Coverage     66.45%   66.50%   +0.04%     
- Complexity      629      632       +3     
============================================
  Files           138      138              
  Lines          6129     6129              
  Branches        179      179              
============================================
+ Hits           4073     4076       +3     
+ Misses         1725     1721       -4     
- Partials        331      332       +1     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.22% <ø> (+0.14%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1927?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...tcher/impl/consumer/UnorderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1927/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9Vbm9yZGVyZWRDb25zdW1lclZlcnRpY2xlLmphdmE=) | `91.30% <0.00%> (+4.34%)` | :arrow_up: |
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1927/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `93.75% <0.00%> (+6.25%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1927?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-broker/pull/1927?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [4b9e5e2...c03a99e](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1927?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).


Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
index c77a7e915..8887c0983 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
@@ -105,6 +105,7 @@ rules:
       - watch
       - patch
       - update
+      - delete
   - apiGroups:
       - "internal.kafka.eventing.knative.dev"
     resources:
@@ -112,7 +113,7 @@ rules:
       - "consumergroups/finalizers"
     verbs:
       - update
-
+      - delete
   # Eventing resources and statuses we care about
   - apiGroups:
       - "eventing.knative.dev"
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_lifecycle.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_lifecycle.go
index 5b804afc7..aa4e5cddf 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_lifecycle.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_lifecycle.go
@@ -25,7 +25,9 @@ import (
 const (
 	ConditionConsumerGroupConsumers          apis.ConditionType = "Consumers"
 	ConditionConsumerGroupConsumersScheduled apis.ConditionType = "ConsumersScheduled"
-	ConsumerSelectorAnnotation                                  = "kafka.eventing.knative.dev/metadata.uid"
+	// Labels
+	KafkaChannelNameLabel      = "kafkachannel-name"
+	ConsumerSelectorAnnotation = "kafka.eventing.knative.dev/metadata.uid"
 )
 
 var (
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
index 8e0210fe3..c95b782b5 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
@@ -28,6 +28,12 @@ var (
 
 func (c *ConsumerGroup) Validate(ctx context.Context) *apis.FieldError {
 	ctx = apis.WithinParent(ctx, c.ObjectMeta)
+	if apis.IsInUpdate(ctx) {
+		err := c.CheckImmutableFields(ctx, apis.GetBaseline(ctx).(*ConsumerGroup).Labels)
+		if err != nil {
+			return err
+		}
+	}
 	return c.Spec.Validate(ctx).ViaField("spec")
 }
 
@@ -48,3 +54,13 @@ func (cts *ConsumerTemplateSpec) Validate(ctx context.Context) *apis.FieldError
 	}
 	return cts.Spec.Validate(specCtx).ViaField("spec")
 }
+
+func (c *ConsumerGroup) CheckImmutableFields(ctx context.Context, original map[string]string) *apis.FieldError {
+	if orig, ok := original[KafkaChannelNameLabel]; ok {
+		if new, ok := c.Labels[KafkaChannelNameLabel]; !ok || orig != new {
+			return ErrImmutableField("Consumer Group Label",
+				"Removing or modifying the consumer group label is unsupported")
+		}
+	}
+	return nil
+}
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2.go b/control-plane/pkg/reconciler/channel/v2/channelv2.go
new file mode 100644
index 000000000..93923915d
--- /dev/null
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2.go
@@ -0,0 +1,634 @@
+/*
+ * 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 v2
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/Shopify/sarama"
+	"go.uber.org/multierr"
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	corelisters "k8s.io/client-go/listers/core/v1"
+
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/reconciler"
+	"knative.dev/pkg/system"
+
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	internalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned"
+	internalslst "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
+	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
+)
+
+const (
+	// TopicPrefix is the Kafka Channel topic prefix - (topic name: knative-messaging-kafka.<channel-namespace>.<channel-name>).
+	TopicPrefix          = "knative-messaging-kafka"
+	DefaultDeliveryOrder = internals.Ordered
+
+	KafkaChannelConditionConsumerGroup apis.ConditionType = "ConsumerGroup" //condition is registered by controller
+)
+
+var (
+	conditionSet = apis.NewLivingConditionSet(
+		KafkaChannelConditionConsumerGroup,
+		base.ConditionTopicReady,
+		base.ConditionConfigMapUpdated,
+		base.ConditionConfigParsed,
+		base.ConditionProbeSucceeded,
+	)
+)
+
+type Reconciler struct {
+	*base.Reconciler
+	*config.Env
+
+	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// mock the function used during the reconciliation loop.
+	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
+
+	// NewKafkaClient creates new sarama Client. It's convenient to add this as Reconciler field so that we can
+	// mock the function used during the reconciliation loop.
+	NewKafkaClient kafka.NewClientFunc
+
+	ConfigMapLister corelisters.ConfigMapLister
+
+	Prober prober.Prober
+
+	IngressHost string
+
+	ConsumerGroupLister internalslst.ConsumerGroupLister
+	InternalsClient     internalsclient.Interface
+}
+
+func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta1.KafkaChannel) reconciler.Event {
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, channel)
+
+	statusConditionManager := base.StatusConditionManager{
+		Object:     channel,
+		SetAddress: channel.Status.SetAddress,
+		Env:        r.Env,
+		Recorder:   controller.GetEventRecorder(ctx),
+	}
+
+	// Get the channel configmap
+	channelConfigMap, err := r.channelConfigMap()
+	if err != nil {
+		return statusConditionManager.FailedToResolveConfig(err)
+	}
+	logger.Debug("configmap read", zap.Any("configmap", channelConfigMap))
+
+	if err := r.TrackConfigMap(channelConfigMap, channel); err != nil {
+		return fmt.Errorf("failed to track broker config: %w", err)
+	}
+
+	// get topic config
+	topicConfig, err := r.topicConfig(logger, channelConfigMap, channel)
+	if err != nil {
+		return statusConditionManager.FailedToResolveConfig(err)
+	}
+	logger.Debug("topic config resolved", zap.Any("config", topicConfig))
+	statusConditionManager.ConfigResolved()
+
+	// get the secret to access Kafka
+	secret, err := security.Secret(ctx, &security.MTConfigMapSecretLocator{ConfigMap: channelConfigMap}, r.SecretProviderFunc())
+	if err != nil {
+		return fmt.Errorf("failed to get secret: %w", err)
+	}
+	if secret != nil {
+		logger.Debug("Secret reference",
+			zap.String("apiVersion", secret.APIVersion),
+			zap.String("name", secret.Name),
+			zap.String("namespace", secret.Namespace),
+			zap.String("kind", secret.Kind),
+		)
+	}
+
+	// get security option for Sarama with secret info in it
+	saramaSecurityOption := security.NewSaramaSecurityOptionFromSecret(secret)
+
+	if err := r.TrackSecret(secret, channel); err != nil {
+		return fmt.Errorf("failed to track secret: %w", err)
+	}
+
+	topicName := kafka.ChannelTopic(TopicPrefix, channel)
+
+	kafkaClusterAdminSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin sarama config: %w", err))
+	}
+
+	// Manually commit the offsets in KafkaChannel controller.
+	// That's because we want to make sure we initialize the offsets within the controller
+	// before dispatcher actually starts consuming messages.
+	kafkaClientSaramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption, kafka.DisableOffsetAutoCommitConfigOption)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin sarama config: %w", err))
+	}
+
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, kafkaClusterAdminSaramaConfig)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
+	}
+	defer kafkaClusterAdminClient.Close()
+
+	kafkaClient, err := r.NewKafkaClient(topicConfig.BootstrapServers, kafkaClientSaramaConfig)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting sarama config: %w", err))
+	}
+	defer kafkaClient.Close()
+
+	// create the topic
+	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
+	if err != nil {
+		return statusConditionManager.FailedToCreateTopic(topic, err)
+	}
+	logger.Debug("Topic created", zap.Any("topic", topic))
+	statusConditionManager.TopicReady(topic)
+
+	// Get data plane config map.
+	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
+	if err != nil {
+		return statusConditionManager.FailedToResolveConfig(err)
+	}
+	logger.Debug("Got contract config map")
+
+	// Get data plane config data.
+	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
+	if err != nil || ct == nil {
+		return statusConditionManager.FailedToGetDataFromConfigMap(err)
+	}
+	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
+
+	// Get resource configuration
+	channelConfig := r.getChannelContractResource(topic, channel, secret, topicConfig)
+
+	subscriptionError := r.reconcileSubscribers(ctx, channel, topicName, topicConfig.BootstrapServers)
+	if subscriptionError != nil {
+		channel.GetConditionSet().Manage(&channel.Status).MarkFalse(KafkaChannelConditionConsumerGroup, "failed to reconcile consumer group", subscriptionError.Error())
+		return subscriptionError
+	}
+	channel.GetConditionSet().Manage(&channel.Status).MarkTrue(KafkaChannelConditionConsumerGroup)
+
+	// Update contract data with the new contract configuration (add/update channel resource)
+	channelIndex := coreconfig.FindResource(ct, channel.UID)
+	changed := coreconfig.AddOrUpdateResourceConfig(ct, channelConfig, channelIndex, logger)
+	logger.Debug("Change detector", zap.Int("changed", changed))
+
+	if changed == coreconfig.ResourceChanged {
+		logger.Debug("Contract changed", zap.Int("changed", changed))
+
+		ct.IncrementGeneration()
+
+		// Update the configuration map with the new contract data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			logger.Error("failed to update data plane config map", zap.Error(
+				statusConditionManager.FailedToUpdateConfigMap(err),
+			))
+			return err
+		}
+		logger.Debug("Contract config map updated")
+	}
+	statusConditionManager.ConfigMapUpdated()
+
+	// We update receiver and dispatcher pods annotation regardless of our contract changed or not due to the fact
+	// that in a previous reconciliation we might have failed to update one of our data plane pod annotation, so we want
+	// to anyway update remaining annotations with the contract generation that was saved in the CM.
+
+	// We reject events to a non-existing Channel, which means that we cannot consider a Channel Ready if all
+	// receivers haven't got the Channel, so update failures to receiver pods is a hard failure.
+	// On the other side, dispatcher pods care about Subscriptions, and the Channel object is used as a configuration
+	// prototype for all associated Subscriptions, so we consider that it's fine on the dispatcher side to receive eventually
+	// the update even if here eventually means seconds or minutes after the actual update.
+
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		logger.Error("Failed to update receiver pod annotation", zap.Error(
+			statusConditionManager.FailedToUpdateReceiverPodsAnnotation(err),
+		))
+		return err
+	}
+	logger.Debug("Updated receiver pod annotation")
+
+	// Update volume generation annotation of dispatcher pods
+	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
+		// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Channel
+		// ready. So, log out the error and move on to the next step.
+		logger.Warn(
+			"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
+			zap.Error(err),
+		)
+
+		statusConditionManager.FailedToUpdateDispatcherPodsAnnotation(err)
+	} else {
+		logger.Debug("Updated dispatcher pod annotation")
+	}
+
+	address := receiver.Address(r.IngressHost, channel)
+	proberAddressable := prober.Addressable{
+		Address: address,
+		ResourceKey: types.NamespacedName{
+			Namespace: channel.GetNamespace(),
+			Name:      channel.GetName(),
+		},
+	}
+
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusReady); status != prober.StatusReady {
+		statusConditionManager.ProbesStatusNotReady(status)
+		return nil // Object will get re-queued once probe status changes.
+	}
+	statusConditionManager.Addressable(address)
+
+	return nil
+}
+
+func (r *Reconciler) FinalizeKind(ctx context.Context, channel *messagingv1beta1.KafkaChannel) reconciler.Event {
+	logger := kafkalogging.CreateFinalizeMethodLogger(ctx, channel)
+
+	// Get contract config map.
+	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get contract config map %s: %w", r.DataPlaneConfigMapAsString(), err)
+	}
+	logger.Debug("Got contract config map")
+
+	// Get contract data.
+	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
+	if err != nil && ct == nil {
+		return fmt.Errorf("failed to get contract: %w", err)
+	}
+	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
+
+	channelIndex := coreconfig.FindResource(ct, channel.UID)
+	if channelIndex != coreconfig.NoResource {
+		coreconfig.DeleteResource(ct, channelIndex)
+
+		logger.Debug("Channel deleted", zap.Int("index", channelIndex))
+
+		// Resource changed, increment contract generation.
+		ct.IncrementGeneration()
+
+		// Update the configuration map with the new contract data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			return err
+		}
+		logger.Debug("Contract config map updated")
+	}
+
+	channel.Status.Address = nil
+
+	// We update receiver and dispatcher pods annotation regardless of our contract changed or not due to the fact
+	// that in a previous reconciliation we might have failed to update one of our data plane pod annotation, so we want
+	// to update anyway remaining annotations with the contract generation that was saved in the CM.
+	// Note: if there aren't changes to be done at the pod annotation level, we just skip the update.
+
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
+	}
+	// Update volume generation annotation of dispatcher pods
+	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
+	}
+
+	//  Rationale: after deleting a topic closing a producer ends up blocking and requesting metadata for max.block.ms
+	//  because topic metadata aren't available anymore.
+	// 	See (under discussions KIPs, unlikely to be accepted as they are):
+	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
+	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
+	address := receiver.Address(r.IngressHost, channel)
+	proberAddressable := prober.Addressable{
+		Address: address,
+		ResourceKey: types.NamespacedName{
+			Namespace: channel.GetNamespace(),
+			Name:      channel.GetName(),
+		},
+	}
+	if status := r.Prober.Probe(ctx, proberAddressable, prober.StatusNotReady); status != prober.StatusNotReady {
+		// Return a requeueKeyError that doesn't generate an event and it re-queues the object
+		// for a new reconciliation.
+		return controller.NewRequeueAfter(5 * time.Second)
+	}
+
+	// get the channel configmap
+	channelConfigMap, err := r.channelConfigMap()
+	if err != nil {
+		return err
+	}
+	logger.Debug("configmap read", zap.Any("configmap", channelConfigMap))
+
+	// get topic config
+	topicConfig, err := r.topicConfig(logger, channelConfigMap, channel)
+	if err != nil {
+		return fmt.Errorf("failed to resolve channel config: %w", err)
+	}
+
+	logger.Debug("topic config resolved", zap.Any("config", topicConfig))
+
+	// get the secret to access Kafka
+	secret, err := security.Secret(ctx, &security.MTConfigMapSecretLocator{ConfigMap: channelConfigMap}, r.SecretProviderFunc())
+	if err != nil {
+		return fmt.Errorf("failed to get secret: %w", err)
+	}
+	if secret != nil {
+		logger.Debug("Secret reference",
+			zap.String("apiVersion", secret.APIVersion),
+			zap.String("name", secret.Name),
+			zap.String("namespace", secret.Namespace),
+			zap.String("kind", secret.Kind),
+		)
+	}
+
+	// get security option for Sarama with secret info in it
+	saramaSecurityOption := security.NewSaramaSecurityOptionFromSecret(secret)
+
+	saramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
+	if err != nil {
+		// even in error case, we return `normal`, since we are fine with leaving the
+		// topic undeleted e.g. when we lose connection
+		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
+	}
+
+	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
+	if err != nil {
+		// even in error case, we return `normal`, since we are fine with leaving the
+		// topic undeleted e.g. when we lose connection
+		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
+	}
+	defer kafkaClusterAdminClient.Close()
+
+	topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, kafka.ChannelTopic(TopicPrefix, channel))
+	if err != nil {
+		return err
+	}
+
+	logger.Debug("Topic deleted", zap.String("topic", topic))
+
+	return nil
+}
+
+func (r *Reconciler) reconcileSubscribers(ctx context.Context, channel *messagingv1beta1.KafkaChannel, topicName string, bootstrapServers []string) error {
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, channel)
+
+	channel.Status.Subscribers = make([]v1.SubscriberStatus, 0)
+	var globalErr error
+	currentCgs := make(map[string]*internalscg.ConsumerGroup)
+	for i := range channel.Spec.Subscribers {
+		s := &channel.Spec.Subscribers[i]
+		logger = logger.With(zap.Any("subscription", s))
+		cg, err := r.reconcileConsumerGroup(ctx, channel, s, topicName, bootstrapServers)
+		if err != nil {
+			logger.Error("error reconciling subscription. marking subscription as not ready", zap.Error(err))
+			channel.Status.Subscribers = append(channel.Status.Subscribers, v1.SubscriberStatus{
+				UID:                s.UID,
+				ObservedGeneration: s.Generation,
+				Ready:              corev1.ConditionFalse,
+				Message:            fmt.Sprint("Subscription not ready. ", err),
+			})
+			globalErr = multierr.Append(globalErr, err)
+		} else {
+			currentCgs[cg.Name] = cg // Adding reconciled consumer group to map
+			if cg.IsReady() {
+				logger.Debug("marking subscription as ready")
+				channel.Status.Subscribers = append(channel.Status.Subscribers, v1.SubscriberStatus{
+					UID:                s.UID,
+					ObservedGeneration: s.Generation,
+					Ready:              corev1.ConditionTrue,
+				})
+			} else {
+				topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
+				if topLevelCondition == nil {
+					msg := fmt.Sprint("Subscription not ready. ", "Failed to reconcile consumer group. ", "Consumer group not ready")
+					channel.Status.Subscribers = append(channel.Status.Subscribers, v1.SubscriberStatus{
+						UID:                s.UID,
+						ObservedGeneration: s.Generation,
+						Ready:              corev1.ConditionUnknown,
+						Message:            msg,
+					})
+					globalErr = multierr.Append(globalErr, errors.New(msg))
+				} else {
+					msg := fmt.Sprint("Subscription not ready. ", topLevelCondition.Reason, topLevelCondition.Message)
+					channel.Status.Subscribers = append(channel.Status.Subscribers, v1.SubscriberStatus{
+						UID:                s.UID,
+						ObservedGeneration: s.Generation,
+						Ready:              corev1.ConditionFalse,
+						Message:            msg,
+					})
+					globalErr = multierr.Append(globalErr, errors.New(msg))
+				}
+			}
+		}
+	}
+
+	// Get all consumer groups associated with this Channel
+	selector := labels.SelectorFromSet(map[string]string{internalscg.KafkaChannelNameLabel: channel.Name})
+	channelCgs, err := r.ConsumerGroupLister.ConsumerGroups(channel.GetNamespace()).List(selector)
+	if err != nil {
+		globalErr = multierr.Append(globalErr, err)
+	}
+	for _, cg := range channelCgs {
+		_, found := currentCgs[cg.Name]
+		if !found { // ConsumerGroup needs to be deleted since it isn't associated with an existing subscriber (subscriber may have been deleted)
+			err := r.finalizeConsumerGroup(ctx, cg)
+			if err != nil {
+				globalErr = multierr.Append(globalErr, err)
+			}
+		}
+	}
+
+	return globalErr
+}
+
+func (r Reconciler) reconcileConsumerGroup(ctx context.Context, channel *messagingv1beta1.KafkaChannel, s *v1.SubscriberSpec, topicName string, bootstrapServers []string) (*internalscg.ConsumerGroup, error) {
+	newcg := &internalscg.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      string(s.UID),
+			Namespace: channel.Namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				*kmeta.NewControllerRef(channel),
+			},
+			Labels: map[string]string{
+				internalscg.KafkaChannelNameLabel: channel.Name, // Identifies the new ConsumerGroup as associated with this channel (same namespace)
+			},
+		},
+		Spec: internalscg.ConsumerGroupSpec{
+			Template: internalscg.ConsumerTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						internalscg.ConsumerSelectorAnnotation: string(s.UID),
+					},
+				},
+				Spec: internalscg.ConsumerSpec{
+					Topics: []string{topicName},
+					Configs: internalscg.ConsumerConfigs{Configs: map[string]string{
+						"group.id":          consumerGroup(channel, s),
+						"bootstrap.servers": strings.Join(bootstrapServers, ","),
+					}},
+					Delivery: &internalscg.DeliverySpec{
+						DeliverySpec: channel.Spec.Delivery,
+						Ordering:     DefaultDeliveryOrder,
+					},
+				},
+			},
+		},
+	}
+
+	cg, err := r.ConsumerGroupLister.ConsumerGroups(channel.GetNamespace()).Get(string(s.UID)) // Get by consumer group id
+	if err != nil && !apierrors.IsNotFound(err) {
+		return nil, err
+	}
+
+	if apierrors.IsNotFound(err) {
+		cg, err = r.InternalsClient.InternalV1alpha1().ConsumerGroups(newcg.GetNamespace()).Create(ctx, newcg, metav1.CreateOptions{})
+		if err != nil && !apierrors.IsAlreadyExists(err) {
+			return nil, fmt.Errorf("failed to create consumer group %s/%s: %w", newcg.GetNamespace(), newcg.GetName(), err)
+		}
+		if apierrors.IsAlreadyExists(err) {
+			return newcg, nil
+		}
+		return cg, nil
+	}
+
+	if equality.Semantic.DeepDerivative(newcg.Spec, cg.Spec) {
+		return cg, nil
+	}
+
+	newCg := &internalscg.ConsumerGroup{
+		TypeMeta:   cg.TypeMeta,
+		ObjectMeta: cg.ObjectMeta,
+		Spec:       newcg.Spec,
+		Status:     cg.Status,
+	}
+	if cg, err = r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
+		return nil, fmt.Errorf("failed to update consumer group %s/%s: %w", newCg.GetNamespace(), newCg.GetName(), err)
+	}
+
+	return cg, nil
+}
+
+func (r *Reconciler) getChannelContractResource(topic string, channel *messagingv1beta1.KafkaChannel, secret *corev1.Secret, config *kafka.TopicConfig) *contract.Resource {
+	resource := &contract.Resource{
+		Uid:    string(channel.UID),
+		Topics: []string{topic},
+		Ingress: &contract.Ingress{
+			IngressType: &contract.Ingress_Path{
+				Path: receiver.PathFromObject(channel),
+			},
+		},
+		BootstrapServers: config.GetBootstrapServers(),
+		Reference: &contract.Reference{
+			Uuid:      string(channel.GetUID()),
+			Namespace: channel.GetNamespace(),
+			Name:      channel.GetName(),
+		},
+	}
+
+	if secret != nil {
+		resource.Auth = &contract.Resource_AuthSecret{
+			AuthSecret: &contract.Reference{
+				Uuid:      string(secret.UID),
+				Namespace: secret.Namespace,
+				Name:      secret.Name,
+				Version:   secret.ResourceVersion,
+			},
+		}
+	}
+
+	return resource
+}
+
+// consumerGroup returns a consumerGroup name for the given channel and subscription
+func consumerGroup(channel *messagingv1beta1.KafkaChannel, s *v1.SubscriberSpec) string {
+	return fmt.Sprintf("kafka.%s.%s.%s", channel.Namespace, channel.Name, string(s.UID))
+}
+
+func (r *Reconciler) channelConfigMap() (*corev1.ConfigMap, error) {
+	// TODO: do we want to support namespaced channels? they're not supported at the moment.
+
+	namespace := system.Namespace()
+	cm, err := r.ConfigMapLister.ConfigMaps(namespace).Get(r.Env.GeneralConfigMapName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, r.Env.GeneralConfigMapName, err)
+	}
+
+	return cm, nil
+}
+
+func (r *Reconciler) topicConfig(logger *zap.Logger, cm *corev1.ConfigMap, channel *messagingv1beta1.KafkaChannel) (*kafka.TopicConfig, error) {
+	bootstrapServers, err := kafka.BootstrapServersFromConfigMap(logger, cm)
+	if err != nil {
+		return nil, fmt.Errorf("unable to get bootstrapServers from configmap: %w - ConfigMap data: %v", err, cm.Data)
+	}
+
+	// Parse & Format the RetentionDuration into Sarama retention.ms string
+	retentionDuration, err := channel.Spec.ParseRetentionDuration()
+	if err != nil {
+		// Should never happen with webhook defaulting and validation in place.
+		logger.Error("Error parsing RetentionDuration, using default instead", zap.String("RetentionDuration", channel.Spec.RetentionDuration), zap.Error(err))
+		retentionDuration = constants.DefaultRetentionDuration
+	}
+	retentionMillisString := strconv.FormatInt(retentionDuration.Milliseconds(), 10)
+
+	return &kafka.TopicConfig{
+		TopicDetail: sarama.TopicDetail{
+			NumPartitions:     channel.Spec.NumPartitions,
+			ReplicationFactor: channel.Spec.ReplicationFactor,
+			ConfigEntries: map[string]*string{
+				constants.KafkaTopicConfigRetentionMs: &retentionMillisString,
+			},
+		},
+		BootstrapServers: bootstrapServers,
+	}, nil
+}
+
+func (r Reconciler) finalizeConsumerGroup(ctx context.Context, cg *internalscg.ConsumerGroup) error {
+	dOpts := metav1.DeleteOptions{
+		Preconditions: &metav1.Preconditions{UID: &cg.UID},
+	}
+	err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Delete(ctx, cg.GetName(), dOpts)
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("failed to remove consumer group %s/%s: %w", cg.GetNamespace(), cg.GetName(), err)
+	}
+	return nil
+}
diff --git a/control-plane/pkg/reconciler/channel/v2/controllerv2.go b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
new file mode 100644
index 000000000..3da615523
--- /dev/null
+++ b/control-plane/pkg/reconciler/channel/v2/controllerv2.go
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v2
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/Shopify/sarama"
+	"k8s.io/client-go/tools/cache"
+
+	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
+	kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/network"
+
+	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
+	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
+
+	"knative.dev/pkg/controller"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
+)
+
+func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
+
+	configmapInformer := configmapinformer.Get(ctx)
+	channelInformer := kafkachannelinformer.Get(ctx)
+	consumerGroupInformer := consumergroupinformer.Get(ctx)
+
+	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(conditionSet)
+
+	reconciler := &Reconciler{
+		Reconciler: &base.Reconciler{
+			KubeClient:                  kubeclient.Get(ctx),
+			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
+			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
+			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+		},
+		NewKafkaClient:             sarama.NewClient,
+		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		Env:                        configs,
+		ConfigMapLister:            configmapInformer.Lister(),
+		ConsumerGroupLister:        consumerGroupInformer.Lister(),
+		InternalsClient:            consumergroupclient.Get(ctx),
+	}
+
+	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
+
+	channelInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		Handler: controller.HandleAll(impl.Enqueue),
+	})
+
+	// ConsumerGroup changes and enqueue associated channel
+	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: consumergroup.Filter("kafkachannel"),
+		Handler:    controller.HandleAll(consumergroup.Enqueue("kafkachannel", impl.EnqueueKey)),
+	})
+
+	return impl
+}
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index e1d50703c..8bdd3a2c5 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -67,8 +67,8 @@ Lists of 158 third-party dependencies.
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.12.1 - http://fabric8.io/kubernetes-server-mock/)
      (Apache License, Version 2.0) Fabric8 :: Mock Web Server (io.fabric8:mockwebserver:0.2.2 - http://fabric8.io/)
      (The Apache Software License, Version 2.0) zjsonpatch (io.fabric8:zjsonpatch:0.3.0 - https://github.com/fabric8io/zjsonpatch/)
-     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.8.2 - https://github.com/micrometer-metrics/micrometer)
-     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.8.2 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.8.3 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.8.3 - https://github.com/micrometer-metrics/micrometer)
      (Apache License, Version 2.0) Netty/Buffer (io.netty:netty-buffer:4.1.74.Final - https://netty.io/netty-buffer/)
      (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.74.Final - https://netty.io/netty-codec/)
      (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.74.Final - https://netty.io/netty-codec-dns/)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index df1bc5fa7..bdc697ad2 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -43,7 +43,7 @@
     <!-- dependencies version -->
     <vertx.version>4.2.5</vertx.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
-    <micrometer.version>1.8.2</micrometer.version>
+    <micrometer.version>1.8.3</micrometer.version>
     <opentelemetry.version>1.11.0</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
     <protobuf.version>3.19.4</protobuf.version>


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
index 504c7e9e8..42a62f714 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"testing"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
@@ -417,6 +418,223 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: "valid with channel label",
+			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						KafkaChannelNameLabel: "channelName", // identifies the new ConsumerGroup as associated with this channel
+					},
+				},
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			}),
+			given: &ConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						KafkaChannelNameLabel: "channelName", // identifies the new ConsumerGroup as associated with this channel
+					},
+				},
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "invalid without channel label",
+			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						KafkaChannelNameLabel: "channelName", // identifies the new ConsumerGroup as associated with this channel
+					},
+				},
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			}),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "invalid with different channel label value",
+			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						KafkaChannelNameLabel: "channelName", // identifies the new ConsumerGroup as associated with this channel
+					},
+				},
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			}),
+			given: &ConsumerGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						KafkaChannelNameLabel: "channelBadName", // identifies the new ConsumerGroup as associated with this channel
+					},
+				},
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
diff --git a/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go b/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go
new file mode 100644
index 000000000..469317c78
--- /dev/null
+++ b/control-plane/pkg/reconciler/channel/v2/controllerv2_test.go
@@ -0,0 +1,70 @@
+/*
+ * 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 v2
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+)
+
+func TestNewController(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	configs := &config.Env{
+		SystemNamespace:      "cm",
+		GeneralConfigMapName: "cm",
+		IngressPodPort:       "8080",
+	}
+
+	ctx, _ = fakekubeclient.With(
+		ctx,
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      configs.GeneralConfigMapName,
+				Namespace: configs.SystemNamespace,
+			},
+		},
+	)
+	dynamicScheme := runtime.NewScheme()
+	_ = fakekubeclientset.AddToScheme(dynamicScheme)
+
+	dynamicclient.With(ctx, dynamicScheme)
+
+	controller := NewController(
+		ctx,
+		configs,
+	)
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}