Repository: knative-sandbox/eventing-kafka-broker
Pull request #1447 (2021-11-09 12:57:45)

Initial Kafka channel reconciler tests
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Pushing for early feedback
- Check individual commits
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

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

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

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

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

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1447/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1458056219544522752/artifacts/line-cov.html#file1) | 0.0% | 39.8% | 39.8

knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-go-coverage | eaa832ab3c80c318af253c1299196e0a3e65f9d9 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1447/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1458056219544522752) | false | `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage`

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

<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/1447?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1447](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (eaa832a) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/f388998ae4cfcf2382846bd178439a1fe4471af6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (f388998) will **increase** coverage by `10.76%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/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/1447?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    #1447       +/-   ##
=============================================
+ Coverage     71.51%   82.28%   +10.76%     
+ Complexity      585      584        -1     
=============================================
  Files           100       65       -35     
  Lines          3964     1925     -2039     
  Branches        166      166               
=============================================
- Hits           2835     1584     -1251     
+ Misses          912      253      -659     
+ Partials        217       88      -129     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.28% <ø> (-0.21%)` | :arrow_down: |

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/1447?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...e/eventing/kafka/broker/core/file/FileWatcher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsZS9GaWxlV2F0Y2hlci5qYXZh) | `75.64% <0.00%> (-5.13%)` | :arrow_down: |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | | |
| [control-plane/pkg/reconciler/channel/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NvbnRyb2xsZXIuZ28=) | | |
| [control-plane/pkg/security/secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0Lmdv) | | |
| [control-plane/pkg/reconciler/source/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvY29udHJvbGxlci5nbw==) | | |
| [control-plane/pkg/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | | |
| [control-plane/pkg/core/config/utils.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvdXRpbHMuZ28=) | | |
| [...ol-plane/pkg/security/secrets\_provider\_net\_spec.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0c19wcm92aWRlcl9uZXRfc3BlYy5nbw==) | | |
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | |
| [...kg/apis/eventing/v1alpha1/kafka\_sink\_conversion.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2NvbnZlcnNpb24uZ28=) | | |
| ... and [26 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447/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/1447?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/1447?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [f388998...eaa832a](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1447?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/control-plane/pkg/prober/cache.go b/control-plane/pkg/prober/cache.go
new file mode 100644
index 000000000..e8e28d312
--- /dev/null
+++ b/control-plane/pkg/prober/cache.go
@@ -0,0 +1,134 @@
+/*
+ * 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 prober
+
+import (
+	"container/list"
+	"context"
+	"sync"
+	"time"
+)
+
+const (
+	// StatusReady signals that a given object is ready.
+	StatusReady Status = iota
+	// StatusUnknown signals that a given object is not ready and its state is unknown.
+	StatusUnknown
+	// StatusNotReady signals that a given object is not ready.
+	StatusNotReady
+)
+
+// Status represents the resource status.
+type Status int
+
+// Cache is a key-status store.
+type Cache interface {
+	// GetStatus retries the status associated with the given key.
+	GetStatus(key string) Status
+	// UpsertStatus add or updates the status associated with the given key.
+	// Once the given key expires the onExpired callback will be called passing the arg parameter.
+	UpsertStatus(key string, status Status, arg interface{}, onExpired ExpirationFunc)
+}
+
+// ExpirationFunc is a callback called once an entry in the cache is expired.
+type ExpirationFunc func(key string, arg interface{})
+
+type localExpiringCache struct {
+	// mu protects targets and entries
+	mu sync.RWMutex
+	// targets is a map of key-pointer to an element in the entries list.
+	targets map[string]*list.Element
+	// entries is a doubly-linked list of all entries present in the cache.
+	// This allows fast deletion of expired entries.
+	entries *list.List
+
+	expiration time.Duration
+}
+
+type value struct {
+	status     Status
+	lastUpsert time.Time
+	key        string
+	arg        interface{}
+	onExpired  ExpirationFunc
+}
+
+func NewLocalExpiringCache(ctx context.Context, expiration time.Duration) Cache {
+	c := &localExpiringCache{
+		mu:         sync.RWMutex{},
+		targets:    make(map[string]*list.Element, 64),
+		entries:    list.New().Init(),
+		expiration: expiration,
+	}
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(expiration):
+				c.removeExpiredEntries(time.Now())
+			}
+		}
+	}()
+	return c
+}
+
+func (c *localExpiringCache) GetStatus(key string) Status {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if v, ok := c.targets[key]; ok {
+		value := v.Value.(*value)
+		if !c.isExpired(value, time.Now()) {
+			return value.status
+		}
+	}
+	return StatusUnknown
+}
+
+func (c *localExpiringCache) UpsertStatus(key string, status Status, arg interface{}, onExpired ExpirationFunc) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if v, ok := c.targets[key]; ok {
+		delete(c.targets, key)
+		c.entries.Remove(v)
+	}
+
+	value := &value{status: status, lastUpsert: time.Now(), key: key, arg: arg, onExpired: onExpired}
+	element := c.entries.PushBack(value)
+	c.targets[key] = element
+}
+
+func (c *localExpiringCache) removeExpiredEntries(now time.Time) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	for curr := c.entries.Front(); curr != nil && c.isExpired(curr.Value.(*value), now); {
+		v := curr.Value.(*value)
+		delete(c.targets, v.key)
+		prev := curr
+		curr = curr.Next()
+		c.entries.Remove(prev)
+
+		v.onExpired(v.key, v.arg)
+	}
+}
+
+func (c *localExpiringCache) isExpired(v *value, now time.Time) bool {
+	return v.lastUpsert.Add(c.expiration).Before(now)
+}
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 589c3d838..3d9350728 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -23,31 +23,35 @@ import (
 	"strings"
 
 	"github.com/Shopify/sarama"
+
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
+
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/retry"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
-	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
-	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
-	"knative.dev/eventing-kafka/pkg/common/constants"
-	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
-	commonsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
-	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/system"
 
+	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+	commonsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 )
 
 const (
@@ -69,6 +73,11 @@ type Reconciler struct {
 	// mock the function used during the reconciliation loop.
 	NewKafkaClient kafka.NewClientFunc
 
+	// InitOffsetsFunc initialize offsets for a provided set of topics and a provided consumer group id.
+	// It's convenient to add this as Reconciler field so that we can mock the function used during the
+	// reconciliation loop.
+	InitOffsetsFunc kafka.InitOffsetsFunc
+
 	ConfigMapLister corelisters.ConfigMapLister
 }
 
@@ -439,7 +448,7 @@ func (r *Reconciler) reconcileInitialOffset(ctx context.Context, channel *messag
 
 	topicName := kafka.ChannelTopic(TopicPrefix, channel)
 	groupID := consumerGroup(channel, sub)
-	_, err := offset.InitOffsets(ctx, kafkaClient, kafkaClusterAdmin, []string{topicName}, groupID)
+	_, err := r.InitOffsetsFunc(ctx, kafkaClient, kafkaClusterAdmin, []string{topicName}, groupID)
 	return err
 }
 
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index c86cc490a..40e8e19c2 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -23,6 +23,7 @@ import (
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/cache"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	kafkachannelinformer "knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel"
@@ -42,6 +43,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
+// TODO: no need for a configmap watcher?
 func NewController(ctx context.Context, watcher configmap.Watcher, configs *config.Env) *controller.Impl {
 
 	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
@@ -62,6 +64,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 		},
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		InitOffsetsFunc:            offset.InitOffsets,
 		Env:                        configs,
 		ConfigMapLister:            configmapInformer.Lister(),
 	}
diff --git a/data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerBenchmark.java b/data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerBenchmark.java
index 926d470ba..e2f8b56a3 100644
--- a/data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerBenchmark.java
+++ b/data-plane/benchmarks/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerBenchmark.java
@@ -19,6 +19,7 @@ import io.cloudevents.CloudEvent;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
 import io.vertx.kafka.client.common.PartitionInfo;
 import io.vertx.kafka.client.common.TopicPartition;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
@@ -27,11 +28,6 @@ import io.vertx.kafka.client.consumer.KafkaReadStream;
 import io.vertx.kafka.client.consumer.OffsetAndMetadata;
 import io.vertx.kafka.client.consumer.OffsetAndTimestamp;
 import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
-import java.time.Duration;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.regex.Pattern;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.openjdk.jmh.annotations.Benchmark;
@@ -41,6 +37,12 @@ import org.openjdk.jmh.annotations.Setup;
 import org.openjdk.jmh.annotations.State;
 import org.openjdk.jmh.infra.Blackhole;
 
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
 public class UnorderedOffsetManagerBenchmark {
 
   @State(Scope.Thread)
@@ -71,96 +73,70 @@ public class UnorderedOffsetManagerBenchmark {
 
   @Benchmark
   public void benchmarkReverseOrder(RecordsState recordsState, Blackhole blackhole) {
-    UnorderedOffsetManager offsetManager = new UnorderedOffsetManager(new MockKafkaConsumer(), null);
+    final OffsetManager offsetManager = new OffsetManager(Vertx.vertx(), new MockKafkaConsumer(), null, 10000L);
 
     int partitions = 100;
     for (int partition = 0; partition < partitions; partition++) {
-      blackhole.consume(
-        offsetManager.recordReceived(recordsState.records[partition][0])
-      );
+      offsetManager.recordReceived(recordsState.records[partition][0]);
     }
 
     for (int offset = 9_999; offset > 0; offset--) {
       for (int partition = 0; partition < partitions; partition++) {
-        blackhole.consume(
-          offsetManager.recordReceived(recordsState.records[partition][offset])
-        );
-        blackhole.consume(
-          offsetManager.successfullySentToSubscriber(recordsState.records[partition][offset])
-        );
+        offsetManager.recordReceived(recordsState.records[partition][offset]);
+        offsetManager.successfullySentToSubscriber(recordsState.records[partition][offset]);
       }
     }
 
     for (int partition = 0; partition < partitions; partition++) {
-      blackhole.consume(
-        offsetManager.successfullySentToSubscriber(recordsState.records[partition][0])
-      );
+      offsetManager.successfullySentToSubscriber(recordsState.records[partition][0]);
     }
   }
 
   @Benchmark
   public void benchmarkOrdered(RecordsState recordsState, Blackhole blackhole) {
-    UnorderedOffsetManager offsetManager = new UnorderedOffsetManager(new MockKafkaConsumer(), null);
+    OffsetManager offsetManager = new OffsetManager(Vertx.vertx(), new MockKafkaConsumer(), null, 10000L);
     int partitions = 100;
 
     for (int offset = 0; offset < 10_000; offset++) {
       for (int partition = 0; partition < partitions; partition++) {
-        blackhole.consume(
-          offsetManager.recordReceived(recordsState.records[partition][offset])
-        );
-        blackhole.consume(
-          offsetManager.successfullySentToSubscriber(recordsState.records[partition][offset])
-        );
+        offsetManager.recordReceived(recordsState.records[partition][offset]);
+        offsetManager.successfullySentToSubscriber(recordsState.records[partition][offset]);
       }
     }
   }
 
   @Benchmark
   public void benchmarkRealisticCase(RecordsState recordsState, Blackhole blackhole) {
-    UnorderedOffsetManager offsetManager = new UnorderedOffsetManager(new MockKafkaConsumer(), null);
+    OffsetManager offsetManager = new OffsetManager(Vertx.vertx(), new MockKafkaConsumer(), null, 10000L);
     int partitions = 10;
 
     for (int partition = 0; partition < partitions; partition++) {
-      blackhole.consume(
-        offsetManager.recordReceived(recordsState.records[partition][0])
-      );
+      offsetManager.recordReceived(recordsState.records[partition][0]);
     }
 
     for (int partition = 0; partition < partitions; partition++) {
-      for (int offset : new int[] {5, 2, 0, 7, 1, 3, 4, 6}) {
-        blackhole.consume(
-          offsetManager.successfullySentToSubscriber(recordsState.records[partition][offset])
-        );
+      for (int offset : new int[]{5, 2, 0, 7, 1, 3, 4, 6}) {
+        offsetManager.successfullySentToSubscriber(recordsState.records[partition][offset]);
       }
     }
   }
 
   @Benchmark
   public void benchmarkMixedABit(RecordsState recordsState, Blackhole blackhole) {
-    UnorderedOffsetManager offsetManager = new UnorderedOffsetManager(new MockKafkaConsumer(), null);
+    OffsetManager offsetManager = new OffsetManager(Vertx.vertx(), new MockKafkaConsumer(), null, 10000L);
     int partitions = 4;
 
     for (int partition = 0; partition < partitions; partition++) {
-      blackhole.consume(
-        offsetManager.recordReceived(recordsState.records[partition][0])
-      );
+      offsetManager.recordReceived(recordsState.records[partition][0]);
     }
 
     for (int i = 0; i < 120; i++) {
       // This will commit in the following order:
       // 1 0 3 2 5 4 ...
-      blackhole.consume(
-        offsetManager.successfullySentToSubscriber(recordsState.records[2][i % 2 == 0 ? i + 1 : i - 1])
-      );
-      blackhole.consume(
-        offsetManager.successfullySentToSubscriber(recordsState.records[1][i % 2 == 0 ? i + 1 : i - 1])
-      );
-      blackhole.consume(
-        offsetManager.successfullySentToSubscriber(recordsState.records[0][i % 2 == 0 ? i + 1 : i - 1])
-      );
-      blackhole.consume(
-        offsetManager.successfullySentToSubscriber(recordsState.records[3][i % 2 == 0 ? i + 1 : i - 1])
-      );
+      offsetManager.successfullySentToSubscriber(recordsState.records[2][i % 2 == 0 ? i + 1 : i - 1]);
+      offsetManager.successfullySentToSubscriber(recordsState.records[1][i % 2 == 0 ? i + 1 : i - 1]);
+      offsetManager.successfullySentToSubscriber(recordsState.records[0][i % 2 == 0 ? i + 1 : i - 1]);
+      offsetManager.successfullySentToSubscriber(recordsState.records[3][i % 2 == 0 ? i + 1 : i - 1]);
     }
   }
 
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
index f047e6ec7..ec5d4745b 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/RecordDispatcherListener.java
@@ -28,7 +28,7 @@ public interface RecordDispatcherListener {
    *
    * @param record record received.
    */
-  Future<Void> recordReceived(KafkaConsumerRecord<?, ?> record);
+  void recordReceived(KafkaConsumerRecord<?, ?> record);
 
   /**
    * The given record cannot be delivered to dead letter sink.
@@ -36,26 +36,26 @@ public interface RecordDispatcherListener {
    * @param record record undeliverable to dead letter sink.
    * @param ex     exception occurred.
    */
-  Future<Void> failedToSendToDeadLetterSink(KafkaConsumerRecord<?, ?> record, Throwable ex);
+  void failedToSendToDeadLetterSink(KafkaConsumerRecord<?, ?> record, Throwable ex);
 
   /**
    * The given event doesn't pass the filter.
    *
    * @param record record discarded.
    */
-  Future<Void> recordDiscarded(KafkaConsumerRecord<?, ?> record);
+  void recordDiscarded(KafkaConsumerRecord<?, ?> record);
 
   /**
    * The given record has been successfully sent to subscriber.
    *
    * @param record record sent to subscriber.
    */
-  Future<Void> successfullySentToSubscriber(KafkaConsumerRecord<?, ?> record);
+  void successfullySentToSubscriber(KafkaConsumerRecord<?, ?> record);
 
   /**
    * The given record has been successfully sent to dead letter sink.
    *
    * @param record record sent to dead letter sink.
    */
-  Future<Void> successfullySentToDeadLetterSink(KafkaConsumerRecord<?, ?> record);
+  void successfullySentToDeadLetterSink(KafkaConsumerRecord<?, ?> record);
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index 0896d038f..7d059f461 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -28,8 +28,10 @@ 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;
 
@@ -139,16 +141,13 @@ public class RecordDispatcherImpl implements RecordDispatcher {
       }
     }
 
-    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
+    recordDispatcherListener.recordReceived(record);
+    // Execute filtering
+    if (filter.test(record.value())) {
+      onFilterMatching(record, finalProm);
+    } else {
+      onFilterNotMatching(record, finalProm);
+    }
   }
 
   private void onFilterMatching(final KafkaConsumerRecord<Object, CloudEvent> record, final Promise<Void> finalProm) {
@@ -161,15 +160,15 @@ public class RecordDispatcherImpl implements RecordDispatcher {
   private void onFilterNotMatching(final KafkaConsumerRecord<Object, CloudEvent> record,
                                    final Promise<Void> finalProm) {
     logDebug("Record doesn't match filtering", record);
-    recordDispatcherListener.recordDiscarded(record)
-      .onComplete(finalProm);
+    recordDispatcherListener.recordDiscarded(record);
+    finalProm.complete();
   }
 
   private void onSubscriberSuccess(final KafkaConsumerRecord<Object, CloudEvent> record,
                                    final Promise<Void> finalProm) {
     logDebug("Successfully sent event to subscriber", record);
-    recordDispatcherListener.successfullySentToSubscriber(record)
-      .onComplete(finalProm);
+    recordDispatcherListener.successfullySentToSubscriber(record);
+    finalProm.complete();
   }
 
   private void onSubscriberFailure(final KafkaConsumerRecord<Object, CloudEvent> record,
@@ -182,15 +181,15 @@ public class RecordDispatcherImpl implements RecordDispatcher {
   private void onDeadLetterSinkSuccess(final KafkaConsumerRecord<Object, CloudEvent> record,
                                        final Promise<Void> finalProm) {
     logDebug("Successfully sent event to the dead letter sink", record);
-    recordDispatcherListener.successfullySentToDeadLetterSink(record)
-      .onComplete(finalProm);
+    recordDispatcherListener.successfullySentToDeadLetterSink(record);
+    finalProm.complete();
   }
 
 
   private void onDeadLetterSinkFailure(final KafkaConsumerRecord<Object, CloudEvent> record, final Throwable exception,
                                        final Promise<Void> finalProm) {
-    recordDispatcherListener.failedToSendToDeadLetterSink(record, exception)
-      .onComplete(finalProm);
+    recordDispatcherListener.failedToSendToDeadLetterSink(record, exception);
+    finalProm.complete();
   }
 
   private static Function<KafkaConsumerRecord<Object, CloudEvent>, Future<Void>> composeSenderAndSinkHandler(
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
new file mode 100644
index 000000000..38a019319
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
@@ -0,0 +1,211 @@
+/*
+ * 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.impl.consumer;
+
+import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
+import io.vertx.core.Vertx;
+import io.vertx.kafka.client.common.TopicPartition;
+import io.vertx.kafka.client.consumer.KafkaConsumer;
+import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
+import io.vertx.kafka.client.consumer.OffsetAndMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Consumer;
+
+/**
+ * This class implements the offset strategy that makes sure that, even unordered, the offset commit is ordered.
+ */
+public final class OffsetManager implements RecordDispatcherListener {
+
+  private static final Logger logger = LoggerFactory.getLogger(OffsetManager.class);
+
+  private final KafkaConsumer<?, ?> consumer;
+
+  private final Map<TopicPartition, OffsetTracker> offsetTrackers;
+
+  private final Consumer<Integer> onCommit;
+
+  /**
+   * All args constructor.
+   *
+   * @param consumer Kafka consumer.
+   * @param onCommit Callback invoked when an offset is actually committed
+   */
+  public OffsetManager(final Vertx vertx,
+                       final KafkaConsumer<?, ?> consumer,
+                       final Consumer<Integer> onCommit,
+                       final long commitIntervalMs) {
+    Objects.requireNonNull(consumer, "provide consumer");
+
+    this.consumer = consumer;
+    this.offsetTrackers = new HashMap<>();
+    this.onCommit = onCommit;
+
+    vertx.setPeriodic(commitIntervalMs, l -> this.offsetTrackers.forEach(this::commit));
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @return
+   */
+  @Override
+  public void recordReceived(final KafkaConsumerRecord<?, ?> record) {
+    final var tp = new TopicPartition(record.topic(), record.partition());
+    if (!offsetTrackers.containsKey(tp)) {
+      // Initialize offset tracker for the given record's topic/partition.
+      offsetTrackers.put(tp, new OffsetTracker(record.offset()));
+    }
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void successfullySentToSubscriber(final KafkaConsumerRecord<?, ?> record) {
+    commit(record);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void successfullySentToDeadLetterSink(final KafkaConsumerRecord<?, ?> record) {
+    commit(record);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void failedToSendToDeadLetterSink(final KafkaConsumerRecord<?, ?> record, final Throwable ex) {
+    commit(record);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void recordDiscarded(final KafkaConsumerRecord<?, ?> record) {
+    commit(record);
+  }
+
+  private void commit(final KafkaConsumerRecord<?, ?> record) {
+    this.offsetTrackers
+      .get(new TopicPartition(record.topic(), record.partition()))
+      .recordNewOffset(record.offset());
+  }
+
+  private synchronized void commit(final TopicPartition topicPartition, final OffsetTracker tracker) {
+    long newOffset = tracker.offsetToCommit();
+    if (newOffset > tracker.getCommitted()) {
+      // Reset the state
+      tracker.setCommitted(newOffset);
+
+      logger.debug("Committing offset for {} offset {}", topicPartition, newOffset);
+
+      // Execute the actual commit
+      consumer.commit(Map.of(topicPartition, new OffsetAndMetadata(newOffset, "")))
+        .onSuccess(ignored -> {
+          if (onCommit != null) {
+            onCommit.accept((int) newOffset);
+          }
+        })
+        .onFailure(cause -> logger.error("failed to commit topic partition {} offset {}", topicPartition, newOffset, cause))
+        .mapEmpty();
+    }
+  }
+
+  /**
+   * This offset tracker keeps track of the committed records.
+   */
+  private static class OffsetTracker {
+
+    // In order to not use a huge amount of memory we cap the BitSet to a _dynamic_ max size governed by this
+    // threshold.
+    private static final int RESET_TRACKER_THRESHOLD = 1_000_000;
+
+    // We store `long` offsets in a `BitSet` that is capable of handling `int` elements.
+    // The BitSet sets a committed bit for an offset `offset` in this way:
+    // bitSetOffset = offset - initialOffset
+    private BitSet committedOffsets;
+
+    // InitialOffset represents the offset committed from where committedOffsets BitSet starts, which means that
+    // the state of committedOffsets[0] is equal to the state of partition[initialOffset].
+    private long initialOffset;
+
+    // CommittedOffsets is the actual offset committed to stable storage.
+    private long committed;
+
+    OffsetTracker(final long initialOffset) {
+      committedOffsets = new BitSet();
+      committed = Math.max(initialOffset, 0);
+      this.initialOffset = committed;
+    }
+
+    synchronized void recordNewOffset(final long offset) {
+      final var bitSetOffset = (int) (offset - initialOffset);
+      committedOffsets.set(bitSetOffset);
+      maybeReset(bitSetOffset);
+    }
+
+    synchronized long offsetToCommit() {
+      return initialOffset + committedOffsets.nextClearBit((int) (committed - initialOffset));
+    }
+
+    synchronized void setCommitted(final long committed) {
+      this.committed = committed;
+    }
+
+    synchronized long getCommitted() {
+      return committed;
+    }
+
+    private void maybeReset(final int offset) {
+      if (offset > RESET_TRACKER_THRESHOLD) {
+        reset();
+      }
+    }
+
+    private void reset() {
+      // To not grow the BitSet indefinitely we create a new BitSet that starts from the committed offset.
+      // Since the delivery might be unordered we should copy the state of the current committedOffset BitSet that goes
+      // from the committed offset to the end of the BitSet.
+
+      final var prevCommittedOffsetsArr = committedOffsets.toLongArray();
+      // Calculate the word index in the long array. Long size is 64.
+      final var relativeOffset = committed - initialOffset;
+      final var wordOfCommitted = (int) (relativeOffset / 64);
+
+      // Copy from wordOfCommitted to the end: [..., wordOfCommitted, ...]
+      final var newCommittedOffsetsArr = Arrays.copyOfRange(
+        prevCommittedOffsetsArr,
+        wordOfCommitted,
+        prevCommittedOffsetsArr.length
+      );
+
+      // Re-create committedOffset BitSet and reset initialOffset.
+      this.committedOffsets = BitSet.valueOf(newCommittedOffsetsArr);
+      this.initialOffset = committed;
+    }
+  }
+}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedOffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedOffsetManager.java
deleted file mode 100644
index 70cffe626..000000000
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedOffsetManager.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- * 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.impl.consumer;
-
-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;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import io.vertx.kafka.client.consumer.OffsetAndMetadata;
-import java.util.Map;
-import java.util.Objects;
-import java.util.function.Consumer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-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 RecordDispatcherListener {
-
-  private static final Logger logger = LoggerFactory
-    .getLogger(OrderedOffsetManager.class);
-
-  private final KafkaConsumer<?, ?> consumer;
-
-  private final Consumer<Integer> onCommit;
-
-  /**
-   * All args constructor.
-   *
-   * @param consumer Kafka consumer.
-   * @param onCommit Callback invoked when an offset is actually committed
-   */
-  public OrderedOffsetManager(final KafkaConsumer<?, ?> consumer, final Consumer<Integer> onCommit) {
-    Objects.requireNonNull(consumer, "provide consumer");
-
-    this.consumer = consumer;
-    this.onCommit = onCommit;
-  }
-
-  /**
-   * {@inheritDoc}
-   *
-   * @return
-   */
-  @Override
-  public Future<Void> recordReceived(final KafkaConsumerRecord<?, ?> record) {
-    return Future.succeededFuture();
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Future<Void> successfullySentToSubscriber(final KafkaConsumerRecord<?, ?> record) {
-    return commit(record);
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Future<Void> successfullySentToDeadLetterSink(final KafkaConsumerRecord<?, ?> record) {
-    return commit(record);
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Future<Void> failedToSendToDeadLetterSink(final KafkaConsumerRecord<?, ?> record, final Throwable ex) {
-    return Future.succeededFuture();
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Future<Void> recordDiscarded(final KafkaConsumerRecord<?, ?> record) {
-    return Future.succeededFuture();
-  }
-
-  private Future<Void> commit(final KafkaConsumerRecord<?, ?> record) {
-    // Execute the actual commit
-    return consumer.commit(Map.of(
-      new TopicPartition(record.topic(), record.partition()),
-      new OffsetAndMetadata(record.offset() + 1, ""))
-    )
-      .onSuccess(ignored -> {
-        if (onCommit != null) {
-          onCommit.accept(1);
-        }
-        logger.debug(
-          "committed {} {} {}",
-          keyValue("topic", record.topic()),
-          keyValue("partition", record.partition()),
-          keyValue("offset", record.offset() + 1)
-        );
-      })
-      .onFailure(cause ->
-        logger.error(
-          "failed to commit {} {} {}",
-          keyValue("topic", record.topic()),
-          keyValue("partition", record.partition()),
-          keyValue("offset", record.offset() + 1),
-          cause
-        )
-      ).mapEmpty();
-  }
-
-}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManager.java
deleted file mode 100644
index 7757815e8..000000000
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManager.java
+++ /dev/null
@@ -1,262 +0,0 @@
-/*
- * 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.impl.consumer;
-
-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;
-import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
-import io.vertx.kafka.client.consumer.OffsetAndMetadata;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Objects;
-import java.util.function.Consumer;
-import org.slf4j.Logger;
-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 RecordDispatcherListener {
-
-  private static final Logger logger = LoggerFactory
-    .getLogger(UnorderedOffsetManager.class);
-
-  private final KafkaConsumer<?, ?> consumer;
-
-  private final Map<TopicPartition, OffsetTracker> offsetTrackers;
-
-  private final Consumer<Integer> onCommit;
-
-  /**
-   * All args constructor.
-   *
-   * @param consumer Kafka consumer.
-   * @param onCommit Callback invoked when an offset is actually committed
-   */
-  public UnorderedOffsetManager(final KafkaConsumer<?, ?> consumer, final Consumer<Integer> onCommit) {
-    Objects.requireNonNull(consumer, "provide consumer");
-
-    this.consumer = consumer;
-    this.offsetTrackers = new HashMap<>();
-    this.onCommit = onCommit;
-  }
-
-  /**
-   * {@inheritDoc}
-   *
-   * @return
-   */
-  @Override
-  public Future<Void> recordReceived(final KafkaConsumerRecord<?, ?> record) {
-    // un-ordered processing doesn't require pause/resume lifecycle.
-
-    // Because recordReceived is guaranteed to be called in order,
-    // we use it to set the last seen acked offset.
-    // TODO If this assumption doesn't work, use this.consumer.committed(new TopicPartition(record.topic(), record.partition()))
-    this.offsetTrackers.computeIfAbsent(new TopicPartition(record.topic(), record.partition()),
-      v -> new OffsetTracker(record.offset() - 1));
-    return Future.succeededFuture();
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Future<Void> successfullySentToSubscriber(final KafkaConsumerRecord<?, ?> record) {
-    return commit(record);
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Future<Void> successfullySentToDeadLetterSink(final KafkaConsumerRecord<?, ?> record) {
-    return commit(record);
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Future<Void> failedToSendToDeadLetterSink(final KafkaConsumerRecord<?, ?> record, final Throwable ex) {
-    this.offsetTrackers.get(new TopicPartition(record.topic(), record.partition()))
-      .recordNewOffset(record.offset());
-    return Future.succeededFuture();
-  }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public Future<Void> recordDiscarded(final KafkaConsumerRecord<?, ?> record) {
-    this.offsetTrackers.get(new TopicPartition(record.topic(), record.partition()))
-      .recordNewOffset(record.offset());
-    return Future.succeededFuture();
-  }
-
-  private Future<Void> commit(final KafkaConsumerRecord<?, ?> record) {
-    TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
-    OffsetTracker tracker = this.offsetTrackers.get(topicPartition);
-    tracker.recordNewOffset(record.offset());
-
-    if (tracker.shouldCommit()) {
-      // Reset the state
-      long newOffset = tracker.offsetToCommit();
-      long uncommittedSize = tracker.uncommittedSize();
-      tracker.reset(newOffset);
-
-      // Execute the actual commit
-      return consumer.commit(Map.of(
-        topicPartition,
-        new OffsetAndMetadata(newOffset, ""))
-      )
-        .onSuccess(ignored -> {
-          if (onCommit != null) {
-            onCommit.accept((int) uncommittedSize);
-          }
-          logger.debug(
-            "committed for topic partition {} {} offset {}",
-            record.topic(),
-            record.partition(),
-            newOffset
-          );
-        })
-        .onFailure(cause ->
-          logger.error(
-            "failed to commit for topic partition {} {} offset {}",
-            record.topic(),
-            record.partition(),
-            newOffset,
-            cause
-          )
-        ).mapEmpty();
-    }
-    return Future.succeededFuture();
-  }
-
-  /**
-   * This offset tracker keeps track of the committed records.
-   *
-   * <h2>Implementation details</h2>
-   *
-   * <p>
-   * For each record, the tracker flip a bit at an index representing the difference between the record's offset and the last acked record offset.
-   * To figure out if we need to commit or not, we just need to check if every bit of the long, up to the greatest uncommitted offset, is 1.
-   * In numerical representation this is equal to 2<sup>(greatestIndex - lastAcked)</sup> - 1. To avoid computing these numbers, we statically initialize
-   * all masks.
-   *
-   * <p>
-   * Because a long is only 64 bit, we use an array of longs (blocks) to represent the eventual uncommitted records.
-   * If blocks are more than one, in order to commit, every block except the last one must have all bits flipped to 1
-   * (in numerical representation -1 because of 2-complement representation of long) and the last block should follow the rule above explained.
-   * Note: in the best and more realistic case the store needs just 1 long, which means that only 64 records are sent unordered.
-   */
-  private static class OffsetTracker {
-
-    private final static int ACKS_GARBAGE_SIZE_THRESHOLD = 16; // Meaning 1024 messages are on hold
-    private final static long[] MASKS = new long[64];
-
-    static {
-      // Initialize MASKS
-      for (int i = 0; i < 64; i++) {
-        long mask = 0;
-        for (int j = 0; j <= i; j++) {
-          mask |= 1L << j;
-        }
-        MASKS[i] = mask;
-      }
-    }
-
-    private long lastAcked;
-    private long[] uncommitted;
-
-    private int greaterBlockIndex;
-    private int greaterBitIndexInGreaterBlock;
-
-    public OffsetTracker(long initialOffset) {
-      this.lastAcked = initialOffset;
-      this.uncommitted = new long[1];
-      this.greaterBlockIndex = -1;
-      this.greaterBitIndexInGreaterBlock = -1;
-    }
-
-    public void recordNewOffset(long offset) {
-      long diffWithLastCommittedOffset = offset - this.lastAcked - 1;
-      int blockIndex = blockIndex(diffWithLastCommittedOffset);
-      int bitIndex = (int) (diffWithLastCommittedOffset % 64); // That's obviously smaller than a long
-
-      checkAcksArraySize(blockIndex);
-
-      // Let's record this bit and update the greater indexes
-      this.uncommitted[blockIndex] |= 1L << bitIndex;
-      if (this.greaterBlockIndex < blockIndex) {
-        this.greaterBlockIndex = blockIndex;
-        this.greaterBitIndexInGreaterBlock = bitIndex;
-      } else if (this.greaterBlockIndex == blockIndex && this.greaterBitIndexInGreaterBlock < bitIndex) {
-        this.greaterBitIndexInGreaterBlock = bitIndex;
-      }
-    }
-
-    public boolean shouldCommit() {
-      // Let's check if we have all the bits to 1, except the last one
-      for (int b = 0; b < this.greaterBlockIndex; b++) {
-        if (this.uncommitted[b] != MASKS[63]) {
-          return false;
-        }
-      }
-
-      return this.uncommitted[this.greaterBlockIndex] == MASKS[this.greaterBitIndexInGreaterBlock];
-    }
-
-    public long uncommittedSize() {
-      return this.greaterBitIndexInGreaterBlock + 1 + (greaterBlockIndex * 64L);
-    }
-
-    public long offsetToCommit() {
-      return this.lastAcked + uncommittedSize() + 1;
-    }
-
-    public void reset(long committed) {
-      this.lastAcked = committed - 1;
-      this.greaterBlockIndex = -1;
-      this.greaterBitIndexInGreaterBlock = -1;
-
-      // Cleanup the acks array or overwrite it, depending on its size
-      if (this.uncommitted.length > ACKS_GARBAGE_SIZE_THRESHOLD) {
-        // No need to keep that big array
-        this.uncommitted = new long[1];
-      } else {
-        Arrays.fill(this.uncommitted, 0L);
-      }
-    }
-
-    private int blockIndex(long val) {
-      return (int) (val >> 6);
-    }
-
-    private void checkAcksArraySize(int blockIndex) {
-      if (blockIndex > this.uncommitted.length - 1) {
-        // Let's make sure we create enough room for more unordered records
-        this.uncommitted = Arrays.copyOf(this.uncommitted, (blockIndex + 1) * 2);
-      }
-    }
-
-  }
-
-}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
index ee17fcedd..af623c6f4 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
@@ -20,14 +20,11 @@ import dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressConfig
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
-import dev.knative.eventing.kafka.broker.core.security.Credentials;
 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.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.ConsumerVerticleFactory;
 import dev.knative.eventing.kafka.broker.dispatcher.DeliveryOrder;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
-import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.KafkaResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.NoopResponseHandler;
@@ -37,10 +34,9 @@ import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.BaseConsumerVe
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.CloudEventOverridesMutator;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.InvalidCloudEventInterceptor;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.KeyDeserializer;
+import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.OffsetManager;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.OrderedConsumerVerticle;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.OrderedOffsetManager;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.UnorderedConsumerVerticle;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.UnorderedOffsetManager;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.AttributesFilter;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.http.WebClientCloudEventSender;
 import io.cloudevents.CloudEvent;
@@ -49,7 +45,6 @@ import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.circuitbreaker.CircuitBreaker;
 import io.vertx.circuitbreaker.CircuitBreakerOptions;
 import io.vertx.core.AbstractVerticle;
-import io.vertx.core.Future;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.tracing.TracingPolicy;
@@ -72,7 +67,6 @@ import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
-import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -178,6 +172,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
           Filter.noop();
 
         final var responseHandler = getNoopResponseHandlerOrDefault(egress, () -> getKafkaResponseHandler(vertx, producerConfigs, resource));
+        final var commitIntervalMs = Integer.parseInt(String.valueOf(consumerConfigs.get(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG)));
 
         final var recordDispatcher = new RecordDispatcherMutatorChain(
           new RecordDispatcherImpl(
@@ -185,7 +180,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
             egressSubscriberSender,
             egressDeadLetterSender,
             responseHandler,
-            getOffsetManager(deliveryOrder, consumer, eventsSentCounter::increment),
+            new OffsetManager(vertx, consumer, eventsSentCounter::increment, commitIntervalMs),
             ConsumerTracer.create(
               ((VertxInternal) vertx).tracer(),
               new KafkaClientOptions()
@@ -305,14 +300,6 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     return !(egressConfig == null || egressConfig.getDeadLetter().isEmpty());
   }
 
-  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);
-    };
-  }
-
   private static AbstractVerticle getConsumerVerticle(final DeliveryOrder type,
                                                       final BaseConsumerVerticle.Initializer initializer,
                                                       final Set<String> topics) {
diff --git a/go.mod b/go.mod
index bda791563..36e640843 100644
--- a/go.mod
+++ b/go.mod
@@ -24,9 +24,9 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421
-	knative.dev/eventing-kafka v0.27.1-0.20211108161601-76680cab3d62
+	knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1
+	knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174
 	knative.dev/hack v0.0.0-20211108170701-96aac1c30be3
-	knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09
+	knative.dev/pkg v0.0.0-20211109100843-91d1932616a7
 	knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974
 )
diff --git a/go.sum b/go.sum
index 77a284b17..1f3079d5e 100644
--- a/go.sum
+++ b/go.sum
@@ -1326,26 +1326,22 @@ k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7/go.mod h1:wXW5VT87nVfh/iL
 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/control-protocol v0.0.0-20211101205041-256f50f9b5a9/go.mod h1:BOIcHyLM9l0pHtMEOcUJo0qRe3Qvrm4p3PXoKhEpeUA=
-knative.dev/eventing v0.27.1-0.20211108073202-ee96d0b3b429/go.mod h1:WsdNr/+tT/IqPeWzZuG3O9/zla/ZFniJC8PgJJWGT58=
-knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421 h1:RSA8U3xuGlX7S8ZF8IZKp6t7kbm4jWpBimpslvO0EI8=
+knative.dev/control-protocol v0.0.0-20211108155701-88c4c1e549c4/go.mod h1:5hOr3rdQDEBE/3LihWgyfkfowQn8V0Vur/qg+g42MXY=
 knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421/go.mod h1:lM+1rTzjcNmrPjTGaEk+QrmQGuM3/O+e3aN96GH4V0Y=
-knative.dev/eventing-kafka v0.27.1-0.20211108161601-76680cab3d62 h1:+6iEOvrEpUUi1IdH8YjynS9n8FYlFcRGiDE312WiMKs=
-knative.dev/eventing-kafka v0.27.1-0.20211108161601-76680cab3d62/go.mod h1:y1hVCjWPPTjTktZ3x+jTAqYq838QIAMuuipYe8YaDTg=
-knative.dev/hack v0.0.0-20211028194650-b96d65a5ff5e/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1 h1:vC/EpcYF1lsvaapxNGmz6vGNZfq94/1Fd770SQTZ9Es=
+knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1/go.mod h1:zHDfFKGqgq9bZvUd3QC6iHRAaj2AbHercYhlxdtc9fA=
+knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174 h1:mMZM9bJesUECCHg2kO5VvdOYK9cEIfAg8NNSNPgcPhY=
+knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174/go.mod h1:6RjAzpeBOW7/fG+2E4d0siS7Zfy0YfxJIC2Qal+3K6M=
 knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211102174540-b284d49386cc/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211105231158-29f86c2653b5/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211108170701-96aac1c30be3 h1:oSvRgnKoU308k7aXbPV3iL5Zh5kBGM2Ptar4hyeda+A=
 knative.dev/hack v0.0.0-20211108170701-96aac1c30be3/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20211102174540-b284d49386cc/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/hack/schema v0.0.0-20211105231158-29f86c2653b5/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20211028235650-5d9d300c2e40/go.mod h1:HyEqMTLzT2hGCisGaRMGzEE8b4Ym+7mhmVx5ygiX+fY=
+knative.dev/hack/schema v0.0.0-20211108170701-96aac1c30be3/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
-knative.dev/pkg v0.0.0-20211103135647-464034912f7e/go.mod h1:a+9AWOb0zOaBVU6xoPgTvEA7vfMnLjPnK4SX3ciSHKE=
-knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09 h1:EuziitWfjhtDEI7vyNV79AtqFia5rACAFOy2rbinVo4=
 knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09/go.mod h1:YiSfFDMdYHZTuO5XS8UzxSeOEmocy+Kb2YOGik9C8lw=
-knative.dev/reconciler-test v0.0.0-20211029073051-cff9b538d33c/go.mod h1:PIwtS0at/rtnsNCxcZUAZUmWKKasTUYJ/w1Mk94bOdA=
+knative.dev/pkg v0.0.0-20211109100843-91d1932616a7 h1:uqo0nLez3NJf03gE9mdYpFxbqZ8iWdVElQa+s8W7VNQ=
+knative.dev/pkg v0.0.0-20211109100843-91d1932616a7/go.mod h1:plYIuKk6WNC9LyiG/9xBnOBYuxO0Ro6xN0z7iXWVUtk=
 knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974 h1:4eo67cEiDvEl1TgITT+3n1ivAcLFCQ6DBJnX855727E=
 knative.dev/reconciler-test v0.0.0-20211101213739-c5f938aa9974/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 39e92aa45..f7d6da2d4 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1104,7 +1104,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.27.1-0.20211108221643-a72cfc91c421
+# knative.dev/eventing v0.27.1-0.20211109172546-31e4e2eb26d1
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1213,7 +1213,7 @@ 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/eventing-kafka v0.27.1-0.20211108161601-76680cab3d62
+# knative.dev/eventing-kafka v0.27.1-0.20211109165044-86d36b670174
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1270,7 +1270,7 @@ knative.dev/eventing-kafka/test/lib/resources
 # knative.dev/hack v0.0.0-20211108170701-96aac1c30be3
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20211108064904-3cc697a3cb09
+# knative.dev/pkg v0.0.0-20211109100843-91d1932616a7
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/control-plane/pkg/prober/cache_test.go b/control-plane/pkg/prober/cache_test.go
new file mode 100644
index 000000000..84a26b949
--- /dev/null
+++ b/control-plane/pkg/prober/cache_test.go
@@ -0,0 +1,93 @@
+/*
+ * 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 prober
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+func TestInMemoryLocalCache(t *testing.T) {
+	d := time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), d*2)
+	defer cancel()
+	c := NewLocalExpiringCache(ctx, d)
+	testCache(t, ctx, c, d)
+}
+
+func testCache(t *testing.T, ctx context.Context, c Cache, d time.Duration) {
+	var wg sync.WaitGroup
+	errors := make(chan error, 1)
+
+	wg.Add(2)
+
+	c.UpsertStatus("key1", StatusUnknown, 4, verifyNoExpired(errors))
+	require.Equal(t, StatusUnknown, c.GetStatus("key1"))
+
+	c.UpsertStatus("key2", StatusNotReady, 42, verifyNoExpired(errors))
+	require.Equal(t, StatusNotReady, c.GetStatus("key2"))
+
+	c.UpsertStatus("key1", StatusReady, 41, verifyOnExpired("key1", 41, &wg, errors))
+	require.Equal(t, StatusReady, c.GetStatus("key1"))
+
+	c.UpsertStatus("key2", StatusReady, 43, verifyOnExpired("key2", 43, &wg, errors))
+	require.Equal(t, StatusReady, c.GetStatus("key2"))
+
+	ctx, cancel := context.WithTimeout(ctx, d*2)
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		done <- struct{}{}
+	}()
+
+	select {
+	case <-ctx.Done():
+		t.Errorf("Timeout waiting for wait group to be done")
+	case err := <-errors:
+		t.Errorf(err.Error())
+	case <-done:
+		// Wait expiration
+		require.Nil(t, wait.PollImmediate(d, d*2, func() (done bool, err error) { return StatusUnknown == c.GetStatus("key1"), nil }))
+		require.Nil(t, wait.PollImmediate(d, d*2, func() (done bool, err error) { return StatusUnknown == c.GetStatus("key2"), nil }))
+	}
+}
+
+func verifyNoExpired(errors chan<- error) func(key string, arg interface{}) {
+	return func(key string, arg interface{}) {
+		errors <- fmt.Errorf("unexpected call to onExpired callback")
+	}
+}
+
+func verifyOnExpired(expectedKey string, expectedArg interface{}, wg *sync.WaitGroup, errors chan<- error) func(key string, arg interface{}) {
+	return func(key string, arg interface{}) {
+		if expectedKey != key {
+			errors <- fmt.Errorf("expected key to be %v got %v", expectedKey, key)
+		}
+		if expectedArg != arg {
+			errors <- fmt.Errorf("expected arg for key %v to be %v got %v", key, expectedArg, arg)
+		}
+		wg.Done()
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 0dbe6043b..4c47235cd 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -116,7 +116,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - no DLS",
 			Objects: []runtime.Object{
 				NewBroker(),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -345,7 +345,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
@@ -392,7 +392,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - config map not readable",
 			Objects: []runtime.Object{
 				NewBroker(),
-				NewConfigMap(&env, []byte(`{"hello": "world"}`)),
+				NewConfigMapWithBinaryData(&env, []byte(`{"hello": "world"}`)),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, nil),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
@@ -887,7 +887,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
@@ -959,7 +959,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				),
 				NewSSLSecret(ConfigMapNamespace, "secret-1"),
 				BrokerConfig(bootstrapServers, 20, 5, BrokerAuthConfig("secret-1")),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
@@ -1741,7 +1741,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			OtherTestData: map[string]interface{}{
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
new file mode 100644
index 000000000..f5ee7e165
--- /dev/null
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -0,0 +1,242 @@
+/*
+ * 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 channel_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/Shopify/sarama"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
+
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/system"
+	"knative.dev/pkg/tracker"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
+	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+
+	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	messagingv1beta1kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+)
+
+// TODO: are we using all of these?
+var configKafka = map[string]string{
+	"version": "1.0.0",
+	"sarama": `
+enableLogging: false
+config: |
+  Version: 2.0.0 # Kafka Version Compatibility From Sarama's Supported List (Major.Minor.Patch)
+  Admin:
+    Timeout: 10000000000  # 10 seconds
+  Net:
+    KeepAlive: 30000000000  # 30 seconds
+    MaxOpenRequests: 1 # Set to 1 for use with Idempotent Producer
+    TLS:
+      Enable: true
+    SASL:
+      Enable: true
+      Version: 1
+  Metadata:
+    RefreshFrequency: 300000000000  # 5 minutes
+`,
+	"eventing-kafka": `
+kafka:
+  authSecretName: kafka-cluster
+  authSecretNamespace: knative-eventing
+  brokers: kafka:9092
+`,
+}
+
+const (
+	finalizerName = "kafkachannels.messaging.knative.dev"
+)
+
+var finalizerUpdatedEvent = Eventf(
+	corev1.EventTypeNormal,
+	"FinalizerUpdate",
+	fmt.Sprintf(`Updated %q finalizers`, ChannelName),
+)
+
+var DefaultEnv = &config.Env{
+	DataPlaneConfigMapNamespace: "knative-eventing",
+	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
+	GeneralConfigMapName:        "kafka-broker-config",
+	IngressName:                 "kafka-channel-ingress",
+	SystemNamespace:             "knative-eventing",
+	DataPlaneConfigFormat:       base.Json,
+}
+
+// TODO: tests with and without subscriptions
+// TODO: tests for things from config-kafka
+// TODO: are we setting a InitialOffsetsCommitted status?
+// TODO: test if things from spec is used properly?
+
+func TestReconcileKind(t *testing.T) {
+
+	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
+
+	env := *DefaultEnv
+	testKey := fmt.Sprintf("%s/%s", ChannelNamespace, ChannelName)
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal - no subscription - no auth",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(ChannelNamespace, ChannelName),
+								},
+							},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(&env, nil),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						//StatusInitialOffsetsCommitted,
+						ChannelAddressable(&env),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+	}
+
+	useTable(t, table, env)
+}
+
+func useTable(t *testing.T, table TableTest, env config.Env) {
+	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		reconciler := &Reconciler{
+			Reconciler: &base.Reconciler{
+				KubeClient:                  kubeclient.Get(ctx),
+				PodLister:                   listers.GetPodLister(),
+				SecretLister:                listers.GetSecretLister(),
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
+				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				SystemNamespace:             env.SystemNamespace,
+				DispatcherLabel:             base.ChannelDispatcherLabel,
+				ReceiverLabel:               base.ChannelReceiverLabel,
+			},
+			Env:             env,
+			ConfigMapLister: listers.GetConfigMapLister(),
+			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
+				return 1, nil
+			},
+			NewKafkaClient: func(addrs []string, config *sarama.Config) (sarama.Client, error) {
+				return &kafkatesting.MockKafkaClient{}, nil
+			},
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopicName: ChannelTopic(),
+					ExpectedTopicDetail: sarama.TopicDetail{
+						NumPartitions:     1,
+						ReplicationFactor: 1,
+					},
+					T: t,
+				}, nil
+			},
+		}
+
+		reconciler.ConfigMapTracker = &FakeTracker{}
+		reconciler.SecretTracker = &FakeTracker{}
+
+		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))
+
+		r := messagingv1beta1kafkachannelreconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakeeventingkafkaclient.Get(ctx),
+			listers.GetKafkaChannelLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+		)
+		return r
+	}))
+}
+
+func patchFinalizers() clientgotesting.PatchActionImpl {
+	action := clientgotesting.PatchActionImpl{}
+	action.Name = ChannelName
+	action.Namespace = ChannelNamespace
+	patch := `{"metadata":{"finalizers":["` + finalizerName + `"],"resourceVersion":""}}`
+	action.Patch = []byte(patch)
+	return action
+}
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 5253258c1..f41000721 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -111,7 +111,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 				NewSink(
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 				),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
@@ -163,7 +163,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 					SinkAuthSecretRef("secret-1"),
 				),
 				NewSSLSecret(SinkNamespace, "secret-1"),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -228,7 +228,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.NumPartitions = nil
 					},
 				),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -294,7 +294,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.NumPartitions = nil
 					},
 				),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -344,7 +344,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						s.Spec.BootstrapServers = []string{"kafka-broker:10000"}
 					},
 				),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
@@ -457,7 +457,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&env, &contract.Contract{
@@ -500,7 +500,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 					StatusControllerOwnsTopic(reconciler.ControllerTopicOwner),
 					BootstrapServers(bootstrapServersArr),
 				),
-				NewConfigMap(&env, []byte(`{"hello": "world"}`)),
+				NewConfigMapWithBinaryData(&env, []byte(`{"hello": "world"}`)),
 				SinkReceiverPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 31fe70c7b..094312360 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -118,7 +118,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -182,7 +182,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -245,7 +245,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -306,7 +306,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -367,7 +367,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -428,7 +428,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 3c1ff5d4b..3d3a0ee67 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -28,8 +28,10 @@ import (
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 	"knative.dev/pkg/reconciler/testing"
 
+	eventingkafkachannels "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	eventingkafkasources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	fakeeventingkafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
+	eventingkafkachannelslisters "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
 	eventingkafkasourceslisters "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
 
 	eventingkafkabroker "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
@@ -121,6 +123,10 @@ func (l *Listers) GetKafkaSourceLister() eventingkafkasourceslisters.KafkaSource
 	return eventingkafkasourceslisters.NewKafkaSourceLister(l.indexerFor(&eventingkafkasources.KafkaSource{}))
 }
 
+func (l *Listers) GetKafkaChannelLister() eventingkafkachannelslisters.KafkaChannelLister {
+	return eventingkafkachannelslisters.NewKafkaChannelLister(l.indexerFor(&eventingkafkachannels.KafkaChannel{}))
+}
+
 func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
 	return l.sorter.IndexerForObjectType(obj)
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
new file mode 100644
index 000000000..1447f43dc
--- /dev/null
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -0,0 +1,149 @@
+/*
+ * 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 testing
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	clientgotesting "k8s.io/client-go/testing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/network"
+)
+
+const (
+	ChannelName             = "kc"
+	ChannelNamespace        = "test-nc"
+	ChannelUUID             = "c1234567-8901-2345-6789-123456789101"
+	ChannelBootstrapServers = "kafka:9092"
+)
+
+func ChannelTopic() string {
+	c := NewChannel().(metav1.Object)
+	return kafka.ChannelTopic(TopicPrefix, c)
+}
+
+func NewChannel(options ...KRShapedOption) runtime.Object {
+	c := &messagingv1beta1.KafkaChannel{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ChannelNamespace,
+			Name:      ChannelName,
+			UID:       ChannelUUID,
+		},
+	}
+	for _, opt := range options {
+		opt(c)
+	}
+	c.SetDefaults(context.Background())
+	return c
+}
+
+func ChannelReceiverPod(namespace string, annotations map[string]string) runtime.Object {
+	return &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "kafka-channel-receiver",
+			Namespace:   namespace,
+			Annotations: annotations,
+			Labels: map[string]string{
+				"app": base.ChannelReceiverLabel,
+			},
+		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
+	}
+}
+
+func ChannelDispatcherPod(namespace string, annotations map[string]string) runtime.Object {
+	return &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        "kafka-channel-dispatcher",
+			Namespace:   namespace,
+			Annotations: annotations,
+			Labels: map[string]string{
+				"app": base.ChannelDispatcherLabel,
+			},
+		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+		},
+	}
+}
+
+func ChannelDispatcherPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Pod",
+		},
+		namespace,
+		ChannelDispatcherPod(namespace, annotations),
+	)
+}
+
+func ChannelReceiverPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
+	return clientgotesting.NewUpdateAction(
+		schema.GroupVersionResource{
+			Group:    "*",
+			Version:  "v1",
+			Resource: "Pod",
+		},
+		namespace,
+		ChannelReceiverPod(namespace, annotations),
+	)
+}
+
+func ChannelAddressable(env *config.Env) func(obj duckv1.KRShaped) {
+
+	return func(obj duckv1.KRShaped) {
+		channel := obj.(*messagingv1beta1.KafkaChannel)
+
+		channel.Status.Address = &duckv1.Addressable{}
+
+		channel.Status.Address.URL = &apis.URL{
+			Scheme: "http",
+			Host:   network.GetServiceHostname(env.IngressName, env.SystemNamespace),
+			Path:   fmt.Sprintf("/%s/%s", channel.Namespace, channel.Name),
+		}
+
+		channel.GetConditionSet().Manage(&channel.Status).MarkTrue(base.ConditionAddressable)
+	}
+}
+
+func WithInitKafkaChannelConditions(obj duckv1.KRShaped) {
+	channel := obj.(*messagingv1beta1.KafkaChannel)
+	channel.Status.InitializeConditions()
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index c4626e39e..ce0f8af11 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -83,7 +83,7 @@ func ServiceURLFrom(ns, name string) string {
 	return fmt.Sprintf("http://%s.%s.svc.cluster.local", name, ns)
 }
 
-func NewConfigMap(env *config.Env, data []byte) runtime.Object {
+func NewConfigMapWithBinaryData(env *config.Env, data []byte) runtime.Object {
 	return reconcilertesting.NewConfigMap(
 		env.DataPlaneConfigMapName,
 		env.DataPlaneConfigMapNamespace,
@@ -99,6 +99,16 @@ func NewConfigMap(env *config.Env, data []byte) runtime.Object {
 	)
 }
 
+func NewConfigMapWithTextData(namespace, name string, data map[string]string) runtime.Object {
+	return reconcilertesting.NewConfigMap(
+		name,
+		namespace,
+		func(configMap *corev1.ConfigMap) {
+			configMap.Data = data
+		},
+	)
+}
+
 func NewConfigMapFromContract(contract *contract.Contract, env *config.Env) runtime.Object {
 	var data []byte
 	var err error
@@ -111,7 +121,7 @@ func NewConfigMapFromContract(contract *contract.Contract, env *config.Env) runt
 		panic(err)
 	}
 
-	return NewConfigMap(env, data)
+	return NewConfigMapWithBinaryData(env, data)
 }
 
 func ConfigMapUpdate(env *config.Env, contract *contract.Contract) clientgotesting.UpdateActionImpl {
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index abaacb823..45ad6dbdf 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -627,7 +627,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 			WantErr:                 true,
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
@@ -658,7 +658,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(BrokerReady),
 				newTrigger(),
 				NewService(),
-				NewConfigMap(&env, nil),
+				NewConfigMapWithBinaryData(&env, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
index 213ca0038..315f3aad5 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
@@ -29,11 +29,12 @@ import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
-import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -236,14 +237,6 @@ public class RecordDispatcherTest {
   }
 
   public static RecordDispatcherListener offsetManagerMock() {
-    final RecordDispatcherListener recordDispatcherListener = mock(RecordDispatcherListener.class);
-
-    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 recordDispatcherListener;
+    return mock(RecordDispatcherListener.class);
   }
 }
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractOffsetManagerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractOffsetManagerTest.java
index 1d9321cd8..971ca11b0 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractOffsetManagerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/AbstractOffsetManagerTest.java
@@ -19,10 +19,12 @@ import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.OffsetAndMetadata;
 import io.vertx.kafka.client.consumer.impl.KafkaConsumerRecordImpl;
+
 import java.util.Collection;
 import java.util.Map;
 import java.util.Set;
@@ -30,6 +32,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
+
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
@@ -45,7 +48,7 @@ import static org.mockito.Mockito.verify;
 
 public abstract class AbstractOffsetManagerTest {
 
-  abstract RecordDispatcherListener createOffsetManager(final KafkaConsumer<?, ?> consumer);
+  abstract RecordDispatcherListener createOffsetManager(final Vertx vertx, final KafkaConsumer<?, ?> consumer);
 
   protected static KafkaConsumerRecord<String, CloudEvent> record(String topic, int partition, long offset) {
     return new KafkaConsumerRecordImpl<>(
@@ -60,7 +63,8 @@ public abstract class AbstractOffsetManagerTest {
   }
 
   protected MapAssert<TopicPartition, Long> assertThatOffsetCommitted(
-    Collection<TopicPartition> partitionsConsumed, Consumer<RecordDispatcherListener> testExecutor) {
+    final Collection<TopicPartition> partitionsConsumed,
+    final Consumer<RecordDispatcherListener> testExecutor) {
     return assertThatOffsetCommittedWithFailures(partitionsConsumed,
       (offsetStrategy, flag) -> testExecutor.accept(offsetStrategy));
   }
@@ -96,10 +100,17 @@ public abstract class AbstractOffsetManagerTest {
       .when(vertxConsumer)
       .commit(any(Map.class));
 
-    testExecutor.accept(createOffsetManager(vertxConsumer), failureFlag);
+    testExecutor.accept(createOffsetManager(Vertx.vertx(), vertxConsumer), failureFlag);
+
+    try {
+      Thread.sleep(1000);
+    } catch (final InterruptedException e) {
+      throw new RuntimeException(e);
+    }
 
+    final var committed = mockConsumer.committed(Set.copyOf(partitionsConsumed));
     return assertThat(
-      mockConsumer.committed(Set.copyOf(partitionsConsumed))
+      committed
         .entrySet()
         .stream()
         .collect(Collectors.toMap(Map.Entry::getKey, v -> v.getValue().offset()))
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManagerTest.java
similarity index 79%
rename from data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerTest.java
rename to data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManagerTest.java
index ce1cfa815..470f7096e 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedOffsetManagerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManagerTest.java
@@ -18,10 +18,15 @@ package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
 import io.cloudevents.CloudEvent;
 import io.micrometer.core.instrument.Counter;
+import io.vertx.core.Vertx;
+import io.vertx.junit5.VertxExtension;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
+
 import java.util.List;
+
 import org.apache.kafka.common.TopicPartition;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.parallel.Execution;
 import org.junit.jupiter.api.parallel.ExecutionMode;
 
@@ -30,12 +35,12 @@ import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 
 @Execution(value = ExecutionMode.CONCURRENT)
-public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
+@ExtendWith(VertxExtension.class)
+public class OffsetManagerTest extends AbstractOffsetManagerTest {
 
   @Override
-  RecordDispatcherListener createOffsetManager(
-    KafkaConsumer<?, ?> consumer) {
-    return new UnorderedOffsetManager(consumer, null);
+  RecordDispatcherListener createOffsetManager(final Vertx vertx, final KafkaConsumer<?, ?> consumer) {
+    return new OffsetManager(vertx, consumer, null, 100L);
   }
 
   @Test
@@ -50,6 +55,46 @@ public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
       .containsEntry(new TopicPartition("aaa", 0), 10L);
   }
 
+  @Test
+  public void shouldCommitAfterSendingEventsOrderedOnTheSamePartitionLongValues() {
+    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
+      for (long i = Integer.MAX_VALUE - 50; i < ((long) Integer.MAX_VALUE) + 50; i++) {
+        var rec = record("aaa", 0, i);
+        offsetStrategy.recordReceived(rec);
+        offsetStrategy.successfullySentToSubscriber(rec);
+      }
+    })
+      .containsEntry(new TopicPartition("aaa", 0), Integer.MAX_VALUE + 50L);
+  }
+
+  @Test
+  public void shouldCommitAfterSendingEventsOrderedOnTheSamePartitionLongPeriod() {
+    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
+      for (long i = 0; i < 2_000_051; i++) {
+
+        var rec = record("aaa", 0, i);
+        offsetStrategy.recordReceived(rec);
+        offsetStrategy.successfullySentToSubscriber(rec);
+      }
+    })
+      .containsEntry(new TopicPartition("aaa", 0), 2_000_051L);
+  }
+
+  @Test
+  public void shouldNotCommitAfterSendingEventsOrderedOnTheSamePartitionBrokenSequence() {
+    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
+      // start number is odd number
+      for (long i = Integer.MAX_VALUE - 50; i < ((long) Integer.MAX_VALUE) + 50; i++) {
+        var rec = record("aaa", 0, i);
+        offsetStrategy.recordReceived(rec);
+        if (i % 2 == 0) {
+          offsetStrategy.successfullySentToSubscriber(rec);
+        }
+      }
+    })
+      .isEmpty();
+  }
+
   @Test
   public void shouldNotCommitAndNotGoOutOfBounds() {
     assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
@@ -148,7 +193,7 @@ public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
       List.of(5L, 2L, 0L, 7L, 1L, 3L, 4L)
         .forEach(offset -> offsetStrategy.successfullySentToSubscriber(record("aaa", 0, offset)));
     })
-      .isEmpty();
+      .containsEntry(new TopicPartition("aaa", 0), 6L);
   }
 
   @Test
@@ -209,10 +254,10 @@ public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
 
   @Test
   @SuppressWarnings("unchecked")
-  public void recordReceived() {
+  public void recordReceived(final Vertx vertx) {
     final KafkaConsumer<String, CloudEvent> consumer = mock(KafkaConsumer.class);
     final Counter eventsSentCounter = mock(Counter.class);
-    new UnorderedOffsetManager(consumer, eventsSentCounter::increment).recordReceived(record("aaa", 0, 0));
+    new OffsetManager(vertx, consumer, eventsSentCounter::increment, 100L).recordReceived(record("aaa", 0, 0));
 
     shouldNeverCommit(consumer);
     shouldNeverPause(consumer);
@@ -221,12 +266,11 @@ public class UnorderedOffsetManagerTest extends AbstractOffsetManagerTest {
 
   @Test
   @SuppressWarnings("unchecked")
-  public void failedToSendToDeadLetterSink() {
+  public void failedToSendToDeadLetterSink(final Vertx vertx) {
     final KafkaConsumer<String, CloudEvent> consumer = mock(KafkaConsumer.class);
     final Counter eventsSentCounter = mock(Counter.class);
 
-    UnorderedOffsetManager strategy =
-      new UnorderedOffsetManager(consumer, eventsSentCounter::increment);
+    OffsetManager strategy = new OffsetManager(vertx, consumer, eventsSentCounter::increment, 100L);
     strategy.recordReceived(record("aaa", 0, 0));
     strategy.failedToSendToDeadLetterSink(record("aaa", 0, 0), null);
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedOffsetManagerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedOffsetManagerTest.java
deleted file mode 100644
index 543b28b66..000000000
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedOffsetManagerTest.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.impl.consumer;
-
-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;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.parallel.Execution;
-import org.junit.jupiter.api.parallel.ExecutionMode;
-
-@Execution(value = ExecutionMode.CONCURRENT)
-public class OrderedOffsetManagerTest extends AbstractOffsetManagerTest {
-
-  @Override
-  RecordDispatcherListener createOffsetManager(
-    KafkaConsumer<?, ?> consumer) {
-    return new OrderedOffsetManager(consumer, null);
-  }
-
-  @Test
-  public void shouldNotCommitAfterRecordReceived() {
-    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
-      offsetStrategy.recordReceived(record("aaa", 0, 0));
-    }).isEmpty();
-  }
-
-  @Test
-  public void shouldNotCommitAfterFailedToSendToDeadLetterSink() {
-    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
-      offsetStrategy.failedToSendToDeadLetterSink(record("aaa", 0, 0), new IllegalStateException());
-    }).isEmpty();
-  }
-
-  @Test
-  public void shouldNotCommitAfterRecordDiscarded() {
-    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
-      offsetStrategy.recordDiscarded(record("aaa", 0, 0));
-    }).isEmpty();
-  }
-
-
-  @Test
-  public void shouldCommitAfterSuccessfullySentToSubscriber() {
-    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
-      offsetStrategy.successfullySentToSubscriber(record("aaa", 0, 0));
-    }).containsEntry(new TopicPartition("aaa", 0), 1L);
-  }
-
-  @Test
-  public void shouldCommitAfterSuccessfullySentToDeadLetterSink() {
-    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
-      offsetStrategy.successfullySentToDeadLetterSink(record("aaa", 0, 0));
-    }).containsEntry(new TopicPartition("aaa", 0), 1L);
-  }
-
-}
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/integration/UnorderedConsumerTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/integration/UnorderedConsumerTest.java
index c76bd90fc..b1efd3aca 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/integration/UnorderedConsumerTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/integration/UnorderedConsumerTest.java
@@ -40,6 +40,8 @@ import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -72,6 +74,7 @@ public class UnorderedConsumerTest {
 
     final var producerConfigs = new Properties();
     final var consumerConfigs = new Properties();
+    consumerConfigs.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100L);
 
     final var consumerVerticleFactoryMock = new ConsumerVerticleFactoryImplMock(
       consumerConfigs,
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 b33f00d3a..a1e9babe1 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
@@ -47,6 +47,12 @@ import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.micrometer.MicrometerMetricsOptions;
 import io.vertx.micrometer.backends.BackendRegistries;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
 import java.io.File;
 import java.io.IOException;
@@ -57,12 +63,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-
 import static java.lang.String.format;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
@@ -143,7 +143,7 @@ public class DataPlaneTest {
    */
   @Test
   @Timeout(timeUnit = TimeUnit.MINUTES, value = 1)
-  public void execute(final Vertx vertx, final VertxTestContext context) throws Exception {
+  public void execute(final Vertx vertx, final VertxTestContext context) {
 
     final var checkpoints = context.checkpoint(3);
 
@@ -309,6 +309,7 @@ public class DataPlaneTest {
     consumerConfigs.put(BOOTSTRAP_SERVERS_CONFIG, format("localhost:%d", KAFKA_PORT));
     consumerConfigs.put(KEY_DESERIALIZER_CLASS_CONFIG, KeyDeserializer.class.getName());
     consumerConfigs.put(VALUE_DESERIALIZER_CLASS_CONFIG, CloudEventDeserializer.class.getName());
+    consumerConfigs.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
 
     final var producerConfigs = producerConfigs();
 
diff --git a/test/kafka/strimzi-cluster-operator.yaml b/test/kafka/strimzi-cluster-operator.yaml
index 3ae5d6887..b25287111 100644
--- a/test/kafka/strimzi-cluster-operator.yaml
+++ b/test/kafka/strimzi-cluster-operator.yaml
@@ -11094,7 +11094,7 @@ spec:
               bootstrapServers:
                 type: string
                 description: Bootstrap servers to connect to. This should be given
-                  as a comma separated list of _<hostname>_:‍_<port>_ pairs.
+                  as a comma separated list of _<hostname>_:_<port>_ pairs.
               tls:
                 type: object
                 properties:
@@ -12984,7 +12984,7 @@ spec:
               bootstrapServers:
                 type: string
                 description: Bootstrap servers to connect to. This should be given
-                  as a comma separated list of _<hostname>_:‍_<port>_ pairs.
+                  as a comma separated list of _<hostname>_:_<port>_ pairs.
               tls:
                 type: object
                 properties:
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index 4278eefdf..b17acc0c1 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20211107-bb5d3e038"
+    kafka.eventing.knative.dev/release: "v20211108-c5e12df13"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/third_party/eventing-kafka-latest/source-crds.yaml b/third_party/eventing-kafka-latest/source-crds.yaml
index 474ec8331..6dbd6efb9 100644
--- a/third_party/eventing-kafka-latest/source-crds.yaml
+++ b/third_party/eventing-kafka-latest/source-crds.yaml
@@ -16,7 +16,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    kafka.eventing.knative.dev/release: "v20211107-bb5d3e038"
+    kafka.eventing.knative.dev/release: "v20211108-c5e12df13"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
   name: kafkabindings.bindings.knative.dev
@@ -86,7 +86,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    kafka.eventing.knative.dev/release: "v20211107-bb5d3e038"
+    kafka.eventing.knative.dev/release: "v20211108-c5e12df13"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index daa8f1aa6..a5dc4975d 100644
--- a/third_party/eventing-latest/eventing-core.yaml
+++ b/third_party/eventing-latest/eventing-core.yaml
@@ -16,8 +16,8 @@ kind: Namespace
 metadata:
   name: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 
 ---
@@ -41,8 +41,8 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -50,8 +50,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -67,8 +67,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -84,8 +84,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -101,8 +101,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -118,8 +118,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -151,8 +151,8 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -160,8 +160,8 @@ kind: ClusterRoleBinding
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -193,8 +193,8 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 ---
 apiVersion: rbac.authorization.k8s.io/v1
@@ -202,8 +202,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -220,8 +220,8 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -237,8 +237,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -254,8 +254,8 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 subjects:
   - kind: ServiceAccount
@@ -287,8 +287,8 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 data:
   channelTemplateSpec: |
@@ -316,8 +316,8 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Configures the default for any Broker that does not specify a spec.config or Broker class.
@@ -350,8 +350,8 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
@@ -385,10 +385,10 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 data:
   _example: |
@@ -432,10 +432,10 @@ metadata:
   name: config-features
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # ALPHA feature: The kreference-group allows you to use the Group field in KReferences.
@@ -472,7 +472,7 @@ metadata:
   name: config-kreference-mapping
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -521,8 +521,8 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "96896b00"
@@ -583,10 +583,10 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 data:
   # Common configuration for all Knative codebase
@@ -637,10 +637,10 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "f46cf09d"
@@ -714,10 +714,10 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     knative.dev/example-checksum: "4002b4c2"
@@ -777,10 +777,10 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/high-availability: "true"
     app.kubernetes.io/name: eventing-controller
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -790,9 +790,9 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20211107-4cc5ecf96"
+        eventing.knative.dev/release: "v20211108-ee96d0b3b"
         app.kubernetes.io/name: eventing-autoscaler
-        app.kubernetes.io/version: "20211107-4cc5ecf96"
+        app.kubernetes.io/version: "20211108-ee96d0b3b"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -810,7 +810,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:e5568b5a0684c8e5b4b3ae6375946feccd013045afb5e77841886b351f292849
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:3e8135959f58a24e7ce1379cc6bbb5f51f4018754d961032b2f3aabb9d5bb3b9
           resources:
             requests:
               cpu: 100m
@@ -828,7 +828,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:ea9578fb84528015b4558b49507758a08915becd855b33a45e8fd07e943b1e1f
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:b5aa0495799dc1123c659a7e5f821e1ae7fd6b51a43ecfc39333645b9c360c1f
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -872,9 +872,9 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     app.kubernetes.io/name: pingsource-mt-adapter
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
@@ -887,9 +887,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211107-4cc5ecf96"
+        eventing.knative.dev/release: "v20211108-ee96d0b3b"
         app.kubernetes.io/name: pingsource-mt-adapter
-        app.kubernetes.io/version: "20211107-4cc5ecf96"
+        app.kubernetes.io/version: "20211108-ee96d0b3b"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       affinity:
@@ -903,7 +903,7 @@ spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:9abaf3cd2e331216f0f027920199e4bafbd7dedd49f3584d3c21782f8971bf6a
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:8483aeb2a3a004643da4911508440306f93b9432388f24a4184633e525baabd9
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -961,9 +961,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   scaleTargetRef:
@@ -987,9 +987,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   minAvailable: 80%
@@ -1018,9 +1018,9 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   selector:
@@ -1031,9 +1031,9 @@ spec:
     metadata:
       labels:
         !!merge <<: *labels
-        eventing.knative.dev/release: "v20211107-4cc5ecf96"
+        eventing.knative.dev/release: "v20211108-ee96d0b3b"
         app.kubernetes.io/name: eventing-webhook
-        app.kubernetes.io/version: "20211107-4cc5ecf96"
+        app.kubernetes.io/version: "20211108-ee96d0b3b"
         app.kubernetes.io/part-of: knative-eventing
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
@@ -1053,7 +1053,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:fe440c8bc793fb6277f0a6861af100bfafac5b18980367a7d055479d28df1c3d
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:466719d83fc413c13ed50308185738a7703676aa65821dba0a1b46fafede62ea
           resources:
             requests:
               # taken from serving.
@@ -1117,10 +1117,10 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     role: eventing-webhook
     app.kubernetes.io/name: eventing-webhook
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   name: eventing-webhook
   namespace: knative-eventing
@@ -1152,11 +1152,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1368,10 +1368,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -1535,11 +1535,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -1807,11 +1807,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -1961,9 +1961,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -2090,10 +2090,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2345,11 +2345,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -2536,10 +2536,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -2855,12 +2855,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -3047,9 +3047,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -3259,9 +3259,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -3444,8 +3444,8 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3458,9 +3458,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3478,9 +3478,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3501,9 +3501,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3528,9 +3528,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3549,9 +3549,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3587,8 +3587,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3614,8 +3614,8 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3632,8 +3632,8 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3666,8 +3666,8 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -3680,9 +3680,9 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     duck.knative.dev/channelable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3720,9 +3720,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3734,9 +3734,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3748,9 +3748,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3762,9 +3762,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3776,9 +3776,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3791,8 +3791,8 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3805,8 +3805,8 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
@@ -3833,8 +3833,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -3969,8 +3969,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4037,8 +4037,8 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4051,9 +4051,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     duck.knative.dev/podspecable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -4099,8 +4099,8 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 aggregationRule:
   clusterRoleSelectors:
@@ -4113,9 +4113,9 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     duck.knative.dev/source: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -4151,8 +4151,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   - apiGroups:
@@ -4238,8 +4238,8 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For watching logging configuration and getting certs.
@@ -4338,8 +4338,8 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 rules:
   # For manipulating certs into secrets.
@@ -4375,8 +4375,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4413,8 +4413,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4447,8 +4447,8 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
@@ -4482,8 +4482,8 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 # The data is populated at install time.
 
@@ -4507,8 +4507,8 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
diff --git a/third_party/eventing-latest/eventing-crds.yaml b/third_party/eventing-latest/eventing-crds.yaml
index 5266fce2b..cb541eb72 100644
--- a/third_party/eventing-latest/eventing-crds.yaml
+++ b/third_party/eventing-latest/eventing-crds.yaml
@@ -17,11 +17,11 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -233,10 +233,10 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -400,11 +400,11 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -672,11 +672,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   name: containersources.sources.knative.dev
 spec:
@@ -826,9 +826,9 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
@@ -955,10 +955,10 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1210,11 +1210,11 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   annotations:
     # TODO add schemas and descriptions
@@ -1401,10 +1401,10 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: flows.knative.dev
@@ -1720,12 +1720,12 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
   name: sinkbindings.sources.knative.dev
 spec:
@@ -1912,9 +1912,9 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: messaging.knative.dev
@@ -2124,9 +2124,9 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20211107-4cc5ecf96"
+    eventing.knative.dev/release: "v20211108-ee96d0b3b"
     knative.dev/crd-install: "true"
-    app.kubernetes.io/version: "20211107-4cc5ecf96"
+    app.kubernetes.io/version: "20211108-ee96d0b3b"
     app.kubernetes.io/part-of: knative-eventing
 spec:
   group: eventing.knative.dev
diff --git a/vendor/knative.dev/eventing-kafka/test/lib/creation.go b/vendor/knative.dev/eventing-kafka/test/lib/creation.go
index d667b3686..bb796517d 100644
--- a/vendor/knative.dev/eventing-kafka/test/lib/creation.go
+++ b/vendor/knative.dev/eventing-kafka/test/lib/creation.go
@@ -31,75 +31,91 @@ import (
 )
 
 func CreateKafkaChannelV1Beta1OrFail(c *testlib.Client, kafkaChannel *channelsv1beta1.KafkaChannel) {
-	kafkaChannelClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	client, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
 		c.T.Fatalf("Failed to create v1beta1 KafkaChannel client: %v", err)
 	}
 
-	kChannels := kafkaChannelClientSet.MessagingV1beta1().KafkaChannels(c.Namespace)
-	if createdKafkaChannel, err := kChannels.Create(context.Background(), kafkaChannel, metav1.CreateOptions{}); err != nil {
-		c.T.Fatalf("Failed to create v1beta1 KafkaChannel %q: %v", kafkaChannel.Name, err)
-	} else {
+	err = c.RetryWebhookErrors(func(i int) error {
+		createdKafkaChannel, err := client.MessagingV1beta1().KafkaChannels(c.Namespace).Create(context.Background(), kafkaChannel, metav1.CreateOptions{})
+		if err != nil {
+			return err
+		}
 		c.Tracker.AddObj(createdKafkaChannel)
+		return nil
+	})
+	if err != nil {
+		c.T.Fatalf("Failed to create v1beta1 KafkaChannel %q: %v", kafkaChannel.Name, err)
 	}
 }
 
 func GetKafkaChannelV1Beta1OrFail(c *testlib.Client, kafkaChannel string) *channelsv1beta1.KafkaChannel {
-	kafkaChannelClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	client, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
 		c.T.Fatalf("Failed to create v1beta1 KafkaChannel client: %v", err)
 	}
 
-	kChannels := kafkaChannelClientSet.MessagingV1beta1().KafkaChannels(c.Namespace)
-	if kcObj, err := kChannels.Get(context.Background(), kafkaChannel, metav1.GetOptions{}); err != nil {
+	var kcObj *channelsv1beta1.KafkaChannel
+	err = c.RetryWebhookErrors(func(i int) error {
+		kcObj, err = client.MessagingV1beta1().KafkaChannels(c.Namespace).Get(context.Background(), kafkaChannel, metav1.GetOptions{})
+		return err
+	})
+	if err != nil {
 		c.T.Fatalf("Failed to get v1beta1 KafkaChannel %q: %v", kafkaChannel, err)
-	} else {
-		return kcObj
 	}
-	return nil
+	return kcObj
 }
 
 func CreateKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource *sourcesv1beta1.KafkaSource) {
-	kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	client, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
 		c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
 	}
 
-	kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
-	if createdKafkaSource, err := kSources.Create(context.Background(), kafkaSource, metav1.CreateOptions{}); err != nil {
-		c.T.Fatalf("Failed to create v1beta1 KafkaSource %q: %v", kafkaSource.Name, err)
-	} else {
+	err = c.RetryWebhookErrors(func(i int) error {
+		createdKafkaSource, err := client.SourcesV1beta1().KafkaSources(c.Namespace).Create(context.Background(), kafkaSource, metav1.CreateOptions{})
+		if err != nil {
+			return err
+		}
 		c.Tracker.AddObj(createdKafkaSource)
+		return nil
+	})
+	if err != nil {
+		c.T.Fatalf("Failed to create v1beta1 KafkaSource %q: %v", kafkaSource.Name, err)
 	}
 }
 
 func GetKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource string) *sourcesv1beta1.KafkaSource {
-	kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	client, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {
 		c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
 	}
 
-	kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
-	if ksObj, err := kSources.Get(context.Background(), kafkaSource, metav1.GetOptions{}); err != nil {
+	var ksObj *sourcesv1beta1.KafkaSource
+	err = c.RetryWebhookErrors(func(i int) error {
+		ksObj, err = client.SourcesV1beta1().KafkaSources(c.Namespace).Get(context.Background(), kafkaSource, metav1.GetOptions{})
+		return err
+	})
+	if err != nil {
 		c.T.Fatalf("Failed to get v1beta1 KafkaSource %q: %v", kafkaSource, err)
-	} else {
-		return ksObj
 	}
-	return nil
+	return ksObj
 }
 
 func UpdateKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource *sourcesv1beta1.KafkaSource) {
-	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		latestKafkaSource := GetKafkaSourceV1Beta1OrFail(c, kafkaSource.Name)
-		kafkaSource.Spec.DeepCopyInto(&latestKafkaSource.Spec)
-		kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
-		if err != nil {
-			c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
-		}
-
-		kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
-		_, err = kSources.Update(context.Background(), latestKafkaSource, metav1.UpdateOptions{})
-		return err
+	err := c.RetryWebhookErrors(func(i int) error {
+		return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+			latestKafkaSource := GetKafkaSourceV1Beta1OrFail(c, kafkaSource.Name)
+			kafkaSource.Spec.DeepCopyInto(&latestKafkaSource.Spec)
+			kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+			if err != nil {
+				c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
+			}
+
+			kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
+			_, err = kSources.Update(context.Background(), latestKafkaSource, metav1.UpdateOptions{})
+			return err
+		})
 	})
 	if err != nil {
 		c.T.Fatalf("Failed to update v1beta1 KafkaSource %q: %v", kafkaSource.Name, err)
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go b/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
index e97c38adf..bc5811242 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/readyness.go
@@ -19,10 +19,12 @@ package broker
 import (
 	"fmt"
 
+	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
 
 	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/delivery"
 	"knative.dev/eventing/test/rekt/resources/trigger"
 	"knative.dev/reconciler-test/resources/svc"
 )
@@ -64,3 +66,22 @@ func GoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
 
 	return f
 }
+
+// GoesReady returns a feature that will create a Broker of the given
+// name and class, with an event receiver as its DLS
+// and confirm it becomes ready with an address.
+func GoesReadyWithProbeReceiver(name, sinkName string, prober *eventshub.EventProber, cfg ...manifest.CfgFn) *feature.Feature {
+	f := new(feature.Feature)
+
+	f.Setup("install probe", prober.ReceiverInstall(sinkName))
+
+	brokerConfig := append(cfg, delivery.WithDeadLetterSink(prober.AsKReference(sinkName), ""))
+	f.Setup(fmt.Sprintf("install broker %q", name), broker.Install(name, brokerConfig...))
+
+	f.Requirement("Broker is ready", broker.IsReady(name))
+
+	f.Stable("broker").
+		Must("be addressable", broker.IsAddressable(name))
+
+	return f
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go b/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
index e287cba97..40907d43e 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
@@ -22,7 +22,6 @@ import (
 
 	"github.com/google/uuid"
 	"knative.dev/eventing/test/rekt/resources/broker"
-	"knative.dev/eventing/test/rekt/resources/delivery"
 	"knative.dev/eventing/test/rekt/resources/eventlibrary"
 	"knative.dev/eventing/test/rekt/resources/flaker"
 	"knative.dev/eventing/test/rekt/resources/trigger"
@@ -32,8 +31,6 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/resources/svc"
 
-	b "knative.dev/eventing/test/rekt/resources/broker"
-
 	. "github.com/cloudevents/sdk-go/v2/test"
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
 )
@@ -79,8 +76,7 @@ func SourceToSink(brokerName string) *feature.Feature {
 //                |
 //                +--[DLQ]--> sink
 //
-func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
-	prober := eventshub.NewProber()
+func SourceToSinkWithDLQ(brokerName, sinkName string, prober *eventshub.EventProber) *feature.Feature {
 	prober.SetTargetResource(broker.GVR(), brokerName)
 
 	via := feature.MakeRandomK8sName("via")
@@ -95,15 +91,6 @@ func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
 		panic(fmt.Errorf("can not find event files: %s", err))
 	}
 
-	// Setup Probes
-	f.Setup("install recorder", prober.ReceiverInstall("sink"))
-
-	// Setup data plane
-	brokerConfig := append(b.WithEnvConfig(), delivery.WithDeadLetterSink(prober.AsKReference("sink"), ""))
-	f.Setup("update broker with DLQ", broker.Install(
-		brokerName,
-		brokerConfig...,
-	))
 	f.Setup("install trigger", trigger.Install(via, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
 
 	// Resources ready.
@@ -118,7 +105,7 @@ func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLQ").
 		Must("accepted all events", prober.AssertSentAll("source")).
-		Must("deliver event to DLQ", prober.AssertReceivedAll("source", "sink"))
+		Must("deliver event to DLQ", prober.AssertReceivedAll("source", sinkName))
 
 	return f
 }
@@ -131,8 +118,7 @@ func SourceToSinkWithDLQ(brokerName string) *feature.Feature {
 //                |
 //                +--[DLQ]--> sink1
 //
-func SourceToTwoSinksWithDLQ(brokerName string) *feature.Feature {
-	prober := eventshub.NewProber()
+func SourceToTwoSinksWithDLQ(brokerName, sink1, sink2 string, prober *eventshub.EventProber) *feature.Feature {
 	prober.SetTargetResource(broker.GVR(), brokerName)
 
 	via1 := feature.MakeRandomK8sName("via")
@@ -149,13 +135,11 @@ func SourceToTwoSinksWithDLQ(brokerName string) *feature.Feature {
 	}
 
 	// Setup Probes
-	f.Setup("install recorder1", prober.ReceiverInstall("sink1"))
-	f.Setup("install recorder2", prober.ReceiverInstall("sink2"))
+	f.Setup("install recorder2", prober.ReceiverInstall(sink2))
 
 	// Setup data plane
-	f.Setup("update broker with DLQ", broker.Install(brokerName, delivery.WithDeadLetterSink(prober.AsKReference("sink1"), "")))
 	f.Setup("install trigger via1", trigger.Install(via1, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
-	f.Setup("install trigger via2", trigger.Install(via2, brokerName, trigger.WithSubscriber(prober.AsKReference("sink2"), "")))
+	f.Setup("install trigger via2", trigger.Install(via2, brokerName, trigger.WithSubscriber(prober.AsKReference(sink2), "")))
 
 	// Resources ready.
 	f.Setup("trigger1 goes ready", trigger.IsReady(via1))
@@ -166,14 +150,14 @@ func SourceToTwoSinksWithDLQ(brokerName string) *feature.Feature {
 
 	// After we have finished sending.
 	f.Requirement("sender is finished", prober.SenderDone("source"))
-	f.Requirement("receiver 1 is finished", prober.ReceiverDone("source", "sink1"))
-	f.Requirement("receiver 2 is finished", prober.ReceiverDone("source", "sink2"))
+	f.Requirement("receiver 1 is finished", prober.ReceiverDone("source", sink1))
+	f.Requirement("receiver 2 is finished", prober.ReceiverDone("source", sink2))
 
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLQ").
 		Must("accepted all events", prober.AssertSentAll("source")).
-		Must("deliver event to DLQ (via1)", prober.AssertReceivedAll("source", "sink1")).
-		Must("deliver event to sink (via2)", prober.AssertReceivedAll("source", "sink2"))
+		Must("deliver event to DLQ (via1)", prober.AssertReceivedAll("source", sink1)).
+		Must("deliver event to sink (via2)", prober.AssertReceivedAll("source", sink2))
 
 	return f
 }
diff --git a/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go b/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
new file mode 100644
index 000000000..6a5e3677d
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
@@ -0,0 +1,143 @@
+/*
+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 trigger
+
+import (
+	"context"
+	"fmt"
+
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/delivery"
+	"knative.dev/eventing/test/rekt/resources/eventlibrary"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+// SourceToTriggerSinkWithDLS tests to see if a Ready Trigger with a DLS defined send
+// failing events to it's DLS.
+//
+// source ---> broker --[trigger]--> bad uri
+//                          |
+//                          +--[DLS]--> sink
+//
+func SourceToTriggerSinkWithDLS(triggerName, brokerName string, prober *eventshub.EventProber) *feature.Feature {
+	prober.SetTargetResource(broker.GVR(), brokerName)
+	f := feature.NewFeature()
+
+	lib := feature.MakeRandomK8sName("lib")
+	f.Setup("install events", eventlibrary.Install(lib))
+	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
+	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
+	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
+		panic(fmt.Errorf("can not find event files: %s", err))
+	}
+
+	// Setup Probes
+	f.Setup("install recorder", prober.ReceiverInstall("sink"))
+
+	// Setup trigger
+	f.Setup("install trigger", trigger.Install(
+		triggerName,
+		brokerName,
+		trigger.WithSubscriber(nil, "bad://uri"),
+		delivery.WithDeadLetterSink(prober.AsKReference("sink"), "")))
+
+	// Resources ready.
+	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
+
+	// Install sender.
+	f.Setup("install source", prober.SenderInstall("source"))
+
+	// After we have finished sending.
+	f.Requirement("sender is finished", prober.SenderDone("source"))
+
+	// Assert events ended up where we expected.
+	f.Stable("trigger with DLS").
+		Must("accepted all events", prober.AssertSentAll("source")).
+		Must("deliver event to DLS", prober.AssertReceivedAll("source", "sink"))
+
+	return f
+}
+
+// SourceToTriggerSinkWithDLSDontUseBrokers tests to see if a Ready Trigger sends
+// failing events to it's DLS even when it's corresponding Ready Broker also have a DLS defined.
+//
+// source ---> broker --[trigger]--> bad uri
+//               |          |
+//               +--[DLS]   +--[DLS]--> sink
+//
+func SourceToTriggerSinkWithDLSDontUseBrokers(triggerName, brokerName, brokerSinkName string, prober *eventshub.EventProber) *feature.Feature {
+	triggerSinkName := "trigger-sink"
+	prober.SetTargetResource(broker.GVR(), brokerName)
+
+	f := feature.NewFeature()
+
+	lib := feature.MakeRandomK8sName("lib")
+	f.Setup("install events", eventlibrary.Install(lib))
+	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
+	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
+	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
+		panic(fmt.Errorf("can not find event files: %s", err))
+	}
+
+	// Setup Probes
+	f.Setup("install trigger recorder", prober.ReceiverInstall(triggerSinkName))
+	f.Setup("install brokers recorder", prober.ReceiverInstall(brokerSinkName))
+
+	// Setup topology
+	brokerConfig := append(
+		broker.WithEnvConfig(),
+		delivery.WithDeadLetterSink(prober.AsKReference(brokerSinkName), ""))
+	f.Setup("install broker with DLS", broker.Install(
+		brokerName,
+		brokerConfig...,
+	))
+
+	f.Setup("install trigger", trigger.Install(
+		triggerName,
+		brokerName,
+		trigger.WithSubscriber(nil, "bad://uri"),
+		delivery.WithDeadLetterSink(prober.AsKReference(triggerSinkName), "")))
+
+	// Resources ready.
+	f.Setup("trigger goes ready", trigger.IsReady(triggerName))
+
+	// Install events after topology is ready.
+	f.Setup("install source", prober.SenderInstall("source"))
+
+	// After we have finished sending.
+	f.Requirement("sender is finished", prober.SenderDone("source"))
+
+	// Assert events ended up where we expected.
+	f.Stable("trigger with a valid DLS ref").
+		Must("accept all events", prober.AssertSentAll("source")).
+		Must("deliver events to trigger DLS", prober.AssertReceivedAll("source", triggerSinkName)).
+		Must("not deliver events to its broker DLS", noEventsToDLS(prober, brokerSinkName))
+
+	return f
+}
+
+func noEventsToDLS(prober *eventshub.EventProber, sinkName string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		if len(prober.ReceivedBy(ctx, sinkName)) == 0 {
+			t.Log("no events were sent to %s DLS", sinkName)
+		} else {
+			t.Errorf("events were received by %s DLS", sinkName)
+		}
+	}
+}