Repository: knative-sandbox/eventing-kafka-broker
Pull request #2523 (2022-08-19 07:51:06)

`pod.Status.Phase == apiv1.PodRunning` is not a good way to test if pods are Running
See https://github.com/kubernetes/client-go/issues/815

I'm using the logic used by https://pkg.go.dev/k8s.io/kubernetes/pkg/api/v1/pod#IsPodReady
but that function is not in a library package so it can't be reused.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2523#" title="Author self-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/main/OWNERS)~~ [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":[]} -->
knative-prow[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 4dde89535d745a6e93d26369afd7249cca765fca | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2523/unit-tests_eventing-kafka-broker_main/1560534820554018816) | true | `/test unit-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2523](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4dde895) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/93778ed644d2f4fdf21a83a6c6991b7833172205?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (93778ed) will **increase** coverage by `18.18%`.
> The diff coverage is `n/a`.

```diff
@@              Coverage Diff              @@
##               main    #2523       +/-   ##
=============================================
+ Coverage     62.99%   81.18%   +18.18%     
- Complexity      704      705        +1     
=============================================
  Files           146       72       -74     
  Lines          9981     2466     -7515     
  Branches        224      224               
=============================================
- Hits           6288     2002     -4286     
+ Misses         3259      330     -2929     
+ Partials        434      134      -300     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.18% <ø> (+0.08%)` | :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/2523?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | | |
| [control-plane/pkg/reconciler/channel/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NvbnRyb2xsZXIuZ28=) | | |
| [control-plane/pkg/contract/serde.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3Qvc2VyZGUuZ28=) | | |
| [...als/kafka/eventing/v1alpha1/consumer\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [...ol-plane/pkg/apis/eventing/v1/broker\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MS9icm9rZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [control-plane/pkg/contract/log\_marshaller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3QvbG9nX21hcnNoYWxsZXIuZ28=) | | |
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | |
| [...ol-plane/pkg/security/secrets\_provider\_net\_spec.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0c19wcm92aWRlcl9uZXRfc3BlYy5nbw==) | | |
| [...trol-plane/pkg/prober/probertesting/mock\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2JlcnRlc3RpbmcvbW9ja19wcm9iZXIuZ28=) | | |
| [...ol-plane/pkg/reconciler/consumer/secret\_locator.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9zZWNyZXRfbG9jYXRvci5nbw==) | | |
| ... and [66 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2523/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

knative-prow[bot]: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 7f1b545ca507675bca685975dfd694844fba3da7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2523/unit-tests_eventing-kafka-broker_main/1560537428970704896) | true | `/test unit-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 9a0949f95..dac3811c6 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -28,7 +28,6 @@ aliases:
   - n3wscott
   docs-wg-leads:
   - abrennan89
-  - csantanapr
   - snneji
   docs-writers:
   - abrennan89
@@ -114,9 +113,18 @@ aliases:
   - lkingland
   - matejvasek
   - nainaz
-  - omerbensaadon
   - salaboy
   - zroubalik
+  func-writers:
+  - lance
+  - lkingland
+  - matejvasek
+  - nainaz
+  - salaboy
+  - zroubalik
+  functions-wg-leads:
+  - lance
+  - salaboy
   homebrew-kn-plugins-approvers:
   - dsimansk
   - maximilien
@@ -138,14 +146,12 @@ aliases:
   - lkingland
   - matejvasek
   - nainaz
-  - omerbensaadon
   - salaboy
   - zroubalik
   kn-plugin-migration-approvers:
   - maximilien
   - zhangtbj
   kn-plugin-operator-approvers:
-  - csantanapr
   - dsimansk
   - houshengbo
   - maximilien
@@ -290,9 +296,7 @@ aliases:
   - zroubalik
   ux-wg-leads:
   - abrennan89
-  - csantanapr
   - snneji
   ux-writers:
   - abrennan89
-  - csantanapr
   - snneji
diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 73eec1726..d20df51b2 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -76,22 +76,21 @@ func main() {
 			},
 		},
 
