Repository: knative-sandbox/eventing-kafka-broker
Pull request #999 (2021-06-15 19:35:03)

Improve data-plane pods annotation update logic
There could be situations where our reconciler failed to update a pod annotation
in a previous reconcile and since there might be multiple pods running we
want to make sure that all pods have the same annotation.

This patch moves the logic to check that we're not performing an
unnecessary update down to when we're trying to update a single pod
annotation.

Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

<!-- Please include the 'why' behind your changes if no issue exists -->


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/999#" 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-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/base/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/999/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1404885402275287040/artifacts/line-cov.html#file0) | 80.5% | 76.8% | -3.7
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/999/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1404885402275287040/artifacts/line-cov.html#file1) | 87.3% | 86.2% | -1.1
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/999/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1404885402275287040/artifacts/line-cov.html#file2) | 83.5% | 82.6% | -1.0

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#999](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (5528958) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/3ec1fab32048a0fbb71e62b123f78392e258954d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (3ec1fab) will **increase** coverage by `4.02%`.
> The diff coverage is `n/a`.

> :exclamation: Current head 5528958 differs from pull request most recent head 21778e7. Consider uploading reports for the commit 21778e7 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/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/999?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main     #999      +/-   ##
============================================
+ Coverage     75.60%   79.62%   +4.02%     
- Complexity      417      419       +2     
============================================
  Files            73       44      -29     
  Lines          2906     1561    -1345     
  Branches        124      124              
============================================
- Hits           2197     1243     -954     
+ Misses          556      255     -301     
+ Partials        153       63      -90     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `79.62% <ø> (+0.19%)` | :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/999?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | | |
| [control-plane/pkg/reconciler/sink/auth\_provider.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2F1dGhfcHJvdmlkZXIuZ28=) | | |
| [control-plane/pkg/receiver/path.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjZWl2ZXIvcGF0aC5nbw==) | | |
| [control-plane/pkg/security/config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvY29uZmlnLmdv) | | |
| [control-plane/pkg/reconciler/sink/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2NvbnRyb2xsZXIuZ28=) | | |
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | |
| [control-plane/pkg/reconciler/kafka/broker\_class.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9icm9rZXJfY2xhc3MuZ28=) | | |
| [...-plane/pkg/reconciler/trigger/trigger\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXJfbGlmZWN5Y2xlLmdv) | | |
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | | |
| [control-plane/pkg/security/secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0Lmdv) | | |
| ... and [18 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/999/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

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

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/base/reconciler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/999/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1404893218079772672/artifacts/line-cov.html#file0) | 80.5% | 76.8% | -3.7
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/999/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1404893218079772672/artifacts/line-cov.html#file1) | 87.3% | 86.2% | -1.1
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/999/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1404893218079772672/artifacts/line-cov.html#file2) | 83.5% | 82.6% | -1.0


Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 8c287a123..0d120a04c 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -54,6 +54,7 @@ aliases:
   - aliok
   - davyodom
   - devguyio
+  - eric-sap
   - lberk
   - matzew
   - phamilton
@@ -194,7 +195,9 @@ aliases:
   operations-wg-leads:
   - houshengbo
   operations-writers:
+  - dprotaso
   - houshengbo
+  - upodroid
   productivity-wg-leads:
   - chizhg
   - n3wscott
diff --git a/control-plane/config/sink/100-webhook-cluster-role.yaml b/control-plane/config/sink/100-webhook-cluster-role.yaml
index d827054c0..6edd095cd 100644
--- a/control-plane/config/sink/100-webhook-cluster-role.yaml
+++ b/control-plane/config/sink/100-webhook-cluster-role.yaml
@@ -81,3 +81,12 @@ rules:
     resources:
       - "leases"
     verbs: *everything
+
+  # finalizers are needed for the owner reference of the webhook
+  - apiGroups:
+      - ""
+    resources:
+      - "namespaces/finalizers"
+    verbs:
+      - "update"
+
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index f8a0bc2b9..d60b53057 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -19,6 +19,7 @@ package base
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -29,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/util/retry"
 	"knative.dev/pkg/tracker"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -196,31 +196,19 @@ func (r *Reconciler) UpdateDataPlaneConfigMap(ctx context.Context, contract *con
 }
 
 func (r *Reconciler) UpdateDispatcherPodsAnnotation(ctx context.Context, logger *zap.Logger, volumeGeneration uint64) error {
-
-	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-
-		labelSelector := r.dispatcherSelector()
-		pods, errors := r.PodLister.Pods(r.SystemNamespace).List(labelSelector)
-		if errors != nil {
-			return fmt.Errorf("failed to list dispatcher pods in namespace %s: %w", r.SystemNamespace, errors)
-		}
-
-		return r.updatePodsAnnotation(ctx, logger, "dispatcher", volumeGeneration, pods)
-	})
+	pods, errors := r.PodLister.Pods(r.SystemNamespace).List(r.dispatcherSelector())
+	if errors != nil {
+		return fmt.Errorf("failed to list dispatcher pods in namespace %s: %w", r.SystemNamespace, errors)
+	}
+	return r.updatePodsAnnotation(ctx, logger, "dispatcher", volumeGeneration, pods)
 }
 
 func (r *Reconciler) UpdateReceiverPodsAnnotation(ctx context.Context, logger *zap.Logger, volumeGeneration uint64) error {
-
-	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-
-		labelSelector := r.receiverSelector()
-		pods, errors := r.PodLister.Pods(r.SystemNamespace).List(labelSelector)
-		if errors != nil {
-			return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.SystemNamespace, errors)
-		}
-
-		return r.updatePodsAnnotation(ctx, logger, "receiver", volumeGeneration, pods)
-	})
+	pods, errors := r.PodLister.Pods(r.SystemNamespace).List(r.receiverSelector())
+	if errors != nil {
+		return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.SystemNamespace, errors)
+	}
+	return r.updatePodsAnnotation(ctx, logger, "receiver", volumeGeneration, pods)
 }
 
 func (r *Reconciler) updatePodsAnnotation(ctx context.Context, logger *zap.Logger, component string, volumeGeneration uint64, pods []*corev1.Pod) error {
@@ -243,6 +231,15 @@ func (r *Reconciler) updatePodsAnnotation(ctx context.Context, logger *zap.Logge
 			annotations = make(map[string]string, 1)
 		}
 
+		// Check whether pod's annotation is the expected one.
+		if v, ok := annotations[VolumeGenerationAnnotationKey]; ok {
+			v, err := strconv.ParseUint(v /* base */, 10 /* bitSize */, 64)
+			if err == nil && v == volumeGeneration {
+				// Volume generation already matches the expected volume generation number.
+				continue
+			}
+		}
+
 		annotations[VolumeGenerationAnnotationKey] = fmt.Sprint(volumeGeneration)
 		pod.SetAnnotations(annotations)
 
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 989019c0c..cc4c87cb5 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -145,10 +145,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		return statusConditionManager.FailedToGetDataFromConfigMap(err)
 	}
 
-	logger.Debug(
-		"Got contract data from config map",
-		zap.Any(base.ContractLogKey, ct),
-	)
+	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
 
 	// Get resource configuration.
 	brokerResource, err := r.getBrokerResource(ctx, topic, broker, secret, topicConfig)
@@ -162,9 +159,10 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 
 	logger.Debug("Change detector", zap.Int("changed", changed))
 
-	coreconfig.IncrementContractGeneration(ct)
-
 	if changed == coreconfig.ResourceChanged {
+		// Resource changed, increment contract generation.
+		coreconfig.IncrementContractGeneration(ct)
+
 		// Update the configuration map with the new contract data.
 		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
 			logger.Error("failed to update data plane config map", zap.Error(
@@ -176,38 +174,39 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	}
 	statusConditionManager.ConfigMapUpdated()
 
-	if changed == coreconfig.ResourceChanged {
+	// We update receiver and dispatcher pods annotation regardless of our contract changed or not due to the fact
+	// that in a previous reconciliation we might have failed to update one of our data plane pod annotation, so we want
+	// to anyway update remaining annotations with the contract generation that was saved in the CM.
+
+	// We reject events to a non-existing Broker, which means that we cannot consider a Broker Ready if all
+	// receivers haven't got the Broker, so update failures to receiver pods is a hard failure.
+	// On the other side, dispatcher pods care about Triggers, and the Broker object is used as a configuration
+	// prototype for all associated Triggers, so we consider that it's fine on the dispatcher side to receive eventually
+	// the update even if here eventually means seconds or minutes after the actual update.
+
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		logger.Error("Failed to update receiver pod annotation", zap.Error(
+			statusConditionManager.FailedToUpdateReceiverPodsAnnotation(err),
+		))
+		return err
+	}
 
-		// After #37 we reject events to a non-existing Broker, which means that we cannot consider a Broker Ready if all
-		// receivers haven't got the Broker, so update failures to receiver pods is a hard failure.
-		// On the other side, dispatcher pods care about Triggers, and the Broker object is used as a configuration
-		// prototype for all associated Triggers, so we consider that it's fine on the dispatcher side to receive eventually
-		// the update even if here eventually means seconds or minutes after the actual update.
+	logger.Debug("Updated receiver pod annotation")
 
-		// Update volume generation annotation of receiver pods
-		if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-			logger.Error("Failed to update receiver pod annotation", zap.Error(
-				statusConditionManager.FailedToUpdateReceiverPodsAnnotation(err),
-			))
-			return err
-		}
+	// Update volume generation annotation of dispatcher pods
+	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
+		// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Broker
+		// ready. So, log out the error and move on to the next step.
+		logger.Warn(
+			"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
+			zap.Error(err),
+		)
 
-		logger.Debug("Updated receiver pod annotation")
-
-		// Update volume generation annotation of dispatcher pods
-		if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-			// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
-			// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Broker
-			// ready. So, log out the error and move on to the next step.
-			logger.Warn(
-				"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
-				zap.Error(err),
-			)
-
-			statusConditionManager.FailedToUpdateDispatcherPodsAnnotation(err)
-		} else {
-			logger.Debug("Updated dispatcher pod annotation")
-		}
+		statusConditionManager.FailedToUpdateDispatcherPodsAnnotation(err)
+	} else {
+		logger.Debug("Updated dispatcher pod annotation")
 	}
 
 	return statusConditionManager.Reconciled()
@@ -236,10 +235,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return fmt.Errorf("failed to get contract: %w", err)
 	}
 
-	logger.Debug(
-		"Got contract data from config map",
-		zap.Any(base.ContractLogKey, ct),
-	)
+	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
 
 	brokerIndex := coreconfig.FindResource(ct, broker.UID)
 	if brokerIndex != coreconfig.NoResource {
@@ -247,19 +243,37 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 
 		logger.Debug("Broker deleted", zap.Int("index", brokerIndex))
 
+		// Resource changed, increment contract generation.
 		coreconfig.IncrementContractGeneration(ct)
 
 		// Update the configuration map with the new contract data.
 		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
 			return err
 		}
-
 		logger.Debug("Contract config map updated")
+	}
+
+	// We update receiver and dispatcher pods annotation regardless of our contract changed or not due to the fact
+	// that in a previous reconciliation we might have failed to update one of our data plane pod annotation, so we want
+	// to update anyway remaining annotations with the contract generation that was saved in the CM.
+	// Note: if there aren't changes to be done at the pod annotation level, we just skip the update.
 
-		// There is no need to update volume generation and dispatcher pod annotation, updates to the config map will
-		// eventually be seen by the dispatcher pod and resources will be deleted accordingly.
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
+	}
+	// Update volume generation annotation of dispatcher pods
+	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
 	}
 
