Repository: knative-sandbox/eventing-kafka-broker
Pull request #1835 (2022-01-31 14:06:14)

[release 1.2] Revert channel parts of #1720
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Same as https://github.com/knative-sandbox/eventing-kafka-broker/pull/1824 but for 1.2 branch
-
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
Reverted the channel related changes done in https://github.com/knative-sandbox/eventing-kafka-broker/pull/1720. The reference information for metrics was not built successfully for channels.
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1835#" title="Author self-approved">aliok</a>*

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

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

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.2/OWNERS)~~ [aliok]

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1835/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1488151649187794944/artifacts/line-cov.html#file0) | 69.8% | 78.8% | 9.0

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1835?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> :exclamation: No coverage uploaded for pull request base (`release-1.2@adb87f1`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1835/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/1835?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@              Coverage Diff               @@
##             release-1.2    #1835   +/-   ##
==============================================
  Coverage               ?   82.01%           
  Complexity             ?      614           
==============================================
  Files                  ?       69           
  Lines                  ?     2024           
  Branches               ?      178           
==============================================
  Hits                   ?     1660           
  Misses                 ?      270           
  Partials               ?       94           
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.01% <0.00%> (?)` | |

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.


------

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

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1835#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1835#pullrequestreview-868042664" title="Approved">pierDipi</a>*

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

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

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.2/OWNERS)~~ [aliok,pierDipi]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 4cdd58912..b1894acd0 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -18,16 +18,14 @@ package channel
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strconv"
 	"time"
 
 	"github.com/Shopify/sarama"
+
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -40,7 +38,6 @@ import (
 	"knative.dev/pkg/system"
 
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
-	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/constants"
@@ -81,8 +78,7 @@ type Reconciler struct {
 	// reconciliation loop.
 	InitOffsetsFunc kafka.InitOffsetsFunc
 
-	ConfigMapLister    corelisters.ConfigMapLister
-	SubscriptionLister messaginglisters.SubscriptionLister
+	ConfigMapLister corelisters.ConfigMapLister
 
 	Prober prober.Prober
 
@@ -483,21 +479,11 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 }
 
 func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (*contract.Egress, error) {
-	subscriptionName, err := r.getSubscriptionName(channel, subscriber)
-	if err != nil {
-		return nil, fmt.Errorf("failed to extract subscription name for subscriber %v: %w", subscriber, err)
-	}
-
 	egress := &contract.Egress{
 		Destination:   subscriber.SubscriberURI.String(),
 		ConsumerGroup: consumerGroup(channel, subscriber),
 		DeliveryOrder: DefaultDeliveryOrder,
 		Uid:           string(subscriber.UID),
-		Reference: &contract.Reference{
-			Uuid:      string(subscriber.UID),
-			Namespace: channel.GetNamespace(),
-			Name:      subscriptionName,
-		},
 		ReplyStrategy: &contract.Egress_DiscardReply{},
 	}
 
@@ -598,21 +584,6 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 	return resource, nil
 }
 
-func (r *Reconciler) getSubscriptionName(channel *messagingv1beta1.KafkaChannel, subscriber *v1.SubscriberSpec) (string, error) {
-	subscriptions, err := r.SubscriptionLister.Subscriptions(channel.GetNamespace()).List(labels.Everything())
-	if err != nil {
-		return "", fmt.Errorf("failed to list subscriptions in namespace %s: %w", channel.GetNamespace(), err)
-	}
-
-	for _, s := range subscriptions {
-		if s.UID == subscriber.UID {
-			return s.Name, nil
-		}
-	}
-
-	return "", errors.New(string(metav1.StatusReasonNotFound))
-}
-
 // consumerGroup returns a consumerGroup name for the given channel and subscription
 func consumerGroup(channel *messagingv1beta1.KafkaChannel, sub *v1.SubscriberSpec) string {
 	return fmt.Sprintf("kafka.%s.%s.%s", channel.Namespace, channel.Name, string(sub.UID))
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index db8296757..8ccfa800b 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -25,8 +25,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
 
-	subscriptioninformer "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
-
 	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"
@@ -67,7 +65,6 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			DispatcherLabel:             base.ChannelDispatcherLabel,
 			ReceiverLabel:               base.ChannelReceiverLabel,
 		},
-		SubscriptionLister:         subscriptioninformer.Get(ctx).Lister(),
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		InitOffsetsFunc:            offset.InitOffsets,
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
deleted file mode 100644
index eb18a1fac..000000000
--- a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake/fake.go
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Code generated by injection-gen. DO NOT EDIT.
-
-package fake
-
-import (
-	context "context"
-
-	fake "knative.dev/eventing/pkg/client/injection/informers/factory/fake"
-	subscription "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription"
-	controller "knative.dev/pkg/controller"
-	injection "knative.dev/pkg/injection"
-)
-
-var Get = subscription.Get
-
-func init() {
-	injection.Fake.RegisterInformer(withInformer)
-}
-
-func withInformer(ctx context.Context) (context.Context, controller.Informer) {
-	f := fake.Get(ctx)
-	inf := f.Messaging().V1().Subscriptions()
-	return context.WithValue(ctx, subscription.Key{}, inf), inf.Informer()
-}
diff --git a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go b/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
deleted file mode 100644
index 37862d8a5..000000000
--- a/vendor/knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/subscription.go
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
-Copyright 2021 The Knative Authors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-// Code generated by injection-gen. DO NOT EDIT.
-
-package subscription
-
-import (
-	context "context"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	labels "k8s.io/apimachinery/pkg/labels"
-	cache "k8s.io/client-go/tools/cache"
-	apismessagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
-	versioned "knative.dev/eventing/pkg/client/clientset/versioned"
-	v1 "knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1"
-	client "knative.dev/eventing/pkg/client/injection/client"
-	factory "knative.dev/eventing/pkg/client/injection/informers/factory"
-	messagingv1 "knative.dev/eventing/pkg/client/listers/messaging/v1"
-	controller "knative.dev/pkg/controller"
-	injection "knative.dev/pkg/injection"
-	logging "knative.dev/pkg/logging"
-)
-
-func init() {
-	injection.Default.RegisterInformer(withInformer)
-	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
-}
-
-// Key is used for associating the Informer inside the context.Context.
-type Key struct{}
-
-func withInformer(ctx context.Context) (context.Context, controller.Informer) {
-	f := factory.Get(ctx)
-	inf := f.Messaging().V1().Subscriptions()
-	return context.WithValue(ctx, Key{}, inf), inf.Informer()
-}
-
-func withDynamicInformer(ctx context.Context) context.Context {
-	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
-	return context.WithValue(ctx, Key{}, inf)
-}
-
-// Get extracts the typed informer from the context.
-func Get(ctx context.Context) v1.SubscriptionInformer {
-	untyped := ctx.Value(Key{})
-	if untyped == nil {
-		logging.FromContext(ctx).Panic(
-			"Unable to fetch knative.dev/eventing/pkg/client/informers/externalversions/messaging/v1.SubscriptionInformer from context.")
-	}
-	return untyped.(v1.SubscriptionInformer)
-}
-
-type wrapper struct {
-	client versioned.Interface
-
-	namespace string
-
-	resourceVersion string
-}
-
-var _ v1.SubscriptionInformer = (*wrapper)(nil)
-var _ messagingv1.SubscriptionLister = (*wrapper)(nil)
-
-func (w *wrapper) Informer() cache.SharedIndexInformer {
-	return cache.NewSharedIndexInformer(nil, &apismessagingv1.Subscription{}, 0, nil)
-}
-
-func (w *wrapper) Lister() messagingv1.SubscriptionLister {
-	return w
-}
-
-func (w *wrapper) Subscriptions(namespace string) messagingv1.SubscriptionNamespaceLister {
-	return &wrapper{client: w.client, namespace: namespace, resourceVersion: w.resourceVersion}
-}
-
-// SetResourceVersion allows consumers to adjust the minimum resourceVersion
-// used by the underlying client.  It is not accessible via the standard
-// lister interface, but can be accessed through a user-defined interface and
-// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
-func (w *wrapper) SetResourceVersion(resourceVersion string) {
-	w.resourceVersion = resourceVersion
-}
-
-func (w *wrapper) List(selector labels.Selector) (ret []*apismessagingv1.Subscription, err error) {
-	lo, err := w.client.MessagingV1().Subscriptions(w.namespace).List(context.TODO(), metav1.ListOptions{
-		LabelSelector:   selector.String(),
-		ResourceVersion: w.resourceVersion,
-	})
-	if err != nil {
-		return nil, err
-	}
-	for idx := range lo.Items {
-		ret = append(ret, &lo.Items[idx])
-	}
-	return ret, nil
-}
-
-func (w *wrapper) Get(name string) (*apismessagingv1.Subscription, error) {
-	return w.client.MessagingV1().Subscriptions(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
-		ResourceVersion: w.resourceVersion,
-	})
-}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index c9575d547..694d7d209 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1179,8 +1179,6 @@ knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger
 knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake
 knative.dev/eventing/pkg/client/injection/informers/factory
 knative.dev/eventing/pkg/client/injection/informers/factory/fake
-knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription
-knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker
 knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger
 knative.dev/eventing/pkg/client/listers/eventing/v1


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index f3c49154b..c5e71fdac 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -22,7 +22,6 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
-	eventingrekttesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 	"knative.dev/pkg/network"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -371,9 +370,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -396,11 +392,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -454,9 +445,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -479,11 +467,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -537,9 +520,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -562,11 +542,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -623,12 +598,6 @@ func TestReconcileKind(t *testing.T) {
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
-				eventingrekttesting.NewSubscription(Subscription2Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription2UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -651,22 +620,12 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}, {
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription2UUID,
 								Destination:   "http://" + Subscription2URI,
 								Uid:           Subscription2UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_DiscardReply{},
-								Reference: &contract.Reference{
-									Name:      Subscription2Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription2UUID,
-								},
 							}},
 						},
 					},
@@ -994,9 +953,6 @@ func TestReconcileKind(t *testing.T) {
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
 				NewSSLSecret(system.Namespace(), "secret-1"),
-				eventingrekttesting.NewSubscription(Subscription1Name, ChannelNamespace,
-					eventingrekttesting.WithSubscriptionUID(Subscription1UUID),
-				),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1027,11 +983,6 @@ func TestReconcileKind(t *testing.T) {
 								Uid:           Subscription1UUID,
 								DeliveryOrder: contract.DeliveryOrder_ORDERED,
 								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
-								Reference: &contract.Reference{
-									Name:      Subscription1Name,
-									Namespace: ChannelNamespace,
-									Uuid:      Subscription1UUID,
-								},
 							}},
 						},
 					},
@@ -1286,9 +1237,8 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 				DispatcherLabel:             base.ChannelDispatcherLabel,
 				ReceiverLabel:               base.ChannelReceiverLabel,
 			},
-			Env:                env,
-			ConfigMapLister:    listers.GetConfigMapLister(),
-			SubscriptionLister: listers.GetSubscriptionLister(),
+			Env:             env,
+			ConfigMapLister: listers.GetConfigMapLister(),
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index f3a49c227..cc3ca16ce 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -24,7 +24,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel/fake"
-	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/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"