Repository: knative-sandbox/eventing-kafka-broker
Pull request #1028 (2021-06-22 15:09:40)

Split RecordDispatcher in interface and implementation class
My plan is to reorganize the code inside the dispatcher package and inside the http package in particular, and decouple a bit more and reorganize the concerns of consumer verticles and record dispatcher. This is one step ahead, next PRs I'll do the actual move of the code. 

## Proposed Changes

- :broom: Split RecordDispatcher in interface and implementation class
- :broom: RecordDispatcher doesn't implement anymore `Function` anymore, nor `Handler`. The `Handler` implementation was a bad idea, because it's hiding a technical detail just for the sake of being easier to use for a single user (the UnorderedConsumerVerticle) 
- :broom: Rename OffsetManager in RecordDispatcherListener, which better resembles the interface purpose.

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/1028#" title="Author self-approved">slinkydeveloper</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)~~ [slinkydeveloper]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: @slinkydeveloper: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 3891811619de8590754060ffdd51cd7c636939cd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1028/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1407355117849546752) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1028). [Your PR dashboard](https://gubernator.knative.dev/pr/slinkydeveloper).

<details>

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

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 3891811619de8590754060ffdd51cd7c636939cd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1028/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1407355117849546752) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 3891811619de8590754060ffdd51cd7c636939cd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1028/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1407355117883101184) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1028). [Your PR dashboard](https://gubernator.knative.dev/pr/slinkydeveloper).

<details>

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/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/1028?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    #1028      +/-   ##
============================================
- Coverage     75.10%   70.36%   -4.74%     
============================================
  Files            75       29      -46     
  Lines          2928     1343    -1585     
  Branches        128        0     -128     
============================================
- Hits           2199      945    -1254     
+ Misses          570      305     -265     
+ Partials        159       93      -66     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

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/1028?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...enting/kafka/broker/core/utils/Configurations.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29uZmlndXJhdGlvbnMuamF2YQ==) | | |
| [...enting/kafka/broker/core/OrderedAsyncExecutor.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvT3JkZXJlZEFzeW5jRXhlY3V0b3IuamF2YQ==) | | |
| [...nting/kafka/broker/core/tracing/TracingConfig.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9UcmFjaW5nQ29uZmlnLmphdmE=) | | |
| [...roker/dispatcher/http/HttpSinkResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaHR0cC9IdHRwU2lua1Jlc3BvbnNlSGFuZGxlci5qYXZh) | | |
| [...fka/broker/core/security/CredentialsValidator.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvQ3JlZGVudGlhbHNWYWxpZGF0b3IuamF2YQ==) | | |
| [.../core/reconciler/impl/ResourcesReconcilerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJJbXBsLmphdmE=) | | |
| [...g/kafka/broker/core/security/KafkaClientsAuth.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS2Fma2FDbGllbnRzQXV0aC5qYXZh) | | |
| [...tive/eventing/kafka/broker/core/utils/Logging.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvTG9nZ2luZy5qYXZh) | | |
| [.../eventing/kafka/broker/receiver/RequestMapper.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9SZXF1ZXN0TWFwcGVyLmphdmE=) | | |
| [...fka/broker/core/security/PlaintextCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvUGxhaW50ZXh0Q3JlZGVudGlhbHMuamF2YQ==) | | |
| ... and [29 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028/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/1028?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/1028?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [f83ac68...3891811](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1028?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).


Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 0d120a04c..24c126c6f 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -1,4 +1,5 @@
-# This file is auto-generated from peribolos
+# This file is auto-generated from peribolos.
+# Do not modify this file, instead modify peribolos/knative-sandbox.yaml
 
 aliases:
   api-core-wg-leads:
diff --git a/control-plane/pkg/core/config/utils.go b/control-plane/pkg/core/config/utils.go
index 7018a28da..efea8777c 100644
--- a/control-plane/pkg/core/config/utils.go
+++ b/control-plane/pkg/core/config/utils.go
@@ -55,7 +55,7 @@ func EgressConfigFromDelivery(
 	defaultBackoffDelayMs uint64,
 ) (*contract.EgressConfig, error) {
 
-	if delivery == nil || (delivery.DeadLetterSink == nil && delivery.Retry == nil) {
+	if delivery == nil || (delivery.DeadLetterSink == nil && delivery.Retry == nil && delivery.Timeout == nil) {
 		return nil, nil
 	}
 
@@ -72,7 +72,7 @@ func EgressConfigFromDelivery(
 	if delivery.Retry != nil {
 		egressConfig.Retry = uint32(*delivery.Retry)
 		var err error
-		delay, err := BackoffDelayFromISO8601String(delivery.BackoffDelay, defaultBackoffDelayMs)
+		delay, err := DurationMillisFromISO8601String(delivery.BackoffDelay, defaultBackoffDelayMs)
 		if err != nil {
 			return nil, fmt.Errorf("failed to parse Spec.Delivery.BackoffDelay: %w", err)
 		}
@@ -80,6 +80,18 @@ func EgressConfigFromDelivery(
 		egressConfig.BackoffPolicy = BackoffPolicyFromString(delivery.BackoffPolicy)
 	}
 
+	if delivery.Timeout != nil {
+		var err error
+		timeout, err := DurationMillisFromISO8601String(
+			delivery.Timeout,
+			0, /* 0 means absent, allowing the data plane to default it */
+		)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse Spec.Delivery.Timeout: %w", err)
+		}
+		egressConfig.Timeout = timeout
+	}
+
 	return egressConfig, nil
 }
 
@@ -102,17 +114,17 @@ func BackoffPolicyFromString(backoffPolicy *duck.BackoffPolicyType) contract.Bac
 	}
 }
 
-// BackoffDelayFromISO8601String returns the BackoffDelay from the given string.
+// DurationMillisFromISO8601String returns the duration in milliseconds from the given string.
 //
 // Default value is the specified defaultDelay.
-func BackoffDelayFromISO8601String(backoffDelay *string, defaultDelay uint64) (uint64, error) {
-	if backoffDelay == nil {
-		return defaultDelay, nil
+func DurationMillisFromISO8601String(durationStr *string, defaultDurationMillis uint64) (uint64, error) {
+	if durationStr == nil {
+		return defaultDurationMillis, nil
 	}
 
-	d, err := period.Parse(*backoffDelay, false)
+	d, err := period.Parse(*durationStr, false)
 	if err != nil {
-		return 0, fmt.Errorf("failed to parse backoffDelay: %w", err)
+		return 0, fmt.Errorf("failed to parse duration string: %w", err)
 	}
 
 	ms, _ := d.Duration()
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index c5fcaf506..c15d6344f 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -66,8 +66,8 @@ Lists of 153 third-party dependencies.
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Server Mock (io.fabric8:kubernetes-server-mock:5.4.1 - http://fabric8.io/kubernetes-server-mock/)
      (Apache License, Version 2.0) Fabric8 :: Mock Web Server (io.fabric8:mockwebserver:0.1.8 - http://fabric8.io/)
      (The Apache Software License, Version 2.0) zjsonpatch (io.fabric8:zjsonpatch:0.3.0 - https://github.com/fabric8io/zjsonpatch/)
-     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.7.0 - https://github.com/micrometer-metrics/micrometer)
-     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.7.0 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.7.1 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.7.1 - https://github.com/micrometer-metrics/micrometer)
      (Apache License, Version 2.0) Netty/Buffer (io.netty:netty-buffer:4.1.65.Final - https://netty.io/netty-buffer/)
      (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.65.Final - https://netty.io/netty-codec/)
      (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.65.Final - https://netty.io/netty-codec-dns/)
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/ResourcesReconciler.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/ResourcesReconciler.java
index ee1d20d0c..9a2c41ddd 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/ResourcesReconciler.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/ResourcesReconciler.java
@@ -16,6 +16,7 @@
 package dev.knative.eventing.kafka.broker.core.reconciler;
 
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerBuilder;
 import io.vertx.core.Future;
 import java.util.Collection;
 
@@ -24,4 +25,8 @@ public interface ResourcesReconciler {
 
   Future<Void> reconcile(Collection<DataPlaneContract.Resource> resources);
 
+  static ResourcesReconcilerBuilder builder() {
+    return new ResourcesReconcilerBuilder();
+  }
+
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerBuilder.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerBuilder.java
new file mode 100644
index 000000000..f3f08be72
--- /dev/null
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerBuilder.java
@@ -0,0 +1,60 @@
+/*
+ * 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.core.reconciler.impl;
+
+import dev.knative.eventing.kafka.broker.core.reconciler.EgressReconcilerListener;
+import dev.knative.eventing.kafka.broker.core.reconciler.IngressReconcilerListener;
+import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
+import io.vertx.core.Vertx;
+import io.vertx.core.eventbus.MessageConsumer;
+import java.util.Objects;
+
+public class ResourcesReconcilerBuilder {
+
+  private IngressReconcilerListener ingressReconcilerListener;
+  private EgressReconcilerListener egressReconcilerListener;
+
+  public ResourcesReconcilerBuilder watchIngress(
+    IngressReconcilerListener ingressReconcilerListener) {
+    Objects.requireNonNull(ingressReconcilerListener);
+    this.ingressReconcilerListener = ingressReconcilerListener;
+    return this;
+  }
+
+  public ResourcesReconcilerBuilder watchEgress(
+    EgressReconcilerListener egressReconcilerListener) {
+    Objects.requireNonNull(egressReconcilerListener);
+    this.egressReconcilerListener = egressReconcilerListener;
+    return this;
+  }
+
+  /**
+   * Build the {@link ResourcesReconciler} and start listening for new contracts using {@link ResourcesReconcilerMessageHandler}.
+   *
+   * @param vertx the vertx object to register the event bus listener
+   * @return the built message consumer
+   */
+  public MessageConsumer<Object> buildAndListen(Vertx vertx) {
+    return ResourcesReconcilerMessageHandler.start(vertx, build());
+  }
+
+  /**
+   * @return the built {@link ResourcesReconciler}
+   */
+  public ResourcesReconciler build() {
+    return new ResourcesReconcilerImpl(ingressReconcilerListener, egressReconcilerListener);
+  }
+}
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
index 7362e20b3..bc8301442 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
@@ -50,7 +50,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
   // egress uid -> <Egress, Resource>
   private final Map<String, Map.Entry<DataPlaneContract.Egress, DataPlaneContract.Resource>> cachedEgresses;
 
-  private ResourcesReconcilerImpl(
+  ResourcesReconcilerImpl(
     IngressReconcilerListener ingressReconcilerListener,
     EgressReconcilerListener egressReconcilerListener) {
     if (ingressReconcilerListener == null && egressReconcilerListener == null) {
@@ -246,33 +246,6 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
       && Objects.equals(e1.getFilter(), e2.getFilter());
   }
 
-  public static Builder builder() {
-    return new Builder();
-  }
-
-  public static class Builder {
-    private IngressReconcilerListener ingressReconcilerListener;
-    private EgressReconcilerListener egressReconcilerListener;
-
-    public Builder watchIngress(
-      IngressReconcilerListener ingressReconcilerListener) {
-      Objects.requireNonNull(ingressReconcilerListener);
-      this.ingressReconcilerListener = ingressReconcilerListener;
-      return this;
-    }
-
-    public Builder watchEgress(
-      EgressReconcilerListener egressReconcilerListener) {
-      Objects.requireNonNull(egressReconcilerListener);
-      this.egressReconcilerListener = egressReconcilerListener;
-      return this;
-    }
-
-    public ResourcesReconcilerImpl build() {
-      return new ResourcesReconcilerImpl(ingressReconcilerListener, egressReconcilerListener);
-    }
-  }
-
   private static void logFailure(final String msg, final DataPlaneContract.Egress egress, final Throwable cause) {
     logger.error(msg + " {} {} {}",
       keyValue("id", egress.getUid()),
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnv.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnv.java
index 68a446a39..4ac5ab8cd 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnv.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/utils/BaseEnv.java
@@ -22,27 +22,27 @@ import static java.util.Objects.requireNonNull;
 public class BaseEnv {
 
   public static final String PRODUCER_CONFIG_FILE_PATH = "PRODUCER_CONFIG_FILE_PATH";
-  public static final String DATA_PLANE_CONFIG_FILE_PATH = "DATA_PLANE_CONFIG_FILE_PATH";
-
-  public static final String METRICS_PORT = "METRICS_PORT";
-  public static final String METRICS_PATH = "METRICS_PATH";
-  public static final String METRICS_PUBLISH_QUANTILES = "METRICS_PUBLISH_QUANTILES";
-  public static final String METRICS_JVM_ENABLED = "METRICS_JVM_ENABLED";
-
-  public static final String CONFIG_TRACING_PATH = "CONFIG_TRACING_PATH";
-
-  public static final String WAIT_STARTUP_SECONDS = "WAIT_STARTUP_SECONDS";
-
   private final String producerConfigFilePath;
+
+  public static final String DATA_PLANE_CONFIG_FILE_PATH = "DATA_PLANE_CONFIG_FILE_PATH";
   private final String dataPlaneConfigFilePath;
 
+  public static final String METRICS_PORT = "METRICS_PORT";
   private final int metricsPort;
+
+  public static final String METRICS_PATH = "METRICS_PATH";
   private final String metricsPath;
+
+  public static final String METRICS_PUBLISH_QUANTILES = "METRICS_PUBLISH_QUANTILES";
   private final boolean metricsPublishQuantiles;
+
+  public static final String METRICS_JVM_ENABLED = "METRICS_JVM_ENABLED";
   private final boolean metricsJvmEnabled;
 
+  public static final String CONFIG_TRACING_PATH = "CONFIG_TRACING_PATH";
   private final String configTracingPath;
 
+  public static final String WAIT_STARTUP_SECONDS = "WAIT_STARTUP_SECONDS";
   private final int waitStartupSeconds;
 
   public BaseEnv(Function<String, String> envProvider) {
diff --git a/data-plane/dispatcher/pom.xml b/data-plane/dispatcher/pom.xml
index 1ea1686e5..1db367507 100644
--- a/data-plane/dispatcher/pom.xml
+++ b/data-plane/dispatcher/pom.xml
@@ -133,9 +133,7 @@
                 <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                 <transformer
                   implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                  <mainClass>
-                    dev.knative.eventing.kafka.broker.dispatcher.Main
-                  </mainClass>
+                  <mainClass>dev.knative.eventing.kafka.broker.dispatcher.main.Main</mainClass>
                 </transformer>
               </transformers>
               <filters>
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/ConsumerDeployerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/ConsumerDeployerVerticle.java
index dd54e5314..04c336c5d 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/ConsumerDeployerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/ConsumerDeployerVerticle.java
@@ -17,8 +17,7 @@ package dev.knative.eventing.kafka.broker.dispatcher;
 
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.reconciler.EgressReconcilerListener;
-import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerImpl;
-import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerMessageHandler;
+import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import dev.knative.eventing.kafka.broker.dispatcher.consumer.ConsumerVerticleFactory;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Future;
@@ -33,11 +32,7 @@ import org.slf4j.LoggerFactory;
 import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
 /**
- * ResourcesManager manages Resource and Egress objects by instantiating and starting verticles based on resources
- * configurations.
- *
- * <p>Note: {@link ConsumerDeployerVerticle} is not thread-safe and it's not supposed to be shared between
- * threads.
+ * This verticle listens on Egress reconciliations by deploying/undeploying new consumer verticles.
  */
 public final class ConsumerDeployerVerticle extends AbstractVerticle implements EgressReconcilerListener {
 
@@ -67,13 +62,10 @@ public final class ConsumerDeployerVerticle extends AbstractVerticle implements
 
   @Override
   public void start() {
-    this.messageConsumer = ResourcesReconcilerMessageHandler.start(
-      vertx,
-      ResourcesReconcilerImpl
-        .builder()
-        .watchEgress(this)
-        .build()
-    );
+    this.messageConsumer = ResourcesReconciler
+      .builder()
+      .watchEgress(this)
+      .buildAndListen(vertx);
   }
 
   @Override
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 76fc13af7..2353d4672 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
@@ -16,239 +16,15 @@
 package dev.knative.eventing.kafka.broker.dispatcher;
 
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
-import dev.knative.eventing.kafka.broker.core.filter.Filter;
-import dev.knative.eventing.kafka.broker.dispatcher.consumer.OffsetManager;
 import io.cloudevents.CloudEvent;
-import io.vertx.core.CompositeFuture;
-import io.vertx.core.Context;
 import io.vertx.core.Future;
-import io.vertx.core.Handler;
-import io.vertx.core.Promise;
-import io.vertx.core.Vertx;
-import io.vertx.kafka.client.common.tracing.ConsumerTracer;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import java.util.Objects;
-import java.util.function.Function;
-import java.util.function.Supplier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
 /**
- * This class implements the core algorithm of the Dispatcher (see {@link
- * RecordDispatcher#handle(KafkaConsumerRecord)}).
- *
- * @see RecordDispatcher#handle(KafkaConsumerRecord)
+ * This interface performs the dispatch of consumed records.
  */
-public class RecordDispatcher implements Handler<KafkaConsumerRecord<String, CloudEvent>>,
-  Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>>, AsyncCloseable {
-
-  private static final Logger logger = LoggerFactory.getLogger(RecordDispatcher.class);
-
-  private final Filter filter;
-  private final Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> subscriberSender;
-  private final Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> dlsSender;
-  private final OffsetManager offsetManager;
-  private final AsyncCloseable closeable;
-  private final ConsumerTracer consumerTracer;
-
-  /**
-   * All args constructor.
-   *
-   * @param filter                event filter
-   * @param subscriberSender      sender to trigger subscriber
-   * @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
-   */
-  public RecordDispatcher(
-    final Filter filter,
-    final ConsumerRecordSender subscriberSender,
-    final ConsumerRecordSender deadLetterSinkSender,
-    final SinkResponseHandler sinkResponseHandler,
-    final OffsetManager offsetManager,
-    final ConsumerTracer consumerTracer) {
-    Objects.requireNonNull(filter, "provide filter");
-    Objects.requireNonNull(subscriberSender, "provide subscriberSender");
-    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.dlsSender = composeSenderAndSinkHandler(deadLetterSinkSender, sinkResponseHandler, "dead letter sink");
-    this.offsetManager = offsetManager;
-    this.closeable = AsyncCloseable.compose(sinkResponseHandler, deadLetterSinkSender, subscriberSender);
-    this.consumerTracer = consumerTracer;
-  }
-
-  /**
-   * Handle the given record and returns a future that completes when the dispatch is completed and the offset is committed
-   *
-   * @param record record to handle.
-   */
-  @Override
-  public Future<Void> apply(KafkaConsumerRecord<String, CloudEvent> record) {
-    Promise<Void> promise = Promise.promise();
-
-    /*
-    That's pretty much what happens here:
-
-    +------------+    +------------+     +-------------+
-    |            |    |            |     |             |
-    |  record    |    |  filter    |     |  subscriber |
-    |  received  +--->+  matching  +---->+  success    +-->end
-    |            |    |            |     |             |
-    +-+-------+--+    +-----+------+     +-----+-------+
-      |       |             |                  |
-      |       v             |                  |             +-------------+
-      |    +--+---------+   |           +------v-------+     |             |
-      |    |            |   |           |  subscriber  +---->+  dls        |
-      |    | filter not |   +---------->+  failure     |     |  success    +---->end
-      |    | matching   |               +----+---------+     |             |
-      |    |            |                    |               +-----+-------+
-      |    +---+--------+              +-----v-------+             |
-      |        |                       | dls failure |             v
-      |        |                       +-------------+----------> end
-      +->end<--+
-     */
-
-    onRecordReceived(record, promise);
-
-    return promise.future();
-  }
-
-  /**
-   * Like {@link #apply(KafkaConsumerRecord)} but discards the returned future
-   */
-  @Override
-  public void handle(final KafkaConsumerRecord<String, CloudEvent> record) {
-    this.apply(record);
-  }
-
-  private void onRecordReceived(final KafkaConsumerRecord<String, CloudEvent> record, Promise<Void> finalProm) {
-    logDebug("Handling record", record);
-
-    // Trace record received event
-    // This needs to be done manually in order to work properly with both ordered and unordered delivery.
-    if (consumerTracer != null) {
-      Context context = Vertx.currentContext();
-      ConsumerTracer.StartedSpan span = consumerTracer.prepareMessageReceived(context, record.record());
-      if (span != null) {
-        finalProm.future()
-          .onComplete(ar -> {
-            if (ar.succeeded()) {
-              span.finish(context);
-            } else {
-              span.fail(context, ar.cause());
-            }
-          });
-      }
-    }
-
-    offsetManager.recordReceived(record)
-      .onSuccess(v -> {
-        // Execute filtering
-        if (filter.test(record.value())) {
-          onFilterMatching(record, finalProm);
-        } else {
-          onFilterNotMatching(record, finalProm);
-        }
-      })
-      .onFailure(finalProm::fail); // This should really never happen
-  }
-
-  private void onFilterMatching(final KafkaConsumerRecord<String, CloudEvent> record, final Promise<Void> finalProm) {
-    logDebug("Record match filtering", record);
-    subscriberSender.apply(record)
-      .onSuccess(res -> onSubscriberSuccess(record, finalProm))
-      .onFailure(ex -> onSubscriberFailure(record, finalProm));
-  }
-
-  private void onFilterNotMatching(final KafkaConsumerRecord<String, CloudEvent> record,
-                                   final Promise<Void> finalProm) {
-    logDebug("Record doesn't match filtering", record);
-    offsetManager.recordDiscarded(record)
-      .onComplete(finalProm);
-  }
-
-  private void onSubscriberSuccess(final KafkaConsumerRecord<String, CloudEvent> record,
-                                   final Promise<Void> finalProm) {
-    logDebug("Successfully sent event to subscriber", record);
-    offsetManager.successfullySentToSubscriber(record)
-      .onComplete(finalProm);
-  }
-
-  private void onSubscriberFailure(final KafkaConsumerRecord<String, CloudEvent> record,
-                                   final Promise<Void> finalProm) {
-    dlsSender.apply(record)
-      .onSuccess(v -> onDeadLetterSinkSuccess(record, finalProm))
-      .onFailure(ex -> onDeadLetterSinkFailure(record, ex, finalProm));
-  }
-
-  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 onDeadLetterSinkFailure(final KafkaConsumerRecord<String, CloudEvent> record, final Throwable exception,
-                                       final Promise<Void> finalProm) {
-    offsetManager.failedToSendToDeadLetterSink(record, exception)
-      .onComplete(finalProm);
-  }
-
-  private static Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> composeSenderAndSinkHandler(
-    ConsumerRecordSender sender, SinkResponseHandler sinkHandler, String senderType) {
-    return rec -> sender.send(rec)
-      .onFailure(ex -> logError("Failed to send event to " + senderType, rec, ex))
-      .compose(res ->
-        sinkHandler.handle(res)
-          .onFailure(ex -> logError("Failed to handle " + senderType + " response", rec, ex))
-      );
-  }
-
-  private static void logError(
-    final String msg,
-    final KafkaConsumerRecord<String, CloudEvent> record,
-    final Throwable cause) {
-
-    if (logger.isDebugEnabled()) {
-      logger.error(msg + " {} {} {} {} {}",
-        keyValue("topic", record.topic()),
-        keyValue("partition", record.partition()),
-        keyValue("headers", record.headers()),
-        keyValue("offset", record.offset()),
-        keyValue("event", record.value()),
-        cause
-      );
-    } else {
-      logger.error(msg + " {} {} {}",
-        keyValue("topic", record.topic()),
-        keyValue("partition", record.partition()),
-        keyValue("offset", record.offset()),
-        cause
-      );
-    }
-  }
-
-  private static void logDebug(
-    final String msg,
-    final KafkaConsumerRecord<String, CloudEvent> record) {
+public interface RecordDispatcher extends AsyncCloseable {
 
-    logger.debug(msg + " {} {} {} {} {}",
-      keyValue("topic", record.topic()),
-      keyValue("partition", record.partition()),
-      keyValue("headers", record.headers()),
-      keyValue("offset", record.offset()),
-      keyValue("event", record.value())
-    );
-  }
+  Future<Void> dispatch(KafkaConsumerRecord<String, CloudEvent> record);
 
-  public Future<Void> close() {
-    return this.closeable.close();
-  }
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherImpl.java
new file mode 100644
index 000000000..faab25cca
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherImpl.java
@@ -0,0 +1,242 @@
+/*
+ * 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.dispatcher;
+
+import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
+import dev.knative.eventing.kafka.broker.core.filter.Filter;
+import io.cloudevents.CloudEvent;
+import io.vertx.core.Context;
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
+import io.vertx.kafka.client.common.tracing.ConsumerTracer;
+import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
+import java.util.Objects;
+import java.util.function.Function;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
+
+/**
+ * This class implements the core algorithm of the Dispatcher (see {@link
+ * RecordDispatcherImpl#dispatch(KafkaConsumerRecord)}).
+ *
+ * @see RecordDispatcherImpl#dispatch(KafkaConsumerRecord)
+ */
+public class RecordDispatcherImpl implements RecordDispatcher {
+
+  private static final Logger logger = LoggerFactory.getLogger(RecordDispatcherImpl.class);
+
+  private final Filter filter;
+  private final Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> subscriberSender;
+  private final Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> dlsSender;
+  private final RecordDispatcherListener recordDispatcherListener;
+  private final AsyncCloseable closeable;
+  private final ConsumerTracer consumerTracer;
+
+  /**
+   * All args constructor.
+   *
+   * @param filter                   event filter
+   * @param subscriberSender         sender to trigger subscriber
+   * @param deadLetterSinkSender     sender to dead letter sink
+   * @param sinkResponseHandler      handler of the response from {@code subscriberSender}
+   * @param recordDispatcherListener hook receiver {@link RecordDispatcherListener}. It allows to plug in custom offset
+   * @param consumerTracer           consumer tracer
+   */
+  public RecordDispatcherImpl(
+    final Filter filter,
+    final ConsumerRecordSender subscriberSender,
+    final ConsumerRecordSender deadLetterSinkSender,
+    final SinkResponseHandler sinkResponseHandler,
+    final RecordDispatcherListener recordDispatcherListener,
+    final ConsumerTracer consumerTracer) {
+    Objects.requireNonNull(filter, "provide filter");
+    Objects.requireNonNull(subscriberSender, "provide subscriberSender");
+    Objects.requireNonNull(deadLetterSinkSender, "provide deadLetterSinkSender");
+    Objects.requireNonNull(recordDispatcherListener, "provide offsetStrategy");
+    Objects.requireNonNull(sinkResponseHandler, "provide sinkResponseHandler");
+
+    this.filter = filter;
+    this.subscriberSender = composeSenderAndSinkHandler(subscriberSender, sinkResponseHandler, "subscriber");
+    this.dlsSender = composeSenderAndSinkHandler(deadLetterSinkSender, sinkResponseHandler, "dead letter sink");
+    this.recordDispatcherListener = recordDispatcherListener;
+    this.closeable = AsyncCloseable.compose(sinkResponseHandler, deadLetterSinkSender, subscriberSender);
+    this.consumerTracer = consumerTracer;
+  }
+
+  /**
+   * Handle the given record and returns a future that completes when the dispatch is completed and the offset is committed
+   *
+   * @param record record to handle.
+   */
+  @Override
+  public Future<Void> dispatch(KafkaConsumerRecord<String, CloudEvent> record) {
+    Promise<Void> promise = Promise.promise();
+
+    /*
+    That's pretty much what happens here:
+
+    +------------+    +------------+     +-------------+
+    |            |    |            |     |             |
+    |  record    |    |  filter    |     |  subscriber |
+    |  received  +--->+  matching  +---->+  success    +-->end
+    |            |    |            |     |             |
+    +-+-------+--+    +-----+------+     +-----+-------+
+      |       |             |                  |
+      |       v             |                  |             +-------------+
+      |    +--+---------+   |           +------v-------+     |             |
+      |    |            |   |           |  subscriber  +---->+  dls        |
+      |    | filter not |   +---------->+  failure     |     |  success    +---->end
+      |    | matching   |               +----+---------+     |             |
+      |    |            |                    |               +-----+-------+
+      |    +---+--------+              +-----v-------+             |
+      |        |                       | dls failure |             v
+      |        |                       +-------------+----------> end
+      +->end<--+
+     */
+
+    onRecordReceived(record, promise);
+
+    return promise.future();
+  }
+
+  private void onRecordReceived(final KafkaConsumerRecord<String, CloudEvent> record, Promise<Void> finalProm) {
+    logDebug("Handling record", record);
+
+    // Trace record received event
+    // This needs to be done manually in order to work properly with both ordered and unordered delivery.
+    if (consumerTracer != null) {
+      Context context = Vertx.currentContext();
+      ConsumerTracer.StartedSpan span = consumerTracer.prepareMessageReceived(context, record.record());
+      if (span != null) {
+        finalProm.future()
+          .onComplete(ar -> {
+            if (ar.succeeded()) {
+              span.finish(context);
+            } else {
+              span.fail(context, ar.cause());
+            }
+          });
+      }
+    }
+
+    recordDispatcherListener.recordReceived(record)
+      .onSuccess(v -> {
+        // Execute filtering
+        if (filter.test(record.value())) {
+          onFilterMatching(record, finalProm);
+        } else {
+          onFilterNotMatching(record, finalProm);
+        }
+      })
+      .onFailure(finalProm::fail); // This should really never happen
+  }
+
+  private void onFilterMatching(final KafkaConsumerRecord<String, CloudEvent> record, final Promise<Void> finalProm) {
+    logDebug("Record match filtering", record);
+    subscriberSender.apply(record)
+      .onSuccess(res -> onSubscriberSuccess(record, finalProm))
+      .onFailure(ex -> onSubscriberFailure(record, finalProm));
+  }
+
+  private void onFilterNotMatching(final KafkaConsumerRecord<String, CloudEvent> record,
+                                   final Promise<Void> finalProm) {
+    logDebug("Record doesn't match filtering", record);
+    recordDispatcherListener.recordDiscarded(record)
+      .onComplete(finalProm);
+  }
+
+  private void onSubscriberSuccess(final KafkaConsumerRecord<String, CloudEvent> record,
+                                   final Promise<Void> finalProm) {
+    logDebug("Successfully sent event to subscriber", record);
+    recordDispatcherListener.successfullySentToSubscriber(record)
+      .onComplete(finalProm);
+  }
+
+  private void onSubscriberFailure(final KafkaConsumerRecord<String, CloudEvent> record,
+                                   final Promise<Void> finalProm) {
+    dlsSender.apply(record)
+      .onSuccess(v -> onDeadLetterSinkSuccess(record, finalProm))
+      .onFailure(ex -> onDeadLetterSinkFailure(record, ex, finalProm));
+  }
+
+  private void onDeadLetterSinkSuccess(final KafkaConsumerRecord<String, CloudEvent> record,
+                                       final Promise<Void> finalProm) {
+    logDebug("Successfully sent event to the dead letter sink", record);
+    recordDispatcherListener.successfullySentToDeadLetterSink(record)
+      .onComplete(finalProm);
+  }
+
+
+  private void onDeadLetterSinkFailure(final KafkaConsumerRecord<String, CloudEvent> record, final Throwable exception,
+                                       final Promise<Void> finalProm) {
+    recordDispatcherListener.failedToSendToDeadLetterSink(record, exception)
+      .onComplete(finalProm);
+  }
+
+  private static Function<KafkaConsumerRecord<String, CloudEvent>, Future<Void>> composeSenderAndSinkHandler(
+    ConsumerRecordSender sender, SinkResponseHandler sinkHandler, String senderType) {
+    return rec -> sender.send(rec)
+      .onFailure(ex -> logError("Failed to send event to " + senderType, rec, ex))
+      .compose(res ->
+        sinkHandler.handle(res)
+          .onFailure(ex -> logError("Failed to handle " + senderType + " response", rec, ex))
+      );
+  }
+
+  private static void logError(
+    final String msg,
+    final KafkaConsumerRecord<String, CloudEvent> record,
+    final Throwable cause) {
+
+    if (logger.isDebugEnabled()) {
+      logger.error(msg + " {} {} {} {} {}",
+        keyValue("topic", record.topic()),
+        keyValue("partition", record.partition()),
+        keyValue("headers", record.headers()),
+        keyValue("offset", record.offset()),
+        keyValue("event", record.value()),
+        cause
+      );
+    } else {
+      logger.error(msg + " {} {} {}",
+        keyValue("topic", record.topic()),
+        keyValue("partition", record.partition()),
+        keyValue("offset", record.offset()),
+        cause
+      );
+    }
+  }
+
+  private static void logDebug(
+    final String msg,
+    final KafkaConsumerRecord<String, CloudEvent> record) {
+
+    logger.debug(msg + " {} {} {} {} {}",
+      keyValue("topic", record.topic()),
+      keyValue("partition", record.partition()),
+      keyValue("headers", record.headers()),
+      keyValue("offset", record.offset()),
+      keyValue("event", record.value())
+    );
+  }
+
+  public Future<Void> close() {
+    return this.closeable.close();
+  }
+}
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/RecordDispatcherListener.java
similarity index 87%
rename from data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/OffsetManager.java
rename to data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
index ba0aa0f2b..f047e6ec7 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/RecordDispatcherListener.java
@@ -13,16 +13,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.consumer;
+package dev.knative.eventing.kafka.broker.dispatcher;
 
-import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
 import io.vertx.core.Future;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 
 /**
- * This class manages the offset of the record consumed by {@link RecordDispatcher}
+ * This class contains hooks for listening events through the {@link dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher} lifecycle.
  */
-public interface OffsetManager {
+public interface RecordDispatcherListener {
 
   /**
    * The given record has been received.
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/SinkResponseHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/SinkResponseHandler.java
index cce52b8b9..50759527b 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/SinkResponseHandler.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/SinkResponseHandler.java
@@ -21,7 +21,7 @@ import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.client.HttpResponse;
 
 /**
- * SinkResponseHandler is responsible for reading the response and acting on it based on its content.
+ * This class reads a sink response and acts on it based on its content.
  */
 public interface SinkResponseHandler extends AsyncCloseable {
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/BaseConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/BaseConsumerVerticle.java
index 6610da341..b820abfd8 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/BaseConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/BaseConsumerVerticle.java
@@ -19,7 +19,6 @@ import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.AbstractVerticle;
-import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
@@ -27,7 +26,6 @@ import io.vertx.kafka.client.consumer.KafkaConsumer;
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.BiFunction;
-import java.util.function.Supplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -76,13 +74,11 @@ public abstract class BaseConsumerVerticle extends AbstractVerticle {
     ).close(stopPromise);
   }
 
-  public void setConsumer(
-    KafkaConsumer<String, CloudEvent> consumer) {
+  public void setConsumer(KafkaConsumer<String, CloudEvent> consumer) {
     this.consumer = consumer;
   }
 
-  public void setRecordDispatcher(
-    RecordDispatcher recordDispatcher) {
+  public void setRecordDispatcher(RecordDispatcher recordDispatcher) {
     this.recordDispatcher = recordDispatcher;
   }
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedConsumerVerticle.java
index 6daba44d5..ddaf5714d 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedConsumerVerticle.java
@@ -122,7 +122,7 @@ public class OrderedConsumerVerticle extends BaseConsumerVerticle {
 
   void enqueueRecord(TopicPartition topicPartition, KafkaConsumerRecord<String, CloudEvent> record) {
     this.recordDispatcherExecutors.computeIfAbsent(topicPartition, (tp) -> new OrderedAsyncExecutor())
-      .offer(() -> this.recordDispatcher.apply(record).onComplete(v -> this.pendingRecords--));
+      .offer(() -> this.recordDispatcher.dispatch(record).onComplete(v -> this.pendingRecords--));
   }
 
   long pollWaitMs() {
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 b3282ab7e..0090a2b15 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
@@ -15,7 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.consumer.impl;
 
-import dev.knative.eventing.kafka.broker.dispatcher.consumer.OffsetManager;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.vertx.core.Future;
 import io.vertx.kafka.client.common.TopicPartition;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
@@ -32,7 +32,7 @@ import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 /**
  * This class implements the offset strategy for the ordered consumer.
  */
-public final class OrderedOffsetManager implements OffsetManager {
+public final class OrderedOffsetManager implements RecordDispatcherListener {
 
   private static final Logger logger = LoggerFactory
     .getLogger(OrderedOffsetManager.class);
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedConsumerVerticle.java
index 662b839b7..e536df937 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedConsumerVerticle.java
@@ -32,7 +32,7 @@ public final class UnorderedConsumerVerticle extends BaseConsumerVerticle {
   @Override
   void startConsumer(Promise<Void> startPromise) {
     this.consumer.exceptionHandler(this::exceptionHandler);
-    this.consumer.handler(this.recordDispatcher);
+    this.consumer.handler(record -> this.recordDispatcher.dispatch(record));
     this.consumer.subscribe(this.topics, startPromise);
   }
 
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 6f53dbfec..c32b849eb 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
@@ -15,7 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.consumer.impl;
 
-import dev.knative.eventing.kafka.broker.dispatcher.consumer.OffsetManager;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.vertx.core.Future;
 import io.vertx.kafka.client.common.TopicPartition;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
@@ -32,7 +32,7 @@ import org.slf4j.LoggerFactory;
 /**
  * This class implements the offset strategy that makes sure that, even unordered, the offset commit is ordered.
  */
-public final class UnorderedOffsetManager implements OffsetManager {
+public final class UnorderedOffsetManager implements RecordDispatcherListener {
 
   private static final Logger logger = LoggerFactory
     .getLogger(UnorderedOffsetManager.class);
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 c0642971a..7ad082ba0 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
@@ -25,10 +25,10 @@ import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.core.security.KafkaClientsAuth;
 import dev.knative.eventing.kafka.broker.core.security.PlaintextCredentials;
 import dev.knative.eventing.kafka.broker.dispatcher.ConsumerRecordSender;
-import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.consumer.ConsumerVerticleFactory;
 import dev.knative.eventing.kafka.broker.dispatcher.consumer.DeliveryOrder;
-import dev.knative.eventing.kafka.broker.dispatcher.consumer.OffsetManager;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import dev.knative.eventing.kafka.broker.dispatcher.consumer.impl.BaseConsumerVerticle;
 import dev.knative.eventing.kafka.broker.dispatcher.consumer.impl.OrderedConsumerVerticle;
 import dev.knative.eventing.kafka.broker.dispatcher.consumer.impl.OrderedOffsetManager;
@@ -164,7 +164,7 @@ public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory {
           new AttributesFilter(egress.getFilter().getAttributesMap()) :
           Filter.noop();
 
-        final RecordDispatcher recordDispatcher = new RecordDispatcher(
+        final RecordDispatcherImpl recordDispatcher = new RecordDispatcherImpl(
           filter,
           egressSubscriberSender,
           egressDeadLetterSender,
@@ -275,8 +275,8 @@ public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory {
     return !(egressConfig == null || egressConfig.getDeadLetter().isEmpty());
   }
 
-  private static OffsetManager getOffsetManager(final DeliveryOrder type, final KafkaConsumer<?, ?> consumer,
-                                                Consumer<Integer> commitHandler) {
+  private static RecordDispatcherListener getOffsetManager(final DeliveryOrder type, final KafkaConsumer<?, ?> consumer,
+                                                           Consumer<Integer> commitHandler) {
     return switch (type) {
       case ORDERED -> new OrderedOffsetManager(consumer, commitHandler);
       case UNORDERED -> new UnorderedOffsetManager(consumer, commitHandler);
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/DispatcherEnv.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/DispatcherEnv.java
similarity index 97%
rename from data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/DispatcherEnv.java
rename to data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/DispatcherEnv.java
index a1e6f6e60..979a40094 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/DispatcherEnv.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/DispatcherEnv.java
@@ -13,21 +13,22 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher;
-
-import static java.util.Objects.requireNonNull;
+package dev.knative.eventing.kafka.broker.dispatcher.main;
 
 import dev.knative.eventing.kafka.broker.core.utils.BaseEnv;
 import java.util.function.Function;
 
+import static java.util.Objects.requireNonNull;
+
 public class DispatcherEnv extends BaseEnv {
 
   public static final String CONSUMER_CONFIG_FILE_PATH = "CONSUMER_CONFIG_FILE_PATH";
-  public static final String WEBCLIENT_CONFIG_FILE_PATH = "WEBCLIENT_CONFIG_FILE_PATH";
-  public static final String EGRESSES_INITIAL_CAPACITY = "EGRESSES_INITIAL_CAPACITY";
-
   private final String consumerConfigFilePath;
+
+  public static final String WEBCLIENT_CONFIG_FILE_PATH = "WEBCLIENT_CONFIG_FILE_PATH";
   private final String webClientConfigFilePath;
+
+  public static final String EGRESSES_INITIAL_CAPACITY = "EGRESSES_INITIAL_CAPACITY";
   private final int egressesInitialCapacity;
 
   public DispatcherEnv(Function<String, String> envProvider) {
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/Main.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/Main.java
similarity index 98%
rename from data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/Main.java
rename to data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/Main.java
index 6002e22c8..04b0f4bef 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/Main.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/Main.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher;
+package dev.knative.eventing.kafka.broker.dispatcher.main;
 
 import dev.knative.eventing.kafka.broker.core.eventbus.ContractMessageCodec;
 import dev.knative.eventing.kafka.broker.core.eventbus.ContractPublisher;
@@ -24,6 +24,7 @@ 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 dev.knative.eventing.kafka.broker.dispatcher.ConsumerDeployerVerticle;
 import dev.knative.eventing.kafka.broker.dispatcher.http.HttpConsumerVerticleFactory;
 import io.cloudevents.kafka.CloudEventDeserializer;
 import io.cloudevents.kafka.CloudEventSerializer;
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 79226a174..32c7146b3 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -43,7 +43,7 @@
     <!-- dependencies version -->
     <vertx.version>4.1.0</vertx.version>
     <cloudevents.sdk.version>2.2.0</cloudevents.sdk.version>
-    <micrometer.version>1.7.0</micrometer.version>
+    <micrometer.version>1.7.1</micrometer.version>
     <opentelemetry.version>1.1.0</opentelemetry.version>
     <jackson.version>2.12.3</jackson.version>
     <protobuf.version>3.17.2</protobuf.version>
diff --git a/data-plane/receiver/pom.xml b/data-plane/receiver/pom.xml
index 2f91078fd..2ecd784a1 100644
--- a/data-plane/receiver/pom.xml
+++ b/data-plane/receiver/pom.xml
@@ -124,9 +124,7 @@
                 <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                 <transformer
                   implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                  <mainClass>
-                    dev.knative.eventing.kafka.broker.receiver.Main
-                  </mainClass>
+                  <mainClass>dev.knative.eventing.kafka.broker.receiver.main.Main</mainClass>
                 </transformer>
               </transformers>
               <filters>
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticle.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticle.java
index 793c106f3..a90a959f0 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticle.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticle.java
@@ -15,8 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.receiver;
 
-import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerImpl;
-import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerMessageHandler;
+import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.CompositeFuture;
 import io.vertx.core.Handler;
@@ -26,7 +25,6 @@ import io.vertx.core.eventbus.MessageConsumer;
 import io.vertx.core.http.HttpServer;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.http.HttpServerRequest;
-
 import java.util.Objects;
 import java.util.function.Function;
 
@@ -63,13 +61,10 @@ public class ReceiverVerticle extends AbstractVerticle {
   public void start(final Promise<Void> startPromise) {
     final var requestMapper = this.requestHandlerFactory.apply(vertx);
 
-    this.messageConsumer = ResourcesReconcilerMessageHandler.start(
-      vertx,
-      ResourcesReconcilerImpl
-        .builder()
-        .watchIngress(requestMapper)
-        .build()
-    );
+    this.messageConsumer = ResourcesReconciler
+      .builder()
+      .watchIngress(requestMapper)
+      .buildAndListen(vertx);
     this.server = vertx.createHttpServer(httpServerOptions);
 
     Handler<HttpServerRequest> requestHandler = requestMapper;
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestMapper.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestMapper.java
index 84edd57a0..9ee1123c5 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestMapper.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestMapper.java
@@ -21,13 +21,16 @@ import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.reconciler.IngressReconcilerListener;
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.core.security.KafkaClientsAuth;
+import dev.knative.eventing.kafka.broker.core.tracing.TracingConfig;
+import dev.knative.eventing.kafka.broker.core.tracing.TracingSpan;
 import dev.knative.eventing.kafka.broker.core.utils.ReferenceCounter;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.message.Encoding;
 import io.cloudevents.jackson.JsonFormat;
 import io.cloudevents.kafka.CloudEventSerializer;
 import io.micrometer.core.instrument.Counter;
-import io.vertx.core.CompositeFuture;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Context;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
@@ -121,30 +124,47 @@ public class RequestMapper implements Handler<HttpServerRequest>, IngressReconci
     }
 
     requestToRecordMapper
-      .recordFromRequest(request, ingressInfo.getTopic())
-      .onSuccess(record -> ingressInfo.getProducer().send(record)
-        .onSuccess(metadata -> {
-          request.response().setStatusCode(RECORD_PRODUCED).end();
-          produceEventsCounter.increment();
-
-          logger.debug("Record produced {} {} {} {} {} {}",
-            keyValue("topic", record.topic()),
-            keyValue("partition", metadata.getPartition()),
-            keyValue("offset", metadata.getOffset()),
-            keyValue("value", record.value()),
-            keyValue("headers", record.headers()),
-            keyValue("path", request.path())
-          );
-        })
-        .onFailure(cause -> {
-          request.response().setStatusCode(FAILED_TO_PRODUCE).end();
-
-          logger.error("Failed to send record {} {}",
-            keyValue("topic", record.topic()),
-            keyValue("path", request.path()),
-            cause
-          );
-        })
+      .requestToRecord(request, ingressInfo.getTopic())
+      .onSuccess(record -> {
+          if (logger.isDebugEnabled()) {
+            final var span = Span.fromContextOrNull(Context.current());
+            if (span != null) {
+              logger.debug("Received event {} {}",
+                keyValue("event", record.value()),
+                keyValue(TracingConfig.TRACE_ID_KEY, span.getSpanContext().getTraceId())
+              );
+            } else {
+              logger.debug("Received event {}", keyValue("event", record.value()));
+            }
+          }
+
+          // Decorate the span with event specific attributed
+          TracingSpan.decorateCurrentWithEvent(record.value());
+
+          ingressInfo.getProducer().send(record)
+            .onSuccess(metadata -> {
+              request.response().setStatusCode(RECORD_PRODUCED).end();
+              produceEventsCounter.increment();
+
+              logger.debug("Record produced {} {} {} {} {} {}",
+                keyValue("topic", record.topic()),
+                keyValue("partition", metadata.getPartition()),
+                keyValue("offset", metadata.getOffset()),
+                keyValue("value", record.value()),
+                keyValue("headers", record.headers()),
+                keyValue("path", request.path())
+              );
+            })
+            .onFailure(cause -> {
+              request.response().setStatusCode(FAILED_TO_PRODUCE).end();
+
+              logger.error("Failed to send record {} {}",
+                keyValue("topic", record.topic()),
+                keyValue("path", request.path()),
+                cause
+              );
+            });
+        }
       )
       .onFailure(cause -> {
         request.response().setStatusCode(MAPPER_FAILED).end();
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestToRecordMapper.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestToRecordMapper.java
index 653c1eaee..d1e53aa54 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestToRecordMapper.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestToRecordMapper.java
@@ -30,7 +30,7 @@ public interface RequestToRecordMapper {
    * @param topic   topic to send the event
    * @return kafka record (record can be null).
    */
-  Future<KafkaProducerRecord<String, CloudEvent>> recordFromRequest(
+  Future<KafkaProducerRecord<String, CloudEvent>> requestToRecord(
     final HttpServerRequest request,
     final String topic
   );
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/CloudEventRequestToRecordMapper.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/StrictRequestToRecordMapper.java
similarity index 53%
rename from data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/CloudEventRequestToRecordMapper.java
rename to data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/StrictRequestToRecordMapper.java
index cd631e8da..eef24d776 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/CloudEventRequestToRecordMapper.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/StrictRequestToRecordMapper.java
@@ -15,30 +15,32 @@
  */
 package dev.knative.eventing.kafka.broker.receiver;
 
-import dev.knative.eventing.kafka.broker.core.tracing.TracingConfig;
-import dev.knative.eventing.kafka.broker.core.tracing.TracingSpan;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.message.MessageReader;
 import io.cloudevents.http.vertx.VertxMessageFactory;
-import io.opentelemetry.api.trace.Span;
-import io.opentelemetry.context.Context;
 import io.vertx.core.Future;
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
+/**
+ * This class implements a strict {@link HttpServerRequest} to {@link KafkaProducerRecord} mapper.
+ * The conversion will fail if the request does not contain a valid {@link CloudEvent}.
+ */
+public class StrictRequestToRecordMapper implements RequestToRecordMapper {
 
-public class CloudEventRequestToRecordMapper implements RequestToRecordMapper {
+  private static class SingletonContainer {
+    private static final StrictRequestToRecordMapper INSTANCE = new StrictRequestToRecordMapper();
+  }
 
-  private static final Logger logger = LoggerFactory.getLogger(CloudEventRequestToRecordMapper.class);
+  public static RequestToRecordMapper getInstance() {
+    return StrictRequestToRecordMapper.SingletonContainer.INSTANCE;
+  }
 
-  public CloudEventRequestToRecordMapper() {
+  private StrictRequestToRecordMapper() {
   }
 
   @Override
-  public Future<KafkaProducerRecord<String, CloudEvent>> recordFromRequest(
+  public Future<KafkaProducerRecord<String, CloudEvent>> requestToRecord(
     final HttpServerRequest request,
     final String topic) {
 
@@ -48,20 +50,6 @@ public class CloudEventRequestToRecordMapper implements RequestToRecordMapper {
         if (event == null) {
           throw new IllegalArgumentException("event cannot be null");
         }
-
-        if (logger.isDebugEnabled()) {
-          final var span = Span.fromContextOrNull(Context.current());
-          if (span != null) {
-            logger.debug("Received event {} {}",
-              keyValue("event", event),
-              keyValue(TracingConfig.TRACE_ID_KEY, span.getSpanContext().getTraceId())
-            );
-          } else {
-            logger.debug("Received event {}", keyValue("event", event));
-          }
-        }
-
-        TracingSpan.decorateCurrentWithEvent(event);
         return KafkaProducerRecord.create(topic, event);
       });
   }
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/Main.java
similarity index 99%
rename from data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java
rename to data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/Main.java
index fc04998c7..da0685a66 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/Main.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.receiver;
+package dev.knative.eventing.kafka.broker.receiver.main;
 
 import dev.knative.eventing.kafka.broker.core.eventbus.ContractMessageCodec;
 import dev.knative.eventing.kafka.broker.core.eventbus.ContractPublisher;
@@ -33,12 +33,6 @@ import io.vertx.core.VertxOptions;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.tracing.TracingPolicy;
 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;
@@ -47,6 +41,11 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Supplier;
+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 static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverEnv.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnv.java
similarity index 97%
rename from data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverEnv.java
rename to data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnv.java
index 27d5cd8a1..352bc5922 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverEnv.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnv.java
@@ -13,13 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.receiver;
-
-import static java.util.Objects.requireNonNull;
+package dev.knative.eventing.kafka.broker.receiver.main;
 
 import dev.knative.eventing.kafka.broker.core.utils.BaseEnv;
 import java.util.function.Function;
 
+import static java.util.Objects.requireNonNull;
+
 class ReceiverEnv extends BaseEnv {
 
   public static final String INGRESS_PORT = "INGRESS_PORT";
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/main/ReceiverVerticleFactory.java
similarity index 84%
rename from data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleFactory.java
rename to data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverVerticleFactory.java
index a6a450668..bc8dfdac3 100644
--- 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/main/ReceiverVerticleFactory.java
@@ -14,14 +14,17 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.receiver;
+package dev.knative.eventing.kafka.broker.receiver.main;
 
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
+import dev.knative.eventing.kafka.broker.receiver.ReceiverVerticle;
+import dev.knative.eventing.kafka.broker.receiver.RequestMapper;
+import dev.knative.eventing.kafka.broker.receiver.SimpleProbeHandlerDecorator;
+import dev.knative.eventing.kafka.broker.receiver.StrictRequestToRecordMapper;
 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;
 
@@ -54,7 +57,7 @@ class ReceiverVerticleFactory implements Supplier<Verticle> {
         v,
         AuthProvider.kubernetes(),
         producerConfigs,
-        new CloudEventRequestToRecordMapper(),
+        StrictRequestToRecordMapper.getInstance(),
         properties -> KafkaProducer.create(v, properties),
         badRequestCounter,
         produceEventsCounter
diff --git a/go.mod b/go.mod
index aec4d028d..5e0e76489 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.20210621164716-b312f0e1c8e5
-	knative.dev/hack v0.0.0-20210614141220-66ab1a098940
-	knative.dev/pkg v0.0.0-20210618060751-f454995ff92b
-	knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7
+	knative.dev/eventing v0.23.1-0.20210624083445-2fcbc0ac500d
+	knative.dev/hack v0.0.0-20210622141627-e28525d8d260
+	knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80
+	knative.dev/reconciler-test v0.0.0-20210623134345-88c84739abd9
 )
diff --git a/go.sum b/go.sum
index 66a57441f..0bb159867 100644
--- a/go.sum
+++ b/go.sum
@@ -6,7 +6,6 @@ cloud.google.com/go v0.44.2/go.mod h1:60680Gw3Yr4ikxnPRS/oxxkBccT6SA1yMk63TGekxK
 cloud.google.com/go v0.45.1/go.mod h1:RpBamKRgapWJb87xiFSdk4g1CME7QZg3uwTez+TSTjc=
 cloud.google.com/go v0.46.3/go.mod h1:a6bKKbmY7er1mI7TEI4lsAkts/mkhTSZK8w33B4RAg0=
 cloud.google.com/go v0.50.0/go.mod h1:r9sluTvynVuxRIOHXQEHMFffphuXHOMZMycpNR5e6To=
-cloud.google.com/go v0.51.0/go.mod h1:hWtGJ6gnXH+KgDv+V0zFGDvpi07n3z8ZNj3T1RW0Gcw=
 cloud.google.com/go v0.52.0/go.mod h1:pXajvRH/6o3+F9jDHZWQ5PbGhn+o8w9qiu/CffaVdO4=
 cloud.google.com/go v0.53.0/go.mod h1:fp/UouUEsRkN6ryDKNW/Upv/JBKnv6WDthjR6+vze6M=
 cloud.google.com/go v0.54.0/go.mod h1:1rq2OEkV3YMf6n/9ZvGWI3GWw0VoqH/1x2nd8Is/bPc=
@@ -46,24 +45,13 @@ contrib.go.opencensus.io/exporter/zipkin v0.1.2/go.mod h1:mP5xM3rrgOjpn79MM8fZbj
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
 github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
-github.com/Azure/go-autorest/autorest v0.9.0/go.mod h1:xyHB1BMZT0cuDHU7I0+g046+BFDTQ8rEZB0s4Yfa6bI=
-github.com/Azure/go-autorest/autorest v0.9.6/go.mod h1:/FALq9T/kS7b5J5qsQ+RSTUdAmGFqi0vUdVNNx8q630=
 github.com/Azure/go-autorest/autorest v0.11.1/go.mod h1:JFgpikqFJ/MleTTxwepExTKnFUKKszPS8UavbQYUMuw=
-github.com/Azure/go-autorest/autorest/adal v0.5.0/go.mod h1:8Z9fGy2MpX0PvDjB1pEgQTmVqjGhiHBW7RJJEciWzS0=
-github.com/Azure/go-autorest/autorest/adal v0.8.2/go.mod h1:ZjhuQClTqx435SRJ2iMlOxPYt3d2C/T/7TiQCVZSn3Q=
 github.com/Azure/go-autorest/autorest/adal v0.9.0/go.mod h1:/c022QCutn2P7uY+/oQWWNcK9YU+MH96NgK+jErpbcg=
 github.com/Azure/go-autorest/autorest/adal v0.9.5/go.mod h1:B7KF7jKIeC9Mct5spmyCB/A8CG/sEz1vwIRGv/bbw7A=
-github.com/Azure/go-autorest/autorest/date v0.1.0/go.mod h1:plvfp3oPSKwf2DNjlBjWF/7vwR+cUD/ELuzDCXwHUVA=
-github.com/Azure/go-autorest/autorest/date v0.2.0/go.mod h1:vcORJHLJEh643/Ioh9+vPmf1Ij9AEBM5FuBIXLmIy0g=
 github.com/Azure/go-autorest/autorest/date v0.3.0/go.mod h1:BI0uouVdmngYNUzGWeSYnokU+TrmwEsOqdt8Y6sso74=
-github.com/Azure/go-autorest/autorest/mocks v0.1.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
-github.com/Azure/go-autorest/autorest/mocks v0.2.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0=
-github.com/Azure/go-autorest/autorest/mocks v0.3.0/go.mod h1:a8FDP3DYzQ4RYfVAxAN3SVSiiO77gL2j2ronKKP0syM=
 github.com/Azure/go-autorest/autorest/mocks v0.4.0/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k=
 github.com/Azure/go-autorest/autorest/mocks v0.4.1/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k=
-github.com/Azure/go-autorest/logger v0.1.0/go.mod h1:oExouG+K6PryycPJfVSxi/koC6LSNgds39diKLz7Vrc=
 github.com/Azure/go-autorest/logger v0.2.0/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZmbF5NWuPV8+WeEW8=
-github.com/Azure/go-autorest/tracing v0.5.0/go.mod h1:r/s2XiOKccPW3HrqB+W0TQzfbtp2fGCgRFtBroKn4Dk=
 github.com/Azure/go-autorest/tracing v0.6.0/go.mod h1:+vhtPC754Xsa23ID7GlGsrdKBpUA79WCAKPPZVC2DeU=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
@@ -71,7 +59,6 @@ github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go
 github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb06e3pkSAbeQ52E9H9iFoQsEEwGN64994WTCIhntQ=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/PuerkitoBio/purell v1.0.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
-github.com/PuerkitoBio/purell v1.1.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/purell v1.1.1 h1:WEQqlqaGbrPkxLJWfBwQmfEAE1Z7ONdDLqrN38tNFfI=
 github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20160726150825-5bd2802263f2/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
@@ -85,7 +72,6 @@ github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWso
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 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=
@@ -95,7 +81,6 @@ github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRF
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
-github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
 github.com/apache/thrift v0.13.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
@@ -104,7 +89,6 @@ github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/aryann/difflib v0.0.0-20170710044230-e206f873d14a/go.mod h1:DAHtR1m6lCRdSC2Tm3DSWRPvIPr6xNKyeHdqDQSQT+A=
-github.com/asaskevich/govalidator v0.0.0-20180720115003-f9ffefc3facf/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
 github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
 github.com/aws/aws-lambda-go v1.13.3/go.mod h1:4UKl9IzQMoD+QF79YdCuzCwp8VbmG4VAQwij/eHl5CU=
 github.com/aws/aws-sdk-go v1.23.20/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
@@ -118,7 +102,6 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
 github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84=
-github.com/blang/semver v3.5.0+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/blang/semver v3.5.1+incompatible h1:cQNTCjp13qL8KC3Nbxr/y2Bqb63oX6wdnnjpJbkM4JQ=
 github.com/blang/semver v3.5.1+incompatible/go.mod h1:kRBLl5iJ+tD4TcOOxsy/0fnwebNt5EWlYSAyrTnjyyk=
 github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=
@@ -181,8 +164,6 @@ github.com/dgryski/go-gk v0.0.0-20200319235926-a69029f61654 h1:XOPLOMn/zT4jIgxfx
 github.com/dgryski/go-gk v0.0.0-20200319235926-a69029f61654/go.mod h1:qm+vckxRlDt0aOla0RYJJVeqHZlWfOm2UIxHaqPB46E=
 github.com/dgryski/go-lttb v0.0.0-20180810165845-318fcdf10a77/go.mod h1:Va5MyIzkU0rAM92tn3hb3Anb7oz7KcnixF49+2wOMe4=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
-github.com/docker/go-units v0.3.3/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
-github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
@@ -226,8 +207,6 @@ github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWo
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/ghodss/yaml v0.0.0-20150909031657-73d445a93680/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
-github.com/globalsign/mgo v0.0.0-20180905125535-1ca0a4f7cbcb/go.mod h1:xkRDCp4j0OGD1HRkm4kmhM+pmpv3AKq5SU7GMg4oO/Q=
-github.com/globalsign/mgo v0.0.0-20181015135952-eeefdecb41b8/go.mod h1:xkRDCp4j0OGD1HRkm4kmhM+pmpv3AKq5SU7GMg4oO/Q=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
@@ -242,58 +221,25 @@ github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7
 github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
 github.com/go-logr/logr v0.4.0 h1:K7/B1jt6fIBQVd4Owv2MqGQClcgf0R266+7C/QjRcLc=
 github.com/go-logr/logr v0.4.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
-github.com/go-openapi/analysis v0.0.0-20180825180245-b006789cd277/go.mod h1:k70tL6pCuVxPJOHXQ+wIac1FUrvNkHolPie/cLEU6hI=
-github.com/go-openapi/analysis v0.17.0/go.mod h1:IowGgpVeD0vNm45So8nr+IcQ3pxVtpRoBWb8PVZO0ik=
-github.com/go-openapi/analysis v0.18.0/go.mod h1:IowGgpVeD0vNm45So8nr+IcQ3pxVtpRoBWb8PVZO0ik=
-github.com/go-openapi/analysis v0.19.2/go.mod h1:3P1osvZa9jKjb8ed2TPng3f0i/UY9snX6gxi44djMjk=
-github.com/go-openapi/analysis v0.19.5/go.mod h1:hkEAkxagaIvIP7VTn8ygJNkd4kAYON2rCu0v0ObL0AU=
-github.com/go-openapi/errors v0.17.0/go.mod h1:LcZQpmvG4wyF5j4IhA73wkLFQg+QJXOQHVjmcZxhka0=
-github.com/go-openapi/errors v0.18.0/go.mod h1:LcZQpmvG4wyF5j4IhA73wkLFQg+QJXOQHVjmcZxhka0=
-github.com/go-openapi/errors v0.19.2/go.mod h1:qX0BLWsyaKfvhluLejVpVNwNRdXZhEbTA4kxxpKBC94=
 github.com/go-openapi/jsonpointer v0.0.0-20160704185906-46af16f9f7b1/go.mod h1:+35s3my2LFTysnkMfxsJBAMHj/DoqoB9knIWoYG/Vk0=
-github.com/go-openapi/jsonpointer v0.17.0/go.mod h1:cOnomiV+CVVwFLk0A/MExoFMjwdsUdVpsRhURCKh+3M=
-github.com/go-openapi/jsonpointer v0.18.0/go.mod h1:cOnomiV+CVVwFLk0A/MExoFMjwdsUdVpsRhURCKh+3M=
 github.com/go-openapi/jsonpointer v0.19.2/go.mod h1:3akKfEdA7DF1sugOqz1dVQHBcuDBPKZGEoHC/NkiQRg=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonpointer v0.19.5 h1:gZr+CIYByUqjcgeLXnQu2gHYQC9o73G2XUeOFYEICuY=
 github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonreference v0.0.0-20160704190145-13c6e3589ad9/go.mod h1:W3Z9FmVs9qj+KR4zFKmDPGiLdk1D9Rlm7cyMvf57TTg=
-github.com/go-openapi/jsonreference v0.17.0/go.mod h1:g4xxGn04lDIRh0GJb5QlpE3HfopLOL6uZrK/VgnsK9I=
-github.com/go-openapi/jsonreference v0.18.0/go.mod h1:g4xxGn04lDIRh0GJb5QlpE3HfopLOL6uZrK/VgnsK9I=
 github.com/go-openapi/jsonreference v0.19.2/go.mod h1:jMjeRr2HHw6nAVajTXJ4eiUwohSTlpa0o73RUL1owJc=
 github.com/go-openapi/jsonreference v0.19.3/go.mod h1:rjx6GuL8TTa9VaixXglHmQmIL98+wF9xc8zWvFonSJ8=
 github.com/go-openapi/jsonreference v0.19.5 h1:1WJP/wi4OjB4iV8KVbH73rQaoialJrqv8gitZLxGLtM=
 github.com/go-openapi/jsonreference v0.19.5/go.mod h1:RdybgQwPxbL4UEjuAruzK1x3nE69AqPYEJeo/TWfEeg=
-github.com/go-openapi/loads v0.17.0/go.mod h1:72tmFy5wsWx89uEVddd0RjRWPZm92WRLhf7AC+0+OOU=
-github.com/go-openapi/loads v0.18.0/go.mod h1:72tmFy5wsWx89uEVddd0RjRWPZm92WRLhf7AC+0+OOU=
-github.com/go-openapi/loads v0.19.0/go.mod h1:72tmFy5wsWx89uEVddd0RjRWPZm92WRLhf7AC+0+OOU=
-github.com/go-openapi/loads v0.19.2/go.mod h1:QAskZPMX5V0C2gvfkGZzJlINuP7Hx/4+ix5jWFxsNPs=
-github.com/go-openapi/loads v0.19.4/go.mod h1:zZVHonKd8DXyxyw4yfnVjPzBjIQcLt0CCsn0N0ZrQsk=
-github.com/go-openapi/runtime v0.0.0-20180920151709-4f900dc2ade9/go.mod h1:6v9a6LTXWQCdL8k1AO3cvqx5OtZY/Y9wKTgaoP6YRfA=
-github.com/go-openapi/runtime v0.19.0/go.mod h1:OwNfisksmmaZse4+gpV3Ne9AyMOlP1lt4sK4FXt0O64=
-github.com/go-openapi/runtime v0.19.4/go.mod h1:X277bwSUBxVlCYR3r7xgZZGKVvBd/29gLDlFGtJ8NL4=
 github.com/go-openapi/spec v0.0.0-20160808142527-6aced65f8501/go.mod h1:J8+jY1nAiCcj+friV/PDoE1/3eeccG9LYBs0tYvLOWc=
-github.com/go-openapi/spec v0.17.0/go.mod h1:XkF/MOi14NmjsfZ8VtAKf8pIlbZzyoTvZsdfssdxcBI=
-github.com/go-openapi/spec v0.18.0/go.mod h1:XkF/MOi14NmjsfZ8VtAKf8pIlbZzyoTvZsdfssdxcBI=
-github.com/go-openapi/spec v0.19.2/go.mod h1:sCxk3jxKgioEJikev4fgkNmwS+3kuYdJtcsZsD5zxMY=
 github.com/go-openapi/spec v0.19.3/go.mod h1:FpwSN1ksY1eteniUU7X0N/BgJ7a4WvBFVA8Lj9mJglo=
 github.com/go-openapi/spec v0.19.6 h1:rMMMj8cV38KVXK7SFc+I2MWClbEfbK705+j+dyqun5g=
 github.com/go-openapi/spec v0.19.6/go.mod h1:Hm2Jr4jv8G1ciIAo+frC/Ft+rR2kQDh8JHKHb3gWUSk=
-github.com/go-openapi/strfmt v0.17.0/go.mod h1:P82hnJI0CXkErkXi8IKjPbNBM6lV6+5pLP5l494TcyU=
-github.com/go-openapi/strfmt v0.18.0/go.mod h1:P82hnJI0CXkErkXi8IKjPbNBM6lV6+5pLP5l494TcyU=
-github.com/go-openapi/strfmt v0.19.0/go.mod h1:+uW+93UVvGGq2qGaZxdDeJqSAqBqBdl+ZPMF/cC8nDY=
-github.com/go-openapi/strfmt v0.19.3/go.mod h1:0yX7dbo8mKIvc3XSKp7MNfxw4JytCfCD6+bY1AVL9LU=
 github.com/go-openapi/swag v0.0.0-20160704191624-1d0bd113de87/go.mod h1:DXUve3Dpr1UfpPtxFw+EFuQ41HhCWZfha5jSVRG7C7I=
-github.com/go-openapi/swag v0.17.0/go.mod h1:AByQ+nYG6gQg71GINrmuDXCPWdL640yX49/kXLo40Tg=
-github.com/go-openapi/swag v0.18.0/go.mod h1:AByQ+nYG6gQg71GINrmuDXCPWdL640yX49/kXLo40Tg=
 github.com/go-openapi/swag v0.19.2/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
-github.com/go-openapi/swag v0.19.7/go.mod h1:ao+8BpOPyKdpQz3AOJfbeEVpLmWAvlT1IfTe5McPyhY=
 github.com/go-openapi/swag v0.19.15 h1:D2NRCBzS9/pEY3gP9Nl8aDqGUcPFrwG2p+CNFrLyrCM=
 github.com/go-openapi/swag v0.19.15/go.mod h1:QYRuS/SOXUCsnplDa677K7+DxSOj6IPNl/eQntq43wQ=
-github.com/go-openapi/validate v0.18.0/go.mod h1:Uh4HdOzKt19xGIGm1qHf/ofbX1YQ4Y+MYsct2VUrAJ4=
-github.com/go-openapi/validate v0.19.2/go.mod h1:1tRCw7m3jtI8eNWEEliiAqUIcBztB2KDnRCRMUi7GTA=
-github.com/go-openapi/validate v0.19.5/go.mod h1:8DJv2CVJQ6kGNpFW6eV9N3JviE1C85nY1c2z52x1Gk4=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -411,7 +357,6 @@ github.com/gorilla/mux v1.7.4/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB7
 github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
 github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
 github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
-github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
@@ -534,12 +479,9 @@ github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0Q
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/mailru/easyjson v0.0.0-20160728113105-d5b7844b561a/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
-github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
-github.com/mailru/easyjson v0.0.0-20190312143242-1de009706dbe/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.7.0/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs=
-github.com/mailru/easyjson v0.7.1-0.20191009090205-6c0755d89d1e/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs=
 github.com/mailru/easyjson v0.7.6/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
@@ -647,7 +589,6 @@ github.com/prometheus/client_golang v1.3.0/go.mod h1:hJaj2vgQTGQmVCsAACORcieXFeD
 github.com/prometheus/client_golang v1.6.0/go.mod h1:ZLOG9ck3JLRdB5MgO8f+lLTe83AXG6ro35rLTxvnIl4=
 github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
 github.com/prometheus/client_golang v1.9.0/go.mod h1:FqZLKOZnGdFAhOK4nqGHa7D66IdsO+O441Eve7ptJDU=
-github.com/prometheus/client_golang v1.10.0/go.mod h1:WJM3cc3yu7XKBKa/I8WeZm+V3eltZnBwfENSU7mdogU=
 github.com/prometheus/client_golang v1.11.0 h1:HNkLOAEQMIDv/K+04rukrLx6ch7msSRwf3/SASFAGtQ=
 github.com/prometheus/client_golang v1.11.0/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqroYurpAkEiz0P2BEV0=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
@@ -665,8 +606,6 @@ github.com/prometheus/common v0.7.0/go.mod h1:DjGbpBbp5NYNiECxcL/VnbXCCaQpKd3tt2
 github.com/prometheus/common v0.9.1/go.mod h1:yhUN8i9wzaXS3w1O07YhxHEBxD+W35wd8bs7vj7HSQ4=
 github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
 github.com/prometheus/common v0.15.0/go.mod h1:U+gB1OBLb1lF3O42bTCL+FK18tX9Oar16Clt/msog/s=
-github.com/prometheus/common v0.18.0/go.mod h1:U+gB1OBLb1lF3O42bTCL+FK18tX9Oar16Clt/msog/s=
-github.com/prometheus/common v0.25.0/go.mod h1:H6QK/N6XVT42whUeIdI3dp36w49c+/iMDk7UAI2qm7Q=
 github.com/prometheus/common v0.26.0 h1:iMAkS2TDoNWnKM+Kopnx/8tnEStIfpYA0ur0xQzzhMQ=
 github.com/prometheus/common v0.26.0/go.mod h1:M7rCNAaPfAosfx8veZJCuw84e35h3Cfd9VFqTh1DIvc=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
@@ -704,7 +643,6 @@ github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
-github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
@@ -720,7 +658,6 @@ github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTd
 github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
 github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
-github.com/spf13/cobra v1.0.0/go.mod h1:/6GTrnGXV9HjY+aR4k0oJ5tcvakLuG6EuKReYlHNrgE=
 github.com/spf13/cobra v1.1.1/go.mod h1:WnodtKOvamDL/PwE2M4iKs8aMDBZ5Q5klgD3qfVJQMI=
 github.com/spf13/cobra v1.1.3/go.mod h1:pGADOWyqRD/YMrPZigI/zbliZ2wVD/23d+is3pSWzOo=
 github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
@@ -730,7 +667,6 @@ github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnIn
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
-github.com/spf13/viper v1.4.0/go.mod h1:PTJ7Z/lr49W6bUbkmS1V3by4uWynFiR9p7+dSq/yZzE=
 github.com/spf13/viper v1.7.0/go.mod h1:8WkrPz2fc9jxqZNCJI/76HCieCp4Q8HaLFoCha5qpdg=
 github.com/sqs/goreturns v0.0.0-20181028201513-538ac6014518/go.mod h1:CKI4AZ4XmGV240rTHfO0hfE83S6/a3/Q1siZJ/vXf7A=
 github.com/streadway/amqp v0.0.0-20190404075320-75d898a42a94/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
@@ -749,20 +685,17 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
-github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tsenart/go-tsz v0.0.0-20180814232043-cdeb9e1e981e/go.mod h1:SWZznP1z5Ki7hDT2ioqiFKEse8K9tU2OUvaRI0NeGQo=
 github.com/tsenart/vegeta/v12 v12.8.3/go.mod h1:ZiJtwLn/9M4fTPdMY7bdbIeyNeFVE8/AHbWFqCsUuho=
 github.com/tsenart/vegeta/v12 v12.8.4 h1:UQ7tG7WkDorKj0wjx78Z4/vsMBP8RJQMGJqRVrkvngg=
 github.com/tsenart/vegeta/v12 v12.8.4/go.mod h1:ZiJtwLn/9M4fTPdMY7bdbIeyNeFVE8/AHbWFqCsUuho=
-github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
 github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
-github.com/vektah/gqlparser v1.1.2/go.mod h1:1ycwN7Ij5njmMkPPAOaRFY4rET2Enx7IkVv3vaXspKw=
 github.com/wavesoftware/go-ensure v1.0.0/go.mod h1:K2UAFSwMTvpiRGay/M3aEYYuurcR8S4A6HkQlJPV8k4=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
@@ -779,11 +712,7 @@ go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 go.etcd.io/etcd v0.0.0-20191023171146-3cf2f69b5738/go.mod h1:dnLIgRNXwCJa5e+c6mIZCrds/GIG4ncV9HhK5PX7jPg=
-go.etcd.io/etcd v0.5.0-alpha.5.0.20200819165624-17cef6e3e9d5/go.mod h1:skWido08r9w6Lq/w70DO5XYIKMu4QFu1+4VsqLQuJy8=
 go.etcd.io/etcd v0.5.0-alpha.5.0.20200910180754-dd1b699fc489/go.mod h1:yVHk9ub3CSBatqGNg7GRmsnfLWtoW60w4eDYfh7vHDg=
-go.mongodb.org/mongo-driver v1.0.3/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
-go.mongodb.org/mongo-driver v1.1.1/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
-go.mongodb.org/mongo-driver v1.1.2/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
 go.opencensus.io v0.20.1/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk=
 go.opencensus.io v0.20.2/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
@@ -821,16 +750,13 @@ golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20190320223903-b7391e95e576/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20190617133340-57b3e21c3d56/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190829043050-9756ffdc2472/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201112155050-0c6587e931a9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -878,7 +804,6 @@ golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-golang.org/x/net v0.0.0-20181005035420-146acd28ed58/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -887,11 +812,9 @@ golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190125091013-d26f9f9a57f3/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190320064053-1272bf9dcd53/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190501004415-9ce7a6920f09/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
 golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -960,7 +883,6 @@ golang.org/x/sys v0.0.0-20181122145206-62eef0e2fa9b/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20190321052220-f7bb7a8bee54/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1006,7 +928,6 @@ golang.org/x/sys v0.0.0-20201201145000-ef89a241ccb3/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20201214210602-f9fddec55a1e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210309074719-68d13333faf2/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1040,7 +961,6 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20181011042414-1f849cf54d09/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
-golang.org/x/tools v0.0.0-20190125232054-d66bd3c5d5a6/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
@@ -1052,7 +972,6 @@ golang.org/x/tools v0.0.0-20190506145303-2d16b83fe98c/go.mod h1:RgjU9mgBXZiqYHBn
 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20190606124116-d0a3d012864b/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190614205625-5aca471b1d59/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
-golang.org/x/tools v0.0.0-20190617190820-da514acc4774/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190624222133-a101b041ded4/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
 golang.org/x/tools v0.0.0-20190628153133-6cdbf07be9d0/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=
@@ -1088,7 +1007,6 @@ golang.org/x/tools v0.0.0-20200505023115-26f46d2f7ef8/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20200512001501-aaeff5de670a/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200512131952-2bc93b1c0c88/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200515010526-7d3b6ebf133d/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
-golang.org/x/tools v0.0.0-20200616133436-c1934b75d054/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200618134242-20370b0cb4b2/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
@@ -1271,29 +1189,22 @@ honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 honnef.co/go/tools v0.0.1-2020.1.5/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
-k8s.io/api v0.19.7/go.mod h1:KTryDUT3l6Mtv7K2J2486PNL9DBns3wOYTkGR+iz63Y=
 k8s.io/api v0.20.7 h1:wOEPJ3NoimUfR9v9sAO2JosPiEP9IGFNplf7zZvYzPU=
 k8s.io/api v0.20.7/go.mod h1:4x0yErUkcEWYG+O0S4QdrYa2+PLEeY2M7aeQe++2nmk=
-k8s.io/apiextensions-apiserver v0.19.7/go.mod h1:XJNNtjISNNePDEUClHt/igzMpQcmjVVh88QH+PKztPU=
 k8s.io/apiextensions-apiserver v0.20.7 h1:bAFRPfYpUvs7U2+SdCZuNFThZm8ZbQhUpEKMe1H2BYw=
 k8s.io/apiextensions-apiserver v0.20.7/go.mod h1:rBGJeRYoDJi1jJFHPA4QWXV6YX/5scZfSdkuMSgWoyA=
 k8s.io/apimachinery v0.19.2/go.mod h1:DnPGDnARWFvYa3pMHgSxtbZb7gpzzAZ1pTfaUNDVlmA=
 k8s.io/apimachinery v0.19.7/go.mod h1:6sRbGRAVY5DOCuZwB5XkqguBqpqLU6q/kOaOdk29z6Q=
 k8s.io/apimachinery v0.20.7 h1:tBfhql7OggSCahvASeEpLRzvxc7FK77wNivi1uXCQWM=
 k8s.io/apimachinery v0.20.7/go.mod h1:ejZXtW1Ra6V1O5H8xPBGz+T3+4gfkTCeExAHKU57MAc=
-k8s.io/apiserver v0.19.7/go.mod h1:DmWVQggNePspa+vSsVytVbS3iBSDTXdJVt0akfHacKk=
 k8s.io/apiserver v0.20.7 h1:kmj4lX5evfdm8h07jRjuSANvRH0kPlXTq6LOSGT6n/k=
 k8s.io/apiserver v0.20.7/go.mod h1:7gbB7UjDdP1/epYBGnIUE6jWY4Wpz99cZ7igfDa9rv4=
-k8s.io/client-go v0.19.7/go.mod h1:iytGI7S3kmv6bWnn+bSQUE4VlrEi4YFssvVB7J7Hvqg=
 k8s.io/client-go v0.20.7 h1:Ot22456XfYAWrCWddw/quevMrFHqP7s1qT499FoumVU=
 k8s.io/client-go v0.20.7/go.mod h1:uGl3qh/Jy3cTF1nDoIKBqUZlRWnj/EM+/leAXETKRuA=
-k8s.io/code-generator v0.19.7/go.mod h1:lwEq3YnLYb/7uVXLorOJfxg+cUu2oihFhHZ0n9NIla0=
 k8s.io/code-generator v0.20.7 h1:iXz1ME6EQqoCkLefa7bcniKHu0SzgbxsFV1RlBcfypc=
 k8s.io/code-generator v0.20.7/go.mod h1:i6FmG+QxaLxvJsezvZp0q/gAEzzOz3U53KFibghWToU=
-k8s.io/component-base v0.19.7/go.mod h1:YX8spPBgwl3I6UGcSdQiEMAqRMSUsGQOW7SEr4+Qa3U=
 k8s.io/component-base v0.20.7/go.mod h1:878UWprXC07P2CWFg+jjvTfxJSlkHp1v2m1MTkNQnJY=
 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=
@@ -1308,21 +1219,18 @@ k8s.io/klog/v2 v2.4.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
 k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H64/qeyjQoUZhGpeFDVdxjTeEVN2o=
 k8s.io/kube-openapi v0.0.0-20201113171705-d219536bb9fd h1:sOHNzJIkytDF6qadMNKhhDRpc6ODik8lVC6nOur7B2c=
 k8s.io/kube-openapi v0.0.0-20201113171705-d219536bb9fd/go.mod h1:WOJ3KddDSol4tAGcJo0Tvi+dK12EcqSLqcWsryKMpfM=
-k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 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.20210621164716-b312f0e1c8e5 h1:zNhrpEfWKq5SMyBetHQbQXw2rc8CAbvS9vFZpErrFJo=
-knative.dev/eventing v0.23.1-0.20210621164716-b312f0e1c8e5/go.mod h1:oy/wTY1ulXVuZD1ROldNYq+xd29Pe2IYmD84JiCzy5o=
-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-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=
+knative.dev/eventing v0.23.1-0.20210624083445-2fcbc0ac500d h1:EKha5vYrqbZo2336nu8APUcIyhVuvt8q2CdO+7b/zSE=
+knative.dev/eventing v0.23.1-0.20210624083445-2fcbc0ac500d/go.mod h1:9xo0SWkIfpXrx0lvGQO7MUlPF8cu+QCMd2gGxj6wxrU=
+knative.dev/hack v0.0.0-20210622141627-e28525d8d260 h1:f2eMtOubAOc/Q7JlvFPDKXiPlJVK+VpX2Cot8hRzCgQ=
+knative.dev/hack v0.0.0-20210622141627-e28525d8d260/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack/schema v0.0.0-20210622141627-e28525d8d260/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
+knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80 h1:GHJ3lglE0/YHfBMMJqluqUNLOmsNXh7s7DBnfrkpRMM=
+knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80/go.mod h1:kGegTnbZ+ljFjAE3E1+8wgaH2LMv8qYi+72o3F3cbdc=
+knative.dev/reconciler-test v0.0.0-20210623134345-88c84739abd9 h1:j+ZLX0o1vBxEpTc7J6QT6sSBblQHpZhJP4vxbFe+C5Y=
+knative.dev/reconciler-test v0.0.0-20210623134345-88c84739abd9/go.mod h1:4wqv2WyWUC5yhTesRUVwgjv/fHTHny1RYBfdB6tVDok=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
@@ -1333,7 +1241,6 @@ pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
 rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
-sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.9/go.mod h1:dzAXnQbTRyDlZPJX2SUPEqvnB+j7AJjtlox7PEwigU0=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.15/go.mod h1:LEScyzhFmoF5pso/YSeBstl57mOzx9xlU9n85RGrDQg=
 sigs.k8s.io/structured-merge-diff/v4 v4.0.1/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK1F7G282QMXDPYydCw=
 sigs.k8s.io/structured-merge-diff/v4 v4.0.2/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK1F7G282QMXDPYydCw=
diff --git a/hack/data-plane.sh b/hack/data-plane.sh
index b5744eb05..f6183e3c5 100755
--- a/hack/data-plane.sh
+++ b/hack/data-plane.sh
@@ -108,7 +108,12 @@ function data_plane_build_push() {
 
   local uuid=${UUID}
   if [ "${uuid}" = "latest" ]; then
-    uuid="$(uuidgen --time)"
+    if [[ "$OSTYPE" == "darwin"* ]]; then
+      # MacOS
+      uuid="$(uuidgen)"
+    else
+      uuid="$(uuidgen --time)"
+    fi
   fi
 
   export KNATIVE_KAFKA_BROKER_RECEIVER_IMAGE="${KO_DOCKER_REPO}"/"${receiver}":"${uuid}"
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1/delivery_types.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1/delivery_types.go
index 0577bd584..b82bbc4f9 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1/delivery_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1/delivery_types.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"github.com/rickb777/date/period"
+	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
@@ -37,6 +38,15 @@ type DeliverySpec struct {
 	// +optional
 	Retry *int32 `json:"retry,omitempty"`
 
+	// Timeout is the timeout of each single request. The value must be greater than 0.
+	// More information on Duration format:
+	//  - https://www.iso.org/iso-8601-date-and-time-format.html
+	//  - https://en.wikipedia.org/wiki/ISO_8601
+	//
+	// Note: This API is EXPERIMENTAL and might break anytime. For more details: https://github.com/knative/eventing/issues/5148
+	// +optional
+	Timeout *string `json:"timeout,omitempty"`
+
 	// BackoffPolicy is the retry backoff policy (linear, exponential).
 	// +optional
 	BackoffPolicy *BackoffPolicyType `json:"backoffPolicy,omitempty"`
@@ -65,6 +75,17 @@ func (ds *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(apis.ErrInvalidValue(*ds.Retry, "retry"))
 	}
 
+	if ds.Timeout != nil {
+		if feature.FromContext(ctx).IsEnabled(feature.DeliveryTimeout) {
+			t, te := period.Parse(*ds.Timeout)
+			if te != nil || t.IsZero() {
+				errs = errs.Also(apis.ErrInvalidValue(*ds.Timeout, "timeout"))
+			}
+		} else {
+			errs = errs.Also(apis.ErrDisallowedFields("timeout"))
+		}
+	}
+
 	if ds.BackoffPolicy != nil {
 		switch *ds.BackoffPolicy {
 		case BackoffPolicyExponential, BackoffPolicyLinear:
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1/zz_generated.deepcopy.go
index 902fb4b1f..0501d0e60 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1/zz_generated.deepcopy.go
@@ -146,6 +146,11 @@ func (in *DeliverySpec) DeepCopyInto(out *DeliverySpec) {
 		*out = new(int32)
 		**out = **in
 	}
+	if in.Timeout != nil {
+		in, out := &in.Timeout, &out.Timeout
+		*out = new(string)
+		**out = **in
+	}
 	if in.BackoffPolicy != nil {
 		in, out := &in.BackoffPolicy, &out.BackoffPolicy
 		*out = new(BackoffPolicyType)
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_conversion.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_conversion.go
index d8a3bc880..d48ccde0c 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_conversion.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_conversion.go
@@ -31,6 +31,7 @@ func (source *DeliverySpec) ConvertTo(ctx context.Context, to apis.Convertible)
 	case *eventingduckv1.DeliverySpec:
 		sink.Retry = source.Retry
 		sink.BackoffDelay = source.BackoffDelay
+		sink.Timeout = source.Timeout
 		if source.BackoffPolicy != nil {
 			if *source.BackoffPolicy == BackoffPolicyLinear {
 				linear := eventingduckv1.BackoffPolicyLinear
@@ -55,6 +56,7 @@ func (sink *DeliverySpec) ConvertFrom(ctx context.Context, from apis.Convertible
 	case *eventingduckv1.DeliverySpec:
 		sink.Retry = source.Retry
 		sink.BackoffDelay = source.BackoffDelay
+		sink.Timeout = source.Timeout
 		if source.BackoffPolicy != nil {
 			if *source.BackoffPolicy == eventingduckv1.BackoffPolicyLinear {
 				linear := BackoffPolicyLinear
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_types.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_types.go
index ac2001704..33ebd3ae2 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_types.go
@@ -38,6 +38,13 @@ type DeliverySpec struct {
 	// +optional
 	Retry *int32 `json:"retry,omitempty"`
 
+	// Timeout is the timeout of each single request.
+	// More information on Duration format:
+	//  - https://www.iso.org/iso-8601-date-and-time-format.html
+	//  - https://en.wikipedia.org/wiki/ISO_8601
+	//
+	Timeout *string `json:"timeout,omitempty"`
+
 	// BackoffPolicy is the retry backoff policy (linear, exponential).
 	// +optional
 	BackoffPolicy *BackoffPolicyType `json:"backoffPolicy,omitempty"`
@@ -66,6 +73,13 @@ func (ds *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(apis.ErrInvalidValue(*ds.Retry, "retry"))
 	}
 
+	if ds.Timeout != nil {
+		_, te := period.Parse(*ds.Timeout)
+		if te != nil {
+			errs = errs.Also(apis.ErrInvalidValue(*ds.Timeout, "timeout"))
+		}
+	}
+
 	if ds.BackoffPolicy != nil {
 		switch *ds.BackoffPolicy {
 		case BackoffPolicyExponential, BackoffPolicyLinear:
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
index c68ae7471..175fbad98 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
@@ -146,6 +146,11 @@ func (in *DeliverySpec) DeepCopyInto(out *DeliverySpec) {
 		*out = new(int32)
 		**out = **in
 	}
+	if in.Timeout != nil {
+		in, out := &in.Timeout, &out.Timeout
+		*out = new(string)
+		**out = **in
+	}
 	if in.BackoffPolicy != nil {
 		in, out := &in.BackoffPolicy, &out.BackoffPolicy
 		*out = new(BackoffPolicyType)
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/api_validation.go b/vendor/knative.dev/eventing/pkg/apis/feature/api_validation.go
new file mode 100644
index 000000000..2a5f9226a
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/api_validation.go
@@ -0,0 +1,85 @@
+/*
+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 feature
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/apis"
+)
+
+// ValidateAPIFields checks that the experimental features fields are disabled if the experimental flag is disabled.
+// experimentalFields can contain a string with dots, to identify sub-structs, like "Destination.Ref.APIVersion"
+func ValidateAPIFields(ctx context.Context, featureName string, object interface{}, experimentalFields ...string) (errs *apis.FieldError) {
+	obj := reflect.ValueOf(object)
+	obj = reflect.Indirect(obj)
+	if obj.Kind() != reflect.Struct {
+		return nil
+	}
+
+	// If feature not enabled, let's check the field is not used
+	if !FromContext(ctx).IsEnabled(featureName) {
+		for _, fieldName := range experimentalFields {
+			fieldVal := walk(obj, strings.Split(fieldName, ".")...)
+
+			if !fieldVal.IsZero() {
+				errs = errs.Also(&apis.FieldError{
+					Message: fmt.Sprintf("Disallowed field because the experimental feature '%s' is disabled", featureName),
+					Paths:   []string{fmt.Sprintf("%s.%s", obj.Type().Name(), fieldName)},
+				})
+			}
+		}
+	}
+
+	return errs
+}
+
+// ValidateAnnotations checks that the experimental features annotations are disabled if the experimental flag is disabled
+func ValidateAnnotations(ctx context.Context, featureName string, object metav1.Object, experimentalAnnotations ...string) (errs *apis.FieldError) {
+	// If feature not enabled, let's check the annotation is not used
+	if !FromContext(ctx).IsEnabled(featureName) {
+		for _, annotation := range experimentalAnnotations {
+			if _, ok := object.GetAnnotations()[annotation]; ok {
+				errs = errs.Also(&apis.FieldError{
+					Message: fmt.Sprintf("Disallowed annotation because the experimental feature '%s' is disabled", featureName),
+					Paths:   []string{annotation},
+				})
+			}
+		}
+	}
+
+	return errs
+}
+
+func walk(value reflect.Value, paths ...string) reflect.Value {
+	switch value.Kind() {
+	case reflect.Struct:
+		newVal := value.FieldByName(paths[0])
+		if len(paths) == 1 {
+			return newVal
+		}
+		return walk(value.FieldByName(paths[0]), paths[1:]...)
+	case reflect.Ptr:
+		return walk(reflect.Indirect(value), paths...)
+	default:
+		return reflect.Zero(value.Type())
+	}
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/features.go b/vendor/knative.dev/eventing/pkg/apis/feature/features.go
new file mode 100644
index 000000000..b12355886
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/features.go
@@ -0,0 +1,80 @@
+/*
+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 feature
+
+import (
+	"fmt"
+	"strings"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
+// Flag is a string value which can be either Enabled, Disabled, or Allowed.
+type Flag string
+
+const (
+	// Enabled turns on an optional behavior.
+	Enabled Flag = "Enabled"
+	// Disabled turns off an optional behavior.
+	Disabled Flag = "Disabled"
+	// Allowed neither explicitly disables or enables a behavior.
+	// eg. allow a client to control behavior with an annotation or allow a new value through validation.
+	Allowed Flag = "Allowed"
+)
+
+// Flags is a map containing all the enabled/disabled flags for the experimental features.
+// Missing entry in the map means feature is equal to feature not enabled.
+type Flags map[string]Flag
+
+// IsEnabled returns true if the feature is enabled
+func (e Flags) IsEnabled(featureName string) bool {
+	return e != nil && e[featureName] == Enabled
+}
+
+// IsAllowed returns true if the feature is enabled or allowed
+func (e Flags) IsAllowed(featureName string) bool {
+	return e.IsEnabled(featureName) || (e != nil && e[featureName] == Allowed)
+}
+
+// NewFlagsConfigFromMap creates a Flags from the supplied Map
+func NewFlagsConfigFromMap(data map[string]string) (Flags, error) {
+	flags := Flags{}
+
+	for k, v := range data {
+		if strings.HasPrefix(k, "_") {
+			// Ignore all the keys starting with _
+			continue
+		}
+		sanitizedKey := strings.TrimSpace(k)
+		if strings.EqualFold(v, string(Allowed)) {
+			flags[sanitizedKey] = Allowed
+		} else if strings.EqualFold(v, string(Disabled)) {
+			flags[sanitizedKey] = Disabled
+		} else if strings.EqualFold(v, string(Enabled)) {
+			flags[sanitizedKey] = Enabled
+		} else {
+			return Flags{}, fmt.Errorf("cannot parse the boolean flag '%s' = '%s'. Allowed values: [true, false]", k, v)
+		}
+	}
+
+	return flags, nil
+}
+
+// NewFlagsConfigFromConfigMap creates a Flags from the supplied configMap
+func NewFlagsConfigFromConfigMap(config *corev1.ConfigMap) (Flags, error) {
+	return NewFlagsConfigFromMap(config.Data)
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go b/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go
new file mode 100644
index 000000000..050316858
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go
@@ -0,0 +1,22 @@
+/*
+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 feature
+
+const (
+	KReferenceGroup = "kreference-group"
+	DeliveryTimeout = "delivery-timeout"
+)
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/store.go b/vendor/knative.dev/eventing/pkg/apis/feature/store.go
new file mode 100644
index 000000000..a22f313c2
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/store.go
@@ -0,0 +1,108 @@
+/*
+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 feature
+
+import (
+	"context"
+
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/configmap"
+)
+
+const (
+	// FlagsConfigName is the name of config map containing the experimental features flags
+	FlagsConfigName = "config-features"
+)
+
+type cfgKey struct{}
+
+// FromContext extracts a Config from the provided context.
+func FromContext(ctx context.Context) Flags {
+	x, ok := ctx.Value(cfgKey{}).(Flags)
+	if ok {
+		return x
+	}
+	return nil
+}
+
+// FromContextOrDefaults is like FromContext, but when no Flags is attached it
+// returns an empty Flags.
+func FromContextOrDefaults(ctx context.Context) Flags {
+	if cfg := FromContext(ctx); cfg != nil {
+		return cfg
+	}
+	return Flags{}
+}
+
+// ToContext attaches the provided Flags to the provided context, returning the
+// new context with the Flags attached.
+func ToContext(ctx context.Context, c Flags) context.Context {
+	return fillContextWithFeatureSpecificFlags(context.WithValue(ctx, cfgKey{}, c), c)
+}
+
+// Store is a typed wrapper around configmap.Untyped store to handle our configmaps.
+// +k8s:deepcopy-gen=false
+type Store struct {
+	*configmap.UntypedStore
+}
+
+// NewStore creates a new store of Configs and optionally calls functions when ConfigMaps are updated.
+func NewStore(logger configmap.Logger, onAfterStore ...func(name string, value interface{})) *Store {
+	store := &Store{
+		UntypedStore: configmap.NewUntypedStore(
+			"feature-flags",
+			logger,
+			configmap.Constructors{
+				FlagsConfigName: NewFlagsConfigFromConfigMap,
+			},
+			onAfterStore...,
+		),
+	}
+
+	return store
+}
+
+// ToContext attaches the current Config state to the provided context.
+func (s *Store) ToContext(ctx context.Context) context.Context {
+	return ToContext(ctx, s.Load())
+}
+
+// IsEnabled is a shortcut for Load().IsEnabled(featureName)
+func (s *Store) IsEnabled(featureName string) bool {
+	return s.Load().IsEnabled(featureName)
+}
+
+// IsAllowed is a shortcut for Load().IsAllowed(featureName)
+func (s *Store) IsAllowed(featureName string) bool {
+	return s.Load().IsAllowed(featureName)
+}
+
+// Load creates a Config from the current config state of the Store.
+func (s *Store) Load() Flags {
+	loaded := s.UntypedLoad(FlagsConfigName)
+	if loaded == nil {
+		return Flags(nil)
+	}
+	return loaded.(Flags)
+}
+
+func fillContextWithFeatureSpecificFlags(ctx context.Context, flags Flags) context.Context {
+	if flags.IsEnabled(KReferenceGroup) {
+		ctx = duckv1.KReferenceGroupAllowed(ctx)
+	}
+	return ctx
+}
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
index 8ff4c6c70..6d223f7da 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
@@ -57,8 +57,18 @@ func (s *HTTPMessageSender) SendWithRetries(req *nethttp.Request, config *RetryC
 		return s.Send(req)
 	}
 
+	client := s.Client
+	if config.RequestTimeout != 0 {
+		client = &nethttp.Client{
+			Transport:     client.Transport,
+			CheckRedirect: client.CheckRedirect,
+			Jar:           client.Jar,
+			Timeout:       config.RequestTimeout,
+		}
+	}
+
 	retryableClient := retryablehttp.Client{
-		HTTPClient:   s.Client,
+		HTTPClient:   client,
 		RetryWaitMin: defaultRetryWaitMin,
 		RetryWaitMax: defaultRetryWaitMax,
 		RetryMax:     config.RetryMax,
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/retries.go b/vendor/knative.dev/eventing/pkg/kncloudevents/retries.go
index b43e93e08..ee816efa1 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/retries.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/retries.go
@@ -63,6 +63,9 @@ type RetryConfig struct {
 
 	CheckRetry CheckRetry
 	Backoff    Backoff
+
+	// RequestTimeout represents the timeout of the single request
+	RequestTimeout time.Duration
 }
 
 func NoRetries() RetryConfig {
@@ -101,6 +104,14 @@ func RetryConfigFromDeliverySpec(spec v1.DeliverySpec) (RetryConfig, error) {
 		}
 	}
 
+	if spec.Timeout != nil {
+		timeout, err := period.Parse(*spec.Timeout)
+		if err != nil {
+			return retryConfig, fmt.Errorf("failed to parse Spec.Timeout: %w", err)
+		}
+		retryConfig.RequestTimeout, _ = timeout.Duration()
+	}
+
 	return retryConfig, nil
 }
 
diff --git a/vendor/knative.dev/pkg/leaderelection/config.go b/vendor/knative.dev/pkg/leaderelection/config.go
index 3d9df1dc7..8d785c522 100644
--- a/vendor/knative.dev/pkg/leaderelection/config.go
+++ b/vendor/knative.dev/pkg/leaderelection/config.go
@@ -25,7 +25,6 @@ import (
 
 	"github.com/kelseyhightower/envconfig"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 
@@ -79,11 +78,6 @@ type Config struct {
 	LeaseDuration time.Duration
 	RenewDeadline time.Duration
 	RetryPeriod   time.Duration
-
-	// This field is deprecated and will be removed once downstream
-	// repositories have removed their validation of it.
-	// TODO(https://github.com/knative/pkg/issues/1478): Remove this field.
-	EnabledComponents sets.String
 }
 
 func (c *Config) GetComponentConfig(name string) ComponentConfig {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 5ea65da69..9d2f43332 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.20210621164716-b312f0e1c8e5
+# knative.dev/eventing v0.23.1-0.20210624083445-2fcbc0ac500d
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1035,6 +1035,7 @@ knative.dev/eventing/pkg/apis/duck/v1beta1
 knative.dev/eventing/pkg/apis/eventing
 knative.dev/eventing/pkg/apis/eventing/v1
 knative.dev/eventing/pkg/apis/eventing/v1beta1
+knative.dev/eventing/pkg/apis/feature
 knative.dev/eventing/pkg/apis/flows
 knative.dev/eventing/pkg/apis/flows/v1
 knative.dev/eventing/pkg/apis/messaging
@@ -1128,10 +1129,10 @@ knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
-# knative.dev/hack v0.0.0-20210614141220-66ab1a098940
+# knative.dev/hack v0.0.0-20210622141627-e28525d8d260
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20210618060751-f454995ff92b
+# knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1222,7 +1223,7 @@ knative.dev/pkg/webhook/certificates/resources
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7
+# knative.dev/reconciler-test v0.0.0-20210623134345-88c84739abd9
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/.github/workflows/knative-go-test.yaml b/.github/workflows/knative-go-test.yaml
index e0ebe43fd..6abb34c2c 100644
--- a/.github/workflows/knative-go-test.yaml
+++ b/.github/workflows/knative-go-test.yaml
@@ -46,6 +46,15 @@ jobs:
 
       - name: Check out code
         uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+
+      - name: Merge upstream
+        run: |
+          git remote add upstream https://github.com/${{ github.repository }}.git
+          git fetch upstream ${{ github.base_ref }}
+          git pull --no-rebase --no-commit upstream ${{ github.base_ref }}
+        shell: bash
 
       - name: Check for .codecov.yaml
         id: codecov-enabled
diff --git a/control-plane/pkg/core/config/utils_test.go b/control-plane/pkg/core/config/utils_test.go
index 6d80c21bb..1317ed6ba 100644
--- a/control-plane/pkg/core/config/utils_test.go
+++ b/control-plane/pkg/core/config/utils_test.go
@@ -111,7 +111,7 @@ func TestBackoffPolicyFromString(t *testing.T) {
 	}
 }
 
-func TestBackoffDelayFromString(t *testing.T) {
+func TestDurationMillisFromISO8601String(t *testing.T) {
 
 	tests := []struct {
 		name         string
@@ -170,12 +170,12 @@ func TestBackoffDelayFromString(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := BackoffDelayFromISO8601String(tt.backoffDelay, tt.defaultDelay)
+			got, err := DurationMillisFromISO8601String(tt.backoffDelay, tt.defaultDelay)
 			if (err != nil) != tt.wantError {
 				t.Errorf("wantError = %v got %v", tt.wantError, err)
 			}
 			if got != tt.want {
-				t.Errorf("BackoffDelayFromISO8601String() = %v, want %v", got, tt.want)
+				t.Errorf("DurationMillisFromISO8601String() = %v, want %v", got, tt.want)
 			}
 		})
 	}
@@ -275,6 +275,7 @@ func TestEgressConfigFromDelivery(t *testing.T) {
 				Retry:          pointer.Int32Ptr(3),
 				BackoffPolicy:  &exponential,
 				BackoffDelay:   pointer.StringPtr("PT1S"),
+				Timeout:        pointer.StringPtr("PT2S"),
 			},
 			defaultBackoffDelayMs: 0,
 			want: &contract.EgressConfig{
@@ -282,6 +283,7 @@ func TestEgressConfigFromDelivery(t *testing.T) {
 				Retry:         3,
 				BackoffPolicy: contract.BackoffPolicy_Exponential,
 				BackoffDelay:  uint64(time.Second.Milliseconds()),
+				Timeout:       uint64(time.Second.Milliseconds() * 2),
 			},
 			wantErr: false,
 		},
@@ -299,6 +301,20 @@ func TestEgressConfigFromDelivery(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name:     "only timeout",
+			ctx:      ctx,
+			resolver: resolver.NewURIResolver(ctx, func(name types.NamespacedName) {}),
+			parent:   &eventing.KafkaSink{},
+			delivery: &eventingduck.DeliverySpec{
+				Timeout: pointer.StringPtr("PT2S"),
+			},
+			defaultBackoffDelayMs: 0,
+			want: &contract.EgressConfig{
+				Timeout: uint64(time.Second.Milliseconds() * 2),
+			},
+			wantErr: false,
+		},
 		{
 			name:     "only retry - use default backoff delay",
 			ctx:      ctx,
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 0b7cd65c6..49cef136e 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -322,6 +322,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 										Retry:         3,
 										BackoffPolicy: contract.BackoffPolicy_Exponential,
 										BackoffDelay:  uint64(time.Second.Milliseconds()),
+										Timeout:       uint64((time.Second * 2).Milliseconds()),
 									},
 								},
 							},
@@ -1408,6 +1409,7 @@ func withDelivery(trigger *eventing.Trigger) {
 		Retry:          pointer.Int32Ptr(3),
 		BackoffPolicy:  &exponential,
 		BackoffDelay:   pointer.StringPtr("PT1S"),
+		Timeout:        pointer.StringPtr("PT2S"),
 	}
 }
 
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourceReconcilerTestRunner.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourceReconcilerTestRunner.java
index d5d8ac1d6..ca9908dce 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourceReconcilerTestRunner.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourceReconcilerTestRunner.java
@@ -16,8 +16,8 @@
 package dev.knative.eventing.kafka.broker.core.reconciler.impl;
 
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import io.vertx.core.Future;
-
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -127,7 +127,7 @@ public class ResourceReconcilerTestRunner {
     final var ingressListener = this.ingressReconcilerListener;
     final var egressListener = this.egressReconcilerListener;
 
-    final var reconcilerBuilder = ResourcesReconcilerImpl
+    final var reconcilerBuilder = ResourcesReconciler
       .builder();
 
     if (ingressListener != null) {
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImplTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImplTest.java
index 761740088..5e55eaf45 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImplTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImplTest.java
@@ -15,14 +15,13 @@
  */
 package dev.knative.eventing.kafka.broker.core.reconciler.impl;
 
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import io.vertx.core.Future;
-import org.junit.jupiter.api.Test;
-
 import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
-
-import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import org.junit.jupiter.api.Test;
 
 import static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Egress;
 import static dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Filter;
@@ -33,7 +32,7 @@ class ResourcesReconcilerImplTest {
 
   @Test
   void nullPointerExceptionWhenNoListenerIsProvided() {
-    assertThatThrownBy(() -> ResourcesReconcilerImpl.builder().build())
+    assertThatThrownBy(() -> ResourcesReconciler.builder().build())
       .isInstanceOf(NullPointerException.class);
   }
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/ConsumerDeployerVerticleTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/ConsumerDeployerVerticleTest.java
index 822018f78..7097fad3b 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/ConsumerDeployerVerticleTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/ConsumerDeployerVerticleTest.java
@@ -15,18 +15,8 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher;
 
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress1;
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress2;
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress3;
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress4;
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress5;
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress6;
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.resource1;
-import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.resource2;
-import static org.assertj.core.api.Assertions.assertThat;
-
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
-import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerImpl;
+import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
@@ -42,6 +32,16 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.parallel.Execution;
 import org.junit.jupiter.api.parallel.ExecutionMode;
 
+import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress1;
+import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress2;
+import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress3;
+import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress4;
+import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress5;
+import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.egress6;
+import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.resource1;
+import static dev.knative.eventing.kafka.broker.core.testing.CoreObjects.resource2;
+import static org.assertj.core.api.Assertions.assertThat;
+
 @ExtendWith(VertxExtension.class)
 @Execution(ExecutionMode.CONCURRENT)
 public class ConsumerDeployerVerticleTest {
@@ -70,7 +70,7 @@ public class ConsumerDeployerVerticleTest {
       .toCompletableFuture()
       .get();
 
-    final var reconciler = ResourcesReconcilerImpl
+    final var reconciler = ResourcesReconciler
       .builder()
       .watchEgress(consumerDeployer)
       .build();
@@ -107,7 +107,7 @@ public class ConsumerDeployerVerticleTest {
       .toCompletableFuture()
       .get();
 
-    final var reconciler = ResourcesReconcilerImpl
+    final var reconciler = ResourcesReconciler
       .builder()
       .watchEgress(consumerDeployer)
       .build();
@@ -156,7 +156,7 @@ public class ConsumerDeployerVerticleTest {
       .toCompletableFuture()
       .get();
 
-    final var reconciler = ResourcesReconcilerImpl
+    final var reconciler = ResourcesReconciler
       .builder()
       .watchEgress(consumerDeployer)
       .build();
@@ -226,7 +226,7 @@ public class ConsumerDeployerVerticleTest {
       .toCompletableFuture()
       .get();
 
-    final var reconciler = ResourcesReconcilerImpl
+    final var reconciler = ResourcesReconciler
       .builder()
       .watchEgress(consumerDeployer)
       .build();
@@ -307,7 +307,7 @@ public class ConsumerDeployerVerticleTest {
       .toCompletableFuture()
       .get();
 
-    final var reconciler = ResourcesReconcilerImpl
+    final var reconciler = ResourcesReconciler
       .builder()
       .watchEgress(consumerDeployer)
       .build();
@@ -380,7 +380,7 @@ public class ConsumerDeployerVerticleTest {
       .toCompletableFuture()
       .get();
 
-    final var reconciler = ResourcesReconcilerImpl
+    final var reconciler = ResourcesReconciler
       .builder()
       .watchEgress(consumerDeployer)
       .build();
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 eaf8e4c60..2c3e38327 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
@@ -17,7 +17,6 @@ package dev.knative.eventing.kafka.broker.dispatcher;
 
 import dev.knative.eventing.kafka.broker.core.filter.Filter;
 import dev.knative.eventing.kafka.broker.core.testing.CoreObjects;
-import dev.knative.eventing.kafka.broker.dispatcher.consumer.OffsetManager;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
 import io.vertx.junit5.VertxExtension;
@@ -45,9 +44,9 @@ public class RecordDispatcherTest {
   @Test
   public void shouldNotSendToSubscriberNorToDeadLetterSinkIfValueDoesntMatch() {
 
-    final OffsetManager receiver = offsetManagerMock();
+    final RecordDispatcherListener receiver = offsetManagerMock();
 
-    final var dispatcherHandler = new RecordDispatcher(
+    final var dispatcherHandler = new RecordDispatcherImpl(
       value -> false,
       ConsumerRecordSender.noop("subscriber send called"),
       ConsumerRecordSender.noop("DLS send called"),
@@ -57,7 +56,7 @@ public class RecordDispatcherTest {
     );
 
     final var record = record();
-    dispatcherHandler.handle(record);
+    dispatcherHandler.dispatch(record);
 
     verify(receiver, times(1)).recordReceived(record);
     verify(receiver, times(1)).recordDiscarded(record);
@@ -70,9 +69,9 @@ public class RecordDispatcherTest {
   public void shouldSendOnlyToSubscriberIfValueMatches() {
 
     final var sendCalled = new AtomicBoolean(false);
-    final OffsetManager receiver = offsetManagerMock();
+    final RecordDispatcherListener receiver = offsetManagerMock();
 
-    final var dispatcherHandler = new RecordDispatcher(
+    final var dispatcherHandler = new RecordDispatcherImpl(
       value -> true, new ConsumerRecordSenderMock(
       record -> {
         sendCalled.set(true);
@@ -90,7 +89,7 @@ public class RecordDispatcherTest {
       null
     );
     final var record = record();
-    dispatcherHandler.handle(record);
+    dispatcherHandler.dispatch(record);
 
     assertTrue(sendCalled.get());
     verify(receiver, times(1)).recordReceived(record);
@@ -105,9 +104,9 @@ public class RecordDispatcherTest {
 
     final var subscriberSenderSendCalled = new AtomicBoolean(false);
     final var dlsSenderSendCalled = new AtomicBoolean(false);
-    final OffsetManager receiver = offsetManagerMock();
+    final RecordDispatcherListener receiver = offsetManagerMock();
 
-    final var dispatcherHandler = new RecordDispatcher(
+    final var dispatcherHandler = new RecordDispatcherImpl(
       value -> true, new ConsumerRecordSenderMock(
       record -> {
         subscriberSenderSendCalled.set(true);
@@ -124,7 +123,7 @@ public class RecordDispatcherTest {
       null
     );
     final var record = record();
-    dispatcherHandler.handle(record);
+    dispatcherHandler.dispatch(record);
 
     assertTrue(subscriberSenderSendCalled.get());
     assertTrue(dlsSenderSendCalled.get());
@@ -140,9 +139,9 @@ public class RecordDispatcherTest {
 
     final var subscriberSenderSendCalled = new AtomicBoolean(false);
     final var dlsSenderSendCalled = new AtomicBoolean(false);
-    final OffsetManager receiver = offsetManagerMock();
+    final RecordDispatcherListener receiver = offsetManagerMock();
 
-    final var dispatcherHandler = new RecordDispatcher(
+    final var dispatcherHandler = new RecordDispatcherImpl(
       value -> true, new ConsumerRecordSenderMock(
       record -> {
         subscriberSenderSendCalled.set(true);
@@ -160,7 +159,7 @@ public class RecordDispatcherTest {
       null
     );
     final var record = record();
-    dispatcherHandler.handle(record);
+    dispatcherHandler.dispatch(record);
 
     assertTrue(subscriberSenderSendCalled.get());
     assertTrue(dlsSenderSendCalled.get());
@@ -174,9 +173,9 @@ public class RecordDispatcherTest {
   @Test
   public void shouldCallFailedToSendToDeadLetterSinkIfValueMatchesAndSubscriberSenderFailsAndNoDeadLetterSinkSender() {
     final var subscriberSenderSendCalled = new AtomicBoolean(false);
-    final OffsetManager receiver = offsetManagerMock();
+    final RecordDispatcherListener receiver = offsetManagerMock();
 
-    final var dispatcherHandler = new RecordDispatcher(
+    final var dispatcherHandler = new RecordDispatcherImpl(
       value -> true,
       new ConsumerRecordSenderMock(
         record -> {
@@ -190,7 +189,7 @@ public class RecordDispatcherTest {
       null
     );
     final var record = record();
-    dispatcherHandler.handle(record);
+    dispatcherHandler.dispatch(record);
 
     assertTrue(subscriberSenderSendCalled.get());
     verify(receiver, times(1)).recordReceived(record);
@@ -212,7 +211,7 @@ public class RecordDispatcherTest {
     final var deadLetterSender = mock(ConsumerRecordSender.class);
     when(deadLetterSender.close()).thenReturn(Future.succeededFuture());
 
-    final RecordDispatcher recordDispatcher = new RecordDispatcher(
+    final RecordDispatcher recordDispatcher = new RecordDispatcherImpl(
       Filter.noop(), subscriberSender,
       deadLetterSender, sinkResponseHandler, offsetManagerMock(), null);
 
@@ -230,15 +229,15 @@ public class RecordDispatcherTest {
     return new KafkaConsumerRecordImpl<>(new ConsumerRecord<>("", 0, 0L, "", CoreObjects.event()));
   }
 
-  public static OffsetManager offsetManagerMock() {
-    final OffsetManager offsetManager = mock(OffsetManager.class);
+  public static RecordDispatcherListener offsetManagerMock() {
+    final RecordDispatcherListener recordDispatcherListener = mock(RecordDispatcherListener.class);
 
-    when(offsetManager.recordReceived(any())).thenReturn(Future.succeededFuture());
-    when(offsetManager.recordDiscarded(any())).thenReturn(Future.succeededFuture());
-    when(offsetManager.successfullySentToDeadLetterSink(any())).thenReturn(Future.succeededFuture());
-    when(offsetManager.successfullySentToSubscriber(any())).thenReturn(Future.succeededFuture());
-    when(offsetManager.failedToSendToDeadLetterSink(any(), any())).thenReturn(Future.succeededFuture());
+    when(recordDispatcherListener.recordReceived(any())).thenReturn(Future.succeededFuture());
+    when(recordDispatcherListener.recordDiscarded(any())).thenReturn(Future.succeededFuture());
+    when(recordDispatcherListener.successfullySentToDeadLetterSink(any())).thenReturn(Future.succeededFuture());
+    when(recordDispatcherListener.successfullySentToSubscriber(any())).thenReturn(Future.succeededFuture());
+    when(recordDispatcherListener.failedToSendToDeadLetterSink(any(), any())).thenReturn(Future.succeededFuture());
 
-    return offsetManager;
+    return recordDispatcherListener;
   }
 }
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 895a8f2e7..9cfb11095 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
@@ -17,7 +17,7 @@ package dev.knative.eventing.kafka.broker.dispatcher.consumer.impl;
 
 import dev.knative.eventing.kafka.broker.dispatcher.ConsumerRecordSender;
 import dev.knative.eventing.kafka.broker.dispatcher.ConsumerRecordSenderMock;
-import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherTest;
 import dev.knative.eventing.kafka.broker.dispatcher.SinkResponseHandlerMock;
 import io.cloudevents.CloudEvent;
@@ -55,7 +55,7 @@ public abstract class AbstractConsumerVerticleTest {
   @SuppressWarnings("unchecked")
   public void subscribedToTopic(final Vertx vertx, final VertxTestContext context) {
     final var consumer = new MockConsumer<String, CloudEvent>(OffsetResetStrategy.LATEST);
-    final var recordDispatcher = new RecordDispatcher(
+    final var recordDispatcher = new RecordDispatcherImpl(
       value -> false,
       ConsumerRecordSender.noop("subscriber send called"),
       ConsumerRecordSender.noop("dead letter sink send called"),
@@ -91,7 +91,7 @@ public abstract class AbstractConsumerVerticleTest {
   @SuppressWarnings("unchecked")
   public void stop(final Vertx vertx, final VertxTestContext context) {
     final var consumer = new MockConsumer<String, CloudEvent>(OffsetResetStrategy.LATEST);
-    final var recordDispatcher = new RecordDispatcher(
+    final var recordDispatcher = new RecordDispatcherImpl(
       value -> false,
       ConsumerRecordSender.noop("subscriber send called"),
       ConsumerRecordSender.noop("dead letter sink send called"),
@@ -165,17 +165,17 @@ public abstract class AbstractConsumerVerticleTest {
     final var dlsSenderClosed = new AtomicBoolean(false);
     final var sinkClosed = new AtomicBoolean(false);
 
-    final var recordDispatcher = new RecordDispatcher(
+    final var recordDispatcher = new RecordDispatcherImpl(
       ce -> true,
       new ConsumerRecordSenderMock(
         record -> Future.succeededFuture(), () -> {
-          consumerRecordSenderClosed.set(true);
-          return Future.succeededFuture();
-        }
+        consumerRecordSenderClosed.set(true);
+        return Future.succeededFuture();
+      }
       ),
       new ConsumerRecordSenderMock(
         record -> Future.succeededFuture(), () -> {
-          dlsSenderClosed.set(true);
+        dlsSenderClosed.set(true);
           return Future.succeededFuture();
         }
       ),
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/AbstractOffsetManagerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/AbstractOffsetManagerTest.java
index 4650a8ca2..ce19db454 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/AbstractOffsetManagerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/AbstractOffsetManagerTest.java
@@ -15,7 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.consumer.impl;
 
-import dev.knative.eventing.kafka.broker.dispatcher.consumer.OffsetManager;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
@@ -45,7 +45,7 @@ import static org.mockito.Mockito.verify;
 
 public abstract class AbstractOffsetManagerTest {
 
-  abstract OffsetManager createOffsetManager(final KafkaConsumer<?, ?> consumer);
+  abstract RecordDispatcherListener createOffsetManager(final KafkaConsumer<?, ?> consumer);
 
   protected static KafkaConsumerRecord<String, CloudEvent> record(String topic, int partition, long offset) {
     return new KafkaConsumerRecordImpl<>(
@@ -60,14 +60,14 @@ public abstract class AbstractOffsetManagerTest {
   }
 
   protected MapAssert<TopicPartition, Long> assertThatOffsetCommitted(
-    Collection<TopicPartition> partitionsConsumed, Consumer<OffsetManager> testExecutor) {
+    Collection<TopicPartition> partitionsConsumed, Consumer<RecordDispatcherListener> testExecutor) {
     return assertThatOffsetCommittedWithFailures(partitionsConsumed,
       (offsetStrategy, flag) -> testExecutor.accept(offsetStrategy));
   }
 
   protected MapAssert<TopicPartition, Long> assertThatOffsetCommittedWithFailures(
     Collection<TopicPartition> partitionsConsumed,
-    BiConsumer<OffsetManager, AtomicBoolean> testExecutor) {
+    BiConsumer<RecordDispatcherListener, AtomicBoolean> testExecutor) {
     final var mockConsumer = new MockConsumer<String, CloudEvent>(OffsetResetStrategy.NONE);
     mockConsumer.assign(partitionsConsumed);
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedConsumerVerticleTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedConsumerVerticleTest.java
index 2b3b3a679..ce88ef85c 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedConsumerVerticleTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/OrderedConsumerVerticleTest.java
@@ -15,7 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.consumer.impl;
 
-import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherImpl;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
@@ -81,8 +81,8 @@ public class OrderedConsumerVerticleTest extends AbstractConsumerVerticleTest {
     // Mock the record dispatcher to count down the latch and save the received records order
     CountDownLatch latch = new CountDownLatch(tasks);
     final Map<TopicPartition, List<Long>> receivedRecords = new HashMap<>(partitions);
-    final var recordDispatcher = mock(RecordDispatcher.class);
-    when(recordDispatcher.apply(any())).then(invocation -> {
+    final var recordDispatcher = mock(RecordDispatcherImpl.class);
+    when(recordDispatcher.dispatch(any())).then(invocation -> {
       final KafkaConsumerRecord<String, CloudEvent> record = invocation.getArgument(0);
       return recordDispatcherLogicMock(vertx, random, delay, latch, record, receivedRecords);
     });
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 4cad6b05d..a63cd78a3 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
@@ -15,7 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.consumer.impl;
 
-import dev.knative.eventing.kafka.broker.dispatcher.consumer.OffsetManager;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
 import java.util.List;
 import org.apache.kafka.common.TopicPartition;
@@ -27,7 +27,7 @@ import org.junit.jupiter.api.parallel.ExecutionMode;
 public class OrderedOffsetManagerTest extends AbstractOffsetManagerTest {
 
   @Override
-  OffsetManager createOffsetManager(
+  RecordDispatcherListener createOffsetManager(
     KafkaConsumer<?, ?> consumer) {
     return new OrderedOffsetManager(consumer, null);
   }
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 ceac82d1b..c74081ea5 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
@@ -15,7 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.dispatcher.consumer.impl;
 
-import dev.knative.eventing.kafka.broker.dispatcher.consumer.OffsetManager;
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.cloudevents.CloudEvent;
 import io.micrometer.core.instrument.Counter;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
@@ -33,7 +33,7 @@ import static org.mockito.Mockito.verify;
 public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
 
   @Override
-  OffsetManager createOffsetManager(
+  RecordDispatcherListener createOffsetManager(
     KafkaConsumer<?, ?> consumer) {
     return new UnorderedOffsetManager(consumer, null);
   }
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverTracingVerticleTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverTracingVerticleTest.java
index cc068e2c3..c20da6972 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverTracingVerticleTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverTracingVerticleTest.java
@@ -103,7 +103,7 @@ public class ReceiverTracingVerticleTest {
       vertx,
       null,
       new Properties(),
-      new CloudEventRequestToRecordMapper(),
+      StrictRequestToRecordMapper.getInstance(),
       properties -> KafkaProducer.create(vertx, mockProducer),
       new CumulativeCounter(mock(Id.class)),
       new CumulativeCounter(mock(Id.class))
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleTest.java
index 52d1d1194..ad4a508e4 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleTest.java
@@ -97,7 +97,7 @@ public class ReceiverVerticleTest {
       vertx,
       null,
       new Properties(),
-      new CloudEventRequestToRecordMapper(),
+      StrictRequestToRecordMapper.getInstance(),
       properties -> producer,
       badRequestCount,
       produceRequestCount
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/RequestMapperTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/RequestMapperTest.java
index 3f54d4f83..a7467e47f 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/RequestMapperTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/RequestMapperTest.java
@@ -15,20 +15,10 @@
  */
 package dev.knative.eventing.kafka.broker.receiver;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.entry;
-import static org.assertj.core.api.InstanceOfAssertFactories.map;
-import static org.junit.jupiter.api.Assertions.fail;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Resource;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
-import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerImpl;
+import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import dev.knative.eventing.kafka.broker.core.testing.CoreObjects;
 import io.cloudevents.CloudEvent;
 import io.micrometer.core.instrument.Counter;
@@ -57,6 +47,16 @@ import org.apache.kafka.clients.producer.MockProducer;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.entry;
+import static org.assertj.core.api.InstanceOfAssertFactories.map;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
 @ExtendWith(VertxExtension.class)
 public class RequestMapperTest {
 
@@ -114,7 +114,8 @@ public class RequestMapperTest {
       mock(Counter.class),
       mock(Counter.class)
     );
-    final var reconciler = ResourcesReconcilerImpl.builder()
+    final var reconciler = ResourcesReconciler
+      .builder()
       .watchIngress(handler)
       .build();
 
@@ -157,7 +158,8 @@ public class RequestMapperTest {
       mock(Counter.class),
       mock(Counter.class)
     );
-    final var reconciler = ResourcesReconcilerImpl.builder()
+    final var reconciler = ResourcesReconciler
+      .builder()
       .watchIngress(handler)
       .build();
 
@@ -428,7 +430,8 @@ public class RequestMapperTest {
       mock(Counter.class),
       mock(Counter.class)
     );
-    final var reconciler = ResourcesReconcilerImpl.builder()
+    final var reconciler = ResourcesReconciler
+      .builder()
       .watchIngress(handler)
       .build();
 
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverEnvTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
similarity index 98%
rename from data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverEnvTest.java
rename to data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
index 3cb818c62..6c961ec90 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverEnvTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.receiver;
+package dev.knative.eventing.kafka.broker.receiver.main;
 
 import dev.knative.eventing.kafka.broker.core.utils.BaseEnv;
 import org.junit.jupiter.api.Test;
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/main/ReceiverVerticleFactoryTest.java
similarity index 95%
rename from data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/ReceiverVerticleFactoryTest.java
rename to data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverVerticleFactoryTest.java
index 183a9c8d0..dc10d9a37 100644
--- 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/main/ReceiverVerticleFactoryTest.java
@@ -14,13 +14,12 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.receiver;
+package dev.knative.eventing.kafka.broker.receiver.main;
 
 import io.micrometer.core.instrument.Counter;
 import io.vertx.core.http.HttpServerOptions;
-import org.junit.jupiter.api.Test;
-
 import java.util.Properties;
+import org.junit.jupiter.api.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
diff --git a/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java b/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
index 1e216438a..93b74edfb 100644
--- a/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
+++ b/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
@@ -23,9 +23,9 @@ import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcile
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.dispatcher.ConsumerDeployerVerticle;
 import dev.knative.eventing.kafka.broker.dispatcher.http.HttpConsumerVerticleFactory;
-import dev.knative.eventing.kafka.broker.receiver.CloudEventRequestToRecordMapper;
 import dev.knative.eventing.kafka.broker.receiver.ReceiverVerticle;
 import dev.knative.eventing.kafka.broker.receiver.RequestMapper;
+import dev.knative.eventing.kafka.broker.receiver.StrictRequestToRecordMapper;
 import io.cloudevents.core.builder.CloudEventBuilder;
 import io.cloudevents.core.message.MessageReader;
 import io.cloudevents.core.v1.CloudEventV1;
@@ -336,7 +336,7 @@ public class DataPlaneTest {
       v,
       null,
       producerConfigs(),
-      new CloudEventRequestToRecordMapper(),
+      StrictRequestToRecordMapper.getInstance(),
       properties -> KafkaProducer.create(v, properties),
       mock(Counter.class),
       mock(Counter.class)
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/.travis.yml b/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/.travis.yml
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/LICENSE b/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/LICENSE
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/README.md b/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/README.md
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/go.mod b/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/go.mod
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/uuid.go b/third_party/VENDOR-LICENSE/github.com/hashicorp/go-uuid/uuid.go
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/.gitignore b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/.gitignore
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/2q.go b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/2q.go
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/LICENSE b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/LICENSE
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/README.md b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/README.md
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/arc.go b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/arc.go
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/doc.go b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/doc.go
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/go.mod b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/go.mod
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/lru.go b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/lru.go
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/simplelru/lru.go b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/simplelru/lru.go
old mode 100755
new mode 100644
diff --git a/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/simplelru/lru_interface.go b/third_party/VENDOR-LICENSE/github.com/hashicorp/golang-lru/simplelru/lru_interface.go
old mode 100755
new mode 100644
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 3146133d7..4bf34ad52 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: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 
 ---
 # Copyright 2018 The Knative Authors
@@ -39,14 +39,14 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -61,7 +61,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -76,7 +76,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -91,7 +91,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -106,7 +106,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -137,14 +137,14 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: pingsource-mt-adapter
@@ -175,14 +175,14 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -198,7 +198,7 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -213,7 +213,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -228,7 +228,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -259,7 +259,7 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 data:
   channelTemplateSpec: |
     apiVersion: messaging.knative.dev/v1
@@ -286,7 +286,7 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 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: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 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: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
 data:
@@ -396,13 +396,16 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     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"
+  # ALPHA feature: The delivery-timeout allows you to use the Timeout field in DeliverySpec.
+  # For more details: https://github.com/knative/eventing/issues/5148
+  delivery-timeout: "disabled"
 
 ---
 # Copyright 2018 The Knative Authors
@@ -425,7 +428,7 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
   annotations:
     knative.dev/example-checksum: "96896b00"
 data:
@@ -485,7 +488,7 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
 data:
@@ -537,7 +540,7 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -612,7 +615,7 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -673,7 +676,7 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/high-availability: "true"
 spec:
   selector:
@@ -683,7 +686,7 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20210618-b164d3c04"
+        eventing.knative.dev/release: "v20210625-c84fe22c2"
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
       affinity:
@@ -700,7 +703,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:9ff12d7bc396bcbf6485afeea36b97cd00009a58356cf4afffbf5f3c2a60e7c9
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:83e211d23e667c9f397ab95e0300b7779261b7c4ce68c409145c867babdd7b13
           resources:
             requests:
               cpu: 100m
@@ -718,7 +721,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:b6f5a6ff3b0de2a0e81c46fcf7bcb9ee9ac013a2c313ebe5416d040258652774
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:20e32d0a81442e85607c41520b6c5a75e154267c4eaa544a6297ad9558b82146
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -762,7 +765,7 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
   replicas: 0
@@ -774,7 +777,7 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20210618-b164d3c04"
+        eventing.knative.dev/release: "v20210625-c84fe22c2"
     spec:
       affinity:
         podAntiAffinity:
@@ -787,7 +790,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:67dec0dfc4f59331503b97b00097e157909778fe9bac5a01cb497c4d41bc46f0
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:17a6f9eea2e3181301017fffdba6a3a322083cb4ce78861d473a971db5373e64
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -845,7 +848,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 spec:
   scaleTargetRef:
     apiVersion: apps/v1
@@ -868,7 +871,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 spec:
   minAvailable: 80%
   selector:
@@ -896,7 +899,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 spec:
   selector:
     matchLabels: &labels
@@ -923,7 +926,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:6112f611a8c46541baa139619ab7d04a4ed0f86a65df36aa5fb752218ab32b4b
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:637508f70218ac33431e9090b2eedd270f2e56c92e50f62ad8655ddb7a6080be
           resources:
             requests:
               # taken from serving.
@@ -987,7 +990,7 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     role: eventing-webhook
   name: eventing-webhook
   namespace: knative-eventing
@@ -1019,7 +1022,7 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1233,7 +1236,7 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1306,6 +1309,7 @@ spec:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
                       format: int32
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
             status:
               description: Status represents the current state of the Broker. This data may be out of date.
               type: object
@@ -1394,7 +1398,7 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -1479,6 +1483,7 @@ spec:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
                       format: int32
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                 subscribers:
                   description: This is the list of subscriptions for this subscribable.
                   type: array
@@ -1522,6 +1527,7 @@ spec:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                        x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature
                       generation:
                         description: Generation of the origin of the subscriber with uid:UID.
                         type: integer
@@ -1662,7 +1668,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1814,7 +1820,7 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -1941,7 +1947,7 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -2005,6 +2011,7 @@ spec:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                        x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                       filter:
                         description: Filter is the expression guarding the branch
                         type: object
@@ -2193,7 +2200,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -2382,7 +2389,7 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -2481,6 +2488,7 @@ spec:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                        x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                       ref:
                         description: Ref points to an Addressable.
                         type: object
@@ -2700,7 +2708,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
@@ -2890,7 +2898,7 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
 spec:
   group: messaging.knative.dev
@@ -2921,6 +2929,7 @@ spec:
                     name:
                       description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
                       type: string
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature
                 delivery:
                   description: Delivery configuration
                   type: object
@@ -2958,6 +2967,7 @@ spec:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
                       format: int32
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                 reply:
                   description: Reply specifies (optionally) how to handle events returned from the Subscriber target.
                   type: object
@@ -3098,7 +3108,7 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -3277,7 +3287,7 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3289,7 +3299,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3307,7 +3317,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3328,7 +3338,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3353,7 +3363,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3372,7 +3382,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: messaging-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3393,7 +3403,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3429,7 +3439,7 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   - apiGroups:
       - ""
@@ -3454,7 +3464,7 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   - apiGroups:
       - ""
@@ -3470,7 +3480,7 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   - apiGroups:
       - ""
@@ -3502,7 +3512,7 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3514,7 +3524,7 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/channelable: "true"
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3551,7 +3561,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3563,7 +3573,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3575,7 +3585,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3587,7 +3597,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3599,7 +3609,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3612,7 +3622,7 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
     resources: ["*"]
@@ -3624,7 +3634,7 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
     resources: ["*"]
@@ -3650,7 +3660,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   - apiGroups:
       - ""
@@ -3784,7 +3794,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   - apiGroups:
       - ""
@@ -3850,7 +3860,7 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3862,7 +3872,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/podspecable: "true"
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -3908,7 +3918,7 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3920,7 +3930,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     duck.knative.dev/source: "true"
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -3956,7 +3966,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   - apiGroups:
       - ""
@@ -4041,7 +4051,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   # For watching logging configuration and getting certs.
   - apiGroups:
@@ -4139,7 +4149,7 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 rules:
   # For manipulating certs into secrets.
   - apiGroups:
@@ -4174,7 +4184,7 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4210,7 +4220,7 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4242,7 +4252,7 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4275,7 +4285,7 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 # The data is populated at install time.
 
 ---
@@ -4298,7 +4308,7 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
diff --git a/third_party/eventing-latest/eventing-crds.yaml b/third_party/eventing-latest/eventing-crds.yaml
index 376663875..704bde201 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: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     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: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -304,6 +304,7 @@ spec:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
                       format: int32
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
             status:
               description: Status represents the current state of the Broker. This data may be out of date.
               type: object
@@ -392,7 +393,7 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -477,6 +478,7 @@ spec:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
                       format: int32
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                 subscribers:
                   description: This is the list of subscriptions for this subscribable.
                   type: array
@@ -520,6 +522,7 @@ spec:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                        x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature
                       generation:
                         description: Generation of the origin of the subscriber with uid:UID.
                         type: integer
@@ -660,7 +663,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -812,7 +815,7 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -939,7 +942,7 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1003,6 +1006,7 @@ spec:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                        x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                       filter:
                         description: Filter is the expression guarding the branch
                         type: object
@@ -1191,7 +1195,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1380,7 +1384,7 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1479,6 +1483,7 @@ spec:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                        x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                       ref:
                         description: Ref points to an Addressable.
                         type: object
@@ -1698,7 +1703,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
@@ -1888,7 +1893,7 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
 spec:
   group: messaging.knative.dev
@@ -1919,6 +1924,7 @@ spec:
                     name:
                       description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
                       type: string
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature
                 delivery:
                   description: Delivery configuration
                   type: object
@@ -1956,6 +1962,7 @@ spec:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
                       format: int32
+                  x-kubernetes-preserve-unknown-fields: true # This is necessary to enable the experimental feature delivery-timeout
                 reply:
                   description: Reply specifies (optionally) how to handle events returned from the Subscriber target.
                   type: object
@@ -2096,7 +2103,7 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210618-b164d3c04"
+    eventing.knative.dev/release: "v20210625-c84fe22c2"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
index 0cc462314..4050e0d37 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/prober.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"strconv"
 	"time"
 
@@ -141,7 +142,7 @@ func (p *EventProber) SenderDone(prefix string) feature.StepFn {
 		interval, timeout := environment.PollTimingsFromContext(ctx)
 		err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 			events := p.SentBy(ctx, prefix)
-			fmt.Println(prefix, "has sent", len(events))
+			log.Println(p.shortNameToName[prefix], "has sent", len(events))
 			if len(events) == len(p.ids) {
 				return true, nil
 			}
@@ -159,13 +160,13 @@ func (p *EventProber) ReceiverDone(from, to string) feature.StepFn {
 		interval, timeout := environment.PollTimingsFromContext(ctx)
 		err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 			sent := p.SentBy(ctx, from)
-			fmt.Println(from, "has sent", len(sent))
+			log.Println(p.shortNameToName[from], "has sent", len(sent))
 
 			received := p.ReceivedBy(ctx, to)
-			fmt.Println(to, "has received", len(received))
+			log.Println(p.shortNameToName[to], "has received", len(received))
 
 			rejected := p.RejectedBy(ctx, to)
-			fmt.Println(to, "has rejected", len(rejected))
+			log.Println(p.shortNameToName[to], "has rejected", len(rejected))
 
 			if len(sent) == len(received)+len(rejected) {
 				return true, nil
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/recorder_vent/recorder.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/recorder_vent/recorder.go
index 64d85c97a..4d516260c 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/recorder_vent/recorder.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/recorder_vent/recorder.go
@@ -42,6 +42,14 @@ type recorder struct {
 }
 
 func (r *recorder) Vent(observed eventshub.EventInfo) error {
+	if observed.Event != nil {
+		var event corev1.Event
+		err := json.Unmarshal(observed.Event.DataEncoded, &event)
+		if err == nil && event.Reason == eventshub.CloudEventObservedReason {
+			// prevent event loops from happening with APIServerSources
+			return nil
+		}
+	}
 	b, err := json.Marshal(observed)
 	if err != nil {
 		return err
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/resources.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/resources.go
index 6530227fb..0dfcbcda1 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/resources.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/resources.go
@@ -57,6 +57,14 @@ func Install(name string, options ...EventsHubOption) feature.StepFn {
 		envs[ConfigLoggingEnv] = knative.LoggingConfigFromContext(ctx)
 		envs[ConfigTracingEnv] = knative.TracingConfigFromContext(ctx)
 
+		// Register the event info store to assert later the events published by the eventshub
+		registerEventsHubStore(
+			k8s.EventListenerFromContext(ctx),
+			t,
+			name,
+			environment.FromContext(ctx).Namespace(),
+		)
+
 		// Deploy
 		if _, err := manifest.InstallYamlFS(ctx, templates, map[string]interface{}{
 			"name": name,
@@ -71,13 +79,5 @@ func Install(name string, options ...EventsHubOption) feature.StepFn {
 		if strings.Contains(envs["EVENT_GENERATORS"], "receiver") {
 			k8s.WaitForServiceEndpointsOrFail(ctx, t, name, 1)
 		}
-
-		// Register the event info store to assert later the events published by the eventshub
-		registerEventsHubStore(
-			k8s.EventListenerFromContext(ctx),
-			t,
-			name,
-			environment.FromContext(ctx).Namespace(),
-		)
 	}
 }