+	// TODO probe (as in #974) and check if status code is 404 otherwise requeue and return.
+	//  Rationale: after deleting a topic closing a producer ends up blocking and requesting metadata for max.block.ms
+	//  because topic metadata aren't available anymore.
+	// 	See (under discussions KIPs, unlikely to be accepted as they are):
+	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
+	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
+
 	topicConfig, brokerConfig, err := r.topicConfig(logger, broker)
 	if err != nil {
 		return fmt.Errorf("failed to resolve broker config: %w", err)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 0a4ebf6cd..0ff56cda0 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -166,9 +166,9 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 	// Update contract data with the new sink configuration.
 	changed := coreconfig.AddOrUpdateResourceConfig(ct, sinkConfig, sinkIndex, logger)
 
-	coreconfig.IncrementContractGeneration(ct)
-
 	if changed == coreconfig.ResourceChanged {
+		// Resource changed, increment contract generation.
+		coreconfig.IncrementContractGeneration(ct)
 		// Update the configuration map with the new contract data.
 		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
 			logger.Error("failed to update data plane config map", zap.Error(
@@ -181,18 +181,21 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 
 	logger.Debug("Config map updated")
 
-	if changed == coreconfig.ResourceChanged {
-		// After #37 we reject events to a non-existing Sink, which means that we cannot consider a Sink Ready if all
-		// receivers haven't got the Sink, so update failures to receiver pods is a hard failure.
+	// We update receiver pods annotation regardless of our contract changed or not due to the fact  that in a previous
+	// reconciliation we might have failed to update one of our data plane pod annotation, so we want to anyway update
+	// remaining annotations with the contract generation that was saved in the CM.
+	// Note: if there aren't changes to be done at the pod annotation level, we just skip the update.
 
-		// Update volume generation annotation of receiver pods
-		if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-			return err
-		}
+	// Since we reject events to a non-existing Sink, which means that we cannot consider a Sink Ready if all
+	// receivers haven't got the Sink, so update failures to receiver pods is a hard failure.
 
-		logger.Debug("Updated receiver pod annotation")
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
 	}
 
+	logger.Debug("Updated receiver pod annotation")
+
 	return statusConditionManager.Reconciled()
 }
 
@@ -230,6 +233,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 
 		logger.Debug("Sink deleted", zap.Int("index", sinkIndex))
 
+		// Resource changed, increment contract generation.
 		coreconfig.IncrementContractGeneration(ct)
 
 		// Update the configuration map with the new contract data.
@@ -240,6 +244,23 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 		logger.Debug("Sinks config map updated")
 	}
 
+	// We update receiver pods annotation regardless of our contract changed or not due to the fact  that in a previous
+	// reconciliation we might have failed to update one of our data plane pod annotation, so we want to anyway update
+	// remaining annotations with the contract generation that was saved in the CM.
+	// Note: if there aren't changes to be done at the pod annotation level, we just skip the update.
+
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
+	}
+
+	// TODO probe (as in #974) and check if status code is 404 otherwise requeue and return.
+	//  Rationale: after deleting a topic closing a producer ends up blocking and requesting metadata for max.block.ms
+	//  because topic metadata aren't available anymore.
+	// 	See (under discussions KIPs, unlikely to be accepted as they are):
+	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
+	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
+
 	if ks.GetStatus().Annotations[base.TopicOwnerAnnotation] == ControllerTopicOwner {
 		securityOption, _, err := security.NewOptionFromSecret(ctx, &SecretLocator{KafkaSink: ks}, r.SecretProviderFunc())
 		if err != nil {
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index fa028e55b..c5fcaf506 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -35,13 +35,13 @@ Lists of 153 third-party dependencies.
      (Unknown license) dispatcher (dev.knative.eventing.kafka.broker:dispatcher:1.0-SNAPSHOT - no url defined)
      (Unknown license) receiver (dev.knative.eventing.kafka.broker:receiver:1.0-SNAPSHOT - no url defined)
      (BSD) Automaton (dk.brics.automaton:automaton:1.11-8 - http://www.brics.dk/automaton/)
-     (The Apache Software License, Version 2.0) CloudEvents - API (io.cloudevents:cloudevents-api:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-api/)
-     (The Apache Software License, Version 2.0) CloudEvents - Core (io.cloudevents:cloudevents-core:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-core/)
-     (The Apache Software License, Version 2.0) CloudEvents - Vert.x Http Binding (io.cloudevents:cloudevents-http-vertx:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-http-vertx/)
-     (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
-     (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
-     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.5.2.Final - https://debezium.io/debezium-parent/debezium-api)
-     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.5.2.Final - https://debezium.io/debezium-parent/debezium-core)
+     (The Apache Software License, Version 2.0) CloudEvents - API (io.cloudevents:cloudevents-api:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-api/)
+     (The Apache Software License, Version 2.0) CloudEvents - Core (io.cloudevents:cloudevents-core:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-core/)
+     (The Apache Software License, Version 2.0) CloudEvents - Vert.x Http Binding (io.cloudevents:cloudevents-http-vertx:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-http-vertx/)
+     (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
+     (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
+     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.5.3.Final - https://debezium.io/debezium-parent/debezium-api)
+     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.5.3.Final - https://debezium.io/debezium-parent/debezium-core)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.4.1 - http://fabric8.io/kubernetes-client/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.4.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.4.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
@@ -125,7 +125,7 @@ Lists of 153 third-party dependencies.
      (Apache License, Version 2.0) Apache ZooKeeper - Server (org.apache.zookeeper:zookeeper:3.5.8 - http://zookeeper.apache.org/zookeeper)
      (Apache License, Version 2.0) Apache ZooKeeper - Jute (org.apache.zookeeper:zookeeper-jute:3.5.8 - http://zookeeper.apache.org/zookeeper-jute)
      (The Apache License, Version 2.0) org.apiguardian:apiguardian-api (org.apiguardian:apiguardian-api:1.1.0 - https://github.com/apiguardian-team/apiguardian)
-     (Apache License, Version 2.0) AssertJ fluent assertions (org.assertj:assertj-core:3.19.0 - https://assertj.github.io/doc/assertj-core/)
+     (Apache License, Version 2.0) AssertJ fluent assertions (org.assertj:assertj-core:3.20.1 - https://assertj.github.io/doc/assertj-core/)
      (Apache 2.0) Awaitility (org.awaitility:awaitility:4.1.0 - http://awaitility.org)
      (GNU General Public License, version 2 (GPL2), with the classpath exception) (The MIT License) Checker Qual (org.checkerframework:checker-compat-qual:2.5.5 - https://checkerframework.org)
      (BSD Licence 3) Hamcrest (org.hamcrest:hamcrest:2.1 - http://hamcrest.org/JavaHamcrest/)
@@ -149,7 +149,7 @@ Lists of 153 third-party dependencies.
      (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.12 - https://www.scala-lang.org/)
      (Apache-2.0) scala-collection-compat (org.scala-lang.modules:scala-collection-compat_2.12:2.2.0 - http://www.scala-lang.org/)
      (Apache-2.0) scala-java8-compat (org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 - http://www.scala-lang.org/)
-     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.30 - http://www.slf4j.org)
-     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.30 - http://www.slf4j.org)
+     (MIT License) SLF4J API Module (org.slf4j:slf4j-api:1.7.31 - http://www.slf4j.org)
+     (MIT License) SLF4J NOP Binding (org.slf4j:slf4j-nop:1.7.31 - http://www.slf4j.org)
      (Apache-2.0) snappy-java (org.xerial.snappy:snappy-java:1.1.7.7 - https://github.com/xerial/snappy-java)
      (Apache License, Version 2.0) SnakeYAML (org.yaml:snakeyaml:1.26 - http://www.snakeyaml.org)
diff --git a/data-plane/config/broker/template/500-dispatcher.yaml b/data-plane/config/broker/template/500-dispatcher.yaml
index df5b1f3c3..2b1739818 100644
--- a/data-plane/config/broker/template/500-dispatcher.yaml
+++ b/data-plane/config/broker/template/500-dispatcher.yaml
@@ -36,7 +36,6 @@ spec:
       serviceAccountName: knative-kafka-data-plane
       securityContext:
         runAsNonRoot: true
-        runAsUser: 65532
       containers:
         - name: kafka-broker-dispatcher
           image: ${KNATIVE_KAFKA_BROKER_DISPATCHER_IMAGE}
diff --git a/data-plane/config/broker/template/500-receiver.yaml b/data-plane/config/broker/template/500-receiver.yaml
index 6ab3e8f54..5ef7cc61f 100644
--- a/data-plane/config/broker/template/500-receiver.yaml
+++ b/data-plane/config/broker/template/500-receiver.yaml
@@ -36,7 +36,6 @@ spec:
       serviceAccountName: knative-kafka-data-plane
       securityContext:
         runAsNonRoot: true
-        runAsUser: 65532
       # To avoid node becoming SPOF, spread our replicas to different nodes.
       affinity:
         podAntiAffinity:
diff --git a/data-plane/config/sink/template/500-receiver.yaml b/data-plane/config/sink/template/500-receiver.yaml
index fbcec601f..edc606e69 100644
--- a/data-plane/config/sink/template/500-receiver.yaml
+++ b/data-plane/config/sink/template/500-receiver.yaml
@@ -36,7 +36,6 @@ spec:
       serviceAccountName: knative-kafka-data-plane
       securityContext:
         runAsNonRoot: true
-        runAsUser: 65532
       # To avoid node becoming SPOF, spread our replicas to different nodes.
       affinity:
         podAntiAffinity:
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
index f6b6cb087..c960fee55 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcher.java
@@ -47,7 +47,7 @@ public class RecordDispatcher implements Handler<KafkaConsumerRecord<String, Clo
 
   private final Filter filter;
   private final Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> subscriberSender;
-  private final Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> dlqSender;
+  private final Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> dlsSender;
   private final OffsetManager offsetManager;
   private final Supplier<Future<?>> closer;
   private final ConsumerTracer consumerTracer;
@@ -57,7 +57,7 @@ public class RecordDispatcher implements Handler<KafkaConsumerRecord<String, Clo
    *
    * @param filter                event filter
    * @param subscriberSender      sender to trigger subscriber
-   * @param deadLetterQueueSender sender to DLQ
+   * @param deadLetterSinkSender  sender to dead letter sink
    * @param sinkResponseHandler   handler of the response from {@code subscriberSender}
    * @param offsetManager         hook receiver {@link OffsetManager}. It allows to plug in custom offset
    * @param consumerTracer        consumer tracer
@@ -65,23 +65,23 @@ public class RecordDispatcher implements Handler<KafkaConsumerRecord<String, Clo
   public RecordDispatcher(
     final Filter filter,
     final ConsumerRecordSender subscriberSender,
-    final ConsumerRecordSender deadLetterQueueSender,
+    final ConsumerRecordSender deadLetterSinkSender,
     final SinkResponseHandler sinkResponseHandler,
     final OffsetManager offsetManager,
     final ConsumerTracer consumerTracer) {
     Objects.requireNonNull(filter, "provide filter");
     Objects.requireNonNull(subscriberSender, "provide subscriberSender");
-    Objects.requireNonNull(deadLetterQueueSender, "provide deadLetterQueueSender");
+    Objects.requireNonNull(deadLetterSinkSender, "provide deadLetterSinkSender");
     Objects.requireNonNull(offsetManager, "provide offsetStrategy");
     Objects.requireNonNull(sinkResponseHandler, "provide sinkResponseHandler");
 
     this.filter = filter;
     this.subscriberSender = composeSenderAndSinkHandler(subscriberSender, sinkResponseHandler, "subscriber");
-    this.dlqSender = composeSenderAndSinkHandler(deadLetterQueueSender, sinkResponseHandler, "dead letter queue");
+    this.dlsSender = composeSenderAndSinkHandler(deadLetterSinkSender, sinkResponseHandler, "dead letter sink");
     this.offsetManager = offsetManager;
     this.closer = () -> CompositeFuture.all(
       sinkResponseHandler.close(),
-      deadLetterQueueSender.close(),
+      deadLetterSinkSender.close(),
       subscriberSender.close()
     );
     this.consumerTracer = consumerTracer;
@@ -108,12 +108,12 @@ public class RecordDispatcher implements Handler<KafkaConsumerRecord<String, Clo
       |       |             |                  |
       |       v             |                  |             +-------------+
       |    +--+---------+   |           +------v-------+     |             |
-      |    |            |   |           |  subscriber  +---->+  dlq        |
+      |    |            |   |           |  subscriber  +---->+  dls        |
       |    | filter not |   +---------->+  failure     |     |  success    +---->end
       |    | matching   |               +----+---------+     |             |
       |    |            |                    |               +-----+-------+
       |    +---+--------+              +-----v-------+             |
-      |        |                       | dlq failure |             v
+      |        |                       | dls failure |             v
       |        |                       +-------------+----------> end
       +->end<--+
      */
@@ -186,21 +186,21 @@ public class RecordDispatcher implements Handler<KafkaConsumerRecord<String, Clo
 
   private void onSubscriberFailure(final KafkaConsumerRecord<String, CloudEvent> record,
                                    final Promise<Void> finalProm) {
-    dlqSender.apply(record)
-      .onSuccess(v -> onDLQSuccess(record, finalProm))
-      .onFailure(ex -> onDLQFailure(record, ex, finalProm));
+    dlsSender.apply(record)
+      .onSuccess(v -> onDeadLetterSinkSuccess(record, finalProm))
+      .onFailure(ex -> onDeadLetterSinkFailure(record, ex, finalProm));
   }
 
-  private void onDLQSuccess(final KafkaConsumerRecord<String, CloudEvent> record, final Promise<Void> finalProm) {
-    logDebug("Successfully sent event to the dlq", record);
-    offsetManager.successfullySentToDLQ(record)
+  private void onDeadLetterSinkSuccess(final KafkaConsumerRecord<String, CloudEvent> record, final Promise<Void> finalProm) {
+    logDebug("Successfully sent event to the dead letter sink", record);
+    offsetManager.successfullySentToDeadLetterSink(record)
       .onComplete(finalProm);
   }
 
 
-  private void onDLQFailure(final KafkaConsumerRecord<String, CloudEvent> record, final Throwable exception,
-                            final Promise<Void> finalProm) {
-    offsetManager.failedToSendToDLQ(record, exception)
+  private void onDeadLetterSinkFailure(final KafkaConsumerRecord<String, CloudEvent> record, final Throwable exception,
+                                       final Promise<Void> finalProm) {
+    offsetManager.failedToSendToDeadLetterSink(record, exception)
       .onComplete(finalProm);
   }
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/OffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/OffsetManager.java
index 6c9a16e79..ba0aa0f2b 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/OffsetManager.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/OffsetManager.java
@@ -32,12 +32,12 @@ public interface OffsetManager {
   Future<Void> recordReceived(KafkaConsumerRecord<?, ?> record);
 
   /**
-   * The given record cannot be delivered to dead letter queue.
+   * The given record cannot be delivered to dead letter sink.
    *
-   * @param record record undeliverable to dead letter queue.
+   * @param record record undeliverable to dead letter sink.
    * @param ex     exception occurred.
    */
-  Future<Void> failedToSendToDLQ(KafkaConsumerRecord<?, ?> record, Throwable ex);
+  Future<Void> failedToSendToDeadLetterSink(KafkaConsumerRecord<?, ?> record, Throwable ex);
 
   /**
    * The given event doesn't pass the filter.
@@ -54,9 +54,9 @@ public interface OffsetManager {
   Future<Void> successfullySentToSubscriber(KafkaConsumerRecord<?, ?> record);
 
   /**
-   * The given record has been successfully sent to dead letter queue.
+   * The given record has been successfully sent to dead letter sink.
    *
-   * @param record record sent to dead letter queue.
+   * @param record record sent to dead letter sink.
    */
-  Future<Void> successfullySentToDLQ(KafkaConsumerRecord<?, ?> record);
+  Future<Void> successfullySentToDeadLetterSink(KafkaConsumerRecord<?, ?> record);
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedOffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedOffsetManager.java
index 72cf0e965..b3282ab7e 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedOffsetManager.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedOffsetManager.java
@@ -76,7 +76,7 @@ public final class OrderedOffsetManager implements OffsetManager {
    * {@inheritDoc}
    */
   @Override
-  public Future<Void> successfullySentToDLQ(final KafkaConsumerRecord<?, ?> record) {
+  public Future<Void> successfullySentToDeadLetterSink(final KafkaConsumerRecord<?, ?> record) {
     return commit(record);
   }
 
@@ -84,7 +84,7 @@ public final class OrderedOffsetManager implements OffsetManager {
    * {@inheritDoc}
    */
   @Override
-  public Future<Void> failedToSendToDLQ(final KafkaConsumerRecord<?, ?> record, final Throwable ex) {
+  public Future<Void> failedToSendToDeadLetterSink(final KafkaConsumerRecord<?, ?> record, final Throwable ex) {
     return Future.succeededFuture();
   }
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManager.java
index 1d194ee7c..6f53dbfec 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManager.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManager.java
@@ -86,7 +86,7 @@ public final class UnorderedOffsetManager implements OffsetManager {
    * {@inheritDoc}
    */
   @Override
-  public Future<Void> successfullySentToDLQ(final KafkaConsumerRecord<?, ?> record) {
+  public Future<Void> successfullySentToDeadLetterSink(final KafkaConsumerRecord<?, ?> record) {
     return commit(record);
   }
 
@@ -94,7 +94,7 @@ public final class UnorderedOffsetManager implements OffsetManager {
    * {@inheritDoc}
    */
   @Override
-  public Future<Void> failedToSendToDLQ(final KafkaConsumerRecord<?, ?> record, final Throwable ex) {
+  public Future<Void> failedToSendToDeadLetterSink(final KafkaConsumerRecord<?, ?> record, final Throwable ex) {
     this.offsetTrackers.get(new TopicPartition(record.topic(), record.partition()))
       .recordNewOffset(record.offset());
     return Future.succeededFuture();
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
index e5e9a7212..532bf0f87 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
@@ -70,8 +70,8 @@ public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory {
 
   private static final Logger logger = LoggerFactory.getLogger(HttpConsumerVerticleFactory.class);
 
-  private final static ConsumerRecordSender NO_DLQ_SENDER =
-    ConsumerRecordSender.create(Future.failedFuture("No DLQ set"), Future.succeededFuture());
+  private final static ConsumerRecordSender NO_DEAD_LETTER_SINK_SENDER =
+    ConsumerRecordSender.create(Future.failedFuture("No dead letter sink set"), Future.succeededFuture());
 
   private final Map<String, Object> consumerConfigs;
   private final WebClientOptions webClientOptions;
@@ -157,7 +157,7 @@ public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory {
 
         final var egressDeadLetterSender = hasDeadLetterSink(egressConfig)
           ? createConsumerRecordSender(vertx, egressConfig.getDeadLetter(), egressConfig)
-          : NO_DLQ_SENDER;
+          : NO_DEAD_LETTER_SINK_SENDER;
 
         final var filter = egress.hasFilter() ?
           new AttributesFilter(egress.getFilter().getAttributesMap()) :
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 8c0f8d58a..79226a174 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -42,21 +42,21 @@
 
     <!-- dependencies version -->
     <vertx.version>4.1.0</vertx.version>
-    <cloudevents.sdk.version>2.1.1</cloudevents.sdk.version>
+    <cloudevents.sdk.version>2.2.0</cloudevents.sdk.version>
     <micrometer.version>1.7.0</micrometer.version>
     <opentelemetry.version>1.1.0</opentelemetry.version>
     <jackson.version>2.12.3</jackson.version>
     <protobuf.version>3.17.2</protobuf.version>
-    <slf4j.version>1.7.30</slf4j.version>
+    <slf4j.version>1.7.31</slf4j.version>
     <ch.qos.logback.version>1.2.3</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>6.6</net.logstash.logback.encoder.version>
-    <assertj.version>3.19.0</assertj.version>
+    <assertj.version>3.20.1</assertj.version>
     <awaitility.version>4.1.0</awaitility.version>
     <junit.jupiter.version>5.7.2</junit.jupiter.version>
     <mokito.junit.jupiter.version>3.11.1</mokito.junit.jupiter.version>
     <fabric8.kubernetes.version>5.4.1</fabric8.kubernetes.version>
     <kafka.version>2.7.0</kafka.version>
-    <debezium.version>1.5.2.Final</debezium.version>
+    <debezium.version>1.5.3.Final</debezium.version>
   </properties>
 
   <modules>
diff --git a/data-plane/profiler/resources/config-sacura-warmup.yaml b/data-plane/profiler/resources/config-sacura-warmup.yaml
index b80b1bb43..b6995a7fb 100644
--- a/data-plane/profiler/resources/config-sacura-warmup.yaml
+++ b/data-plane/profiler/resources/config-sacura-warmup.yaml
@@ -14,11 +14,11 @@
 
 sender:
   target: http://localhost:8080/attack-ingress-single
-  frequency: 1000
+  frequency: 10000
   workers: 15
   keepAlive: true
 receiver:
   port: 8081
-  timeout: 0m
+  timeout: 2m
 duration: 5m
 timeout: 0m
diff --git a/data-plane/profiler/resources/config-sacura.yaml b/data-plane/profiler/resources/config-sacura.yaml
index 086b73e80..d1742f7de 100644
--- a/data-plane/profiler/resources/config-sacura.yaml
+++ b/data-plane/profiler/resources/config-sacura.yaml
@@ -14,11 +14,11 @@
 
 sender:
   target: http://localhost:8080/attack-ingress-single
-  frequency: 1000
+  frequency: 10000
   workers: 15
   keepAlive: true
 receiver:
   port: 8081
-  timeout: 0m
+  timeout: 2m
 duration: 10m
-timeout: 0m
+timeout: 1m
diff --git a/data-plane/profiler/run.sh b/data-plane/profiler/run.sh
index a9615a49d..31f693929 100755
--- a/data-plane/profiler/run.sh
+++ b/data-plane/profiler/run.sh
@@ -15,6 +15,8 @@
 # limitations under the License.
 #
 
+set -e
+
 # Set runtime variables
 # echo 1 > /proc/sys/kernel/perf_event_paranoid
 # echo 0 > /proc/sys/kernel/kptr_restrict
@@ -40,6 +42,9 @@ PROJECT_ROOT_DIR=$(dirname $0)/..
 RESOURCES_DIR="$(dirname $0)"/resources
 ASYNC_PROFILER_URL="https://github.com/jvm-profiling-tools/async-profiler/releases/download/v2.0/async-profiler-2.0-linux-x64.tar.gz"
 KAFKA_URL="https://archive.apache.org/dist/kafka/2.6.0/kafka_2.13-2.6.0.tgz"
+LOG_DIR=${LOG_DIR:-"/tmp/eventing-kafka-broker-logs/profiler"}
+
+rm -rf "${LOG_DIR}" && mkdir -p "${LOG_DIR}"
 
 echo "Project root dir: ${PROJECT_ROOT_DIR}"
 echo "Resource dir: ${RESOURCES_DIR}"
@@ -50,28 +55,26 @@ echo "Kafka URL: ${KAFKA_URL}"
 cd "${PROJECT_ROOT_DIR}" && ./mvnw package -DskipTests -Dlicense.skip -Deditorconfig.skip -B -U --no-transfer-progress && cd - || exit 1
 
 # Download async profiler.
-rm async-profiler
-rm async-profiler.tgz
+rm -rf async-profiler
+rm -rf async-profiler.tgz
 mkdir async-profiler
 wget -O - ${ASYNC_PROFILER_URL} >async-profiler.tgz
 tar xzvf async-profiler.tgz -C async-profiler --strip-components=1
 
 # Download Apache Kafka.
-rm kafka
-rm kafka.tgz
-rm -r /tmp/kafka-logs/
-rm -r /tmp/zookeeper/
+rm -rf kafka
+rm -rf kafka.tgz
+rm -rf /tmp/kafka-logs/
+rm -rf /tmp/zookeeper/
 mkdir kafka
 wget -O - ${KAFKA_URL} >kafka.tgz
 tar xzvf kafka.tgz -C kafka --strip-components=1
 
-ls -al
-
 # Start Zookeeper and Kafka.
 cd kafka || exit 1
-./bin/zookeeper-server-start.sh config/zookeeper.properties &
+./bin/zookeeper-server-start.sh config/zookeeper.properties >"${LOG_DIR}/zookeeper.log" &
 zookeeper_pid=$!
-./bin/kafka-server-start.sh config/server.properties &
+./bin/kafka-server-start.sh config/server.properties >"${LOG_DIR}/kafka.log" &
 kafka_pid=$!
 
 # Create our Kafka topic.
@@ -106,7 +109,7 @@ java \
   -XX:+UnlockDiagnosticVMOptions \
   -XX:+DebugNonSafepoints \
   -Dlogback.configurationFile="${RESOURCES_DIR}"/config-logging.xml \
-  -jar "${PROJECT_ROOT_DIR}"/receiver/target/receiver-1.0-SNAPSHOT.jar &
+  -jar "${PROJECT_ROOT_DIR}"/receiver/target/receiver-1.0-SNAPSHOT.jar >"${LOG_DIR}/receiver.log" &
 receiver_pid=$!
 
 # Define expected env variables.
@@ -121,7 +124,7 @@ java \
   -XX:+UnlockDiagnosticVMOptions \
   -XX:+DebugNonSafepoints \
   -Dlogback.configurationFile="${RESOURCES_DIR}"/config-logging.xml \
-  -jar "${PROJECT_ROOT_DIR}"/dispatcher/target/dispatcher-1.0-SNAPSHOT.jar &
+  -jar "${PROJECT_ROOT_DIR}"/dispatcher/target/dispatcher-1.0-SNAPSHOT.jar >"${LOG_DIR}/dispatcher.log" &
 dispatcher_pid=$!
 
 # Download Sacura
@@ -155,6 +158,6 @@ kill $dispatcher_pid
 kill -9 $kafka_pid
 kill -9 $zookeeper_pid
 
-rm -r kafka kafka.tgz async-profiler async-profiler.tgz
+rm -rf kafka kafka.tgz async-profiler async-profiler.tgz
 
 exit 0
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java
index d0d42789b..fc04998c7 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java
@@ -20,19 +20,25 @@ import dev.knative.eventing.kafka.broker.core.eventbus.ContractPublisher;
 import dev.knative.eventing.kafka.broker.core.file.FileWatcher;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerMessageHandler;
-import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.core.tracing.TracingConfig;
 import dev.knative.eventing.kafka.broker.core.utils.Configurations;
 import dev.knative.eventing.kafka.broker.core.utils.Shutdown;
 import io.cloudevents.kafka.CloudEventSerializer;
 import io.cloudevents.kafka.PartitionKeyExtensionInterceptor;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.vertx.core.DeploymentOptions;
+import io.vertx.core.Verticle;
 import io.vertx.core.Vertx;
 import io.vertx.core.VertxOptions;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.tracing.TracingPolicy;
-import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.tracing.opentelemetry.OpenTelemetryOptions;
+import net.logstash.logback.encoder.LogstashEncoder;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.ClosedWatchServiceException;
@@ -40,12 +46,7 @@ import java.nio.file.FileSystems;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.function.Function;
-import net.logstash.logback.encoder.LogstashEncoder;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.function.Supplier;
 
 import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
@@ -110,34 +111,19 @@ public class Main {
       producerConfigs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CloudEventSerializer.class);
       producerConfigs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, PartitionKeyExtensionInterceptor.class.getName());
 
-      final Function<Vertx, RequestMapper> handlerFactory = v -> new RequestMapper(
-        v,
-        AuthProvider.kubernetes(),
-        producerConfigs,
-        new CloudEventRequestToRecordMapper(),
-        properties -> KafkaProducer.create(v, properties),
-        badRequestCounter,
-        produceEventsCounter
-      );
-
       final var httpServerOptions = new HttpServerOptions(
         Configurations.getPropertiesAsJson(env.getHttpServerConfigFilePath())
       );
       httpServerOptions.setPort(env.getIngressPort());
       httpServerOptions.setTracingPolicy(TracingPolicy.PROPAGATE);
 
-      final var verticle = new ReceiverVerticle(
-        httpServerOptions,
-        handlerFactory,
-        h -> new SimpleProbeHandlerDecorator(
-          env.getLivenessProbePath(),
-          env.getReadinessProbePath(),
-          h
-        )
-      );
+      final Supplier<Verticle> verticle = new ReceiverVerticleFactory(env, producerConfigs, badRequestCounter, produceEventsCounter, httpServerOptions);
+
+      final var deploymentOptions = new DeploymentOptions()
+        .setInstances(Runtime.getRuntime().availableProcessors());
 
       final var waitVerticle = new CountDownLatch(1);
-      vertx.deployVerticle(verticle)
+      vertx.deployVerticle(verticle, deploymentOptions)
         .onSuccess(v -> {
           logger.info("Receiver started");
           waitVerticle.countDown();
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleFactory.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleFactory.java
new file mode 100644
index 000000000..a6a450668
--- /dev/null
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleFactory.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.knative.eventing.kafka.broker.receiver;
+
+import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
+import io.micrometer.core.instrument.Counter;
+import io.vertx.core.Verticle;
+import io.vertx.core.http.HttpServerOptions;
+import io.vertx.kafka.client.producer.KafkaProducer;
+
+import java.util.Properties;
+import java.util.function.Supplier;
+
+class ReceiverVerticleFactory implements Supplier<Verticle> {
+
+  private final ReceiverEnv env;
+  private final Properties producerConfigs;
+  private final Counter badRequestCounter;
+  private final Counter produceEventsCounter;
+  private final HttpServerOptions httpServerOptions;
+
+  ReceiverVerticleFactory(final ReceiverEnv env,
+                          final Properties producerConfigs,
+                          final Counter badRequestCounter,
+                          final Counter produceEventsCounter,
+                          final HttpServerOptions httpServerOptions) {
+
+    this.env = env;
+    this.producerConfigs = producerConfigs;
+    this.badRequestCounter = badRequestCounter;
+    this.produceEventsCounter = produceEventsCounter;
+    this.httpServerOptions = httpServerOptions;
+  }
+
+  @Override
+  public Verticle get() {
+    return new ReceiverVerticle(
+      httpServerOptions,
+      v -> new RequestMapper(
+        v,
+        AuthProvider.kubernetes(),
+        producerConfigs,
+        new CloudEventRequestToRecordMapper(),
+        properties -> KafkaProducer.create(v, properties),
+        badRequestCounter,
+        produceEventsCounter
+      ),
+      h -> new SimpleProbeHandlerDecorator(
+        env.getLivenessProbePath(),
+        env.getReadinessProbePath(),
+        h
+      )
+    );
+  }
+}
diff --git a/go.mod b/go.mod
index 3dedbc6ac..933a53520 100644
--- a/go.mod
+++ b/go.mod
@@ -21,8 +21,8 @@ require (
 	k8s.io/apiserver v0.20.7
 	k8s.io/client-go v0.20.7
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.23.1-0.20210615125721-af6744d48af9
+	knative.dev/eventing v0.23.1-0.20210618081650-b164d3c042d2
 	knative.dev/hack v0.0.0-20210614141220-66ab1a098940
-	knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73
+	knative.dev/pkg v0.0.0-20210618060751-f454995ff92b
 	knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7
 )
diff --git a/go.sum b/go.sum
index 65d979384..b3a3c7293 100644
--- a/go.sum
+++ b/go.sum
@@ -86,6 +86,7 @@ github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMx
 github.com/VividCortex/gohistogram v1.0.0/go.mod h1:Pf5mBqqDxYaXu3hDrrU+w6nw50o/4+TcAqDqk/vUH7g=
 github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5/go.mod h1:SkGFH1ia65gfNATL8TAiHDNxPzPdmEL5uirI2Uyuz6c=
 github.com/agnivade/levenshtein v1.0.1/go.mod h1:CURSv5d9Uaml+FovSIICkLbAUZ9S4RqaHDIsdSBg7lM=
+github.com/ahmetb/gen-crd-api-reference-docs v0.3.1-0.20210420163308-c1402a70e2f1/go.mod h1:TdjdkYhlOifCQWPs1UdTma97kQQMozf5h26hTuG70u8=
 github.com/alecthomas/jsonschema v0.0.0-20180308105923-f2c93856175a/go.mod h1:qpebaTNSsyUn5rPSJMsfqEtDw71TTggXM6stUDI16HA=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafoB+tBA3gMyHYHrpOtNuDiK/uB5uXxq5wM=
@@ -1294,8 +1295,10 @@ k8s.io/component-base v0.20.7/go.mod h1:878UWprXC07P2CWFg+jjvTfxJSlkHp1v2m1MTkNQ
 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-20201113003025-83324d819ded/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
+k8s.io/gengo v0.0.0-20201203183100-97869a43a9d9/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
 k8s.io/gengo v0.0.0-20210203185629-de9496dff47b h1:bAU8IlrMA6KbP0dIg/sVSJn95pDCUHDZx0DpTGrf2v4=
 k8s.io/gengo v0.0.0-20210203185629-de9496dff47b/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAEV2be7d5xI0vBa/VySYy3E=
+k8s.io/klog v0.2.0/go.mod h1:Gq+BEi5rUBO/HRz0bTSXDUcqjScdoY3a9IHpCEIOOfk=
 k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8=
 k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
@@ -1309,16 +1312,16 @@ k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.23.1-0.20210615125721-af6744d48af9 h1:iQOiKuphVaCgyDJcmoAjLtlcFvfCA978wd/PiHyPGgY=
-knative.dev/eventing v0.23.1-0.20210615125721-af6744d48af9/go.mod h1:3i9HgqhC4/3EqiS82CHmhL5XbGiz9911L9rkrAyN28w=
+knative.dev/eventing v0.23.1-0.20210618081650-b164d3c042d2 h1:Vz1hXjF7CQ6YMMQXi1XfkModc34WQQTc/hmkKk9+p9M=
+knative.dev/eventing v0.23.1-0.20210618081650-b164d3c042d2/go.mod h1:8fnNPdQgBT7NT+yCFgE1kuvN0mMsfnCfPIwnlyPrxy0=
 knative.dev/hack v0.0.0-20210601210329-de04b70e00d0/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20210614141220-66ab1a098940 h1:Aao7SsI087Pa8x+LDLaBdj/ylduPh2bJ8HbB7UjXElE=
 knative.dev/hack v0.0.0-20210614141220-66ab1a098940/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210614141220-66ab1a098940/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20210602095030-0e61d6763dd6/go.mod h1:/hT1IXHPs8ZwQ8yTJtfaYWpLmsLhC2s6/v/ydF7PoiE=
-knative.dev/pkg v0.0.0-20210615092720-192b0c9d6e56/go.mod h1:QVfbwA5ovshGnXDcx54l2QPYhGKppw/JgLYJWe7sA6A=
-knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73 h1:p7KMCn1+n9QtJ/YZLncpMwSlWKv8vUA5IpiCaCxJWYY=
 knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73/go.mod h1:QVfbwA5ovshGnXDcx54l2QPYhGKppw/JgLYJWe7sA6A=
+knative.dev/pkg v0.0.0-20210618060751-f454995ff92b h1:oPoZUyOvI3fNrxDlpRdB3eW4lPiLIwbNBm+25nEnVco=
+knative.dev/pkg v0.0.0-20210618060751-f454995ff92b/go.mod h1:QVfbwA5ovshGnXDcx54l2QPYhGKppw/JgLYJWe7sA6A=
 knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7 h1:BNP1vs/FNDfpqFb+FaDzwZ9qnr3X+ms5u3lZDZk2ivw=
 knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7/go.mod h1:Ggh7/FyaVr9R1kQiuB9KpNsvaKVoPlZ0w8OoxUGi9Ys=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
diff --git a/hack/tools.go b/hack/tools.go
index 896ffd623..49137d8b0 100644
--- a/hack/tools.go
+++ b/hack/tools.go
@@ -24,6 +24,7 @@ import (
 	_ "knative.dev/pkg/hack"
 
 	// Test images from eventing
+	_ "knative.dev/eventing/test/test_images/event-library"
 	_ "knative.dev/eventing/test/test_images/event-sender"
 	_ "knative.dev/eventing/test/test_images/heartbeats"
 	_ "knative.dev/eventing/test/test_images/performance"
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 14b250f88..ae9ff4d4c 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1026,7 +1026,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.23.1-0.20210615125721-af6744d48af9
+# knative.dev/eventing v0.23.1-0.20210618081650-b164d3c042d2
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1121,6 +1121,7 @@ knative.dev/eventing/test/rekt/resources/broker
 knative.dev/eventing/test/rekt/resources/delivery
 knative.dev/eventing/test/rekt/resources/trigger
 knative.dev/eventing/test/test_images
+knative.dev/eventing/test/test_images/event-library
 knative.dev/eventing/test/test_images/event-sender
 knative.dev/eventing/test/test_images/heartbeats
 knative.dev/eventing/test/test_images/performance
@@ -1130,7 +1131,7 @@ knative.dev/eventing/test/test_images/request-sender
 # knative.dev/hack v0.0.0-20210614141220-66ab1a098940
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73
+# knative.dev/pkg v0.0.0-20210618060751-f454995ff92b
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/.github/workflows/kind-e2e.yaml b/.github/workflows/kind-e2e.yaml
index 0e4617b02..d4d97d158 100644
--- a/.github/workflows/kind-e2e.yaml
+++ b/.github/workflows/kind-e2e.yaml
@@ -33,6 +33,7 @@ jobs:
 
         test-suite:
           - ./test/e2e
+          - ./test/e2e_new
 
         # Map between K8s and KinD versions.
         # This is attempting to make it a bit clearer what's being tested.
@@ -48,9 +49,6 @@ jobs:
             kind-version: v0.11.1
             kind-image-sha: sha256:69860bda5563ac81e3c0057d654b5253219618a22ec3a346306239bba8cfa1a6
 
-          # Add the flags we use for each of these test suites.
-          - test-suite: ./test/e2e
-
     env:
       GOPATH: ${{ github.workspace }}
       GO111MODULE: off
@@ -76,10 +74,10 @@ jobs:
           sudo mv ko /usr/local/bin
           echo '::endgroup::'
 
-      - name: Set up Go 1.15.x
+      - name: Set up Go 1.16.x
         uses: actions/setup-go@v2
         with:
-          go-version: 1.15.x
+          go-version: 1.16.x
 
       - name: Check out code onto GOPATH
         uses: actions/checkout@v2
@@ -129,8 +127,14 @@ jobs:
 
       - name: Run e2e Tests
         run: |
+          export KO_FLAGS="--platform=linux/amd64"
+          export GO111MODULE=on
+          export GOFLAGS=-mod=vendor
+          export SYSTEM_NAMESPACE=knative-eventing
+          export CLUSTER_DOMAIN=${CLUSTER_SUFFIX}
+
           # Run the tests tagged as e2e on the KinD cluster.
-          go test -race -count=1 -timeout=1h -short -tags=e2e \
+          go test -race -count=1 -timeout=1h -v -short -tags=e2e \
              ${{ matrix.test-suite }} ${{ matrix.extra-test-flags }}
 
       - name: Teardown Eventing Kafka Broker
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 461fbf2ba..4526a8e22 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -107,11 +107,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				NewConfigMap(&configs, nil),
 				NewService(),
 				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "2",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -169,8 +169,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "3"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -817,11 +817,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				NewConfigMap(&configs, nil),
 				NewService(),
 				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "2",
+					base.VolumeGenerationAnnotationKey: "3",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -889,12 +889,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				NewConfigMap(&configs, nil),
 				NewService(),
 				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
-					"annotation_to_preserve":           "value_to_preserve",
+					"annotation_to_preserve": "value_to_preserve",
 				}),
 				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "2",
-					"annotation_to_preserve":           "value_to_preserve",
+					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
@@ -1201,8 +1199,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1265,8 +1263,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1329,8 +1327,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1390,8 +1388,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1462,8 +1460,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1490,6 +1488,56 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				BootstrapServersConfigMapKey: bootstrapServers,
 			},
 		},
+		{
+			Name: "Reconciled normal - unchanged contract - changed data plane pods annotation",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig:     &contract.EgressConfig{DeadLetter: "http://test-service.test-service-namespace.svc.cluster.local/"},
+						},
+					},
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
 	}
 
 	for i := range table {
@@ -1610,8 +1658,6 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 					WithDelivery(),
 				),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index b39980929..a89531352 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -102,8 +102,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
-					"annotation_to_preserve":           "value_to_preserve",
+					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
@@ -156,7 +155,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				NewSSLSecret(SinkNamespace, "secret-1"),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -221,7 +220,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -334,7 +333,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -723,16 +722,75 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 							BootstrapServers: bootstrapServers,
 						},
 					},
+					Generation: 1,
 				}, &configs),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "2",
+					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
-			WantUpdates: []clientgotesting.UpdateActionImpl{},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						SinkControllerOwnsTopic,
+						InitSinkConditions,
+						SinkDataPlaneAvailable,
+						SinkConfigParsed,
+						BootstrapServers(bootstrapServersArr),
+						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
+						SinkAddressable(&configs.Env),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - unchanged contract - changed receiver pod annotation",
+			Objects: []runtime.Object{
+				NewSink(
+					SinkControllerOwnsTopic,
+					BootstrapServers(bootstrapServersArr),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
+							Topics:  []string{"my-existing-topic-a"},
+							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+						},
+						{
+							Uid:    "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
+							Topics: []string{"my-existing-topic-b"},
+						},
+						{
+							Uid:              SinkUUID,
+							Topics:           []string{SinkTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}, &configs),
+				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherTest.java
index 20a3f0fc0..031d1e3a6 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherTest.java
@@ -43,14 +43,14 @@ import static org.mockito.Mockito.when;
 public class RecordDispatcherTest {
 
   @Test
-  public void shouldNotSendToSubscriberNorToDLQIfValueDoesntMatch() {
+  public void shouldNotSendToSubscriberNorToDeadLetterSinkIfValueDoesntMatch() {
 
     final OffsetManager receiver = offsetManagerMock();
 
     final var dispatcherHandler = new RecordDispatcher(
       value -> false,
       ConsumerRecordSender.create(Future.failedFuture("subscriber send called"), Future.succeededFuture()),
-      ConsumerRecordSender.create(Future.failedFuture("DLQ send called"), Future.succeededFuture()),
+      ConsumerRecordSender.create(Future.failedFuture("DLS send called"), Future.succeededFuture()),
       new SinkResponseHandlerMock(
         Future::succeededFuture,
         response -> Future.succeededFuture()
@@ -62,8 +62,8 @@ public class RecordDispatcherTest {
     verify(receiver, times(1)).recordReceived(record);
     verify(receiver, times(1)).recordDiscarded(record);
     verify(receiver, never()).successfullySentToSubscriber(any());
-    verify(receiver, never()).successfullySentToDLQ(any());
-    verify(receiver, never()).failedToSendToDLQ(any(), any());
+    verify(receiver, never()).successfullySentToDeadLetterSink(any());
+    verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
   }
 
   @Test
@@ -83,7 +83,7 @@ public class RecordDispatcherTest {
       new ConsumerRecordSenderMock(
         Future::succeededFuture,
         record -> {
-          fail("DLQ send called");
+          fail("DLS send called");
           return Future.succeededFuture();
         }
       ), new SinkResponseHandlerMock(
@@ -96,16 +96,16 @@ public class RecordDispatcherTest {
     assertTrue(sendCalled.get());
     verify(receiver, times(1)).recordReceived(record);
     verify(receiver, times(1)).successfullySentToSubscriber(record);
-    verify(receiver, never()).successfullySentToDLQ(any());
-    verify(receiver, never()).failedToSendToDLQ(any(), any());
+    verify(receiver, never()).successfullySentToDeadLetterSink(any());
+    verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
     verify(receiver, never()).recordDiscarded(any());
   }
 
   @Test
-  public void shouldSendToDLQIfValueMatchesAndSubscriberSenderFails() {
+  public void shouldSendToDeadLetterSinkIfValueMatchesAndSubscriberSenderFails() {
 
     final var subscriberSenderSendCalled = new AtomicBoolean(false);
-    final var DLQSenderSendCalled = new AtomicBoolean(false);
+    final var dlsSenderSendCalled = new AtomicBoolean(false);
     final OffsetManager receiver = offsetManagerMock();
 
     final var dispatcherHandler = new RecordDispatcher(
@@ -119,7 +119,7 @@ public class RecordDispatcherTest {
       new ConsumerRecordSenderMock(
         Future::succeededFuture,
         record -> {
-          DLQSenderSendCalled.set(true);
+          dlsSenderSendCalled.set(true);
           return Future.succeededFuture();
         }
       ), new SinkResponseHandlerMock(
@@ -130,19 +130,19 @@ public class RecordDispatcherTest {
     dispatcherHandler.handle(record);
 
     assertTrue(subscriberSenderSendCalled.get());
-    assertTrue(DLQSenderSendCalled.get());
+    assertTrue(dlsSenderSendCalled.get());
     verify(receiver, times(1)).recordReceived(record);
-    verify(receiver, times(1)).successfullySentToDLQ(record);
+    verify(receiver, times(1)).successfullySentToDeadLetterSink(record);
     verify(receiver, never()).successfullySentToSubscriber(any());
-    verify(receiver, never()).failedToSendToDLQ(any(), any());
+    verify(receiver, never()).failedToSendToDeadLetterSink(any(), any());
     verify(receiver, never()).recordDiscarded(any());
   }
 
   @Test
-  public void shouldCallFailedToSendToDLQIfValueMatchesAndSubscriberAndDLQSenderFail() {
+  public void shouldCallFailedToSendToDeadLetterSinkIfValueMatchesAndSubscriberAndDeadLetterSinkSenderFail() {
 
     final var subscriberSenderSendCalled = new AtomicBoolean(false);
-    final var DLQSenderSendCalled = new AtomicBoolean(false);
+    final var dlsSenderSendCalled = new AtomicBoolean(false);
     final OffsetManager receiver = offsetManagerMock();
 
     final var dispatcherHandler = new RecordDispatcher(
@@ -156,7 +156,7 @@ public class RecordDispatcherTest {
       new ConsumerRecordSenderMock(
         Future::succeededFuture,
         record -> {
-          DLQSenderSendCalled.set(true);
+          dlsSenderSendCalled.set(true);
           return Future.failedFuture("");
         }
       ), new SinkResponseHandlerMock(
@@ -167,16 +167,16 @@ public class RecordDispatcherTest {
     dispatcherHandler.handle(record);
 
     assertTrue(subscriberSenderSendCalled.get());
-    assertTrue(DLQSenderSendCalled.get());
+    assertTrue(dlsSenderSendCalled.get());
     verify(receiver, times(1)).recordReceived(record);
-    verify(receiver, times(1)).failedToSendToDLQ(eq(record), any());
-    verify(receiver, never()).successfullySentToDLQ(any());
+    verify(receiver, times(1)).failedToSendToDeadLetterSink(eq(record), any());
+    verify(receiver, never()).successfullySentToDeadLetterSink(any());
     verify(receiver, never()).successfullySentToSubscriber(any());
     verify(receiver, never()).recordDiscarded(any());
   }
 
   @Test
-  public void shouldCallFailedToSendToDLQIfValueMatchesAndSubscriberSenderFailsAndNoDLQSender() {
+  public void shouldCallFailedToSendToDeadLetterSinkIfValueMatchesAndSubscriberSenderFailsAndNoDeadLetterSinkSender() {
     final var subscriberSenderSendCalled = new AtomicBoolean(false);
     final OffsetManager receiver = offsetManagerMock();
 
@@ -189,7 +189,7 @@ public class RecordDispatcherTest {
           return Future.failedFuture("");
         }
       ),
-      ConsumerRecordSender.create(Future.failedFuture("No DLQ configured"), Future.succeededFuture()),
+      ConsumerRecordSender.create(Future.failedFuture("No DLS configured"), Future.succeededFuture()),
       new SinkResponseHandlerMock(
         Future::succeededFuture,
         response -> Future.succeededFuture()
@@ -200,14 +200,14 @@ public class RecordDispatcherTest {
 
     assertTrue(subscriberSenderSendCalled.get());
     verify(receiver, times(1)).recordReceived(record);
-    verify(receiver, times(1)).failedToSendToDLQ(eq(record), any());
-    verify(receiver, never()).successfullySentToDLQ(any());
+    verify(receiver, times(1)).failedToSendToDeadLetterSink(eq(record), any());
+    verify(receiver, never()).successfullySentToDeadLetterSink(any());
     verify(receiver, never()).successfullySentToSubscriber(any());
     verify(receiver, never()).recordDiscarded(any());
   }
 
   @Test
-  public void shouldCloseSinkResponseHandlerSubscriberSenderAndDLSSender(final VertxTestContext context) {
+  public void shouldCloseSinkResponseHandlerSubscriberSenderAndDeadLetterSinkSender(final VertxTestContext context) {
 
     final var subscriberSender = mock(ConsumerRecordSender.class);
     when(subscriberSender.close()).thenReturn(Future.succeededFuture());
@@ -241,9 +241,9 @@ public class RecordDispatcherTest {
 
     when(offsetManager.recordReceived(any())).thenReturn(Future.succeededFuture());
     when(offsetManager.recordDiscarded(any())).thenReturn(Future.succeededFuture());
-    when(offsetManager.successfullySentToDLQ(any())).thenReturn(Future.succeededFuture());
+    when(offsetManager.successfullySentToDeadLetterSink(any())).thenReturn(Future.succeededFuture());
     when(offsetManager.successfullySentToSubscriber(any())).thenReturn(Future.succeededFuture());
-    when(offsetManager.failedToSendToDLQ(any(), any())).thenReturn(Future.succeededFuture());
+    when(offsetManager.failedToSendToDeadLetterSink(any(), any())).thenReturn(Future.succeededFuture());
 
     return offsetManager;
   }
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/AbstractConsumerVerticleTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/AbstractConsumerVerticleTest.java
index 1b699ab04..79a684e61 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/AbstractConsumerVerticleTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/AbstractConsumerVerticleTest.java
@@ -58,7 +58,7 @@ public abstract class AbstractConsumerVerticleTest {
     final var recordDispatcher = new RecordDispatcher(
       value -> false,
       ConsumerRecordSender.create(Future.failedFuture("subscriber send called"), Future.succeededFuture()),
-      ConsumerRecordSender.create(Future.failedFuture("DLQ send called"), Future.succeededFuture()),
+      ConsumerRecordSender.create(Future.failedFuture("dead letter sink send called"), Future.succeededFuture()),
       new SinkResponseHandlerMock(
         Future::succeededFuture,
         response -> Future.succeededFuture()
@@ -97,7 +97,7 @@ public abstract class AbstractConsumerVerticleTest {
     final var recordDispatcher = new RecordDispatcher(
       value -> false,
       ConsumerRecordSender.create(Future.failedFuture("subscriber send called"), Future.succeededFuture()),
-      ConsumerRecordSender.create(Future.failedFuture("DLQ send called"), Future.succeededFuture()),
+      ConsumerRecordSender.create(Future.failedFuture("dead letter sink send called"), Future.succeededFuture()),
       new SinkResponseHandlerMock(
         Future::succeededFuture,
         response -> Future.succeededFuture()
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedOffsetManagerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedOffsetManagerTest.java
index 8f04a65e3..4cad6b05d 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedOffsetManagerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedOffsetManagerTest.java
@@ -40,9 +40,9 @@ public class OrderedOffsetManagerTest extends AbstractOffsetManagerTest {
   }
 
   @Test
-  public void shouldNotCommitAfterFailedToSendToDLQ() {
+  public void shouldNotCommitAfterFailedToSendToDeadLetterSink() {
     assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
-      offsetStrategy.failedToSendToDLQ(record("aaa", 0, 0), new IllegalStateException());
+      offsetStrategy.failedToSendToDeadLetterSink(record("aaa", 0, 0), new IllegalStateException());
     }).isEmpty();
   }
 
@@ -62,9 +62,9 @@ public class OrderedOffsetManagerTest extends AbstractOffsetManagerTest {
   }
 
   @Test
-  public void shouldCommitAfterSuccessfullySentToDLQ() {
+  public void shouldCommitAfterSuccessfullySentToDeadLetterSink() {
     assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
-      offsetStrategy.successfullySentToDLQ(record("aaa", 0, 0));
+      offsetStrategy.successfullySentToDeadLetterSink(record("aaa", 0, 0));
     }).containsEntry(new TopicPartition("aaa", 0), 1L);
   }
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManagerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManagerTest.java
index c5fb5ca28..ceac82d1b 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManagerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManagerTest.java
@@ -164,11 +164,11 @@ public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
   }
 
   @Test
-  public void shouldCommitSuccessfullyOnSuccessfullySentToDLQ() {
+  public void shouldCommitSuccessfullyOnSuccessfullySentToDeadLetterSink() {
     assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
       var rec = record("aaa", 0, 0);
       offsetStrategy.recordReceived(rec);
-      offsetStrategy.successfullySentToDLQ(rec);
+      offsetStrategy.successfullySentToDeadLetterSink(rec);
     })
       .containsEntry(new TopicPartition("aaa", 0), 1L);
   }
@@ -185,11 +185,11 @@ public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
   }
 
   @Test
-  public void shouldCommitSuccessfullyWithRecordFailedToDLQInTheMiddle() {
+  public void shouldCommitSuccessfullyWithRecordFailedToDeadLetterSinkInTheMiddle() {
     assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
       var rec = record("aaa", 0, 0);
       offsetStrategy.recordReceived(rec);
-      offsetStrategy.failedToSendToDLQ(rec, new IllegalStateException());
+      offsetStrategy.failedToSendToDeadLetterSink(rec, new IllegalStateException());
       offsetStrategy.successfullySentToSubscriber(record("aaa", 0, 1));
     })
       .containsEntry(new TopicPartition("aaa", 0), 2L);
@@ -221,14 +221,14 @@ public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
 
   @Test
   @SuppressWarnings("unchecked")
-  public void failedToSendToDLQ() {
+  public void failedToSendToDeadLetterSink() {
     final KafkaConsumer<String, CloudEvent> consumer = mock(KafkaConsumer.class);
     final Counter eventsSentCounter = mock(Counter.class);
 
     UnorderedOffsetManager strategy =
       new UnorderedOffsetManager(consumer, eventsSentCounter::increment);
     strategy.recordReceived(record("aaa", 0, 0));
-    strategy.failedToSendToDLQ(record("aaa", 0, 0), null);
+    strategy.failedToSendToDeadLetterSink(record("aaa", 0, 0), null);
 
     shouldNeverCommit(consumer);
     shouldNeverPause(consumer);
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactoryTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactoryTest.java
index 03123eb44..8aef49bd5 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactoryTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactoryTest.java
@@ -97,7 +97,7 @@ public class HttpConsumerVerticleFactoryTest {
   }
 
   @Test
-  public void shouldNotThrowIllegalArgumentExceptionIfNotDLQ() {
+  public void shouldNotThrowIllegalArgumentExceptionIfNotDeadLetterSink() {
 
     final var consumerProperties = new Properties();
     consumerProperties.setProperty(BOOTSTRAP_SERVERS_CONFIG, "0.0.0.0:9092");
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleFactoryTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleFactoryTest.java
new file mode 100644
index 000000000..183a9c8d0
--- /dev/null
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleFactoryTest.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.knative.eventing.kafka.broker.receiver;
+
+import io.micrometer.core.instrument.Counter;
+import io.vertx.core.http.HttpServerOptions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+public class ReceiverVerticleFactoryTest {
+
+  @Test
+  public void shouldCreateMultipleReceiverVerticleInstances() {
+    final var supplier = new ReceiverVerticleFactory(
+      mock(ReceiverEnv.class),
+      mock(Properties.class),
+      mock(Counter.class),
+      mock(Counter.class),
+      mock(HttpServerOptions.class)
+    );
+
+    assertThat(supplier.get()).isNotSameAs(supplier.get());
+  }
+}
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index ac6c56b99..3146133d7 100644
--- a/third_party/eventing-latest/eventing-core.yaml
+++ b/third_party/eventing-latest/eventing-core.yaml
@@ -16,7 +16,7 @@ kind: Namespace
 metadata:
   name: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 
 ---
 # Copyright 2018 The Knative Authors
@@ -39,14 +39,14 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -61,7 +61,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -76,7 +76,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -91,7 +91,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -106,7 +106,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -137,14 +137,14 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: pingsource-mt-adapter
@@ -175,14 +175,14 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -198,7 +198,7 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -213,7 +213,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -228,7 +228,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -259,7 +259,7 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 data:
   channelTemplateSpec: |
     apiVersion: messaging.knative.dev/v1
@@ -286,7 +286,7 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
   default-br-config: |
@@ -318,7 +318,7 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
   default-ch-config: |
@@ -351,7 +351,7 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
 data:
@@ -396,10 +396,13 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
 data:
+  # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
+  # For more details: https://github.com/knative/eventing/issues/5086
+  kreference-group: "disabled"
 
 ---
 # Copyright 2018 The Knative Authors
@@ -422,7 +425,7 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
   annotations:
     knative.dev/example-checksum: "96896b00"
 data:
@@ -482,7 +485,7 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
 data:
@@ -534,7 +537,7 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -609,7 +612,7 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -670,7 +673,7 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/high-availability: "true"
 spec:
   selector:
@@ -680,7 +683,7 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20210614-393e4119b"
+        eventing.knative.dev/release: "v20210618-b164d3c04"
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
       affinity:
@@ -697,7 +700,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:36498faab932d0facb9258444cad7665d1ee3cb4abe503eecf96f81c7765cb05
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:9ff12d7bc396bcbf6485afeea36b97cd00009a58356cf4afffbf5f3c2a60e7c9
           resources:
             requests:
               cpu: 100m
@@ -715,7 +718,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:9b359077ac6e68230219b4b8815cf02b937812b8bcf05f517dedd84e8526af7d
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:b6f5a6ff3b0de2a0e81c46fcf7bcb9ee9ac013a2c313ebe5416d040258652774
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -759,7 +762,7 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
   replicas: 0
@@ -771,7 +774,7 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20210614-393e4119b"
+        eventing.knative.dev/release: "v20210618-b164d3c04"
     spec:
       affinity:
         podAntiAffinity:
@@ -784,7 +787,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:5792ff450e1b55fe16ddf1f73e884e8c4eb9c17621e8251a429bf48bba66a148
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:67dec0dfc4f59331503b97b00097e157909778fe9bac5a01cb497c4d41bc46f0
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -842,7 +845,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 spec:
   scaleTargetRef:
     apiVersion: apps/v1
@@ -865,7 +868,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 spec:
   minAvailable: 80%
   selector:
@@ -893,7 +896,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 spec:
   selector:
     matchLabels: &labels
@@ -920,7 +923,7 @@ spec:
           terminationMessagePolicy: FallbackToLogsOnError
           # This is the Go import path for the binary that is containerized
           # and substituted here.
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:3d197fb76f7798bb619eef673dad8768affb3cdf926ef01cec9dfbe9482ca693
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:6112f611a8c46541baa139619ab7d04a4ed0f86a65df36aa5fb752218ab32b4b
           resources:
             requests:
               # taken from serving.
@@ -984,7 +987,7 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     role: eventing-webhook
   name: eventing-webhook
   namespace: knative-eventing
@@ -1016,7 +1019,7 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1230,7 +1233,7 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1391,7 +1394,7 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -1659,7 +1662,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1811,7 +1814,7 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -1938,7 +1941,7 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -2190,7 +2193,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -2379,7 +2382,7 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -2697,7 +2700,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
@@ -2887,7 +2890,7 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
 spec:
   group: messaging.knative.dev
@@ -2998,6 +3001,7 @@ spec:
                         namespace:
                           description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                           type: string
+                      x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature
                     uri:
                       description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
@@ -3094,7 +3098,7 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -3273,7 +3277,7 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3285,7 +3289,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3303,7 +3307,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3324,7 +3328,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3349,7 +3353,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3368,7 +3372,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: messaging-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3389,7 +3393,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3425,7 +3429,7 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   - apiGroups:
       - ""
@@ -3450,7 +3454,7 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   - apiGroups:
       - ""
@@ -3466,7 +3470,7 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   - apiGroups:
       - ""
@@ -3498,7 +3502,7 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3510,7 +3514,7 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/channelable: "true"
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3547,7 +3551,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3559,7 +3563,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3571,7 +3575,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3583,7 +3587,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3595,7 +3599,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3608,7 +3612,7 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
     resources: ["*"]
@@ -3620,7 +3624,7 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
     resources: ["*"]
@@ -3646,7 +3650,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   - apiGroups:
       - ""
@@ -3780,7 +3784,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   - apiGroups:
       - ""
@@ -3846,7 +3850,7 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3858,7 +3862,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/podspecable: "true"
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -3904,7 +3908,7 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3916,7 +3920,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     duck.knative.dev/source: "true"
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -3952,7 +3956,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   - apiGroups:
       - ""
@@ -4037,7 +4041,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   # For watching logging configuration and getting certs.
   - apiGroups:
@@ -4060,6 +4064,13 @@ rules:
       - "list"
       - "watch"
       - "patch"
+  # finalizers are needed for the owner reference of the webhook
+  - apiGroups:
+      - ""
+    resources:
+      - "namespaces/finalizers"
+    verbs:
+      - "update"
   # For getting our Deployment so we can decorate with ownerref.
   - apiGroups:
       - "apps"
@@ -4128,7 +4139,7 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 rules:
   # For manipulating certs into secrets.
   - apiGroups:
@@ -4163,7 +4174,7 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4199,7 +4210,7 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4231,7 +4242,7 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4264,7 +4275,7 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 # The data is populated at install time.
 
 ---
@@ -4287,7 +4298,7 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
diff --git a/third_party/eventing-latest/eventing-crds.yaml b/third_party/eventing-latest/eventing-crds.yaml
index 0fc167948..376663875 100644
--- a/third_party/eventing-latest/eventing-crds.yaml
+++ b/third_party/eventing-latest/eventing-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -231,7 +231,7 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -392,7 +392,7 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -660,7 +660,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -812,7 +812,7 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -939,7 +939,7 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1191,7 +1191,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1380,7 +1380,7 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1698,7 +1698,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
@@ -1888,7 +1888,7 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
 spec:
   group: messaging.knative.dev
@@ -1999,6 +1999,7 @@ spec:
                         namespace:
                           description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                           type: string
+                      x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature
                     uri:
                       description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
@@ -2095,7 +2096,7 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210614-393e4119b"
+    eventing.knative.dev/release: "v20210618-b164d3c04"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go
index d384bfd62..753d0644d 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/factory.go
@@ -58,7 +58,12 @@ func MakeFactory(ctor Ctor, unstructured bool, logger *zap.SugaredLogger) Factor
 	return func(t *testing.T, r *TableRow) (controller.Reconciler, ActionRecorderList, EventList) {
 		ls := NewListers(r.Objects)
 
-		ctx := context.Background()
+		var ctx context.Context
+		if r.Ctx != nil {
+			ctx = r.Ctx
+		} else {
+			ctx = context.Background()
+		}
 		ctx = logging.WithLogger(ctx, logger)
 
 		ctx, kubeClient := fakekubeclient.With(ctx, ls.GetKubeObjects()...)
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
index b4dd4083a..2d665b254 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
@@ -138,6 +138,19 @@ func WithSubscriptionSubscriberRef(gvk metav1.GroupVersionKind, name, namespace
 	}
 }
 
+func WithSubscriptionSubscriberRefUsingGroup(gvk metav1.GroupVersionKind, name, namespace string) SubscriptionOption {
+	return func(s *v1.Subscription) {
+		s.Spec.Subscriber = &duckv1.Destination{
+			Ref: &duckv1.KReference{
+				Group:     gvk.Group,
+				Kind:      gvk.Kind,
+				Name:      name,
+				Namespace: namespace,
+			},
+		}
+	}
+}
+
 func WithSubscriptionDeliveryRef(gvk metav1.GroupVersionKind, name, namespace string) SubscriptionOption {
 	return func(s *v1.Subscription) {
 		s.Spec.Delivery = &eventingduckv1.DeliverySpec{
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/channel_spec_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/channel_spec_test_helper.go
index 2c790a886..a784e5866 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/channel_spec_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/channel_spec_test_helper.go
@@ -39,32 +39,32 @@ func ChannelSpecTestHelperWithChannelTestRunner(
 	options ...testlib.SetupClientOption,
 ) {
 
-	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(st *testing.T, channel metav1.TypeMeta) {
-		client := testlib.Setup(st, true, options...)
-		defer testlib.TearDown(client)
-
+	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(t *testing.T, channel metav1.TypeMeta) {
 		t.Run("Channel spec allows subscribers", func(t *testing.T) {
+			client := testlib.Setup(t, true, options...)
+			defer testlib.TearDown(client)
+
 			if channel == channelv1 {
 				t.Skip("Not running spec.subscribers array test for generic Channel")
 			}
-			channelSpecAllowsSubscribersArray(st, client, channel)
+			channelSpecAllowsSubscribersArray(t, client, channel)
 		})
 	})
 }
 
-func channelSpecAllowsSubscribersArray(st *testing.T, client *testlib.Client, channel metav1.TypeMeta, options ...testlib.SetupClientOption) {
-	st.Logf("Running channel spec conformance test with channel %s", channel)
+func channelSpecAllowsSubscribersArray(t *testing.T, client *testlib.Client, channel metav1.TypeMeta, options ...testlib.SetupClientOption) {
+	t.Logf("Running channel spec conformance test with channel %s", channel)
 
 	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
 		dtsv, err := getChannelDuckTypeSupportVersionFromTypeMeta(client, channel)
 		if err != nil {
-			st.Fatalf("Unable to check Channel duck type support version for %s: %s", channel, err)
+			t.Fatalf("Unable to check Channel duck type support version for %s: %s", channel, err)
 		}
 		if dtsv != "v1" {
-			st.Fatalf("Unexpected duck type version, wanted [v1] got: %s", dtsv)
+			t.Fatalf("Unexpected duck type version, wanted [v1] got: %s", dtsv)
 		}
 		channelName := names.SimpleNameGenerator.GenerateName("channel-spec-subscribers-")
-		client.T.Logf("Creating channel %+v-%s", channel, channelName)
+		t.Logf("Creating channel %+v-%s", channel, channelName)
 		client.CreateChannelOrFail(channelName, &channel)
 		client.WaitForResourceReadyOrFail(channelName, &channel)
 
@@ -75,7 +75,7 @@ func channelSpecAllowsSubscribersArray(st *testing.T, client *testlib.Client, ch
 
 		channelable, err := getChannelAsChannelable(channelName, client, channel)
 		if err != nil {
-			st.Fatalf("Unable to get channel %s to v1 duck type: %s", channel, err)
+			t.Fatalf("Unable to get channel %s to v1 duck type: %s", channel, err)
 		}
 
 		// SPEC: each channel CRD MUST contain an array of subscribers: spec.subscribers
@@ -90,25 +90,25 @@ func channelSpecAllowsSubscribersArray(st *testing.T, client *testlib.Client, ch
 
 		raw, err := json.Marshal(ch)
 		if err != nil {
-			st.Fatalf("Error marshaling %s: %s", channel, err)
+			t.Fatalf("Error marshaling %s: %s", channel, err)
 		}
 		u := &unstructured.Unstructured{}
 		err = json.Unmarshal(raw, u)
 		if err != nil {
-			st.Fatalf("Error unmarshaling %s: %s", u, err)
+			t.Fatalf("Error unmarshaling %s: %s", u, err)
 		}
 
 		err = client.RetryWebhookErrors(func(attempt int) error {
 			_, e := client.Dynamic.Resource(gvr).Namespace(client.Namespace).Update(context.Background(), u, metav1.UpdateOptions{})
 			if e != nil {
-				client.T.Logf("Failed to update channel spec at attempt %d %q %q: %v", attempt, channel.Kind, channelName, e)
+				t.Logf("Failed to update channel spec at attempt %d %q %q: %v", attempt, channel.Kind, channelName, e)
 			}
 			return e
 		})
 		return err
 	})
 	if err != nil {
-		st.Fatalf("Error updating %s with subscribers in spec: %s", channel, err)
+		t.Fatalf("Error updating %s with subscribers in spec: %s", channel, err)
 	}
 }
 
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
index 684ad5d00..509ae6484 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
@@ -39,23 +39,23 @@ func ChannelStatusSubscriberTestHelperWithChannelTestRunner(
 	options ...testlib.SetupClientOption,
 ) {
 
-	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(st *testing.T, channel metav1.TypeMeta) {
-		client := testlib.Setup(st, true, options...)
-		defer testlib.TearDown(client)
-
+	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(t *testing.T, channel metav1.TypeMeta) {
 		t.Run("Channel has required status subscriber fields", func(t *testing.T) {
-			channelHasRequiredSubscriberStatus(ctx, st, client, channel, options...)
+			client := testlib.Setup(t, true, options...)
+			defer testlib.TearDown(client)
+
+			channelHasRequiredSubscriberStatus(ctx, t, client, channel, options...)
 		})
 	})
 }
 
-func channelHasRequiredSubscriberStatus(ctx context.Context, st *testing.T, client *testlib.Client, channel metav1.TypeMeta, options ...testlib.SetupClientOption) {
-	st.Logf("Running channel subscriber status conformance test with channel %q", channel)
+func channelHasRequiredSubscriberStatus(ctx context.Context, t *testing.T, client *testlib.Client, channel metav1.TypeMeta, options ...testlib.SetupClientOption) {
+	t.Logf("Running channel subscriber status conformance test with channel %q", channel)
 
 	channelName := "channel-req-status-subscriber"
 	subscriberServiceName := "channel-req-status-subscriber-svc"
 
-	client.T.Logf("Creating channel %+v-%s", channel, channelName)
+	t.Logf("Creating channel %+v-%s", channel, channelName)
 	client.CreateChannelOrFail(channelName, &channel)
 	client.WaitForResourceReadyOrFail(channelName, &channel)
 
@@ -73,29 +73,29 @@ func channelHasRequiredSubscriberStatus(ctx context.Context, st *testing.T, clie
 
 	dtsv, err := getChannelDuckTypeSupportVersion(channelName, client, &channel)
 	if err != nil {
-		st.Fatalf("Unable to check Channel duck type support version for %q: %q", channel, err)
+		t.Fatalf("Unable to check Channel duck type support version for %q: %q", channel, err)
 	}
 	if dtsv != "v1" {
-		st.Fatalf("Unexpected duck type version, wanted [v1] got: %s", dtsv)
+		t.Fatalf("Unexpected duck type version, wanted [v1] got: %s", dtsv)
 	}
 
 	channelable, err := getChannelAsChannelable(channelName, client, channel)
 	if err != nil {
-		st.Fatalf("Unable to get channel %q to v1 duck type: %q", channel, err)
+		t.Fatalf("Unable to get channel %q to v1 duck type: %q", channel, err)
 	}
 
 	// SPEC: Each subscription to a channel is added to the channel status.subscribers automatically.
 	if channelable.Status.Subscribers == nil {
-		st.Fatalf("%q does not have status.subscribers", channel)
+		t.Fatalf("%q does not have status.subscribers", channel)
 	}
 	ss := findSubscriberStatusV1(channelable.Status.Subscribers, subscription)
 	if ss == nil {
-		st.Fatalf("No subscription status found for channel %q and subscription %v", channel, subscription)
+		t.Fatalf("No subscription status found for channel %q and subscription %v", channel, subscription)
 	}
 
 	// SPEC: The ready field of the subscriber identified by its uid MUST be set to True when the subscription is ready to be processed.
 	if ss.Ready != corev1.ConditionTrue {
-		st.Fatalf("Subscription not ready found for channel %q and subscription %v", channel, subscription)
+		t.Fatalf("Subscription not ready found for channel %q and subscription %v", channel, subscription)
 	}
 }
 
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_test_helper.go
index 7ef45ab31..86686b38e 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_test_helper.go
@@ -32,46 +32,46 @@ func ChannelStatusTestHelperWithChannelTestRunner(
 	options ...testlib.SetupClientOption,
 ) {
 
-	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(st *testing.T, channel metav1.TypeMeta) {
-		client := testlib.Setup(st, true, options...)
-		defer testlib.TearDown(client)
-
+	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(t *testing.T, channel metav1.TypeMeta) {
 		t.Run("Channel has required status fields", func(t *testing.T) {
-			channelHasRequiredStatus(st, client, channel, options...)
+			client := testlib.Setup(t, true, options...)
+			defer testlib.TearDown(client)
+
+			channelHasRequiredStatus(t, client, channel, options...)
 		})
 	})
 }
 
-func channelHasRequiredStatus(st *testing.T, client *testlib.Client, channel metav1.TypeMeta, options ...testlib.SetupClientOption) {
-	st.Logf("Running channel status conformance test with channel %q", channel)
+func channelHasRequiredStatus(t *testing.T, client *testlib.Client, channel metav1.TypeMeta, options ...testlib.SetupClientOption) {
+	t.Logf("Running channel status conformance test with channel %q", channel)
 
 	channelName := "channel-req-status"
 
-	client.T.Logf("Creating channel %+v-%s", channel, channelName)
+	t.Logf("Creating channel %+v-%s", channel, channelName)
 	client.CreateChannelOrFail(channelName, &channel)
 	client.WaitForResourceReadyOrFail(channelName, &channel)
 
 	dtsv, err := getChannelDuckTypeSupportVersion(channelName, client, &channel)
 	if err != nil {
-		st.Fatalf("Unable to check Channel duck type support version for %q: %q", channel, err)
+		t.Fatalf("Unable to check Channel duck type support version for %q: %q", channel, err)
 	}
 	if dtsv != "v1" {
-		st.Fatalf("Unexpected duck type version, wanted [v1] got: %s", dtsv)
+		t.Fatalf("Unexpected duck type version, wanted [v1] got: %s", dtsv)
 	}
 
 	channelable, err := getChannelAsChannelable(channelName, client, channel)
 	if err != nil {
-		st.Fatalf("Unable to get channel %q to v1 duck type: %q", channel, err)
+		t.Fatalf("Unable to get channel %q to v1 duck type: %q", channel, err)
 	}
 
 	// SPEC: Channel CRD MUST have a status subresource which contains address
 	if channelable.Status.AddressStatus.Address == nil {
-		st.Fatalf("%q does not have status.address", channel)
+		t.Fatalf("%q does not have status.address", channel)
 	}
 
 	// SPEC: When the channel instance is ready to receive events status.address.hostname and
 	// status.address.url MUST be populated
 	if channelable.Status.Address.URL.IsEmpty() {
-		st.Fatalf("No hostname found for %q", channel)
+		t.Fatalf("No hostname found for %q", channel)
 	}
 }
diff --git a/vendor/knative.dev/eventing/test/lib/test_runner.go b/vendor/knative.dev/eventing/test/lib/test_runner.go
index 4066da298..986e58598 100644
--- a/vendor/knative.dev/eventing/test/lib/test_runner.go
+++ b/vendor/knative.dev/eventing/test/lib/test_runner.go
@@ -18,10 +18,13 @@ package lib
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"os"
 	"path/filepath"
 	"sort"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -29,10 +32,7 @@ import (
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apiserver/pkg/storage/names"
-
 	pkgTest "knative.dev/pkg/test"
-	"knative.dev/pkg/test/helpers"
 	"knative.dev/pkg/test/prow"
 
 	"knative.dev/eventing/pkg/utils"
@@ -47,6 +47,15 @@ import (
 const (
 	podLogsDir         = "pod-logs"
 	testPullSecretName = "kn-eventing-test-pull-secret"
+	MaxNamespaceSkip   = 20
+	MaxRetries         = 5
+	RetrySleepDuration = 2 * time.Second
+)
+
+var (
+	nsMutex        sync.Mutex
+	namespaceCount int
+	ReuseNamespace bool
 )
 
 // ComponentsTestRunner is used to run tests against different eventing components.
@@ -148,19 +157,16 @@ var SetupClientOptionNoop SetupClientOption = func(*Client) {
 // Setup creates the client objects needed in the e2e tests,
 // and does other setups, like creating namespaces, set the test case to run in parallel, etc.
 func Setup(t *testing.T, runInParallel bool, options ...SetupClientOption) *Client {
-	// Create a new namespace to run this test case.
-	namespace := makeK8sNamespace(t.Name())
-	t.Logf("namespace is : %q", namespace)
-	client, err := NewClient(
-		pkgTest.Flags.Kubeconfig,
-		pkgTest.Flags.Cluster,
-		namespace,
-		t)
+	client, err := CreateNamespacedClient(t)
 	if err != nil {
 		t.Fatal("Couldn't initialize clients:", err)
 	}
 
-	CreateNamespaceIfNeeded(t, client, namespace)
+	// If namespaces are re-used the pull-secret is supposed to be created in advance.
+	if !ReuseNamespace {
+		SetupServiceAccount(t, client)
+		SetupPullSecret(t, client)
+	}
 
 	// Run the test case in parallel if needed.
 	if runInParallel {
@@ -178,9 +184,73 @@ func Setup(t *testing.T, runInParallel bool, options ...SetupClientOption) *Clie
 	return client
 }
 
-func makeK8sNamespace(baseFuncName string) string {
-	base := helpers.MakeK8sNamePrefix(baseFuncName)
-	return names.SimpleNameGenerator.GenerateName(base + "-")
+func CreateNamespacedClient(t *testing.T) (*Client, error) {
+	ns := ""
+	// Try next MaxNamespaceSkip namespaces before giving up. This should address the issue with
+	// development cycles when namespaces from previous runs were not cleaned properly.
+	for i := 0; i < MaxNamespaceSkip; i++ {
+		ns = NextNamespace()
+		client, err := NewClient(
+			pkgTest.Flags.Kubeconfig,
+			pkgTest.Flags.Cluster,
+			ns,
+			t)
+		if err != nil {
+			return nil, err
+		}
+		if ReuseNamespace {
+			// Re-using existing namespace, no need to create it.
+			// The namespace is supposed to be created in advance.
+			return client, nil
+		} else {
+			// The test is supposed to create a new test namespace for itself.
+			// Keep trying until we find a namespace that doesn't exist yet.
+			if err := CreateNamespaceWithRetry(client, ns); err != nil {
+				if apierrs.IsAlreadyExists(err) {
+					continue
+				}
+				return nil, err
+			}
+		}
+		return client, nil
+	}
+	return nil, errors.New("unable to find available namespace")
+}
+
+// NextNamespace returns the next unique namespace.
+func NextNamespace() string {
+	ns := os.Getenv("EVENTING_E2E_NAMESPACE")
+	if ns == "" {
+		ns = "eventing-e2e"
+	}
+	return fmt.Sprintf("%s%d", ns, GetNextNamespaceId())
+}
+
+// GetNextNamespaceId return the next unique ID for the next namespace.
+func GetNextNamespaceId() int {
+	nsMutex.Lock()
+	defer nsMutex.Unlock()
+	current := namespaceCount
+	namespaceCount++
+	return current
+}
+
+// CreateNamespaceWithRetry creates the given namespace with retries.
+func CreateNamespaceWithRetry(client *Client, namespace string) error {
+	var (
+		retries int
+		err     error
+	)
+	for retries < MaxRetries {
+		nsSpec := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
+		if _, err = client.Kube.CoreV1().Namespaces().
+			Create(context.Background(), nsSpec, metav1.CreateOptions{}); err == nil {
+			return nil
+		}
+		retries++
+		time.Sleep(RetrySleepDuration)
+	}
+	return err
 }
 
 // TearDown will delete created names using clients.
@@ -227,8 +297,11 @@ func TearDown(client *Client) {
 	}
 
 	client.Tracker.Clean(true)
-	if err := DeleteNameSpace(client); err != nil {
-		client.T.Logf("Could not delete the namespace %q: %v", client.Namespace, err)
+	// If we're reusing existing namespaces leave the deletion to the creator.
+	if !ReuseNamespace {
+		if err := DeleteNameSpace(client); err != nil {
+			client.T.Logf("Could not delete the namespace %q: %v", client.Namespace, err)
+		}
 	}
 }
 
@@ -253,33 +326,25 @@ func formatEvent(e *corev1.Event) string {
 	}, "\n")
 }
 
-// CreateNamespaceIfNeeded creates a new namespace if it does not exist.
-func CreateNamespaceIfNeeded(t *testing.T, client *Client, namespace string) {
-	_, err := client.Kube.CoreV1().Namespaces().Get(context.Background(), namespace, metav1.GetOptions{})
-
-	if err != nil && apierrs.IsNotFound(err) {
-		nsSpec := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
-		_, err = client.Kube.CoreV1().Namespaces().Create(context.Background(), nsSpec, metav1.CreateOptions{})
-
-		if err != nil {
-			t.Fatalf("Failed to create Namespace: %s; %v", namespace, err)
-		}
-
-		// https://github.com/kubernetes/kubernetes/issues/66689
-		// We can only start creating pods after the default ServiceAccount is created by the kube-controller-manager.
-		err = waitForServiceAccountExists(client, "default", namespace)
-		if err != nil {
-			t.Fatal("The default ServiceAccount was not created for the Namespace:", namespace)
-		}
+// SetupServiceAccount creates a new namespace if it does not exist.
+func SetupServiceAccount(t *testing.T, client *Client) {
+	// https://github.com/kubernetes/kubernetes/issues/66689
+	// We can only start creating pods after the default ServiceAccount is created by the kube-controller-manager.
+	err := waitForServiceAccountExists(client, "default", client.Namespace)
+	if err != nil {
+		t.Fatal("The default ServiceAccount was not created for the Namespace:", client.Namespace)
+	}
+}
 
-		// If the "default" Namespace has a secret called
-		// "kn-eventing-test-pull-secret" then use that as the ImagePullSecret
-		// on the "default" ServiceAccount in this new Namespace.
-		// This is needed for cases where the images are in a private registry.
-		_, err := utils.CopySecret(client.Kube.CoreV1(), "default", testPullSecretName, namespace, "default")
-		if err != nil && !apierrs.IsNotFound(err) {
-			t.Fatalf("error copying the secret into ns %q: %s", namespace, err)
-		}
+// SetupPullSecret sets up kn-eventing-test-pull-secret on the client namespace.
+func SetupPullSecret(t *testing.T, client *Client) {
+	// If the "default" Namespace has a secret called
+	// "kn-eventing-test-pull-secret" then use that as the ImagePullSecret
+	// on the "default" ServiceAccount in this new Namespace.
+	// This is needed for cases where the images are in a private registry.
+	_, err := utils.CopySecret(client.Kube.CoreV1(), "default", testPullSecretName, client.Namespace, "default")
+	if err != nil && !apierrs.IsNotFound(err) {
+		t.Fatalf("error copying the secret into ns %q: %s", client.Namespace, err)
 	}
 }
 
diff --git a/vendor/knative.dev/eventing/test/test_images/event-library/main.go b/vendor/knative.dev/eventing/test/test_images/event-library/main.go
new file mode 100644
index 000000000..ee1388b0c
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/test_images/event-library/main.go
@@ -0,0 +1,45 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/kelseyhightower/envconfig"
+	_ "knative.dev/pkg/system/testing"
+)
+
+type envConfig struct {
+	DataPath string `envconfig:"KO_DATA_PATH" default:"/var/run/ko/" required:"true"`
+	Port     int    `envconfig:"PORT" default:"8080" required:"true"`
+}
+
+var env envConfig
+
+func main() {
+	if err := envconfig.Process("", &env); err != nil {
+		log.Fatalf("failed to process env var: %s", err)
+	}
+
+	m := http.NewServeMux()
+	m.Handle("/", http.FileServer(http.Dir(env.DataPath)))
+
+	log.Println("Listening on", env.Port)
+	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", env.Port), m))
+}
diff --git a/vendor/knative.dev/pkg/test/environment/config.go b/vendor/knative.dev/pkg/test/environment/config.go
index 39cfe9014..6591a4637 100644
--- a/vendor/knative.dev/pkg/test/environment/config.go
+++ b/vendor/knative.dev/pkg/test/environment/config.go
@@ -17,7 +17,7 @@ limitations under the License.
 // This file contains logic to encapsulate flags which are needed to specify
 // what cluster, etc. to use for e2e tests.
 
-package flags
+package environment
 
 import (
 	"flag"