Repository: knative-sandbox/eventing-kafka-broker
Pull request #1928 (2022-02-17 21:43:55)

Evictor fix for patching consumergroup when removing evicted pod placement
Signed-off-by: aavarghese <avarghese@us.ibm.com>

Part of #1537

Fixes issue documented in https://github.com/knative-sandbox/eventing-kafka-broker/issues/1537#issuecomment-1041891419
<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Modify consumer group status placement to remove evicted pod via Patch rather than Update


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/1928#" title="Author self-approved">aavarghese</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"]} -->
aavarghese: /cc @pierDipi 
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/consumergroup/evictor.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1928/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1494427419396804608/artifacts/line-cov.html#file0) | 87.5% | 89.7% | 2.2


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/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 2fe24b639..516eee779 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -106,20 +106,22 @@ func isAtLeastOneRunning(pods []*corev1.Pod) bool {
 	return false
 }
 
-func (r *Reconciler) GetOrCreateDataPlaneConfigMap(ctx context.Context) (*corev1.ConfigMap, error) {
+type ConfigMapOption func(cm *corev1.ConfigMap)
+
+func (r *Reconciler) GetOrCreateDataPlaneConfigMap(ctx context.Context, options ...ConfigMapOption) (*corev1.ConfigMap, error) {
 
 	cm, err := r.KubeClient.CoreV1().
 		ConfigMaps(r.DataPlaneConfigMapNamespace).
 		Get(ctx, r.DataPlaneConfigMapName, metav1.GetOptions{})
 
 	if apierrors.IsNotFound(err) {
-		cm, err = r.createDataPlaneConfigMap(ctx)
+		cm, err = r.createDataPlaneConfigMap(ctx, options)
 	}
 
 	return cm, err
 }
 
-func (r *Reconciler) createDataPlaneConfigMap(ctx context.Context) (*corev1.ConfigMap, error) {
+func (r *Reconciler) createDataPlaneConfigMap(ctx context.Context, options []ConfigMapOption) (*corev1.ConfigMap, error) {
 	cm := &corev1.ConfigMap{
 		TypeMeta: metav1.TypeMeta{},
 		ObjectMeta: metav1.ObjectMeta{
@@ -130,6 +132,11 @@ func (r *Reconciler) createDataPlaneConfigMap(ctx context.Context) (*corev1.Conf
 			ConfigMapDataKey: []byte(""),
 		},
 	}
+
+	for _, opt := range options {
+		opt(cm)
+	}
+
 	return r.KubeClient.CoreV1().ConfigMaps(r.DataPlaneConfigMapNamespace).Create(ctx, cm, metav1.CreateOptions{})
 }
 
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/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index a06ab9b23..e0143dcb5 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -23,8 +23,10 @@ import (
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/utils/pointer"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/reconciler"
@@ -327,7 +329,7 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 
 	b := r.commonReconciler(p, cmName)
 
-	cm, err := b.GetOrCreateDataPlaneConfigMap(ctx)
+	cm, err := b.GetOrCreateDataPlaneConfigMap(ctx, podOwnerReference(p))
 	if err != nil {
 		return false, fmt.Errorf("failed to get or create data plane ConfigMap %s/%s: %w", p.GetNamespace(), cmName, err)
 	}
@@ -387,3 +389,16 @@ func cmNameFromPod(p *corev1.Pod, c *kafkainternals.Consumer) (string, error) {
 	}
 	return vDp.ConfigMap.Name, nil
 }
+
+func podOwnerReference(p *corev1.Pod) base.ConfigMapOption {
+	return func(cm *corev1.ConfigMap) {
+		cm.OwnerReferences = append(cm.OwnerReferences, metav1.OwnerReference{
+			APIVersion:         p.APIVersion,
+			Kind:               p.Kind,
+			Name:               p.Name,
+			UID:                p.UID,
+			Controller:         pointer.Bool(true),
+			BlockOwnerDeletion: pointer.Bool(true),
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor.go b/control-plane/pkg/reconciler/consumergroup/evictor.go
index 8e5d10675..9ca0a573e 100644
--- a/control-plane/pkg/reconciler/consumergroup/evictor.go
+++ b/control-plane/pkg/reconciler/consumergroup/evictor.go
@@ -24,9 +24,11 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
 	"knative.dev/eventing/pkg/scheduler"
+	"knative.dev/pkg/apis/duck"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
 
@@ -69,7 +71,7 @@ func (e evictor) evict(pod *corev1.Pod, vpod scheduler.VPod, from *eventingduckv
 		return fmt.Errorf("failed to mark pod unschedulable: %w", err)
 	}
 
-	cg, err := e.InternalsClient.
+	cgBefore, err := e.InternalsClient.
 		ConsumerGroups(key.Namespace).
 		Get(e.ctx, key.Name, metav1.GetOptions{})
 	if apierrors.IsNotFound(err) {
@@ -80,21 +82,34 @@ func (e evictor) evict(pod *corev1.Pod, vpod scheduler.VPod, from *eventingduckv
 	}
 
 	// Do not evict when the consumer group isn't scheduled yet.
-	if cg.IsNotScheduled() {
+	if cgBefore.IsNotScheduled() {
 		return nil
 	}
 
-	cg.Status.Placements = removePlacement(cg.GetPlacements(), from)
+	cgAfter := cgBefore.DeepCopy()
+	cgAfter.Status.Placements = removePlacement(cgAfter.GetPlacements(), from)
 
-	_, err = e.InternalsClient.
-		ConsumerGroups(cg.GetNamespace()).
-		Update(e.ctx, cg, metav1.UpdateOptions{})
-	if apierrors.IsNotFound(err) {
+	jsonPatch, err := duck.CreatePatch(cgBefore, cgAfter)
+	if err != nil {
+		return err
+	}
+
+	// If there is nothing to patch, we are good, just return.
+	// Empty patch is [], hence we check for that.
+	if len(jsonPatch) == 0 {
 		return nil
 	}
+
+	patch, err := jsonPatch.MarshalJSON()
+	if err != nil {
+		return fmt.Errorf("marshaling JSON patch: %w", err)
+	}
+
+	_, err = e.InternalsClient.ConsumerGroups(key.Namespace).Patch(e.ctx, key.Name, types.JSONPatchType, patch, metav1.PatchOptions{}, "status")
 	if err != nil {
-		return fmt.Errorf("failed to update consumer group %s/%s: %w", cg.GetNamespace(), cg.GetName(), err)
+		return fmt.Errorf("failed patching: %w", err)
 	}
+	logger.Debug("Patched consumer group with placement changes", zap.Any("patch", patch))
 
 	return nil
 }
@@ -119,7 +134,7 @@ func (e *evictor) disablePodScheduling(logger *zap.Logger, pod *corev1.Pod) erro
 }
 
 func removePlacement(before []eventingduckv1alpha1.Placement, toRemove *eventingduckv1alpha1.Placement) []eventingduckv1alpha1.Placement {
-	after := make([]eventingduckv1alpha1.Placement, 0, len(before))
+	after := make([]eventingduckv1alpha1.Placement, 0, len(before)-1)
 	for _, p := range before {
 		if p.PodName != toRemove.PodName {
 			after = append(after, p)
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>
diff --git a/hack/release.sh b/hack/release.sh
index b47f0f00d..8f626793d 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -35,6 +35,7 @@ function build_release() {
   [ -f "${EVENTING_KAFKA_BROKER_ARTIFACT}" ] && rm "${EVENTING_KAFKA_BROKER_ARTIFACT}"
   [ -f "${EVENTING_KAFKA_SINK_ARTIFACT}" ] && rm "${EVENTING_KAFKA_SINK_ARTIFACT}"
   [ -f "${EVENTING_KAFKA_CHANNEL_ARTIFACT}" ] && rm "${EVENTING_KAFKA_CHANNEL_ARTIFACT}"
+  [ -f "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}" ] && rm "${EVENTING_KAFKA_POST_INSTALL_ARTIFACT}"
 
   control_plane_setup
   if [[ $? -ne 0 ]]; then


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>")
+	}
+}
diff --git a/control-plane/pkg/reconciler/consumer/consumer_test.go b/control-plane/pkg/reconciler/consumer/consumer_test.go
index 4d8c05f4d..3a2ba0469 100644
--- a/control-plane/pkg/reconciler/consumer/consumer_test.go
+++ b/control-plane/pkg/reconciler/consumer/consumer_test.go
@@ -91,7 +91,9 @@ func TestReconcileKind(t *testing.T) {
 					DataPlaneConfigFormat:       base.Json,
 					DataPlaneConfigMapNamespace: SystemNamespace,
 					DataPlaneConfigMapName:      "p1",
-				}, []byte("")),
+				}, []byte(""),
+					DispatcherPodAsOwnerReference("p1"),
+				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&config.Env{
@@ -129,7 +131,9 @@ func TestReconcileKind(t *testing.T) {
 							},
 						},
 					},
-				}),
+				},
+					DispatcherPodAsOwnerReference("p1"),
+				),
 				{Object: NewDispatcherPod("p1",
 					PodRunning(),
 					PodAnnotations(map[string]string{
@@ -204,7 +208,10 @@ func TestReconcileKind(t *testing.T) {
 					DataPlaneConfigFormat:       base.Json,
 					DataPlaneConfigMapNamespace: SystemNamespace,
 					DataPlaneConfigMapName:      "p1",
-				}, []byte("")),
+				},
+					[]byte(""),
+					DispatcherPodAsOwnerReference("p1"),
+				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&config.Env{
@@ -248,7 +255,9 @@ func TestReconcileKind(t *testing.T) {
 							},
 						},
 					},
-				}),
+				},
+					DispatcherPodAsOwnerReference("p1"),
+				),
 				{Object: NewDispatcherPod("p1",
 					PodRunning(),
 					PodAnnotations(map[string]string{
diff --git a/control-plane/pkg/reconciler/consumergroup/evictor_test.go b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
index e334ede0d..f587e4f6a 100644
--- a/control-plane/pkg/reconciler/consumergroup/evictor_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
@@ -45,6 +45,11 @@ func TestEvictorEvictSuccess(t *testing.T) {
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
 	}
+
+	pod1 := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns1", Name: "name1"},
+	}
+
 	cg := &kafkainternals.ConsumerGroup{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
 		Status: kafkainternals.ConsumerGroupStatus{
@@ -52,6 +57,7 @@ func TestEvictorEvictSuccess(t *testing.T) {
 				MaxAllowedVReplicas: pointer.Int32(1),
 				Placements: []eventingduckv1alpha1.Placement{
 					{PodName: pod.GetName(), VReplicas: 1},
+					{PodName: pod1.GetName(), VReplicas: 1},
 				},
 			}},
 		},
@@ -59,7 +65,43 @@ func TestEvictorEvictSuccess(t *testing.T) {
 	cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
 	cg.MarkScheduleSucceeded()
 
-	placement := &eventingduckv1alpha1.Placement{PodName: pod.GetName(), VReplicas: 1}
+	placement := &eventingduckv1alpha1.Placement{PodName: pod1.GetName(), VReplicas: 1}
+
+	ctx, _ = kubeclient.With(ctx, pod)
+	ctx, _ = kafkainternalsclient.With(ctx, cg)
+
+	e := newEvictor(ctx)
+	err := e.evict(pod, cg, placement)
+
+	require.Nil(t, err)
+}
+
+func TestEvictorNoEviction(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
+	}
+
+	pod1 := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns1", Name: "name1"},
+	}
+
+	cg := &kafkainternals.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "ns-1", Name: "cg-name"},
+		Status: kafkainternals.ConsumerGroupStatus{
+			PlaceableStatus: eventingduckv1alpha1.PlaceableStatus{Placeable: eventingduckv1alpha1.Placeable{
+				MaxAllowedVReplicas: pointer.Int32(1),
+				Placements: []eventingduckv1alpha1.Placement{
+					{PodName: pod.GetName(), VReplicas: 1},
+				},
+			}},
+		},
+	}
+	cg.GetConditionSet().Manage(cg.GetStatus()).InitializeConditions()
+	cg.MarkScheduleSucceeded()
+
+	placement := &eventingduckv1alpha1.Placement{PodName: pod1.GetName(), VReplicas: 1}
 
 	ctx, _ = kubeclient.With(ctx, pod)
 	ctx, _ = kafkainternalsclient.With(ctx, cg)
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 5030042f0..3fc025a45 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -28,6 +28,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
+	"k8s.io/utils/pointer"
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
@@ -55,6 +56,8 @@ const (
 	SecretUUID            = "a7185016-5d98-4b54-84e8-3b1cd4acc6b6"
 
 	SystemNamespace = "knative-eventing"
+
+	DispatcherPodUUID = "a7185016-5d98-4b54-84e8-3b1cd4acc6bp"
 )
 
 var (
@@ -107,11 +110,11 @@ func ServiceURLFrom(ns, name string) string {
 	return fmt.Sprintf("http://%s.%s.svc.cluster.local", name, ns)
 }
 
-func NewConfigMapWithBinaryData(env *config.Env, data []byte) runtime.Object {
+func NewConfigMapWithBinaryData(env *config.Env, data []byte, options ...reconcilertesting.ConfigMapOption) runtime.Object {
 	return reconcilertesting.NewConfigMap(
 		env.DataPlaneConfigMapName,
 		env.DataPlaneConfigMapNamespace,
-		func(configMap *corev1.ConfigMap) {
+		append(options, func(configMap *corev1.ConfigMap) {
 			if configMap.BinaryData == nil {
 				configMap.BinaryData = make(map[string][]byte, 1)
 			}
@@ -119,7 +122,7 @@ func NewConfigMapWithBinaryData(env *config.Env, data []byte) runtime.Object {
 				data = []byte("")
 			}
 			configMap.BinaryData[base.ConfigMapDataKey] = data
-		},
+		})...,
 	)
 }
 
@@ -133,7 +136,7 @@ func NewConfigMapWithTextData(namespace, name string, data map[string]string) ru
 	)
 }
 