-		// TODO: enable later
-		//// Namespaced broker controller
-		//injection.NamedControllerConstructor{
-		//	Name: "broker-namespaced-controller",
-		//	ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-		//		return broker.NewNamespacedController(ctx, watcher, brokerEnv)
-		//	},
-		//},
-		//
-		//// Namespaced trigger controller
-		//injection.NamedControllerConstructor{
-		//	Name: "trigger-namespaced-controller",
-		//	ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-		//		return trigger.NewNamespacedController(ctx, watcher, brokerEnv)
-		//	},
-		//},
+		// Namespaced broker controller
+		injection.NamedControllerConstructor{
+			Name: "broker-namespaced-controller",
+			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+				return broker.NewNamespacedController(ctx, watcher, brokerEnv)
+			},
+		},
+
+		// Namespaced trigger controller
+		injection.NamedControllerConstructor{
+			Name: "trigger-namespaced-controller",
+			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+				return trigger.NewNamespacedController(ctx, watcher, brokerEnv)
+			},
+		},
 
 		// Channel controller
 		injection.NamedControllerConstructor{
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 852d884b8..666d47275 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
@@ -38,6 +38,7 @@ rules:
       - watch
       - update
       - create
+      - delete
   - apiGroups:
       - "*"
     resources:
@@ -98,6 +99,15 @@ rules:
       - watch
       - update
       - create
+      - delete
+  - apiGroups:
+      - "rbac.authorization.k8s.io"
+    resources:
+      - clusterrolebindings
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
     - "*"
     resources:
@@ -108,6 +118,7 @@ rules:
       - watch
       - update
       - create
+      - delete
   - apiGroups:
       - "apps"
     resources:
@@ -118,6 +129,7 @@ rules:
       - watch
       - update
       - create
+      - delete
 
   # Scheduler permissions
   - apiGroups:
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 5d30180ac..3f3cca0d0 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -101,7 +101,16 @@ func (r *Reconciler) IsDispatcherRunning() bool {
 
 func isAtLeastOneRunning(pods []*corev1.Pod) bool {
 	for _, pod := range pods {
-		if pod.Status.Phase == corev1.PodRunning {
+		if isReady(pod) {
+			return true
+		}
+	}
+	return false
+}
+
+func isReady(pod *corev1.Pod) bool {
+	for _, c := range pod.Status.Conditions {
+		if c.Type == corev1.PodReady && c.Status == corev1.ConditionTrue {
 			return true
 		}
 	}
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 34ccf87ed..c5c63c1be 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -87,6 +87,16 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		Recorder:   controller.GetEventRecorder(ctx),
 	}
 
+	// Get contract config map. Do this in advance, otherwise
+	// the dataplane pods that need volume mounts to the contract configmap
+	// will get stuck and will never be ready.
+	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
+	if err != nil {
+		return statusConditionManager.FailedToGetConfigMap(err)
+	}
+
+	logger.Debug("Got contract config map")
+
 	if !r.IsReceiverRunning() || !r.IsDispatcherRunning() {
 		return statusConditionManager.DataPlaneNotAvailable()
 	}
@@ -133,14 +143,6 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		return err
 	}
 
-	// Get contract config map.
-	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
-	if err != nil {
-		return statusConditionManager.FailedToGetConfigMap(err)
-	}
-
-	logger.Debug("Got contract config map")
-
 	// Get contract data.
 	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
 	if err != nil && ct == nil {
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker.go b/control-plane/pkg/reconciler/broker/namespaced_broker.go
index 05dba7910..943dd23b4 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_broker.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker.go
@@ -19,6 +19,7 @@ package broker
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	mf "github.com/manifestival/manifestival"
 	"go.uber.org/zap"
@@ -60,6 +61,7 @@ type NamespacedReconciler struct {
 
 	ConfigMapLister          corelisters.ConfigMapLister
 	ServiceAccountLister     corelisters.ServiceAccountLister
+	ServiceLister            corelisters.ServiceLister
 	ClusterRoleBindingLister rbaclisters.ClusterRoleBindingLister
 	DeploymentLister         appslisters.DeploymentLister
 
@@ -83,24 +85,13 @@ func (r *NamespacedReconciler) ReconcileKind(ctx context.Context, broker *eventi
 
 	br := r.createReconcilerForBrokerInstance(broker)
 
-	manifest, err := r.getManifest(broker)
+	manifest, err := r.getManifest(ctx, broker)
 	if err != nil {
-		return fmt.Errorf("unable to transform dataplane manifest. namespace: %s, broker: %v, error: %v", broker.Namespace, broker, err)
+		return propagateErrorCondition(broker, fmt.Errorf("unable to transform dataplane manifest: %w", err))
 	}
 
 	if err = manifest.Apply(); err != nil {
-		return fmt.Errorf("unable to apply dataplane manifest. namespace: %s, broker: %v, error: %w", broker.Namespace, broker, err)
-	}
-
-	logger := logging.FromContext(ctx).Desugar()
-	logger.Debug("Data plane reconciled")
-
-	for _, res := range manifest.Resources() {
-		logger.Debug("Resource reconciled",
-			zap.String("gvk", res.GroupVersionKind().String()),
-			zap.String("resource.name", res.GetName()),
-			zap.String("resource.namespace", res.GetNamespace()),
-		)
+		return propagateErrorCondition(broker, fmt.Errorf("unable to apply dataplane manifest: %w", err))
 	}
 
 	return br.ReconcileKind(ctx, broker)
@@ -148,7 +139,7 @@ func (r *NamespacedReconciler) createReconcilerForBrokerInstance(broker *eventin
 
 // getManifest returns the manifest that is transformed from the BaseDataPlaneManifest with the changes
 // for the given broker
-func (r *NamespacedReconciler) getManifest(broker *eventing.Broker) (mf.Manifest, error) {
+func (r *NamespacedReconciler) getManifest(ctx context.Context, broker *eventing.Broker) (mf.Manifest, error) {
 	var resources []unstructured.Unstructured
 
 	additionalConfigMaps, err := r.configMapsFromSystemNamespace(broker)
@@ -169,6 +160,12 @@ func (r *NamespacedReconciler) getManifest(broker *eventing.Broker) (mf.Manifest
 	}
 	resources = append(resources, additionalServiceAccounts...)
 
+	additionalServices, err := r.servicesFromSystemNamespace(broker)
+	if err != nil {
+		return mf.Manifest{}, err
+	}
+	resources = append(resources, additionalServices...)
+
 	additionalRoleBindings, err := r.roleBindingsFromSystemNamespace(broker)
 	if err != nil {
 		return mf.Manifest{}, err
@@ -185,7 +182,32 @@ func (r *NamespacedReconciler) getManifest(broker *eventing.Broker) (mf.Manifest
 		return mf.Manifest{}, fmt.Errorf("unable to transform base dataplane manifest with namespace injection. namespace: %s, error: %v", broker.Namespace, err)
 	}
 
-	return manifest.Transform(appendNewOwnerRefsToPersisted(manifest.Client, broker))
+	manifest, err = manifest.Transform(appendNewOwnerRefsToPersisted(manifest.Client, broker))
+	if err != nil {
+		return mf.Manifest{}, fmt.Errorf("unable to append owner ref: %w", err)
+	}
+
+	manifest, err = manifest.Transform(filterMetadata())
+	if err != nil {
+		return mf.Manifest{}, fmt.Errorf("unable to filter metadata: %w", err)
+	}
+
+	manifest, err = manifest.Transform(setLabel())
+	if err != nil {
+		return mf.Manifest{}, fmt.Errorf("unable to set label: %w", err)
+	}
+
+	logger := logging.FromContext(ctx).Desugar()
+
+	for _, res := range manifest.Resources() {
+		logger.Debug("Resource",
+			zap.String("gvk", res.GroupVersionKind().String()),
+			zap.String("resource.name", res.GetName()),
+			zap.String("resource.namespace", res.GetNamespace()),
+		)
+	}
+
+	return manifest, nil
 }
 
 func (r *NamespacedReconciler) deploymentsFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
@@ -193,7 +215,7 @@ func (r *NamespacedReconciler) deploymentsFromSystemNamespace(broker *eventing.B
 		"kafka-broker-receiver",
 		"kafka-broker-dispatcher",
 	}
-	resources := make([]unstructured.Unstructured, len(deployments))
+	resources := make([]unstructured.Unstructured, 0, len(deployments))
 	for _, name := range deployments {
 		resource, err := r.createManifestFromSystemDeployment(broker, name)
 		if err != nil {
@@ -210,7 +232,7 @@ func (r *NamespacedReconciler) configMapsFromSystemNamespace(broker *eventing.Br
 		"config-tracing",
 		"kafka-config-logging",
 	}
-	resources := make([]unstructured.Unstructured, len(configMaps))
+	resources := make([]unstructured.Unstructured, 0, len(configMaps))
 	for _, name := range configMaps {
 		resource, err := r.createResourceFromSystemConfigMap(broker, name)
 		if err != nil {
@@ -265,7 +287,7 @@ func (r *NamespacedReconciler) serviceAccountsFromSystemNamespace(broker *eventi
 	serviceAccounts := []string{
 		"knative-kafka-broker-data-plane",
 	}
-	resources := make([]unstructured.Unstructured, len(serviceAccounts))
+	resources := make([]unstructured.Unstructured, 0, len(serviceAccounts))
 	for _, name := range serviceAccounts {
 		resource, err := r.createManifestFromSystemServiceAccount(broker, name)
 		if err != nil {
@@ -297,11 +319,50 @@ func (r *NamespacedReconciler) createManifestFromSystemServiceAccount(broker *ev
 	return unstructuredFromObject(cm)
 }
 
+func (r *NamespacedReconciler) servicesFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
+	services := []string{
+		"kafka-broker-ingress",
+	}
+	resources := make([]unstructured.Unstructured, 0, len(services))
+	for _, name := range services {
+		resource, err := r.createManifestFromSystemService(broker, name)
+		if err != nil {
+			return nil, err
+		}
+		resources = append(resources, resource)
+	}
+	return resources, nil
+}
+
+func (r *NamespacedReconciler) createManifestFromSystemService(broker *eventing.Broker, name string) (unstructured.Unstructured, error) {
+	sysService, err := r.ServiceLister.Services(r.SystemNamespace).Get(name)
+	if err != nil {
+		return unstructured.Unstructured{}, fmt.Errorf("failed to get Service %s/%s: %w", r.SystemNamespace, name, err)
+	}
+
+	o := &corev1.Service{
+		TypeMeta: metav1.TypeMeta{Kind: "Service", APIVersion: corev1.SchemeGroupVersion.String()},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:   broker.GetNamespace(),
+			Name:        sysService.Name,
+			Labels:      sysService.Labels,
+			Annotations: sysService.Annotations,
+		},
+		Spec: corev1.ServiceSpec{
+			Ports:    sysService.Spec.Ports,
+			Selector: sysService.Spec.Selector,
+			// leave other fields empty on purpose
+		},
+	}
+
+	return unstructuredFromObject(o)
+}
+
 func (r *NamespacedReconciler) roleBindingsFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
 	clusterRoleBindings := []string{
 		"knative-kafka-broker-data-plane",
 	}
-	resources := make([]unstructured.Unstructured, len(clusterRoleBindings))
+	resources := make([]unstructured.Unstructured, 0, len(clusterRoleBindings))
 	for _, name := range clusterRoleBindings {
 		resource, err := r.createManifestFromClusterRoleBinding(broker, name)
 		if err != nil {
@@ -395,3 +456,35 @@ func appendOwnerRef(refs []metav1.OwnerReference, broker *eventing.Broker) ([]me
 
 	return append(refs, newRef), true
 }
+
+func filterMetadata() mf.Transformer {
+	return func(u *unstructured.Unstructured) error {
+		u.SetLabels(filterMetadataMap(u.GetLabels()))
+		u.SetAnnotations(filterMetadataMap(u.GetAnnotations()))
+		return nil
+	}
+}
+
+func setLabel() mf.Transformer {
+	return func(u *unstructured.Unstructured) error {
+		labels := u.GetLabels()
+		labels[kafka.NamespacedBrokerDataplaneLabelKey] = kafka.NamespacedBrokerDataplaneLabelValue
+		u.SetLabels(labels)
+		return nil
+	}
+}
+
+func filterMetadataMap(metadata map[string]string) map[string]string {
+	r := make(map[string]string, len(metadata))
+	for k, v := range metadata {
+		if strings.Contains(k, "knative") || k == "app" {
+			r[k] = v
+		}
+	}
+	return r
+}
+
+func propagateErrorCondition(broker *eventing.Broker, err error) error {
+	broker.GetConditionSet().Manage(broker.GetStatus()).MarkFalse(base.ConditionDataPlaneAvailable, "CreateDataPlane", err.Error())
+	return err
+}
diff --git a/control-plane/pkg/reconciler/broker/namespaced_controller.go b/control-plane/pkg/reconciler/broker/namespaced_controller.go
index 7404ee91f..7879fd5b6 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_controller.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_controller.go
@@ -23,18 +23,22 @@ import (
 	"github.com/Shopify/sarama"
 	mfclient "github.com/manifestival/client-go-client"
 	"go.uber.org/zap"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/client-go/tools/cache"
 
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/logging"
+	pkgreconciler "knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
+
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 
 	brokerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker"
@@ -43,6 +47,9 @@ import (
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
+	serviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
+	serviceaccountinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount"
+	clusterrolebindinginformer "knative.dev/pkg/client/injection/kube/informers/rbac/v1/clusterrolebinding"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
@@ -76,6 +83,10 @@ func NewNamespacedController(ctx context.Context, watcher configmap.Watcher, env
 		},
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
 		ConfigMapLister:            configmapInformer.Lister(),
+		ServiceAccountLister:       serviceaccountinformer.Get(ctx).Lister(),
+		ServiceLister:              serviceinformer.Get(ctx).Lister(),
+		ClusterRoleBindingLister:   clusterrolebindinginformer.Get(ctx).Lister(),
+		DeploymentLister:           deploymentinformer.Get(ctx).Lister(),
 		Env:                        env,
 		ManifestivalClient:         mfc,
 	}
@@ -96,7 +107,10 @@ func NewNamespacedController(ctx context.Context, watcher configmap.Watcher, env
 	})
 
 	reconciler.SecretTracker = impl.Tracker
-	secretinformer.Get(ctx).Informer().AddEventHandler(controller.HandleAll(reconciler.SecretTracker.OnChanged))
+	secretinformer.Get(ctx).Informer().AddEventHandler(controller.HandleAll(controller.EnsureTypeMeta(
+		reconciler.SecretTracker.OnChanged,
+		corev1.SchemeGroupVersion.WithKind("Secret"),
+	)))
 
 	globalResync := func(_ interface{}) {
 		impl.GlobalResync(brokerInformer.Informer())
@@ -119,7 +133,41 @@ func NewNamespacedController(ctx context.Context, watcher configmap.Watcher, env
 			kafka.FilterWithLabel("app", "kafka-broker-dispatcher"),
 			kafka.FilterWithLabel("app", "kafka-broker-receiver"),
 		),
-		Handler: controller.HandleAll(globalResync),
+		Handler: controller.HandleAll(controller.EnsureTypeMeta(
+			globalResync,
+			appsv1.SchemeGroupVersion.WithKind("Deployment"),
+		)),
+	})
+
+	// we set a label for each resource we create and filter things based on that
+	filterFunc := pkgreconciler.LabelFilterFunc(
+		kafka.NamespacedBrokerDataplaneLabelKey,
+		kafka.NamespacedBrokerDataplaneLabelValue,
+		false, // allowUnset
+	)
+
+	clusterrolebindinginformer.Get(ctx).Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: filterFunc,
+		Handler: controller.HandleAll(controller.EnsureTypeMeta(
+			globalResync,
+			rbacv1.SchemeGroupVersion.WithKind("ClusterRoleBinding"),
+		)),
+	})
+
+	serviceaccountinformer.Get(ctx).Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: filterFunc,
+		Handler: controller.HandleAll(controller.EnsureTypeMeta(
+			globalResync,
+			corev1.SchemeGroupVersion.WithKind("ServiceAccount"),
+		)),
+	})
+
+	serviceinformer.Get(ctx).Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: filterFunc,
+		Handler: controller.HandleAll(controller.EnsureTypeMeta(
+			globalResync,
+			corev1.SchemeGroupVersion.WithKind("Service"),
+		)),
 	})
 
 	reconciler.ConfigMapTracker = impl.Tracker
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
index 7fcf35c9d..25855515f 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
@@ -19,7 +19,6 @@ import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.tracing.TracingSpan;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.ResponseFailureException;
-import dev.knative.eventing.kafka.broker.dispatcher.main.ConsumerVerticleFactoryImpl;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.http.vertx.VertxMessageFactory;
 import io.cloudevents.rw.CloudEventRWException;
@@ -124,11 +123,15 @@ public final class WebClientCloudEventSender implements CloudEventSender {
       },
       // Retry and failures
       cause -> {
+        if (cause instanceof ResponseFailureException) {
+          final var response = ((ResponseFailureException) cause).getResponse();
+          if (isRetryableStatusCode(response.statusCode()) && retryCounter < egress.getRetry()) {
+            return retry(retryCounter, event);
+          }
+        }
+
         if (retryCounter < egress.getRetry()) {
-          Promise<HttpResponse<Buffer>> r = Promise.promise();
-          final var delay = retryPolicyFunc.apply(retryCounter + 1);
-          vertx.setTimer(delay, v -> send(event, retryCounter + 1).onComplete(r));
-          return r.future();
+          return retry(retryCounter, event);
         }
 
         return Future.failedFuture(cause);
@@ -136,6 +139,13 @@ public final class WebClientCloudEventSender implements CloudEventSender {
     );
   }
 
+  private Future<HttpResponse<Buffer>> retry(int retryCounter, CloudEvent event) {
+    Promise<HttpResponse<Buffer>> r = Promise.promise();
+    final var delay = retryPolicyFunc.apply(retryCounter + 1);
+    vertx.setTimer(delay, v -> send(event, retryCounter + 1).onComplete(r));
+    return r.future();
+  }
+
   private void requestCompleted() {
     inFlightRequests.decrementAndGet();
     if (closed.get() && inFlightRequests.get() == 0) {
@@ -159,8 +169,8 @@ public final class WebClientCloudEventSender implements CloudEventSender {
       })
       .onSuccess(response -> {
 
-        if (isRetryableStatusCode(response.statusCode())) {
-          logError(event, response);
+        if (response.statusCode() >= 300) {
+          logError("Received a non-retryable status code that is not 2xx.", event, response);
           breaker.tryFail(new ResponseFailureException(
             response,
             "Received failure response, status code: " + response.statusCode())
@@ -168,10 +178,6 @@ public final class WebClientCloudEventSender implements CloudEventSender {
           return;
         }
 
-        if (response.statusCode() >= 300) {
-          logError("Received a non-retryable status code that is not 2xx.", event, response);
-        }
-
         breaker.tryComplete(response);
       });
   }
diff --git a/go.mod b/go.mod
index 73e33cfd0..90cf94f27 100644
--- a/go.mod
+++ b/go.mod
@@ -29,11 +29,11 @@ require (
 	k8s.io/apiserver v0.23.9
 	k8s.io/client-go v0.23.9
 	k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
-	knative.dev/eventing v0.33.1-0.20220817092547-cb3901fa91f2
-	knative.dev/eventing-kafka v0.33.1-0.20220817023247-6b51416d16cb
+	knative.dev/eventing v0.33.1-0.20220822135655-5ee615866b28
+	knative.dev/eventing-kafka v0.33.1-0.20220822070354-97a8afd1af90
 	knative.dev/hack v0.0.0-20220815132133-e9a8475f4329
 	knative.dev/pkg v0.0.0-20220818004048-4a03844c0b15
-	knative.dev/reconciler-test v0.0.0-20220816163747-93a10831a971
+	knative.dev/reconciler-test v0.0.0-20220818122349-177f8264c28c
 	sigs.k8s.io/yaml v1.3.0
 )
 
diff --git a/go.sum b/go.sum
index 380291241..2c2d33d85 100644
--- a/go.sum
+++ b/go.sum
@@ -1504,44 +1504,36 @@ k8s.io/api v0.15.7/go.mod h1:a/tUxscL+UxvYyA7Tj5DRc8ivYqJIO1Y5KDdlI6wSvo=
 k8s.io/api v0.19.2/go.mod h1:IQpK0zFQ1xc5iNIQPqzgoOwuFugaYHK4iCknlAQP9nI=
 k8s.io/api v0.19.7/go.mod h1:KTryDUT3l6Mtv7K2J2486PNL9DBns3wOYTkGR+iz63Y=
 k8s.io/api v0.22.5/go.mod h1:mEhXyLaSD1qTOf40rRiKXkc+2iCem09rWLlFwhCEiAs=
-k8s.io/api v0.23.8/go.mod h1:26VNldp5fRyiZ/tEO/3V8RJW+In1Aay3YkxsDzkvuWU=
 k8s.io/api v0.23.9 h1:v7Ee2CZuyb6rVm1q4bUe7ZonWleLsrvgcOTxPGjQVa4=
 k8s.io/api v0.23.9/go.mod h1:r4g0GrGdLgwSYB90qgO4tBrbKtALBhUfut+oFt4ikCc=
 k8s.io/apiextensions-apiserver v0.19.2/go.mod h1:EYNjpqIAvNZe+svXVx9j4uBaVhTB4C94HkY3w058qcg=
-k8s.io/apiextensions-apiserver v0.23.8/go.mod h1:pcYtXkSRncXGgCCQSbTqppO3XMaLWdqlDisD4dUKQ/U=
 k8s.io/apiextensions-apiserver v0.23.9 h1:q6X/HhgKo7/Up1p1TeNGNAwqcaBcxaxjxxwXa/5ht+E=
 k8s.io/apiextensions-apiserver v0.23.9/go.mod h1:uu79PjF1T6YbfFqL5kVTmEdxb40Z0eHM7MfHDHz9dho=
 k8s.io/apimachinery v0.15.7/go.mod h1:Xc10RHc1U+F/e9GCloJ8QAeCGevSVP5xhOhqlE+e1kM=
 k8s.io/apimachinery v0.19.2/go.mod h1:DnPGDnARWFvYa3pMHgSxtbZb7gpzzAZ1pTfaUNDVlmA=
 k8s.io/apimachinery v0.19.7/go.mod h1:6sRbGRAVY5DOCuZwB5XkqguBqpqLU6q/kOaOdk29z6Q=
 k8s.io/apimachinery v0.22.5/go.mod h1:xziclGKwuuJ2RM5/rSFQSYAj0zdbci3DH8kj+WvyN0U=
-k8s.io/apimachinery v0.23.8/go.mod h1:BEuFMMBaIbcOqVIJqNZJXGFTP4W6AycEpb5+m/97hrM=
 k8s.io/apimachinery v0.23.9 h1:u9Pu7Ffe+9+QJUemtNjuCwvHSnOUeYEwgSHV+88Ne0g=
 k8s.io/apimachinery v0.23.9/go.mod h1:BEuFMMBaIbcOqVIJqNZJXGFTP4W6AycEpb5+m/97hrM=
 k8s.io/apiserver v0.19.2/go.mod h1:FreAq0bJ2vtZFj9Ago/X0oNGC51GfubKK/ViOKfVAOA=
-k8s.io/apiserver v0.23.8/go.mod h1:/WVQG619hfUCgp3pazpqVspyC83NJemH5B0qxlZjNpw=
 k8s.io/apiserver v0.23.9 h1:SyqC4vrduSS9hkWLHww8kuFL9mVRal798kRHwGHESRM=
 k8s.io/apiserver v0.23.9/go.mod h1:vIXpgCnHep34bP/y+wGhYdn1NgxAvWtntxfEjst0e74=
 k8s.io/client-go v0.15.7/go.mod h1:QMNB76d3lKPvPQdOOnnxUF693C3hnCzUbC2umg70pWA=
 k8s.io/client-go v0.19.2/go.mod h1:S5wPhCqyDNAlzM9CnEdgTGV4OqhsW3jGO1UM1epwfJA=
 k8s.io/client-go v0.19.7/go.mod h1:iytGI7S3kmv6bWnn+bSQUE4VlrEi4YFssvVB7J7Hvqg=
 k8s.io/client-go v0.22.5/go.mod h1:cs6yf/61q2T1SdQL5Rdcjg9J1ElXSwbjSrW2vFImM4Y=
-k8s.io/client-go v0.23.8/go.mod h1:m9GosuTn6NlShCjb5XvrRlNDuHeIbGrsDLxjDoS6gec=
 k8s.io/client-go v0.23.9 h1:OKxNCL+nhw7UBB5b01OVuAV4Db/AdBdaV6/GYpucuOw=
 k8s.io/client-go v0.23.9/go.mod h1:sNo0X0MZqo4Uu0qDY5Fl5Y60cJFinBDWWUBOAM5JUCM=
 k8s.io/code-generator v0.19.2/go.mod h1:moqLn7w0t9cMs4+5CQyxnfA/HV8MF6aAVENF+WZZhgk=
-k8s.io/code-generator v0.23.8/go.mod h1:S0Q1JVA+kSzTI1oUvbKAxZY/DYbA/ZUb4Uknog12ETk=
 k8s.io/code-generator v0.23.9 h1:nCvstXOrfWAbxcmrUzJ+TpRsbPHh2XVDNKxqzkH/nH4=
 k8s.io/code-generator v0.23.9/go.mod h1:S0Q1JVA+kSzTI1oUvbKAxZY/DYbA/ZUb4Uknog12ETk=
 k8s.io/component-base v0.19.2/go.mod h1:g5LrsiTiabMLZ40AR6Hl45f088DevyGY+cCE2agEIVo=
-k8s.io/component-base v0.23.8/go.mod h1:rCj6EeaYLsNneVoFuSPL/AlEWmomc39j9M9i4NpR8r0=
 k8s.io/component-base v0.23.9/go.mod h1:WUNtIRIMd9WBS2r5LCSNZoh6f/Uh+8O+aGuZUG5t428=
 k8s.io/gengo v0.0.0-20190128074634-0689ccc1d7d6/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/gengo v0.0.0-20200428234225-8167cfdcfc14/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/gengo v0.0.0-20201203183100-97869a43a9d9/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/gengo v0.0.0-20210813121822-485abfe95c7c/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
-k8s.io/gengo v0.0.0-20220307231824-4627b89bbf1b/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/gengo v0.0.0-20220613173612-397b4ae3bce7 h1:RGb68G3yotdQggcyenx9y0+lnVJCXXcLa6geXOMlf5o=
 k8s.io/gengo v0.0.0-20220613173612-397b4ae3bce7/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/klog v0.0.0-20181102134211-b9b56d5dfc92/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
@@ -1553,7 +1545,6 @@ k8s.io/klog/v2 v2.2.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
 k8s.io/klog/v2 v2.9.0/go.mod h1:hy9LJ/NvuK+iVyP4Ehqva4HxZG/oXyIS3n3Jmire4Ec=
 k8s.io/klog/v2 v2.30.0/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
 k8s.io/klog/v2 v2.40.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
-k8s.io/klog/v2 v2.60.1-0.20220317184644-43cc75f9ae89/go.mod h1:N3kgBtsFxMb4nQ0eBDgbHEt/dtxBuTkSFQ+7K5OUoz4=
 k8s.io/klog/v2 v2.70.2-0.20220707122935-0990e81f1a8f h1:dltw7bAn8bCrQ2CmzzhgoieUZEbWqrvIGVdHGioP5nY=
 k8s.io/klog/v2 v2.70.2-0.20220707122935-0990e81f1a8f/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
 k8s.io/kube-openapi v0.0.0-20190228160746-b3a7cee44a30/go.mod h1:BXM9ceUBTj2QnfH2MK1odQs778ajze1RxcmP6S8RVVc=
@@ -1570,28 +1561,21 @@ k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20211116205334-6203023598ed/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9 h1:HNSDgDCrr/6Ly3WEGKZftiE7IY19Vz2GdbOCyI4qqhc=
 k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20220728062837-75cf857da76a/go.mod h1:+5+qcOXiZN0gs7a8a4NaTJd5sAMNipzD41MNzxRKrQs=
-knative.dev/eventing v0.33.1-0.20220728144837-15dd7ca8c811/go.mod h1:rxOHyuk8ssl+ymaSpeA1ZsRJcG6gSlgje2+di6LoFmg=
-knative.dev/eventing v0.33.1-0.20220817092547-cb3901fa91f2 h1:15CD9JuxGozvC85ac6XdApu0mdeQYdtMohpyBnp9O1U=
-knative.dev/eventing v0.33.1-0.20220817092547-cb3901fa91f2/go.mod h1:SkjRpdGliaVo3lD171ZYSFRqk5+fnIE0p8cXrJsScB0=
-knative.dev/eventing-kafka v0.33.1-0.20220817023247-6b51416d16cb h1:ObQB8+nAP9wrtpmB7Woj9MDX7hPMFK/rMMQzbWQ5yAk=
-knative.dev/eventing-kafka v0.33.1-0.20220817023247-6b51416d16cb/go.mod h1:TuF7LsYUU/YzZmdGEITGJLPd/2L/mxHY6LbrqRbUMIA=
-knative.dev/hack v0.0.0-20220701014203-65c463ac8c98/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20220725145124-782bbaabb8a1/go.mod h1:t/azP8I/Cygaw+87O7rkAPrNRjCelmtfSzWzu/9TM7I=
-knative.dev/hack v0.0.0-20220728013938-9dabf7cf62e3/go.mod h1:t/azP8I/Cygaw+87O7rkAPrNRjCelmtfSzWzu/9TM7I=
+knative.dev/control-protocol v0.0.0-20220818153549-f18dbde7d9bd/go.mod h1:vO3Xc0k0h6fFVsVG9kNMUMcVKG7MAx7jMbZDvgSuzwI=
+knative.dev/eventing v0.33.1-0.20220818050948-6eb4c785ae43/go.mod h1:+kn5/HpGP1fXAai69THMxfkm7L33Z19pcyUWOpWZi2Y=
+knative.dev/eventing v0.33.1-0.20220822135655-5ee615866b28 h1:nupMPxXvdrpahgnp9LnU12Przw4u6rGc0pzCsDDSjX4=
+knative.dev/eventing v0.33.1-0.20220822135655-5ee615866b28/go.mod h1:+kn5/HpGP1fXAai69THMxfkm7L33Z19pcyUWOpWZi2Y=
+knative.dev/eventing-kafka v0.33.1-0.20220822070354-97a8afd1af90 h1:NaYfH9qxVea0kCMA3ZH9QM6jvbGZHXbouK6GM+lqNtA=
+knative.dev/eventing-kafka v0.33.1-0.20220822070354-97a8afd1af90/go.mod h1:kIbGKy5CIAscoPHpiO0laSvQtWLH3WBRw7hW3teYk0M=
 knative.dev/hack v0.0.0-20220815132133-e9a8475f4329 h1:/X969GuiLDMgb+tyx5qZDLxGhENkS3OLPj/VCSW+Ul0=
 knative.dev/hack v0.0.0-20220815132133-e9a8475f4329/go.mod h1:t/azP8I/Cygaw+87O7rkAPrNRjCelmtfSzWzu/9TM7I=
-knative.dev/hack/schema v0.0.0-20220701014203-65c463ac8c98/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20220705130606-e60d250dc637/go.mod h1:MxsV3ZGdrH87zv4O9HtxzbNwmiKcGwEP7UdriLINSn0=
-knative.dev/pkg v0.0.0-20220728013737-449f7e6dab63/go.mod h1:KBzRmnxrSjsJ17ktIYzfdbCs0HrqmY0MYpAOM2b8XNE=
-knative.dev/pkg v0.0.0-20220802185824-a01dfedb0486/go.mod h1:nBMKMJvyoaJdkpUrjwLVs/DwaP6d73R3UkXK6lblJyE=
+knative.dev/hack/schema v0.0.0-20220815132133-e9a8475f4329/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
+knative.dev/pkg v0.0.0-20220816153547-f78a00694307/go.mod h1:YLjXbkQLlGHok+u0FLfMbBHFzY9WGu3GHhnrptoAy8I=
 knative.dev/pkg v0.0.0-20220818004048-4a03844c0b15 h1:GNmzHVaUo3zoi/wtIN71LPQaWy6DdoYzmb+GIq2s4fw=
 knative.dev/pkg v0.0.0-20220818004048-4a03844c0b15/go.mod h1:YLjXbkQLlGHok+u0FLfMbBHFzY9WGu3GHhnrptoAy8I=
-knative.dev/reconciler-test v0.0.0-20220722111920-0cfde63b455a/go.mod h1:Awq+u0SA8Cr6JAqef7JxGH1AQk73bllmugV83a60r+E=
-knative.dev/reconciler-test v0.0.0-20220728013938-124ae0741d42/go.mod h1:Awq+u0SA8Cr6JAqef7JxGH1AQk73bllmugV83a60r+E=
-knative.dev/reconciler-test v0.0.0-20220803140519-da2ce0e66cb9/go.mod h1:/RUSXB2r9cUregJDlSTxKxqDMoeKPi93zq2AoAvUjqo=
-knative.dev/reconciler-test v0.0.0-20220816163747-93a10831a971 h1:ShhVNhRaV1NNm1kGl3wYclzapbLtBYqJJkajaa1/XPU=
 knative.dev/reconciler-test v0.0.0-20220816163747-93a10831a971/go.mod h1:YQhGT2cxDyHMbVniFDDt7/Oq/6vgOEYh5Utnh/kmavY=
+knative.dev/reconciler-test v0.0.0-20220818122349-177f8264c28c h1:wWtcZ1ZyP+mXx4xQmcKzKyXkk/t07/iSq/jqhqxoyCM=
+knative.dev/reconciler-test v0.0.0-20220818122349-177f8264c28c/go.mod h1:A437yxlDVDVKQv779WlB9Nj9lWAMoOKHQXFXls24Sps=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/vendor/knative.dev/eventing/cmd/event_display/main.go b/vendor/knative.dev/eventing/cmd/event_display/main.go
index 3b423038f..e18adfbd9 100644
--- a/vendor/knative.dev/eventing/cmd/event_display/main.go
+++ b/vendor/knative.dev/eventing/cmd/event_display/main.go
@@ -75,9 +75,11 @@ func run(ctx context.Context) {
 	if err != nil {
 		log.Printf("Failed to read tracing config, using the no-op default: %v", err)
 	}
-	if err := tracing.SetupStaticPublishing(zap.L().Sugar(), "", conf); err != nil {
+	tracer, err := tracing.SetupPublishingWithStaticConfig(zap.L().Sugar(), "", conf)
+	if err != nil {
 		log.Fatalf("Failed to initialize tracing: %v", err)
 	}
+	defer tracer.Shutdown(context.Background())
 
 	if err := c.StartReceiver(ctx, display); err != nil {
 		log.Fatal("Error during receiver's runtime: ", err)
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount/serviceaccount.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount/serviceaccount.go
new file mode 100644
index 000000000..4d9bbd809
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount/serviceaccount.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2022 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 serviceaccount
+
+import (
+	context "context"
+
+	apicorev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	v1 "k8s.io/client-go/informers/core/v1"
+	kubernetes "k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/listers/core/v1"
+	cache "k8s.io/client-go/tools/cache"
+	client "knative.dev/pkg/client/injection/kube/client"
+	factory "knative.dev/pkg/client/injection/kube/informers/factory"
+	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.Core().V1().ServiceAccounts()
+	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.ServiceAccountInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch k8s.io/client-go/informers/core/v1.ServiceAccountInformer from context.")
+	}
+	return untyped.(v1.ServiceAccountInformer)
+}
+
+type wrapper struct {
+	client kubernetes.Interface
+
+	namespace string
+
+	resourceVersion string
+}
+
+var _ v1.ServiceAccountInformer = (*wrapper)(nil)
+var _ corev1.ServiceAccountLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apicorev1.ServiceAccount{}, 0, nil)
+}
+
+func (w *wrapper) Lister() corev1.ServiceAccountLister {
+	return w
+}
+
+func (w *wrapper) ServiceAccounts(namespace string) corev1.ServiceAccountNamespaceLister {
+	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 []*apicorev1.ServiceAccount, err error) {
+	lo, err := w.client.CoreV1().ServiceAccounts(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) (*apicorev1.ServiceAccount, error) {
+	return w.client.CoreV1().ServiceAccounts(w.namespace).Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/rbac/v1/clusterrolebinding/clusterrolebinding.go b/vendor/knative.dev/pkg/client/injection/kube/informers/rbac/v1/clusterrolebinding/clusterrolebinding.go
new file mode 100644
index 000000000..7c2081d55
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/rbac/v1/clusterrolebinding/clusterrolebinding.go
@@ -0,0 +1,110 @@
+/*
+Copyright 2022 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 clusterrolebinding
+
+import (
+	context "context"
+
+	apirbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	v1 "k8s.io/client-go/informers/rbac/v1"
+	kubernetes "k8s.io/client-go/kubernetes"
+	rbacv1 "k8s.io/client-go/listers/rbac/v1"
+	cache "k8s.io/client-go/tools/cache"
+	client "knative.dev/pkg/client/injection/kube/client"
+	factory "knative.dev/pkg/client/injection/kube/informers/factory"
+	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.Rbac().V1().ClusterRoleBindings()
+	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.ClusterRoleBindingInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch k8s.io/client-go/informers/rbac/v1.ClusterRoleBindingInformer from context.")
+	}
+	return untyped.(v1.ClusterRoleBindingInformer)
+}
+
+type wrapper struct {
+	client kubernetes.Interface
+
+	resourceVersion string
+}
+
+var _ v1.ClusterRoleBindingInformer = (*wrapper)(nil)
+var _ rbacv1.ClusterRoleBindingLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apirbacv1.ClusterRoleBinding{}, 0, nil)
+}
+
+func (w *wrapper) Lister() rbacv1.ClusterRoleBindingLister {
+	return w
+}
+
+// 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 []*apirbacv1.ClusterRoleBinding, err error) {
+	lo, err := w.client.RbacV1().ClusterRoleBindings().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) (*apirbacv1.ClusterRoleBinding, error) {
+	return w.client.RbacV1().ClusterRoleBindings().Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 9bd7d14c7..c43d1ff38 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1284,7 +1284,7 @@ k8s.io/utils/internal/third_party/forked/golang/net
 k8s.io/utils/net
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.33.1-0.20220817092547-cb3901fa91f2
+# knative.dev/eventing v0.33.1-0.20220822135655-5ee615866b28
 ## explicit; go 1.18
 knative.dev/eventing/cmd/event_display
 knative.dev/eventing/pkg/apis/config
@@ -1435,7 +1435,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/eventing-kafka v0.33.1-0.20220817023247-6b51416d16cb
+# knative.dev/eventing-kafka v0.33.1-0.20220822070354-97a8afd1af90
 ## explicit; go 1.16
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1539,8 +1539,10 @@ knative.dev/pkg/client/injection/kube/informers/core/v1/secret
 knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/service
 knative.dev/pkg/client/injection/kube/informers/core/v1/service/fake
+knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount
 knative.dev/pkg/client/injection/kube/informers/factory
 knative.dev/pkg/client/injection/kube/informers/factory/fake
+knative.dev/pkg/client/injection/kube/informers/rbac/v1/clusterrolebinding
 knative.dev/pkg/codegen/cmd/injection-gen
 knative.dev/pkg/codegen/cmd/injection-gen/args
 knative.dev/pkg/codegen/cmd/injection-gen/generators
@@ -1611,7 +1613,7 @@ knative.dev/pkg/webhook/json
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20220816163747-93a10831a971
+# knative.dev/reconciler-test v0.0.0-20220818122349-177f8264c28c
 ## explicit; go 1.18
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/.github/workflows/kind-e2e.yaml b/.github/workflows/kind-e2e.yaml
index 366a29592..628f35ce7 100644
--- a/.github/workflows/kind-e2e.yaml
+++ b/.github/workflows/kind-e2e.yaml
@@ -25,6 +25,7 @@ jobs:
 
         broker-class:
           - Kafka
+          - KafkaNamespaced
 
         test-suite:
           - ./test/e2e
@@ -49,6 +50,20 @@ jobs:
           - test-suite: ./test/e2e_channel
             extra-test-flags: -channels=messaging.knative.dev/v1beta1:KafkaChannel
 
+        exclude:
+          - test-suite: ./test/e2e
+            broker-class: KafkaNamespaced
+          - test-suite: ./test/e2e_sink
+            broker-class: KafkaNamespaced
+          - test-suite: ./test/e2e_source
+            broker-class: KafkaNamespaced
+          - test-suite: ./test/e2e_channel
+            broker-class: KafkaNamespaced
+          - test-suite: ./test/e2e_new_channel
+            broker-class: KafkaNamespaced
+          - test-suite: ./test/experimental
+            broker-class: KafkaNamespaced
+
     env:
       KO_DOCKER_REPO: localhost:5001
       REGISTRY_PORT: 5001
diff --git a/control-plane/pkg/reconciler/base/reconciler_test.go b/control-plane/pkg/reconciler/base/reconciler_test.go
index decde42c3..2b5af4bc0 100644
--- a/control-plane/pkg/reconciler/base/reconciler_test.go
+++ b/control-plane/pkg/reconciler/base/reconciler_test.go
@@ -284,7 +284,10 @@ func addRunningPod(store cache.Store, kc kubernetes.Interface, label string) {
 			Namespace: "ns",
 			Labels:    map[string]string{"app": label},
 		},
-		Status: corev1.PodStatus{Phase: corev1.PodRunning},
+		Status: corev1.PodStatus{
+			Phase:      corev1.PodRunning,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
+		},
 	}
 
 	if err := store.Add(pod); err != nil {
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 87be93963..239f090e8 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -262,6 +262,10 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					[]string{"my-not-present-topic"}, "my-not-present-topic",
 				),
 			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
@@ -542,6 +546,10 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					BrokerTopic(), createTopicError,
 				),
 			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
@@ -1298,6 +1306,10 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					fmt.Sprintf(`failed to get contract configuration: failed to get configmap %s/%s: configmap %q not found`, ConfigMapNamespace, ConfigMapName, ConfigMapName),
 				),
 			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
@@ -1348,6 +1360,10 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					"failed to get contract configuration: supported config Kind: ConfigMap - got Pod",
 				),
 			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
@@ -1496,6 +1512,10 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					fmt.Sprintf("%s: %s", base.ReasonDataPlaneNotAvailable, base.MessageDataPlaneNotAvailable),
 				),
 			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName, nil),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index cea56d912..54df4d70d 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -303,7 +303,8 @@ func BrokerDispatcherPod(namespace string, annotations map[string]string) runtim
 			},
 		},
 		Status: corev1.PodStatus{
-			Phase: corev1.PodRunning,
+			Phase:      corev1.PodRunning,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
 		},
 	}
 }
@@ -323,7 +324,8 @@ func BrokerReceiverPod(namespace string, annotations map[string]string) runtime.
 			},
 		},
 		Status: corev1.PodStatus{
-			Phase: corev1.PodRunning,
+			Phase:      corev1.PodRunning,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
 		},
 	}
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index 2982313c9..b1937ed6d 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -135,7 +135,8 @@ func ChannelReceiverPod(namespace string, annotations map[string]string) runtime
 			},
 		},
 		Status: corev1.PodStatus{
-			Phase: corev1.PodRunning,
+			Phase:      corev1.PodRunning,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
 		},
 	}
 }
@@ -155,7 +156,8 @@ func ChannelDispatcherPod(namespace string, annotations map[string]string) runti
 			},
 		},
 		Status: corev1.PodStatus{
-			Phase: corev1.PodRunning,
+			Phase:      corev1.PodRunning,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
 		},
 	}
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_sink.go b/control-plane/pkg/reconciler/testing/objects_sink.go
index d2cb723b0..2766d4fe2 100644
--- a/control-plane/pkg/reconciler/testing/objects_sink.go
+++ b/control-plane/pkg/reconciler/testing/objects_sink.go
@@ -150,7 +150,8 @@ func SinkReceiverPod(namespace string, annotations map[string]string) runtime.Ob
 			},
 		},
 		Status: corev1.PodStatus{
-			Phase: corev1.PodRunning,
+			Phase:      corev1.PodRunning,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
 		},
 	}
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index f3ee2ac14..a4af5bb6d 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -143,7 +143,8 @@ func SourceDispatcherPod(namespace string, annotations map[string]string) runtim
 			},
 		},
 		Status: corev1.PodStatus{
-			Phase: corev1.PodRunning,
+			Phase:      corev1.PodRunning,
+			Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}},
 		},
 	}
 }
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
index 5f4621686..d9526e895 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
@@ -184,6 +184,58 @@ public class WebClientCloudEventSenderTest {
     sender.close().onSuccess(v -> context.completeNow());
   }
 
+  @Test
+  @Timeout(value = 20000)
+  public void shouldNotRetryAndFail(final Vertx vertx, final VertxTestContext context) throws ExecutionException, InterruptedException {
+
+    final var port = 12345;
+    final var retry = 5;
+    final var event = CloudEventBuilder.v1()
+      .withId(UUID.randomUUID().toString())
+      .withSource(URI.create("/api/v1/orders"))
+      .withType("dev.knative.eventing.created")
+      .build();
+
+    final var counter = new LongAdder();
+
+    vertx.createHttpServer()
+      .requestHandler(r -> {
+        // non-retry status code
+        r.response().setStatusCode(422).end();
+      })
+      .listen(port, "localhost")
+      .toCompletionStage()
+      .toCompletableFuture()
+      .get();
+
+    final var sender = new WebClientCloudEventSender(
+      vertx,
+      WebClient.create(vertx),
+      "http://localhost:" + port,
+      DataPlaneContract.EgressConfig.newBuilder()
+        .setBackoffDelay(100L)
+        .setTimeout(100L)
+        .setBackoffPolicy(DataPlaneContract.BackoffPolicy.Linear)
+        .setRetry(retry)
+        .build()
+    );
+
+    final var success = new AtomicBoolean(true);
+
+    sender.send(event)
+      .onFailure(v -> success.set(false))
+      .onSuccess(v -> success.set(true));
+
+    await().untilFalse(success);
+    await().untilAdder(counter, is(equalTo(0L)));
+
+    // Verify that after some time counter is still equal to 0.
+    Thread.sleep(10000L);
+    await().untilAdder(counter, is(equalTo(0L)));
+
+    sender.close().onSuccess(v -> context.completeNow());
+  }
+
   @Test
   @Timeout(value = 20000)
   public void shouldTimeoutAndFail(final Vertx vertx, final VertxTestContext context) throws ExecutionException, InterruptedException {
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index 98040afcf..debfab5d1 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -28,9 +28,6 @@ if [ "${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO:-""}" != "" ]; then
   success
 fi
 
-# TODO: going to be reverted once we have the Prow changes in https://github.com/knative/test-infra
-export BROKER_CLASS="Kafka"
-
 if [[ -z "${BROKER_CLASS}" ]]; then
   fail_test "Broker class is not defined. Specify it with 'BROKER_CLASS' env var."
 else
diff --git a/test/e2e_new/dls_test.go b/test/e2e_new/dls_test.go
index 23a828981..c3733f519 100644
--- a/test/e2e_new/dls_test.go
+++ b/test/e2e_new/dls_test.go
@@ -48,6 +48,7 @@ func TestDeadLetterSink(t *testing.T) {
 		environment.Managed(t),
 	)
 
+	env.Test(ctx, t, SendsEventErrorWithoutRetries())
 	env.Test(ctx, t, SendsEventWithRetries())
 	env.Test(ctx, t, SendsEventNoRetries())
 }
@@ -112,6 +113,67 @@ func SendsEventWithRetries() *feature.Feature {
 	return f
 }
 
+func SendsEventErrorWithoutRetries() *feature.Feature {
+	f := feature.NewFeature()
+
+	sourceName := feature.MakeRandomK8sName("source")
+	sinkName := feature.MakeRandomK8sName("sink")
+	deadLetterSinkName := feature.MakeRandomK8sName("dls")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	ev := cetest.FullEvent()
+
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		broker.WithBrokerClass(broker.EnvCfg.BrokerClass),
+	))
+
+	f.Setup("broker is ready", broker.IsReady(brokerName))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName))
+
+	f.Setup("install sink", eventshub.Install(
+		sinkName,
+		eventshub.StartReceiver,
+		eventshub.DropFirstN(1),
+		eventshub.DropEventsResponseBody("bad response"),
+		eventshub.DropEventsResponseCode(422), // non-retry error
+	))
+
+	f.Setup("install dead letter sink", eventshub.Install(
+		deadLetterSinkName,
+		eventshub.StartReceiver,
+	))
+
+	f.Setup("install trigger", trigger.Install(
+		triggerName,
+		brokerName,
+		trigger.WithSubscriber(svc.AsKReference(sinkName), ""),
+		trigger.WithDeadLetterSink(svc.AsKReference(deadLetterSinkName), ""),
+	))
+	f.Setup("trigger is ready", trigger.IsReady(triggerName))
+
+	f.Setup("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(broker.GVR(), brokerName),
+		eventshub.InputEvent(ev),
+	))
+
+	f.Assert("receive event on sink",
+		OnStore(sinkName).
+			MatchEvent(cetest.HasId(ev.ID())).
+			Exact(1),
+	)
+
+	f.Assert("receive event on dead letter sink",
+		OnStore(deadLetterSinkName).
+			MatchEvent(cetest.HasId(ev.ID())).
+			Exact(1),
+	)
+
+	return f
+}
+
 func SendsEventNoRetries() *feature.Feature {
 	f := feature.NewFeature()
 
diff --git a/test/reconciler-tests.sh b/test/reconciler-tests.sh
index 758e2f32a..8b9e0d2b9 100755
--- a/test/reconciler-tests.sh
+++ b/test/reconciler-tests.sh
@@ -28,9 +28,6 @@ if [ "${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO:-""}" != "" ]; then
   success
 fi
 
-# TODO: going to be reverted once we have the Prow changes in https://github.com/knative/test-infra
-export BROKER_CLASS="Kafka"
-
 if [[ -z "${BROKER_CLASS}" ]]; then
   fail_test "Broker class is not defined. Specify it with 'BROKER_CLASS' env var."
 else
diff --git a/vendor/knative.dev/eventing/test/test_images/event-sender/main.go b/vendor/knative.dev/eventing/test/test_images/event-sender/main.go
index 99d14d78e..f46f42e03 100644
--- a/vendor/knative.dev/eventing/test/test_images/event-sender/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/event-sender/main.go
@@ -31,6 +31,7 @@ import (
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
 	"go.opencensus.io/plugin/ochttp"
 	"go.uber.org/zap"
+	"knative.dev/pkg/tracing"
 	"knative.dev/pkg/tracing/propagation/tracecontextb3"
 
 	"knative.dev/eventing/test/lib/sender"
@@ -111,12 +112,14 @@ func main() {
 	}
 
 	var c cloudevents.Client
+	var tracer tracing.Tracer
 	if addTracing {
 		log.Println("Adding tracing")
 		logger, _ := zap.NewDevelopment()
-		if err := test_images.ConfigureTracing(logger.Sugar(), ""); err != nil {
+		if tracer, err = test_images.ConfigureTracing(logger.Sugar(), ""); err != nil {
 			log.Fatalf("Unable to setup trace publishing: %v", err)
 		}
+		defer tracer.Shutdown(context.Background())
 		c, err = cloudevents.NewClient(t, client.WithObservabilityService(obsclient.New()))
 	} else {
 		c, err = cloudevents.NewClient(t)
diff --git a/vendor/knative.dev/eventing/test/test_images/recordevents/main.go b/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
index ca718ddc2..9027bb51c 100644
--- a/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
@@ -50,9 +50,11 @@ func main() {
 	ctx, _ := injection.EnableInjectionOrDie(nil, cfg)
 	ctx = test_images.ConfigureLogging(ctx, "recordevents")
 
-	if err := test_images.ConfigureTracing(logging.FromContext(ctx), ""); err != nil {
+	tracer, err := test_images.ConfigureTracing(logging.FromContext(ctx), "")
+	if err != nil {
 		logging.FromContext(ctx).Fatal("Unable to setup trace publishing", err)
 	}
+	defer tracer.Shutdown(context.Background())
 
 	var env envConfig
 	if err := envconfig.Process("", &env); err != nil {
diff --git a/vendor/knative.dev/eventing/test/test_images/utils.go b/vendor/knative.dev/eventing/test/test_images/utils.go
index 2771d2ce4..13efeeaa6 100644
--- a/vendor/knative.dev/eventing/test/test_images/utils.go
+++ b/vendor/knative.dev/eventing/test/test_images/utils.go
@@ -57,13 +57,13 @@ const (
 )
 
 // ConfigureTracing can be used in test-images to configure tracing.
-func ConfigureTracing(logger *zap.SugaredLogger, serviceName string) error {
+func ConfigureTracing(logger *zap.SugaredLogger, serviceName string) (tracing.Tracer, error) {
 	tracingEnv := os.Getenv(ConfigTracingEnv)
 	conf, err := config.JSONToTracingConfig(tracingEnv)
 	if err != nil {
 		logger.Warn("Error while trying to read the tracing config, using NoopConfig: ", err)
 	}
-	return tracing.SetupStaticPublishing(logger, serviceName, conf)
+	return tracing.SetupPublishingWithStaticConfig(logger, serviceName, conf)
 }
 
 // ConfigureLogging can be used in test-images to configure logging.
diff --git a/vendor/knative.dev/eventing/test/test_images/wathola-forwarder/main.go b/vendor/knative.dev/eventing/test/test_images/wathola-forwarder/main.go
index 5f45b2be6..348787857 100644
--- a/vendor/knative.dev/eventing/test/test_images/wathola-forwarder/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/wathola-forwarder/main.go
@@ -22,4 +22,5 @@ var instance forwarder.Forwarder
 func main() {
 	instance = forwarder.New()
 	instance.Forward()
+	forwarder.Shutdown()
 }
diff --git a/vendor/knative.dev/eventing/test/test_images/wathola-receiver/main.go b/vendor/knative.dev/eventing/test/test_images/wathola-receiver/main.go
index 719946cf3..da7bea927 100644
--- a/vendor/knative.dev/eventing/test/test_images/wathola-receiver/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/wathola-receiver/main.go
@@ -15,11 +15,14 @@
 
 package main
 
-import "knative.dev/eventing/test/upgrade/prober/wathola/receiver"
+import (
+	"knative.dev/eventing/test/upgrade/prober/wathola/receiver"
+)
 
 var instance receiver.Receiver
 
 func main() {
 	instance = receiver.New()
 	instance.Receive()
+	receiver.Shutdown()
 }
diff --git a/vendor/knative.dev/eventing/test/test_images/wathola-sender/main.go b/vendor/knative.dev/eventing/test/test_images/wathola-sender/main.go
index 21e97c42f..8016a9a3f 100644
--- a/vendor/knative.dev/eventing/test/test_images/wathola-sender/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/wathola-sender/main.go
@@ -15,8 +15,12 @@
 
 package main
 
-import "knative.dev/eventing/test/upgrade/prober/wathola/sender"
+import (
+	"knative.dev/eventing/test/upgrade/prober/wathola/sender"
+)
 
 func main() {
-	sender.New().SendContinually()
+	s := sender.New()
+	s.SendContinually()
+	sender.Shutdown()
 }
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go b/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
index 94259d787..ab7dbd0aa 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/configuration.go
@@ -180,17 +180,11 @@ func (p *prober) compileTemplate(templateName string, endpoint interface{}, trac
 	var buff bytes.Buffer
 	data := struct {
 		*Config
-		// Deprecated: use ForwarderTarget
-		Namespace string
-		// Deprecated: use Endpoint
-		BrokerURL       string
 		Endpoint        interface{}
 		TracingConfig   string
 		ForwarderTarget string
 	}{
 		p.config,
-		p.client.Namespace,
-		fmt.Sprintf("%v", endpoint),
 		endpoint,
 		tracingConfig,
 		fmt.Sprintf(forwarderTargetFmt, p.client.Namespace),
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/tracing.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/tracing.go
index 79f5adbc5..20f6839ec 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/tracing.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/config/tracing.go
@@ -17,17 +17,29 @@ limitations under the License.
 package config
 
 import (
+	"context"
+
 	"go.uber.org/zap"
 	"knative.dev/pkg/tracing"
 	tracingconfig "knative.dev/pkg/tracing/config"
 )
 
+var tracer tracing.Tracer
+
 func SetupTracing() {
 	config, err := tracingconfig.JSONToTracingConfig(Instance.TracingConfig)
 	if err != nil {
 		Log.Warn("Tracing configuration is invalid, using the no-op default", zap.Error(err))
 	}
-	if err = tracing.SetupStaticPublishing(Log, "", config); err != nil {
+	if tracer, err = tracing.SetupPublishingWithStaticConfig(Log, "", config); err != nil {
 		Log.Fatal("Error setting up trace publishing", zap.Error(err))
 	}
 }
+
+func ShutdownTracing() {
+	if tracer != nil {
+		if err := tracer.Shutdown(context.Background()); err != nil {
+			Log.Warn("Failed to shutdown tracing")
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/forwarder/services.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/forwarder/services.go
index 582154234..1a86445a7 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/forwarder/services.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/forwarder/services.go
@@ -47,6 +47,10 @@ func New() Forwarder {
 	return f
 }
 
+func Shutdown() {
+	config.ShutdownTracing()
+}
+
 func (f *forwarder) Forward() {
 	port := config.Instance.Forwarder.Port
 	client.Receive(port, Canceling, f.forwardEvent)
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/receiver/services.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/receiver/services.go
index eec2a0e25..7a4861492 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/receiver/services.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/receiver/services.go
@@ -50,6 +50,10 @@ func New() Receiver {
 	return r
 }
 
+func Shutdown() {
+	config.ShutdownTracing()
+}
+
 func (r receiver) Receive() {
 	port := config.Instance.Receiver.Port
 	client.Receive(port, Canceling, r.receiveEvent, r.reportMiddleware)
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/operations.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/operations.go
index 259d375dd..8cbc03a98 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/operations.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/operations.go
@@ -31,6 +31,10 @@ func New() Sender {
 	}
 }
 
+func Shutdown() {
+	config.ShutdownTracing()
+}
+
 // NewEventID creates new event ID
 func NewEventID() string {
 	return randString(16)
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/services.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/services.go
index febb4115e..7827e12ad 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/services.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/services.go
@@ -48,7 +48,6 @@ var (
 		"supported by any registered event sender")
 	log                     = config.Log
 	senderConfig            = &config.Instance.Sender
-	eventSenders            = make([]EventSender, 0, 1)
 	eventSendersWithContext = make([]EventSenderWithContext, 0, 1)
 )
 
@@ -65,9 +64,6 @@ func (s *sender) SendContinually() {
 	var shutdownCh = make(chan struct{})
 	defer func() {
 		s.sendFinished()
-		// Give time to send tracing information.
-		// https://github.com/knative/pkg/issues/2475
-		time.Sleep(5 * time.Second)
 	}()
 
 	go func() {
@@ -143,16 +139,9 @@ func NewCloudEvent(data interface{}, typ string) cloudevents.Event {
 
 // ResetEventSenders will reset configured event senders to defaults.
 func ResetEventSenders() {
-	eventSenders = make([]EventSender, 0, 1)
 	eventSendersWithContext = make([]EventSenderWithContext, 0, 1)
 }
 
-// RegisterEventSender will register a EventSender to be used.
-// Deprecated. Use RegisterEventSenderWithContext.
-func RegisterEventSender(es EventSender) {
-	eventSenders = append(eventSenders, es)
-}
-
 // RegisterEventSenderWithContext will register EventSenderWithContext to be used.
 func RegisterEventSenderWithContext(es EventSenderWithContext) {
 	eventSendersWithContext = append(eventSendersWithContext, es)
@@ -162,7 +151,7 @@ func RegisterEventSenderWithContext(es EventSenderWithContext) {
 func SendEvent(ctx context.Context, ce cloudevents.Event, endpoint interface{}) error {
 	sendersWithCtx := make([]EventSenderWithContext, 0, len(eventSendersWithContext)+1)
 	sendersWithCtx = append(sendersWithCtx, eventSendersWithContext...)
-	if len(eventSendersWithContext) == 0 && len(eventSenders) == 0 {
+	if len(eventSendersWithContext) == 0 {
 		sendersWithCtx = append(sendersWithCtx, httpSender{})
 	}
 	for _, eventSender := range sendersWithCtx {
@@ -170,15 +159,6 @@ func SendEvent(ctx context.Context, ce cloudevents.Event, endpoint interface{})
 			return eventSender.SendEventWithContext(ctx, ce, endpoint)
 		}
 	}
-	// Backwards compatibility.
-	// TODO: Remove when downstream repositories start using EventSenderWithContext.
-	senders := make([]EventSender, 0, len(eventSenders)+1)
-	senders = append(senders, eventSenders...)
-	for _, eventSender := range senders {
-		if eventSender.Supports(endpoint) {
-			return eventSender.SendEvent(ce, endpoint)
-		}
-	}
 	return fmt.Errorf("%w: endpoint is %#v", ErrEndpointTypeNotSupported, endpoint)
 }
 
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/types.go b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/types.go
index bdf94eb75..1bbae5c23 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/types.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/wathola/sender/types.go
@@ -33,14 +33,6 @@ type EndpointSupporter interface {
 	Supports(endpoint interface{}) bool
 }
 
-// EventSender will be used to send events to configured endpoint.
-// Deprecated. Use EventSenderWithContext.
-type EventSender interface {
-	EndpointSupporter
-	// SendEvent will send event to given endpoint.
-	SendEvent(ce cloudevents.Event, endpoint interface{}) error
-}
-
 // EventSenderWithContext will be used to send events to configured endpoint, passing a context.
 type EventSenderWithContext interface {
 	EndpointSupporter