-func NewConfigMapFromContract(contract *contract.Contract, env *config.Env) runtime.Object {
+func NewConfigMapFromContract(contract *contract.Contract, env *config.Env, options ...reconcilertesting.ConfigMapOption) runtime.Object {
 	var data []byte
 	var err error
 	if env.DataPlaneConfigFormat == base.Protobuf {
@@ -145,10 +148,10 @@ func NewConfigMapFromContract(contract *contract.Contract, env *config.Env) runt
 		panic(err)
 	}
 
-	return NewConfigMapWithBinaryData(env, data)
+	return NewConfigMapWithBinaryData(env, data, options...)
 }
 
-func ConfigMapUpdate(env *config.Env, contract *contract.Contract) clientgotesting.UpdateActionImpl {
+func ConfigMapUpdate(env *config.Env, contract *contract.Contract, options ...reconcilertesting.ConfigMapOption) clientgotesting.UpdateActionImpl {
 	return clientgotesting.NewUpdateAction(
 		schema.GroupVersionResource{
 			Group:    "*",
@@ -156,7 +159,7 @@ func ConfigMapUpdate(env *config.Env, contract *contract.Contract) clientgotesti
 			Resource: "ConfigMap",
 		},
 		env.DataPlaneConfigMapNamespace,
-		NewConfigMapFromContract(contract, env),
+		NewConfigMapFromContract(contract, env, options...),
 	)
 }
 
@@ -324,9 +327,14 @@ type PodOption func(pod *corev1.Pod)
 
 func NewDispatcherPod(name string, options ...PodOption) *corev1.Pod {
 	p := &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Pod",
+		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
 			Namespace: SystemNamespace,
+			UID:       DispatcherPodUUID,
 		},
 		Spec: corev1.PodSpec{
 			Volumes: []corev1.Volume{{
@@ -349,6 +357,20 @@ func NewDispatcherPod(name string, options ...PodOption) *corev1.Pod {
 	return p
 }
 
+func DispatcherPodAsOwnerReference(name string) reconcilertesting.ConfigMapOption {
+	d := NewDispatcherPod(name)
+	return func(configMap *corev1.ConfigMap) {
+		configMap.OwnerReferences = append(configMap.OwnerReferences, metav1.OwnerReference{
+			APIVersion:         d.APIVersion,
+			Kind:               d.Kind,
+			Name:               d.Name,
+			UID:                d.UID,
+			Controller:         pointer.Bool(true),
+			BlockOwnerDeletion: pointer.Bool(true),
+		})
+	}
+}
+
 func PodRunning() PodOption {
 	return func(pod *corev1.Pod) {
 		pod.Status.Phase = corev1.PodRunning