Repository: knative-sandbox/eventing-kafka-broker
Pull request #889 (2021-05-09 22:25:00)

Skip TestTriggerNoFinalizerOnBrokerNotFound test
Skip TestTriggerNoFinalizerOnBrokerNotFound test since it's flaky

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

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** after the PR has been reviewed.

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).

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

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

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":["pierdipi"]} -->
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/trigger/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/889/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1391519612893401088/artifacts/line-cov.html#file0) | 60.0% | 63.6% | 3.6


Patch:
diff --git a/.github/workflows/build-dependabot-pr.yaml b/.github/workflows/build-dependabot-pr.yaml
index 568915b7a..e283a6bbf 100644
--- a/.github/workflows/build-dependabot-pr.yaml
+++ b/.github/workflows/build-dependabot-pr.yaml
@@ -29,15 +29,15 @@ jobs:
         run: echo ${{ github.ref }} | sed 's/refs\/heads\///' > ./dependabot-pr/BRANCH_REF
       - name: Configure git
         run: |
-          git config --local user.email "41898282+github-actions[bot]@users.noreply.github.com"
-          git config --local user.name "github-actions[bot]"
+          git config --local user.email "automation@knative.team"
+          git config --local user.name "Knative Automation"
       # NOTE: Prefixing/appending commit messages with `[dependabot skip]` allows
       # dependabot to rebase/update the pull request, force-pushing over any changes
       - name: Commit and copy license changes
         run: |
           if [[ $(git status | grep 'data-plane/THIRD-PARTY.txt') ]]; then
             git add data-plane/THIRD-PARTY.txt
-            git commit -m "[dependabot skip] Update data-plane/THIRD-PARTY.txt"
+            git commit --signoff -m "[dependabot skip] Update data-plane/THIRD-PARTY.txt"
             git format-patch -n HEAD^ --binary --stdout > ./dependabot-pr/changes.patch
           else
             echo "Licenses were not changed"
diff --git a/.github/workflows/knative-boilerplate.yaml b/.github/workflows/knative-boilerplate.yaml
index c0e648b1c..9d4093ea6 100644
--- a/.github/workflows/knative-boilerplate.yaml
+++ b/.github/workflows/knative-boilerplate.yaml
@@ -42,10 +42,10 @@ jobs:
 
     steps:
 
-      - name: Set up Go 1.15.x
+      - name: Set up Go 1.16.x
         uses: actions/setup-go@v2
         with:
-          go-version: 1.15.x
+          go-version: 1.16.x
         id: go
 
       - name: Check out code
diff --git a/.github/workflows/knative-go-build.yaml b/.github/workflows/knative-go-build.yaml
index 7bd13004a..da2ffd01a 100644
--- a/.github/workflows/knative-go-build.yaml
+++ b/.github/workflows/knative-go-build.yaml
@@ -27,7 +27,7 @@ jobs:
     name: Build
     strategy:
       matrix:
-        go-version: [1.15.x]
+        go-version: [1.16.x]
         platform: [ubuntu-latest]
 
     runs-on: ${{ matrix.platform }}
@@ -43,7 +43,13 @@ jobs:
       - name: Check out code
         uses: actions/checkout@v2
 
+      - id: go_mod
+        uses: andstor/file-existence-action@v1
+        with:
+          files: go.mod
+
       - name: Build
+        if: ${{ steps.go_mod.outputs.files_exists == 'true' }}
         run: |
           tags="$(grep -I  -r '// +build' . | \
                 grep -v '^./vendor/' | \
@@ -55,5 +61,4 @@ jobs:
                 tr '\n' ' ')"
 
           echo "Building with tags: ${tags}"
-          go test -vet=off -tags "${tags}" -run=^$ ./... | grep -v "no test" || true
-
+          go test -vet=off -tags "${tags}" -exec echo ./...
diff --git a/.github/workflows/knative-releasability.yaml b/.github/workflows/knative-releasability.yaml
index c4f5fb732..5e064c4c4 100644
--- a/.github/workflows/knative-releasability.yaml
+++ b/.github/workflows/knative-releasability.yaml
@@ -54,10 +54,10 @@ jobs:
             echo "SLACK_WEBHOOK=exists" >> $GITHUB_ENV
           fi
 
-      - name: Set up Go 1.15.x
+      - name: Set up Go 1.16.x
         uses: actions/setup-go@v2
         with:
-          go-version: 1.15.x
+          go-version: 1.16.x
 
       - name: Install Dependencies
         run: GO111MODULE=on go get knative.dev/test-infra/buoy@main
diff --git a/.github/workflows/knative-style.yaml b/.github/workflows/knative-style.yaml
index c5770a4b0..754b91736 100644
--- a/.github/workflows/knative-style.yaml
+++ b/.github/workflows/knative-style.yaml
@@ -40,10 +40,10 @@ jobs:
           importpath: golang.org/x/tools/cmd/goimports
 
     steps:
-      - name: Set up Go 1.15.x
+      - name: Set up Go 1.16.x
         uses: actions/setup-go@v2
         with:
-          go-version: 1.15.x
+          go-version: 1.16.x
         id: go
 
       - name: Check out code
@@ -89,10 +89,10 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - name: Set up Go 1.15.x
+      - name: Set up Go 1.16.x
         uses: actions/setup-go@v2
         with:
-          go-version: 1.15.x
+          go-version: 1.16.x
         id: go
 
       - name: Check out code
diff --git a/.github/workflows/knative-verify.yaml b/.github/workflows/knative-verify.yaml
index 5f20c548b..e18937e17 100644
--- a/.github/workflows/knative-verify.yaml
+++ b/.github/workflows/knative-verify.yaml
@@ -27,7 +27,7 @@ jobs:
     name: Verify Deps and Codegen
     strategy:
       matrix:
-        go-version: [1.15.x]
+        go-version: [1.16.x]
         platform: [ubuntu-latest]
 
     runs-on: ${{ matrix.platform }}
diff --git a/.github/workflows/update-dependabot-pr.yaml b/.github/workflows/update-dependabot-pr.yaml
index cb36b0c9e..8154837a7 100644
--- a/.github/workflows/update-dependabot-pr.yaml
+++ b/.github/workflows/update-dependabot-pr.yaml
@@ -49,8 +49,8 @@ jobs:
           ref: ${{ env.BRANCH_REF }}
       - name: Configure git
         run: |
-          git config --local user.email "41898282+github-actions[bot]@users.noreply.github.com"
-          git config --local user.name "github-actions[bot]"
+          git config --local user.email "automation@knative.team"
+          git config --local user.name "Knative Automation"
       - name: Commit license changes
         run: |
           CHANGES_PATH=/home/runner/work/dependabot-pr/changes.patch
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 9ba802cba..4be251020 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -54,6 +54,7 @@ aliases:
   eventing-kafka-approvers:
   - aliok
   - davyodom
+  - devguyio
   - lberk
   - matzew
   - phamilton
diff --git a/control-plane/pkg/kafka/consumer_group_lag.go b/control-plane/pkg/kafka/consumer_group_lag.go
new file mode 100644
index 000000000..a7ccb8ec9
--- /dev/null
+++ b/control-plane/pkg/kafka/consumer_group_lag.go
@@ -0,0 +1,245 @@
+/*
+ * 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 kafka
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/Shopify/sarama"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+const (
+	invalidOffset = -1
+)
+
+// ConsumerGroupLagProvider provides consumer group lags.
+type ConsumerGroupLagProvider interface {
+	// GetLag returns consumer group lag for a given topic and a given consumer group.
+	GetLag(topic, consumerGroup string) (ConsumerGroupLag, error)
+
+	// Close closes the consumer group lag provider.
+	Close() error
+}
+
+// ConsumerGroupLag contains partition lag of a topic.
+type ConsumerGroupLag struct {
+	Topic         string
+	ConsumerGroup string
+	ByPartition   []PartitionLag
+}
+
+// PartitionLag contains consumer lag information of a partition.
+type PartitionLag struct {
+	// Offset that will be produced next.
+	LatestOffset int64
+	// Offset that will be consumed next.
+	ConsumerOffset int64
+	// Signal whether a consumer made a fetch request or not to the leader of this partition.
+	//
+	// Note: a committed offset is the offset that will be consumed next.
+	//
+	// OffsetCommitted = false, no fetch request has been made by a consumer to the leader of this partition.
+	// OffsetCommitted = true, a fetch request has been made by a consumer to the leader of this partition.
+	OffsetCommitted bool
+}
+
+type adminFunc func(client sarama.Client) (sarama.ClusterAdmin, error)
+
+type consumerGroupLagProvider struct {
+	client         sarama.Client
+	adminFunc      adminFunc
+	offsetStrategy int64
+}
+
+var (
+	offsetStrategies = sets.NewInt64(sarama.OffsetNewest, sarama.OffsetOldest)
+)
+
+const (
+	defaultOffsetStrategy = sarama.OffsetOldest
+)
+
+// NewConsumerGroupLagProvider creates a new ConsumerGroupLagProvider.
+func NewConsumerGroupLagProvider(client sarama.Client, adminFunc adminFunc, saramaOffsetStrategy int64) ConsumerGroupLagProvider {
+	if !offsetStrategies.Has(saramaOffsetStrategy) {
+		saramaOffsetStrategy = defaultOffsetStrategy
+	}
+	return &consumerGroupLagProvider{client: client, adminFunc: adminFunc, offsetStrategy: saramaOffsetStrategy}
+}
+
+// GetLag returns consumer group lag for a given group.
+func (p *consumerGroupLagProvider) GetLag(topic, consumerGroup string) (ConsumerGroupLag, error) {
+
+	admin, err := p.adminFunc(p.client)
+	if err != nil {
+		return ConsumerGroupLag{}, fmt.Errorf("failed to create admin client: %w", err)
+	}
+	// Note: Do not close admin since it closes the underlying client.
+
+	partitions, err := getPartitionsForTopic(admin, topic)
+	if err != nil {
+		return ConsumerGroupLag{}, err
+	}
+
+	offsets, err := admin.ListConsumerGroupOffsets(consumerGroup, map[string][]int32{topic: partitions})
+	if err != nil {
+		return ConsumerGroupLag{}, fmt.Errorf("failed to list consumer group offsets: %w", err)
+	}
+
+	consumerGroupLag := ConsumerGroupLag{
+		Topic:         topic,
+		ConsumerGroup: consumerGroup,
+		ByPartition:   make([]PartitionLag, len(partitions)),
+	}
+	for _, partition := range partitions {
+		partitionLag, err := p.getPartitionLag(partition, topic, offsets)
+		if err != nil {
+			return consumerGroupLag, fmt.Errorf("failed to get lag for partition %d: %w", partition, err)
+		}
+		consumerGroupLag.ByPartition[partition] = partitionLag
+	}
+	return consumerGroupLag, nil
+}
+
+func getPartitionsForTopic(admin sarama.ClusterAdmin, topic string) ([]int32, error) {
+
+	topicsMetadata, err := admin.DescribeTopics([]string{topic})
+	if err != nil {
+		return nil, fmt.Errorf("failed to describe topic %s: %w", topic, err)
+	}
+	if len(topicsMetadata) != 1 {
+		return nil, fmt.Errorf("unexpected number of topic metadata for topic %s: %d", topic, len(topicsMetadata))
+	}
+
+	partitionMetadata := topicsMetadata[0].Partitions
+	partitions := make([]int32, 0, len(partitionMetadata))
+	for _, p := range partitionMetadata {
+		partitions = append(partitions, p.ID)
+	}
+	return partitions, nil
+}
+
+func (p *consumerGroupLagProvider) getPartitionLag(partition int32, topic string, offsets *sarama.OffsetFetchResponse) (PartitionLag, error) {
+
+	// Get the offset of the message that will be consumed next.
+	block := offsets.GetBlock(topic, partition)
+	if block == nil {
+		return PartitionLag{}, fmt.Errorf("failed to find offset block for partition %d of topic %s", partition, topic)
+	}
+	consumerOffset := block.Offset
+
+	// Get the offset of the message that will be produced next.
+	latestOffset, err := p.client.GetOffset(topic, partition, sarama.OffsetNewest)
+	if err != nil {
+		return PartitionLag{}, fmt.Errorf("failed to find latest offset for topic %s and partition %d", topic, partition)
+	}
+	latestOffset = max(0, latestOffset) // latest offset should always be greater or equal to 0.
+
+	offsetCommitted := true
+	if consumerOffset <= invalidOffset {
+		// When we receive an invalid consumer offset, it means no offset has yet been committed.
+		offsetCommitted = false
+
+		if p.offsetStrategy == sarama.OffsetOldest {
+			// Set consumer offset to the first offset.
+			consumerOffset = 0
+		} else if p.offsetStrategy == sarama.OffsetNewest {
+			// Set consumer offset to the offset that will be produced next.
+			consumerOffset = latestOffset
+		} else {
+			return PartitionLag{}, fmt.Errorf(
+				"received invalid offset (%d) and unknown offset strategy (%d) configured",
+				consumerOffset, p.offsetStrategy,
+			)
+		}
+	}
+
+	pl := PartitionLag{
+		LatestOffset:    latestOffset,
+		ConsumerOffset:  consumerOffset,
+		OffsetCommitted: offsetCommitted,
+	}
+	return pl, nil
+}
+
+func max(offset int64, x int64) int64 {
+	if offset > x {
+		return offset
+	}
+	return x
+}
+
+func (p *consumerGroupLagProvider) Close() error {
+	if err := p.client.Close(); err != nil && !errors.Is(err, sarama.ErrClosedClient) {
+		return err
+	}
+	return nil
+}
+
+// Total returns the sum of each partition lag.
+func (cgl ConsumerGroupLag) Total() uint64 {
+	var total uint64
+	for _, lag := range cgl.ByPartition {
+		total += uint64(lag.Lag())
+	}
+	return total
+}
+
+func (cgl ConsumerGroupLag) String() string {
+	sb := &strings.Builder{}
+
+	writeSeparator := func(n int, sb *strings.Builder) {
+		sb.WriteString(strings.Repeat("-", n))
+		sb.WriteRune('\n')
+	}
+
+	// Write header
+	// --------------
+	// partition: lag
+	// --------------
+	header := "partition: lag\n"
+	n := len(header)
+
+	sb.WriteString(fmt.Sprintf("Topic: %s\n", cgl.Topic))
+	sb.WriteString(fmt.Sprintf("Consumer group: %s\n", cgl.ConsumerGroup))
+	writeSeparator(n, sb)
+	sb.WriteString(header)
+	writeSeparator(n, sb)
+
+	for p, l := range cgl.ByPartition {
+		sb.WriteString(fmt.Sprintf("%d: %s\n", p, l))
+	}
+	sb.WriteString(fmt.Sprintf("Total lag: %d\n", cgl.Total()))
+	return sb.String()
+}
+
+// Lag returns LatestOffset - ConsumerOffset.
+func (pl PartitionLag) Lag() int64 {
+	return pl.LatestOffset - pl.ConsumerOffset
+}
+
+func (pl PartitionLag) String() string {
+	return fmt.Sprintf("latest offset %d consumer offset %d offsetCommitted %v lag %d",
+		pl.LatestOffset,
+		pl.ConsumerOffset,
+		pl.OffsetCommitted,
+		pl.Lag(),
+	)
+}
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index 2d03cd585..004303311 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -104,12 +104,8 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 	configmapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.DataPlaneConfigMapName),
 		Handler: cache.ResourceEventHandlerFuncs{
-			AddFunc: func(obj interface{}) {
-				globalResync(obj)
-			},
-			DeleteFunc: func(obj interface{}) {
-				globalResync(obj)
-			},
+			AddFunc:    globalResync,
+			DeleteFunc: globalResync,
 		},
 	})
 
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 1b015b94c..49d26db2f 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -16,12 +16,12 @@ Lists of 150 third-party dependencies.
      (Apache 2.0) Gson (com.google.code.gson:gson:2.8.6 - https://github.com/google/gson/gson)
      (Apache 2.0) error-prone annotations (com.google.errorprone:error_prone_annotations:2.3.4 - http://nexus.sonatype.org/oss-repository-hosting.html/error_prone_parent/error_prone_annotations)
      (The Apache Software License, Version 2.0) Guava InternalFutureFailureAccess and InternalFutures (com.google.guava:failureaccess:1.0.1 - https://github.com/google/guava/failureaccess)
-     (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.0-android - https://github.com/google/guava/guava)
      (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.0-jre - https://github.com/google/guava/guava)
+     (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.1-android - https://github.com/google/guava/guava)
      (The Apache Software License, Version 2.0) Guava ListenableFuture only (com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava - https://github.com/google/guava/listenablefuture)
      (The Apache Software License, Version 2.0) J2ObjC Annotations (com.google.j2objc:j2objc-annotations:1.3 - https://github.com/google/j2objc/)
-     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.15.8 - https://developers.google.com/protocol-buffers/protobuf-java/)
-     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.15.8 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
+     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.16.0 - https://developers.google.com/protocol-buffers/protobuf-java/)
+     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.16.0 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
      (Apache 2.0) OkHttp Logging Interceptor (com.squareup.okhttp3:logging-interceptor:3.12.12 - https://github.com/square/okhttp/logging-interceptor)
      (Apache 2.0) MockWebServer (com.squareup.okhttp3:mockwebserver:3.12.6 - https://github.com/square/okhttp/mockwebserver)
      (Apache 2.0) OkHttp (com.squareup.okhttp3:okhttp:3.14.9 - https://github.com/square/okhttp/okhttp)
@@ -35,11 +35,11 @@ Lists of 150 third-party dependencies.
      (Unknown license) dispatcher (dev.knative.eventing.kafka.broker:dispatcher:1.0-SNAPSHOT - no url defined)
      (Unknown license) receiver (dev.knative.eventing.kafka.broker:receiver:1.0-SNAPSHOT - no url defined)
      (BSD) Automaton (dk.brics.automaton:automaton:1.11-8 - http://www.brics.dk/automaton/)
-     (The Apache Software License, Version 2.0) CloudEvents - API (io.cloudevents:cloudevents-api:2.1.0 - https://cloudevents.github.io/sdk-java/cloudevents-api/)
-     (The Apache Software License, Version 2.0) CloudEvents - Core (io.cloudevents:cloudevents-core:2.1.0 - https://cloudevents.github.io/sdk-java/cloudevents-core/)
-     (The Apache Software License, Version 2.0) CloudEvents - Vert.x Http Binding (io.cloudevents:cloudevents-http-vertx:2.1.0 - https://cloudevents.github.io/sdk-java/cloudevents-http-vertx/)
-     (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.1.0 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
-     (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.1.0 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
+     (The Apache Software License, Version 2.0) CloudEvents - API (io.cloudevents:cloudevents-api:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-api/)
+     (The Apache Software License, Version 2.0) CloudEvents - Core (io.cloudevents:cloudevents-core:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-core/)
+     (The Apache Software License, Version 2.0) CloudEvents - Vert.x Http Binding (io.cloudevents:cloudevents-http-vertx:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-http-vertx/)
+     (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
+     (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.1.1 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
      (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.5.0.Final - https://debezium.io/debezium-parent/debezium-api)
      (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.5.0.Final - https://debezium.io/debezium-parent/debezium-core)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.3.1 - http://fabric8.io/kubernetes-client/)
@@ -123,7 +123,7 @@ Lists of 150 third-party dependencies.
      (Apache License, Version 2.0) Apache ZooKeeper - Jute (org.apache.zookeeper:zookeeper-jute:3.5.8 - http://zookeeper.apache.org/zookeeper-jute)
      (The Apache License, Version 2.0) org.apiguardian:apiguardian-api (org.apiguardian:apiguardian-api:1.1.0 - https://github.com/apiguardian-team/apiguardian)
      (Apache License, Version 2.0) AssertJ fluent assertions (org.assertj:assertj-core:3.19.0 - https://assertj.github.io/doc/assertj-core/)
-     (Apache 2.0) Awaitility (org.awaitility:awaitility:4.0.3 - http://awaitility.org)
+     (Apache 2.0) Awaitility (org.awaitility:awaitility:4.1.0 - http://awaitility.org)
      (GNU General Public License, version 2 (GPL2), with the classpath exception) (The MIT License) Checker Qual (org.checkerframework:checker-compat-qual:2.5.5 - https://checkerframework.org)
      (BSD Licence 3) Hamcrest (org.hamcrest:hamcrest:2.1 - http://hamcrest.org/JavaHamcrest/)
      (New BSD License) Hamcrest Core (org.hamcrest:hamcrest-core:1.3 - https://github.com/hamcrest/JavaHamcrest/hamcrest-core)
@@ -139,8 +139,8 @@ Lists of 150 third-party dependencies.
      (The MIT License) mockito-core (org.mockito:mockito-core:3.9.0 - https://github.com/mockito/mockito)
      (The MIT License) mockito-junit-jupiter (org.mockito:mockito-junit-jupiter:3.9.0 - https://github.com/mockito/mockito)
      (Apache License, Version 2.0) Objenesis (org.objenesis:objenesis:3.2 - http://objenesis.org/objenesis)
-     (GNU General Public License (GPL), version 2, with the Classpath exception) JMH Core (org.openjdk.jmh:jmh-core:1.29 - http://openjdk.java.net/projects/code-tools/jmh/jmh-core/)
-     (GNU General Public License (GPL), version 2, with the Classpath exception) JMH Generators: Annotation Processors (org.openjdk.jmh:jmh-generator-annprocess:1.29 - http://openjdk.java.net/projects/code-tools/jmh/jmh-generator-annprocess/)
+     (GNU General Public License (GPL), version 2, with the Classpath exception) JMH Core (org.openjdk.jmh:jmh-core:1.31 - http://openjdk.java.net/projects/code-tools/jmh/jmh-core/)
+     (GNU General Public License (GPL), version 2, with the Classpath exception) JMH Generators: Annotation Processors (org.openjdk.jmh:jmh-generator-annprocess:1.31 - http://openjdk.java.net/projects/code-tools/jmh/jmh-generator-annprocess/)
      (The Apache License, Version 2.0) org.opentest4j:opentest4j (org.opentest4j:opentest4j:1.2.0 - https://github.com/ota4j-team/opentest4j)
      (Apache-2.0) Scala Library (org.scala-lang:scala-library:2.12.12 - https://www.scala-lang.org/)
      (Apache-2.0) Scala Compiler (org.scala-lang:scala-reflect:2.12.12 - https://www.scala-lang.org/)
diff --git a/data-plane/benchmarks/pom.xml b/data-plane/benchmarks/pom.xml
index eda86c52c..eb3568c01 100644
--- a/data-plane/benchmarks/pom.xml
+++ b/data-plane/benchmarks/pom.xml
@@ -32,7 +32,7 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <jmh.version>1.29</jmh.version>
+    <jmh.version>1.31</jmh.version>
     <javac.target>15</javac.target>
     <uberjar.name>benchmarks</uberjar.name>
   </properties>
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index ae4075053..9e2feef2e 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -43,16 +43,16 @@
 
     <!-- dependencies version -->
     <vertx.version>4.0.3</vertx.version>
-    <cloudevents.sdk.version>2.1.0</cloudevents.sdk.version>
+    <cloudevents.sdk.version>2.1.1</cloudevents.sdk.version>
     <micrometer.version>1.6.6</micrometer.version>
     <opentelemetry.version>0.15.0</opentelemetry.version>
     <jackson.version>2.12.3</jackson.version>
-    <protobuf.version>3.15.8</protobuf.version>
+    <protobuf.version>3.16.0</protobuf.version>
     <slf4j.version>1.7.30</slf4j.version>
     <ch.qos.logback.version>1.2.3</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>6.6</net.logstash.logback.encoder.version>
     <assertj.version>3.19.0</assertj.version>
-    <awaitility.version>4.0.3</awaitility.version>
+    <awaitility.version>4.1.0</awaitility.version>
     <junit.jupiter.version>5.7.1</junit.jupiter.version>
     <mokito.junit.jupiter.version>3.9.0</mokito.junit.jupiter.version>
     <fabric8.kubernetes.version>5.3.1</fabric8.kubernetes.version>
diff --git a/go.mod b/go.mod
index 4f6df2a4d..79d704d5a 100644
--- a/go.mod
+++ b/go.mod
@@ -21,8 +21,8 @@ require (
 	k8s.io/apiserver v0.19.7
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.22.1-0.20210504001137-8a094e6031f2
+	knative.dev/eventing v0.22.1-0.20210511131446-68934518c68f
 	knative.dev/hack v0.0.0-20210428122153-93ad9129c268
-	knative.dev/pkg v0.0.0-20210428141353-878c85083565
-	knative.dev/reconciler-test v0.0.0-20210428195854-3db87491edb3
+	knative.dev/pkg v0.0.0-20210510175900-4564797bf3b7
+	knative.dev/reconciler-test v0.0.0-20210506205310-ed3c37806817
 )
diff --git a/go.sum b/go.sum
index fde61dca1..2df69c999 100644
--- a/go.sum
+++ b/go.sum
@@ -139,7 +139,6 @@ github.com/cloudevents/sdk-go/observability/opencensus/v2 v2.4.1/go.mod h1:lhEpx
 github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2 v2.2.0 h1:PB+tnuauZWE2RnfDJnyv8wuLc1tkWk/p8mYiy2xzdvQ=
 github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2 v2.2.0/go.mod h1:XbBXL/a5TGNGs5N4UreDCIK7F71MNrXURJeVXury+XY=
 github.com/cloudevents/sdk-go/v2 v2.0.0/go.mod h1:3CTrpB4+u7Iaj6fd7E2Xvm5IxMdRoaAhqaRVnOr2rCU=
-github.com/cloudevents/sdk-go/v2 v2.3.1 h1:QRTu0yRA4FbznjRSds0/4Hy6cVYpWV2wInlNJSHWAtw=
 github.com/cloudevents/sdk-go/v2 v2.3.1/go.mod h1:4fO2UjPMYYR1/7KPJQCwTPb0lFA8zYuitkUpAZFSY1Q=
 github.com/cloudevents/sdk-go/v2 v2.4.1 h1:rZJoz9QVLbWQmnvLPDFEmv17Czu+CfSPwMO6lhJ72xQ=
 github.com/cloudevents/sdk-go/v2 v2.4.1/go.mod h1:MZiMwmAh5tGj+fPFvtHv9hKurKqXtdB9haJYMJ/7GJY=
@@ -318,7 +317,6 @@ github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:W
 github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
 github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
-github.com/golang/protobuf v1.4.3 h1:JjCZWpVbqXDqFVmTfYWEVTMIYrL/NPdPSCHPJ0T/raM=
 github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
@@ -572,7 +570,6 @@ github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+W
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.10.2/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.11.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
-github.com/onsi/ginkgo v1.12.1 h1:mFwc4LvZ0xpSvDZ3E+k8Yte0hLOMxXUlP+yXtJqkYfQ=
 github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
 github.com/onsi/ginkgo v1.14.2 h1:8mVmC9kjFFmA8H4pKMUhcblgifdkOIXPvbhN1T36q1M=
 github.com/onsi/ginkgo v1.14.2/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
@@ -581,7 +578,6 @@ github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1Cpa
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
-github.com/onsi/gomega v1.10.2 h1:aY/nuoWlKJud2J6U0E3NWsjlg+0GtwXxgEqthRdzlcs=
 github.com/onsi/gomega v1.10.2/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.10.4 h1:NiTx7EEvBzu9sFOD1zORteLSt3o8gnlvZZwSE9TnY9U=
 github.com/onsi/gomega v1.10.4/go.mod h1:g/HbgYopi++010VEqkFgJHKC09uJiW9UkXvMUuKHUCQ=
@@ -837,7 +833,6 @@ golang.org/x/lint v0.0.0-20190909230951-414d861bb4ac/go.mod h1:6SW0HCj/g11FgYtHl
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f/go.mod h1:5qLYkcX4OjUUV8bRuDixDT3tpyyb+LUpUlRWLxfhWrs=
 golang.org/x/lint v0.0.0-20200130185559-910be7a94367/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
-golang.org/x/lint v0.0.0-20200302205851-738671d3881b h1:Wh+f8QHJXR411sJR8/vRBTZ7YapZaRvUcLFFJhusH0k=
 golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
 golang.org/x/lint v0.0.0-20201208152925-83fdc39ff7b5 h1:2M3HP5CCK1Si9FQhwnzYhXdG6DXeebvUHFpre8QvbyI=
 golang.org/x/lint v0.0.0-20201208152925-83fdc39ff7b5/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
@@ -897,7 +892,6 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY
 golang.org/x/net v0.0.0-20201031054903-ff519b6c9102/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20210119194325-5f4716e94777 h1:003p0dJM77cxMSyCPFphvZf/Y5/NXf5fzg6ufd1/Oew=
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
@@ -921,7 +915,6 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201207232520-09787c993a3a h1:DcqTD9SDLc+1P/r1EmRBwnVsrOwW+kk2vWf9n+1sGhs=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -996,7 +989,6 @@ golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.5 h1:i6eZZ+zk0SOf0xgBpEpPD18qWcJda6q1sxt3S0kzyUQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -1173,7 +1165,6 @@ google.golang.org/grpc v1.30.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM
 google.golang.org/grpc v1.31.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.31.1/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.36.1 h1:cmUfbeGKnz9+2DD/UYsMQXeqbHZqZDs4eQwW0sFOpBY=
 google.golang.org/grpc v1.36.1/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
 google.golang.org/grpc v1.37.0 h1:uSZWeQJX5j11bIQ4AJoj+McDBo29cY1MCoC1wO3ts+c=
 google.golang.org/grpc v1.37.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
@@ -1186,7 +1177,6 @@ google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGjtUeSXeh4=
-google.golang.org/protobuf v1.25.0 h1:Ejskq+SyPohKW+1uil0JJMtmHCgJPJ/qWTxr8qp+R4c=
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0 h1:bxAC2xTBsZGibn2RTntX0oH50xLsqy1OxA9tTL3p/lk=
@@ -1271,19 +1261,19 @@ k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.22.1-0.20210504001137-8a094e6031f2 h1:m1Thy5RduRcPY60QRWXRHO5P2gZDex9uk7aF4rnvd4k=
-knative.dev/eventing v0.22.1-0.20210504001137-8a094e6031f2/go.mod h1:p7FKFj0dYMKI1Ji0GSKdRg2YzLMWVXABm0QHN2RtkSI=
-knative.dev/hack v0.0.0-20210325223819-b6ab329907d3 h1:km0Rrh0T9/wA2pivQm1hqSPVwgNgGCHC2WNn3GakZmE=
+knative.dev/eventing v0.22.1-0.20210511131446-68934518c68f h1:3bYpG0sg4qjVilSFOlt5BBGR0czbeU4ApkxtEdilOB8=
+knative.dev/eventing v0.22.1-0.20210511131446-68934518c68f/go.mod h1:V2vHWE8p2p6sokoNVXafQLiWGJ7hQpw0IzI+w5RqOaA=
 knative.dev/hack v0.0.0-20210325223819-b6ab329907d3/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20210428122153-93ad9129c268 h1:lBIj9Epd9UQ55NEaHzAdY/UZbuaegCdGPKVC2+Z68Q0=
 knative.dev/hack v0.0.0-20210428122153-93ad9129c268/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210428122153-93ad9129c268/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/pkg v0.0.0-20210420053235-1afd04993622 h1:wSyDPp/LuOLeDCpvUHgKXqb4DfmCEPelsaYzC0Fojm0=
 knative.dev/pkg v0.0.0-20210420053235-1afd04993622/go.mod h1:UtcSLHy2XIz5blWoPTA40F87zk4O7erxkCwv+7Tsmws=
-knative.dev/pkg v0.0.0-20210428141353-878c85083565 h1:4I8Pm2IlSJbdJ1R9fC18kOlZlfCZkB59JraRGratgnY=
 knative.dev/pkg v0.0.0-20210428141353-878c85083565/go.mod h1:fIl4l4OmZodkElyaHoT0LCF5wT+3+P/kinawQ4XlLtE=
-knative.dev/reconciler-test v0.0.0-20210428195854-3db87491edb3 h1:nYSGqYmXe6zIGX5uzYrAl73fn5yPfxtQNg5KIIDu/gE=
+knative.dev/pkg v0.0.0-20210510175900-4564797bf3b7 h1:i4P8emOPrLctmbaPHp5eRIOqz+XTOkit7KgZeS+onKs=
+knative.dev/pkg v0.0.0-20210510175900-4564797bf3b7/go.mod h1:fIl4l4OmZodkElyaHoT0LCF5wT+3+P/kinawQ4XlLtE=
 knative.dev/reconciler-test v0.0.0-20210428195854-3db87491edb3/go.mod h1:yZihS1XoBt7oxU6Jq+U2hMKmUvfKFEaj3vMqOMBt/tI=
+knative.dev/reconciler-test v0.0.0-20210506205310-ed3c37806817 h1:St9fyMUnKmumy+5jlawD/lfdBSnAVY/8bgoqOUdfFHs=
+knative.dev/reconciler-test v0.0.0-20210506205310-ed3c37806817/go.mod h1:pUozU+xgCWG9eH+4bYBXM0zWPl8lwvan34gCi/ZDP8o=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
 modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1/channelable_types.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1/channelable_types.go
index b65b1ab64..2d03a3f87 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1/channelable_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1/channelable_types.go
@@ -59,6 +59,7 @@ type ChannelableStatus struct {
 	// * Conditions - the latest available observations of a resource's current state.
 	duckv1.Status `json:",inline"`
 	// AddressStatus is the part where the Channelable fulfills the Addressable contract.
+	// +optional
 	duckv1.AddressStatus `json:",inline"`
 	// Subscribers is populated with the statuses of each of the Channelable's subscribers.
 	SubscribableStatus `json:",inline"`
diff --git a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/broker_types.go b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/broker_types.go
index f968fdc7e..d03749bd1 100644
--- a/vendor/knative.dev/eventing/pkg/apis/eventing/v1/broker_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/eventing/v1/broker_types.go
@@ -90,6 +90,7 @@ type BrokerStatus struct {
 
 	// Broker is Addressable. It exposes the endpoint as an URI to get events
 	// delivered into the Broker mesh.
+	// +optional
 	Address duckv1.Addressable `json:"address,omitempty"`
 }
 
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/channel_types.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/channel_types.go
index f541348fd..5df5d71b6 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/channel_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/channel_types.go
@@ -29,7 +29,8 @@ import (
 // +genreconciler
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
-// Channel represents a generic Channel. It is normally used when we want a Channel, but don't need a specific Channel implementation.
+// Channel represents a generic Channel. It is normally used when we want a
+// Channel, but do not need a specific Channel implementation.
 type Channel struct {
 	metav1.TypeMeta `json:",inline"`
 	// +optional
@@ -38,8 +39,8 @@ type Channel struct {
 	// Spec defines the desired state of the Channel.
 	Spec ChannelSpec `json:"spec,omitempty"`
 
-	// Status represents the current state of the Channel. This data may be out of
-	// date.
+	// Status represents the current state of the Channel. This data may be out
+	// of date.
 	// +optional
 	Status ChannelStatus `json:"status,omitempty"`
 }
@@ -61,12 +62,15 @@ var (
 	_ duckv1.KRShaped = (*Channel)(nil)
 )
 
-// ChannelSpec defines which subscribers have expressed interest in receiving events from this Channel.
-// It also defines the ChannelTemplate to use in order to create the CRD Channel backing this Channel.
+// ChannelSpec defines which subscribers have expressed interest in receiving
+// events from this Channel. It also defines the ChannelTemplate to use in
+// order to create the CRD Channel backing this Channel.
 type ChannelSpec struct {
-	// ChannelTemplate specifies which Channel CRD to use to create the CRD Channel backing this Channel.
-	// This is immutable after creation. Normally this is set by the Channel defaulter, not directly by the user.
-	ChannelTemplate *ChannelTemplateSpec `json:"channelTemplate"`
+	// ChannelTemplate specifies which Channel CRD to use to create the CRD
+	// Channel backing this Channel. This is immutable after creation.
+	// Normally this is set by the Channel defaulter, not directly by the user.
+	// +optional
+	ChannelTemplate *ChannelTemplateSpec `json:"channelTemplate,omitempty"`
 
 	// Channel conforms to ChannelableSpec
 	eventingduckv1.ChannelableSpec `json:",inline"`
@@ -78,6 +82,7 @@ type ChannelStatus struct {
 	eventingduckv1.ChannelableStatus `json:",inline"`
 
 	// Channel is an KReference to the Channel CRD backing this Channel.
+	// +optional
 	Channel *duckv1.KReference `json:"channel,omitempty"`
 }
 
@@ -91,7 +96,8 @@ type ChannelList struct {
 	Items           []Channel `json:"items"`
 }
 
-// GetStatus retrieves the status of the Channel. Implements the KRShaped interface.
+// GetStatus retrieves the status of the Channel. Implements the KRShaped
+// interface.
 func (t *Channel) GetStatus() *duckv1.Status {
 	return &t.Status.Status
 }
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/subscription_types.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/subscription_types.go
index 17304013a..187bc8fff 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/subscription_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/subscription_types.go
@@ -122,12 +122,15 @@ type SubscriptionStatus struct {
 // Subscription.
 type SubscriptionStatusPhysicalSubscription struct {
 	// SubscriberURI is the fully resolved URI for spec.subscriber.
+	// +optional
 	SubscriberURI *apis.URL `json:"subscriberUri,omitempty"`
 
 	// ReplyURI is the fully resolved URI for the spec.reply.
+	// +optional
 	ReplyURI *apis.URL `json:"replyUri,omitempty"`
 
 	// ReplyURI is the fully resolved URI for the spec.delivery.deadLetterSink.
+	// +optional
 	DeadLetterSinkURI *apis.URL `json:"deadLetterSinkUri,omitempty"`
 }
 
diff --git a/vendor/knative.dev/pkg/resolver/addressable_resolver.go b/vendor/knative.dev/pkg/resolver/addressable_resolver.go
index 4407f80be..a5701ea44 100644
--- a/vendor/knative.dev/pkg/resolver/addressable_resolver.go
+++ b/vendor/knative.dev/pkg/resolver/addressable_resolver.go
@@ -21,26 +21,27 @@ import (
 	"errors"
 	"fmt"
 
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/cache"
+	"knative.dev/pkg/client/injection/ducks/duck/v1/addressable"
+	"knative.dev/pkg/controller"
+
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
-	"k8s.io/apimachinery/pkg/types"
-
 	"knative.dev/pkg/apis"
 	pkgapisduck "knative.dev/pkg/apis/duck"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
-	"knative.dev/pkg/controller"
 	"knative.dev/pkg/network"
 	"knative.dev/pkg/tracker"
-
-	"knative.dev/pkg/client/injection/ducks/duck/v1/addressable"
 )
 
 // URIResolver resolves Destinations and ObjectReferences into a URI.
 type URIResolver struct {
-	tracker         tracker.Interface
-	informerFactory pkgapisduck.InformerFactory
+	tracker       tracker.Interface
+	listerFactory func(schema.GroupVersionResource) (cache.GenericLister, error)
 }
 
 // NewURIResolver constructs a new URIResolver with context and a callback
@@ -49,13 +50,19 @@ func NewURIResolver(ctx context.Context, callback func(types.NamespacedName)) *U
 	ret := &URIResolver{}
 
 	ret.tracker = tracker.New(callback, controller.GetTrackerLease(ctx))
-	ret.informerFactory = &pkgapisduck.CachedInformerFactory{
+
+	informerFactory := &pkgapisduck.CachedInformerFactory{
 		Delegate: &pkgapisduck.EnqueueInformerFactory{
 			Delegate:     addressable.Get(ctx),
 			EventHandler: controller.HandleAll(ret.tracker.OnChanged),
 		},
 	}
 
+	ret.listerFactory = func(gvr schema.GroupVersionResource) (cache.GenericLister, error) {
+		_, l, err := informerFactory.Get(ctx, gvr)
+		return l, err
+	}
+
 	return ret
 }
 
@@ -163,7 +170,7 @@ func (r *URIResolver) URIFromObjectReference(ctx context.Context, ref *corev1.Ob
 		return url, nil
 	}
 
-	_, lister, err := r.informerFactory.Get(ctx, gvr)
+	lister, err := r.listerFactory(gvr)
 	if err != nil {
 		return nil, apierrs.NewNotFound(gvr.GroupResource(), "Lister")
 	}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 96e00e96b..0784f46ed 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1015,7 +1015,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.22.1-0.20210504001137-8a094e6031f2
+# knative.dev/eventing v0.22.1-0.20210511131446-68934518c68f
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1117,7 +1117,7 @@ knative.dev/eventing/test/test_images/request-sender
 # knative.dev/hack v0.0.0-20210428122153-93ad9129c268
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20210428141353-878c85083565
+# knative.dev/pkg v0.0.0-20210510175900-4564797bf3b7
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1206,7 +1206,7 @@ knative.dev/pkg/webhook/certificates/resources
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20210428195854-3db87491edb3
+# knative.dev/reconciler-test v0.0.0-20210506205310-ed3c37806817
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/.github/workflows/knative-go-test.yaml b/.github/workflows/knative-go-test.yaml
index 965df1e74..79432a7a3 100644
--- a/.github/workflows/knative-go-test.yaml
+++ b/.github/workflows/knative-go-test.yaml
@@ -31,7 +31,7 @@ jobs:
     name: Unit Tests
     strategy:
       matrix:
-        go-version: [1.15.x]
+        go-version: [1.16.x]
         platform: [ubuntu-latest]
 
     runs-on: ${{ matrix.platform }}
diff --git a/control-plane/pkg/kafka/consumer_group_lag_test.go b/control-plane/pkg/kafka/consumer_group_lag_test.go
new file mode 100644
index 000000000..e2b5526a6
--- /dev/null
+++ b/control-plane/pkg/kafka/consumer_group_lag_test.go
@@ -0,0 +1,490 @@
+/*
+ * 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 kafka
+
+import (
+	"io"
+	"io/ioutil"
+	"math"
+	"net"
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+func TestConsumerGroupLagProvider(t *testing.T) {
+
+	const group = "group-1"
+
+	brokerAddr := "localhost:43245"
+	broker, closeFunc := fakeKafkaBrokerListener(t, brokerAddr)
+	defer closeFunc()
+
+	tt := []struct {
+		name    string
+		client  *saramaClientMock
+		wantLag uint64
+	}{
+		{
+			name: "No Lag",
+			client: &saramaClientMock{
+				topics: []string{"topic-1"},
+				partitionsByTopic: map[string][]Partition{
+					"topic-1": {
+						{ID: 0, ConsumerOffset: 10, LatestOffset: 10},
+						{ID: 1, ConsumerOffset: 0, LatestOffset: 0},
+						{ID: 2, ConsumerOffset: 3, LatestOffset: 3},
+						{ID: 3, ConsumerOffset: 20, LatestOffset: 20},
+						{ID: 4, ConsumerOffset: 12, LatestOffset: 12},
+					},
+				},
+				groups: []string{group},
+				broker: broker,
+			},
+		},
+		{
+			name: "Single partition Lag",
+			client: &saramaClientMock{
+				topics: []string{"topic-1", "topic-2", "topic-3"},
+				partitionsByTopic: map[string][]Partition{
+					"topic-1": {
+						{ID: 0, ConsumerOffset: 10, LatestOffset: 10},
+						{ID: 1, ConsumerOffset: 0, LatestOffset: 3},
+						{ID: 2, ConsumerOffset: 3, LatestOffset: 3},
+						{ID: 3, ConsumerOffset: 20, LatestOffset: 20},
+						{ID: 4, ConsumerOffset: 12, LatestOffset: 12},
+					},
+					"topic-2": {
+						{ID: 0, ConsumerOffset: 10, LatestOffset: 10},
+						{ID: 1, ConsumerOffset: 3, LatestOffset: 3},
+						{ID: 2, ConsumerOffset: 3, LatestOffset: 3},
+						{ID: 3, ConsumerOffset: 20, LatestOffset: 20},
+						{ID: 4, ConsumerOffset: 1, LatestOffset: 12},
+					},
+					"topic-3": {
+						{ID: 0, ConsumerOffset: 1, LatestOffset: 7},
+						{ID: 1, ConsumerOffset: 3, LatestOffset: 3},
+						{ID: 2, ConsumerOffset: 3, LatestOffset: 3},
+						{ID: 3, ConsumerOffset: 20, LatestOffset: 20},
+						{ID: 4, ConsumerOffset: 12, LatestOffset: 12},
+					},
+				},
+				groups: []string{group},
+				broker: broker,
+			},
+		},
+		{
+			name: "Partitions Lag",
+			client: &saramaClientMock{
+				topics: []string{"topic-1", "topic-2", "topic-3", "topic-4"},
+				partitionsByTopic: map[string][]Partition{
+					"topic-1": {
+						{ID: 0, ConsumerOffset: 7, LatestOffset: 10},
+						{ID: 1, ConsumerOffset: 0, LatestOffset: 3},
+						{ID: 2, ConsumerOffset: 2, LatestOffset: 3},
+						{ID: 3, ConsumerOffset: 20, LatestOffset: 20},
+						{ID: 4, ConsumerOffset: 1, LatestOffset: 12},
+					},
+					"topic-2": {
+						{ID: 0, ConsumerOffset: 7, LatestOffset: 10},
+					},
+					"topic-3": {
+						{ID: 0, ConsumerOffset: 0, LatestOffset: 10},
+						{ID: 1, ConsumerOffset: 0, LatestOffset: 3},
+						{ID: 2, ConsumerOffset: 0, LatestOffset: 3},
+						{ID: 3, ConsumerOffset: 0, LatestOffset: 20},
+						{ID: 4, ConsumerOffset: 0, LatestOffset: 12},
+					},
+					"topic-4": {
+						{ID: 0, ConsumerOffset: 0, LatestOffset: math.MaxInt64},
+						{ID: 1, ConsumerOffset: 0, LatestOffset: math.MaxInt64},
+						{ID: 2, ConsumerOffset: 0, LatestOffset: math.MaxInt64},
+						{ID: 3, ConsumerOffset: 0, LatestOffset: math.MaxInt64},
+						{ID: 4, ConsumerOffset: 0, LatestOffset: math.MaxInt64},
+					},
+				},
+				groups: []string{group},
+				broker: broker,
+			},
+		},
+	}
+
+	t.Parallel()
+	for _, tc := range tt {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			tc.client.t = t
+
+			for _, topic := range tc.client.topics {
+				t.Run(topic, func(t *testing.T) {
+					provider := NewConsumerGroupLagProvider(tc.client, func(sarama.Client) (sarama.ClusterAdmin, error) { return tc.client, nil }, -100)
+
+					lag, err := provider.GetLag(topic, group)
+					require.Nil(t, err)
+
+					wantLag := uint64(0)
+					for _, p := range tc.client.partitionsByTopic[topic] {
+						wantLag += uint64(p.LatestOffset - p.ConsumerOffset)
+					}
+
+					require.Equal(t, wantLag, lag.Total())
+
+					err = provider.Close()
+					require.Nil(t, err)
+					require.True(t, tc.client.closed)
+				})
+			}
+		})
+	}
+
+}
+
+func TestConsumerGroupLagTotal(t *testing.T) {
+	tests := []struct {
+		name string
+		cgl  ConsumerGroupLag
+		want uint64
+	}{
+		{
+			name: "empty",
+			cgl: ConsumerGroupLag{
+				Topic:       "topic",
+				ByPartition: []PartitionLag{},
+			},
+			want: 0,
+		},
+		{
+			name: "consumer group lags",
+			cgl: ConsumerGroupLag{
+				Topic: "topic",
+				ByPartition: []PartitionLag{
+					{LatestOffset: 23, ConsumerOffset: 3},
+					{LatestOffset: 22, ConsumerOffset: 1},
+					{LatestOffset: 22, ConsumerOffset: 0},
+					{LatestOffset: 2, ConsumerOffset: 1},
+				},
+			},
+			want: 20 + 21 + 22 + 1,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.cgl.Total(); got != tt.want {
+				t.Errorf("Total() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestConsumerGroupLagString(t *testing.T) {
+	cgl := ConsumerGroupLag{}
+	require.NotEmpty(t, cgl.String())
+}
+
+type Partition struct {
+	ConsumerOffset int64
+	LatestOffset   int64
+	ID             int32
+}
+
+type saramaClientMock struct {
+	t                 *testing.T
+	topics            []string
+	groups            []string
+	partitionsByTopic map[string][]Partition
+	broker            *sarama.Broker
+	closed            bool
+}
+
+func (s saramaClientMock) CreateTopic(topic string, _ *sarama.TopicDetail, _ bool) error {
+	s.hasTopic(topic)
+	return nil
+}
+
+func (s saramaClientMock) ListTopics() (map[string]sarama.TopicDetail, error) {
+	m := make(map[string]sarama.TopicDetail, len(s.topics))
+	for _, t := range s.topics {
+		m[t] = sarama.TopicDetail{
+			NumPartitions:     int32(len(s.partitionsByTopic[t])),
+			ReplicationFactor: 1,
+			ReplicaAssignment: map[int32][]int32{},
+			ConfigEntries:     map[string]*string{},
+		}
+	}
+	return m, nil
+}
+
+func (s saramaClientMock) DescribeTopics(topics []string) (metadata []*sarama.TopicMetadata, err error) {
+	for _, t := range topics {
+		s.hasTopic(t)
+		partitions := make([]*sarama.PartitionMetadata, 0, len(s.partitionsByTopic[t]))
+		for _, p := range s.partitionsByTopic[t] {
+			partitions = append(partitions, &sarama.PartitionMetadata{ID: p.ID})
+		}
+		metadata = append(metadata, &sarama.TopicMetadata{
+			Err:        0,
+			Name:       t,
+			IsInternal: false,
+			Partitions: partitions,
+		})
+	}
+	return
+}
+
+func (s saramaClientMock) DeleteTopic(topic string) error {
+	s.hasTopic(topic)
+	return nil
+}
+
+func (s saramaClientMock) CreatePartitions(topic string, _ int32, _ [][]int32, _ bool) error {
+	s.hasTopic(topic)
+	return nil
+}
+
+func (s saramaClientMock) AlterPartitionReassignments(topic string, _ [][]int32) error {
+	s.hasTopic(topic)
+	return nil
+}
+
+func (s saramaClientMock) ListPartitionReassignments(topic string, _ []int32) (topicStatus map[string]map[int32]*sarama.PartitionReplicaReassignmentsStatus, err error) {
+	s.hasTopic(topic)
+	return
+}
+
+func (s saramaClientMock) DeleteRecords(topic string, _ map[int32]int64) error {
+	s.hasTopic(topic)
+	return nil
+}
+
+func (s saramaClientMock) DescribeConfig(sarama.ConfigResource) ([]sarama.ConfigEntry, error) {
+	panic("implement me")
+}
+
+func (s saramaClientMock) AlterConfig(sarama.ConfigResourceType, string, map[string]*string, bool) error {
+	panic("implement me")
+}
+
+func (s saramaClientMock) CreateACL(sarama.Resource, sarama.Acl) error {
+	panic("implement me")
+}
+
+func (s saramaClientMock) ListAcls(sarama.AclFilter) ([]sarama.ResourceAcls, error) {
+	panic("implement me")
+}
+
+func (s saramaClientMock) DeleteACL(sarama.AclFilter, bool) ([]sarama.MatchingAcl, error) {
+	panic("implement me")
+}
+
+func (s saramaClientMock) ListConsumerGroups() (map[string]string, error) {
+	panic("implement me")
+}
+
+func (s saramaClientMock) DescribeConsumerGroups([]string) ([]*sarama.GroupDescription, error) {
+	panic("implement me")
+}
+
+func (s saramaClientMock) ListConsumerGroupOffsets(group string, topicPartitions map[string][]int32) (*sarama.OffsetFetchResponse, error) {
+	s.hasGroup(group)
+
+	s.hasTopicPartitions(topicPartitions)
+
+	response := &sarama.OffsetFetchResponse{
+		Version: 2,
+		Blocks:  map[string]map[int32]*sarama.OffsetFetchResponseBlock{},
+	}
+	for _, t := range s.topics {
+		response.Blocks[t] = map[int32]*sarama.OffsetFetchResponseBlock{}
+		for _, p := range s.partitionsByTopic[t] {
+			response.Blocks[t][p.ID] = &sarama.OffsetFetchResponseBlock{Offset: p.ConsumerOffset}
+		}
+	}
+	return response, nil
+}
+
+func (s saramaClientMock) hasTopicPartitions(topicPartitions map[string][]int32) {
+	for topic, partitions := range topicPartitions {
+		s.hasTopic(topic)
+		for _, p := range partitions {
+			s.hasPartition(topic, p)
+		}
+	}
+}
+
+func (s saramaClientMock) DeleteConsumerGroup(group string) error {
+	s.hasGroup(group)
+	return nil
+}
+
+func (s saramaClientMock) DescribeCluster() (brokers []*sarama.Broker, controllerID int32, err error) {
+	panic("implement me")
+}
+
+func (s saramaClientMock) DescribeLogDirs([]int32) (map[int32][]sarama.DescribeLogDirsResponseDirMetadata, error) {
+	panic("implement me")
+}
+
+func (s saramaClientMock) Config() *sarama.Config {
+	return sarama.NewConfig()
+}
+
+func (s saramaClientMock) Controller() (*sarama.Broker, error) {
+	return s.broker, nil
+}
+
+func (s saramaClientMock) RefreshController() (*sarama.Broker, error) {
+	return s.broker, nil
+}
+
+func (s saramaClientMock) Brokers() []*sarama.Broker {
+	return []*sarama.Broker{s.broker}
+}
+
+func (s saramaClientMock) Broker(int32) (*sarama.Broker, error) {
+	return s.broker, nil
+}
+
+func (s saramaClientMock) Topics() ([]string, error) {
+	return s.topics, nil
+}
+
+func (s saramaClientMock) Partitions(topic string) ([]int32, error) {
+	s.hasTopic(topic)
+	return s.getPartitions(topic), nil
+}
+
+func (s saramaClientMock) WritablePartitions(topic string) ([]int32, error) {
+	s.hasTopic(topic)
+	return s.getPartitions(topic), nil
+}
+
+func (s saramaClientMock) Leader(topic string, partitionID int32) (*sarama.Broker, error) {
+	s.hasTopic(topic)
+	s.hasPartition(topic, partitionID)
+	return s.broker, nil
+}
+
+func (s saramaClientMock) Replicas(topic string, partitionID int32) ([]int32, error) {
+	s.hasTopic(topic)
+	s.hasPartition(topic, partitionID)
+	panic("implement me")
+}
+
+func (s saramaClientMock) hasPartition(topic string, partitionID int32) {
+	require.True(s.t, sets.NewInt32(s.getPartitions(topic)...).Has(partitionID))
+}
+
+func (s saramaClientMock) hasTopic(topic string) {
+	require.True(s.t, sets.NewString(s.topics...).Has(topic))
+}
+
+func (s saramaClientMock) hasGroup(group string) {
+	require.True(s.t, sets.NewString(s.groups...).Has(group))
+}
+
+func (s saramaClientMock) InSyncReplicas(topic string, partitionID int32) ([]int32, error) {
+	s.hasTopic(topic)
+	s.hasPartition(topic, partitionID)
+	panic("implement me")
+}
+
+func (s saramaClientMock) OfflineReplicas(topic string, partitionID int32) ([]int32, error) {
+	s.hasTopic(topic)
+	s.hasPartition(topic, partitionID)
+	panic("implement me")
+}
+
+func (s saramaClientMock) RefreshBrokers([]string) error {
+	return nil
+}
+
+func (s saramaClientMock) RefreshMetadata(topics ...string) error {
+	for _, t := range topics {
+		s.hasTopic(t)
+	}
+	return nil
+}
+
+func (s saramaClientMock) GetOffset(topic string, partitionID int32, time int64) (int64, error) {
+	require.Equal(s.t, sarama.OffsetNewest, time)
+	s.hasTopic(topic)
+	s.hasPartition(topic, partitionID)
+
+	for _, p := range s.partitionsByTopic[topic] {
+		if p.ID == partitionID {
+			return p.LatestOffset, nil
+		}
+	}
+	return -1, nil
+}
+
+func (s saramaClientMock) Coordinator(string) (*sarama.Broker, error) {
+	return s.broker, nil
+}
+
+func (s saramaClientMock) RefreshCoordinator(string) error {
+	return nil
+}
+
+func (s saramaClientMock) InitProducerID() (*sarama.InitProducerIDResponse, error) {
+	panic("implement me")
+}
+
+func (s *saramaClientMock) Close() error {
+	if !s.closed {
+		s.closed = true
+	}
+	return nil
+}
+
+func (s saramaClientMock) Closed() bool {
+	return s.closed
+}
+
+func (s saramaClientMock) getPartitions(topic string) []int32 {
+	partitions := make([]int32, 0, len(s.partitionsByTopic[topic]))
+	for _, p := range s.partitionsByTopic[topic] {
+		partitions = append(partitions, p.ID)
+	}
+	return partitions
+}
+
+func fakeKafkaBrokerListener(t *testing.T, addr string) (*sarama.Broker, func()) {
+	// Fake Kafka broker
+	listener, err := net.Listen("tcp", addr)
+	require.Nil(t, err)
+
+	go func() {
+		conn, err := listener.Accept()
+		if err != nil {
+			return
+		}
+		_, _ = io.Copy(ioutil.Discard, conn)
+	}()
+
+	broker := sarama.NewBroker(addr)
+	err = broker.Open(sarama.NewConfig())
+	require.Nil(t, err)
+
+	return broker, func() {
+		_ = broker.Close()
+		_ = listener.Close()
+	}
+}
diff --git a/test/e2e/consumer_group_lag_provider_test.go b/test/e2e/consumer_group_lag_provider_test.go
new file mode 100644
index 000000000..09410bd9c
--- /dev/null
+++ b/test/e2e/consumer_group_lag_provider_test.go
@@ -0,0 +1,57 @@
+// +build e2e
+
+/*
+ * 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 e2e
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/types"
+	testlib "knative.dev/eventing/test/lib"
+
+	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
+	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
+)
+
+func TestConsumerGroupLagProvider(t *testing.T) {
+	t.Skip("Only useful for debugging the consumer group lag provider control-plane/pkg/kafka/consumer_group_lag.go")
+
+	const (
+		consumerGroupLagJobName = "consumer-group-lag-provider-test"
+	)
+
+	client := testlib.Setup(t, false)
+	defer testlib.TearDown(client)
+
+	ctx := context.Background()
+
+	err := kafkatest.VerifyConsumerGroupLag(
+		client.Kube,
+		client.Tracker,
+		types.NamespacedName{
+			Namespace: client.Namespace,
+			Name:      consumerGroupLagJobName,
+		},
+	)
+
+	pkgtesting.LogJobOutput(t, ctx, client.Kube, client.Namespace, consumerGroupLagJobName)
+
+	require.Nil(t, err)
+}
diff --git a/test/e2e/sacura_test.go b/test/e2e/sacura_test.go
index 4c76e4909..48c9621a9 100644
--- a/test/e2e/sacura_test.go
+++ b/test/e2e/sacura_test.go
@@ -26,17 +26,23 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	batchv1 "k8s.io/api/batch/v1"
-	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	testlib "knative.dev/eventing/test/lib"
+
+	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
+	pkgtesting "knative.dev/eventing-kafka-broker/test/pkg/testing"
 )
 
 const (
-	app       = "sacura"
-	namespace = "sacura"
+	app                            = "sacura"
+	namespace                      = "sacura"
+	sacuraVerifyCommittedOffsetJob = "verify-committed-offset"
+	sacuraTriggerName              = "trigger"
+	sacuraTopic                    = "knative-broker-sacura-sacura"
 
 	pollTimeout  = 30 * time.Minute
 	pollInterval = 10 * time.Second
@@ -56,11 +62,35 @@ func TestSacuraJob(t *testing.T) {
 		return isJobSucceeded(job)
 	})
 
-	logJobOutput(t, app, c, namespace, ctx)
+	pkgtesting.LogJobOutput(t, ctx, c.Kube, namespace, app)
 
 	if jobPollError != nil {
 		t.Fatal(jobPollError)
 	}
+
+	t.Log(strings.Repeat("-", 30))
+	t.Log("Verify committed offset")
+	t.Log(strings.Repeat("-", 30))
+
+	trigger, err := c.Eventing.EventingV1().Triggers(namespace).Get(ctx, sacuraTriggerName, metav1.GetOptions{})
+	require.Nil(t, err, "Failed to get trigger %s/%s: %v", namespace, sacuraTriggerName)
+
+	t.Run("verify committed offset", func(t *testing.T) {
+		err = kafkatest.VerifyCommittedOffset(
+			c.Kube,
+			c.Tracker,
+			types.NamespacedName{
+				Namespace: namespace,
+				Name:      sacuraVerifyCommittedOffsetJob,
+			},
+			&kafkatest.AdminConfig{
+				BootstrapServers: pkgtesting.BootstrapServersPlaintext,
+				Topic:            sacuraTopic,
+				Group:            string(trigger.UID),
+			},
+		)
+		require.Nil(t, err, "Failed to verify committed offset")
+	})
 }
 
 func isJobSucceeded(job *batchv1.Job) (bool, error) {
@@ -70,24 +100,3 @@ func isJobSucceeded(job *batchv1.Job) (bool, error) {
 
 	return job.Status.Succeeded > 0, nil
 }
-
-func logJobOutput(t *testing.T, app string, c *testlib.Client, namespace string, ctx context.Context) {
-
-	selector := labels.SelectorFromSet(map[string]string{"app": app})
-
-	pods, err := c.Kube.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector.String()})
-	assert.Nil(t, err)
-	assert.Greater(t, len(pods.Items), 0)
-
-	logs := make([]string, 0, len(pods.Items))
-	for _, pod := range pods.Items {
-		if l, err := c.Kube.CoreV1().Pods(namespace).GetLogs(pod.Name, &corev1.PodLogOptions{}).DoRaw(ctx); err != nil {
-			logs = append(logs, err.Error())
-		} else {
-			logs = append(logs, string(l))
-		}
-	}
-
-	t.Log(strings.Join(logs, "\n"))
-	testlib.ExportLogs(namespace, namespace)
-}
diff --git a/test/e2e_new/trigger_finalizer_test.go b/test/e2e_new/trigger_finalizer_test.go
index b8cd3f255..5ef060171 100644
--- a/test/e2e_new/trigger_finalizer_test.go
+++ b/test/e2e_new/trigger_finalizer_test.go
@@ -37,6 +37,7 @@ import (
 
 func TestTriggerNoFinalizerOnBrokerNotFound(t *testing.T) {
 
+	t.Skip("Flaky https://github.com/knative-sandbox/eventing-kafka-broker/issues/885")
 	t.Parallel()
 
 	ctx, env := global.Environment(
@@ -46,6 +47,8 @@ func TestTriggerNoFinalizerOnBrokerNotFound(t *testing.T) {
 		k8s.WithEventListener,
 	)
 
+	t.Logf("Namespace is %s", env.Namespace())
+
 	env.Test(ctx, t, triggerNoFinalizerOnBrokerNotFound())
 }
 
diff --git a/test/pkg/kafka/util.go b/test/pkg/kafka/util.go
new file mode 100644
index 000000000..7e114d14d
--- /dev/null
+++ b/test/pkg/kafka/util.go
@@ -0,0 +1,57 @@
+/*
+ * 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 kafka
+
+import (
+	"context"
+	"fmt"
+
+	batchv1 "k8s.io/api/batch/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	testlib "knative.dev/eventing/test/lib"
+)
+
+func verifyJobSucceeded(
+	ctx context.Context,
+	client kubernetes.Interface,
+	tracker *testlib.Tracker,
+	namespacedName types.NamespacedName,
+	job *batchv1.Job) error {
+
+	job, err := client.BatchV1().Jobs(namespacedName.Namespace).Create(ctx, job, metav1.CreateOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to create job: %w", err)
+	}
+
+	gvr, _ := meta.UnsafeGuessKindToResource(job.GroupVersionKind())
+	tracker.Add(gvr.Group, gvr.Version, gvr.Resource, namespacedName.Namespace, namespacedName.Name)
+
+	return wait.PollImmediate(interval, timeout, func() (bool, error) {
+		job, err := client.BatchV1().Jobs(namespacedName.Namespace).Get(ctx, namespacedName.Name, metav1.GetOptions{})
+		if err != nil {
+			return false, fmt.Errorf("failed to get job: %w", err)
+		}
+		if job.Status.Succeeded >= 1 {
+			return true, nil
+		}
+		return false, nil
+	})
+}
diff --git a/test/pkg/kafka/verify_committed_offset.go b/test/pkg/kafka/verify_committed_offset.go
new file mode 100644
index 000000000..ab23f9690
--- /dev/null
+++ b/test/pkg/kafka/verify_committed_offset.go
@@ -0,0 +1,94 @@
+/*
+ * 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 kafka
+
+import (
+	"context"
+
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/utils/pointer"
+	testlib "knative.dev/eventing/test/lib"
+	pkgtest "knative.dev/pkg/test"
+)
+
+const (
+	committedOffsetImage = "committed-offset"
+)
+
+type AdminConfig struct {
+	BootstrapServers string `json:"bootstrapServers" required:"true" split_words:"true"`
+	Topic            string `json:"topic" required:"true" split_words:"true"`
+	Group            string `json:"group" required:"true" split_words:"true"`
+}
+
+func VerifyCommittedOffset(
+	client kubernetes.Interface,
+	tracker *testlib.Tracker,
+	namespacedName types.NamespacedName,
+	config *AdminConfig) error {
+
+	ctx := context.Background()
+
+	job := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespacedName.Namespace,
+			Name:      namespacedName.Name,
+			Labels: map[string]string{
+				"app": namespacedName.Name,
+			},
+		},
+		Spec: batchv1.JobSpec{
+			BackoffLimit: pointer.Int32Ptr(2),
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"app": namespacedName.Name,
+					},
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:            namespacedName.Name,
+							Image:           pkgtest.ImagePath(committedOffsetImage),
+							ImagePullPolicy: corev1.PullIfNotPresent,
+							Env: []corev1.EnvVar{
+								{
+									Name:  "BOOTSTRAP_SERVERS",
+									Value: config.BootstrapServers,
+								},
+								{
+									Name:  "TOPIC",
+									Value: config.Topic,
+								},
+								{
+									Name:  "GROUP",
+									Value: config.Group,
+								},
+							},
+						},
+					},
+					RestartPolicy: "Never",
+				},
+			},
+		},
+	}
+	return verifyJobSucceeded(ctx, client, tracker, namespacedName, job)
+}
diff --git a/test/pkg/kafka/verify_consumer_group_lag.go b/test/pkg/kafka/verify_consumer_group_lag.go
new file mode 100644
index 000000000..71ab0727e
--- /dev/null
+++ b/test/pkg/kafka/verify_consumer_group_lag.go
@@ -0,0 +1,73 @@
+/*
+ * 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 kafka
+
+import (
+	"context"
+
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/utils/pointer"
+	testlib "knative.dev/eventing/test/lib"
+	pkgtest "knative.dev/pkg/test"
+)
+
+const (
+	consumerGroupLagImage = "consumer-group-lag-provider-test"
+)
+
+func VerifyConsumerGroupLag(
+	client kubernetes.Interface,
+	tracker *testlib.Tracker,
+	namespacedName types.NamespacedName) error {
+
+	ctx := context.Background()
+
+	job := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespacedName.Namespace,
+			Name:      namespacedName.Name,
+			Labels: map[string]string{
+				"app": namespacedName.Name,
+			},
+		},
+		Spec: batchv1.JobSpec{
+			BackoffLimit: pointer.Int32Ptr(0),
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"app": namespacedName.Name,
+					},
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:            namespacedName.Name,
+							Image:           pkgtest.ImagePath(consumerGroupLagImage),
+							ImagePullPolicy: corev1.PullIfNotPresent,
+						},
+					},
+					RestartPolicy: "Never",
+				},
+			},
+		},
+	}
+	return verifyJobSucceeded(ctx, client, tracker, namespacedName, job)
+}
diff --git a/test/pkg/kafka/verify_messages_topic.go b/test/pkg/kafka/verify_messages_topic.go
index d137d2ab3..88809ac5f 100644
--- a/test/pkg/kafka/verify_messages_topic.go
+++ b/test/pkg/kafka/verify_messages_topic.go
@@ -18,14 +18,11 @@ package kafka
 
 import (
 	"context"
-	"fmt"
 
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/pointer"
 	testlib "knative.dev/eventing/test/lib"
@@ -90,23 +87,5 @@ func VerifyMessagesInTopic(
 			},
 		},
 	}
-	job, err := client.BatchV1().Jobs(namespacedName.Namespace).Create(ctx, job, metav1.CreateOptions{})
-
-	if err != nil {
-		return fmt.Errorf("failed to create job: %w", err)
-	}
-
-	gvr, _ := meta.UnsafeGuessKindToResource(job.GroupVersionKind())
-	tracker.Add(gvr.Group, gvr.Version, gvr.Resource, namespacedName.Namespace, namespacedName.Name)
-
-	return wait.PollImmediate(interval, timeout, func() (bool, error) {
-		job, err := client.BatchV1().Jobs(namespacedName.Namespace).Get(ctx, namespacedName.Name, metav1.GetOptions{})
-		if err != nil {
-			return false, fmt.Errorf("failed to get job: %w", err)
-		}
-		if job.Status.Succeeded >= 1 {
-			return true, nil
-		}
-		return false, nil
-	})
+	return verifyJobSucceeded(ctx, client, tracker, namespacedName, job)
 }
diff --git a/test/pkg/kafka/verify_num_partitions_replication.go b/test/pkg/kafka/verify_num_partitions_replication.go
index d552a4e67..d6198482c 100644
--- a/test/pkg/kafka/verify_num_partitions_replication.go
+++ b/test/pkg/kafka/verify_num_partitions_replication.go
@@ -23,9 +23,8 @@ import (
 
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/pointer"
 	testlib "knative.dev/eventing/test/lib"
@@ -97,23 +96,5 @@ func VerifyNumPartitionAndReplicationFactor(
 		},
 		Status: batchv1.JobStatus{},
 	}
-
-	job, err := client.BatchV1().Jobs(namespace).Create(ctx, job, metav1.CreateOptions{})
-	if err != nil {
-		return fmt.Errorf("failed to create job: %w", err)
-	}
-
-	gvr, _ := meta.UnsafeGuessKindToResource(job.GroupVersionKind())
-	tracker.Add(gvr.Group, gvr.Version, gvr.Resource, namespace, name)
-
-	return wait.PollImmediate(interval, timeout, func() (bool, error) {
-		job, err := client.BatchV1().Jobs(namespace).Get(ctx, name, metav1.GetOptions{})
-		if err != nil {
-			return false, fmt.Errorf("failed to get job: %w", err)
-		}
-		if job.Status.Succeeded >= 1 {
-			return true, nil
-		}
-		return false, nil
-	})
+	return verifyJobSucceeded(ctx, client, tracker, types.NamespacedName{Namespace: namespace, Name: name}, job)
 }
diff --git a/test/pkg/testing/log.go b/test/pkg/testing/log.go
new file mode 100644
index 000000000..091166a7e
--- /dev/null
+++ b/test/pkg/testing/log.go
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2020 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"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
+	testlib "knative.dev/eventing/test/lib"
+)
+
+func LogJobOutput(t *testing.T, ctx context.Context, c kubernetes.Interface, namespace string, app string) {
+
+	selector := labels.SelectorFromSet(map[string]string{"app": app})
+
+	pods, err := c.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector.String()})
+	assert.Nil(t, err)
+	assert.Greater(t, len(pods.Items), 0)
+
+	logs := make([]string, 0, len(pods.Items))
+	for _, pod := range pods.Items {
+		if l, err := c.CoreV1().Pods(namespace).GetLogs(pod.Name, &corev1.PodLogOptions{}).DoRaw(ctx); err != nil {
+			logs = append(logs, err.Error())
+		} else {
+			logs = append(logs, string(l))
+		}
+	}
+
+	t.Log(strings.Join(logs, "\n"))
+	testlib.ExportLogs(namespace, namespace)
+}
diff --git a/test/test_images/committed-offset/admin.go b/test/test_images/committed-offset/admin.go
new file mode 100644
index 000000000..905d6333e
--- /dev/null
+++ b/test/test_images/committed-offset/admin.go
@@ -0,0 +1,59 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"encoding/json"
+	"log"
+
+	"github.com/Shopify/sarama"
+	"github.com/kelseyhightower/envconfig"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	kafkatest "knative.dev/eventing-kafka-broker/test/pkg/kafka"
+)
+
+func main() {
+
+	adminConfig := sarama.NewConfig()
+	adminConfig.Version = sarama.V2_0_0_0
+
+	envConfig := &kafkatest.AdminConfig{}
+
+	if err := envconfig.Process("", envConfig); err != nil {
+		log.Fatal("Failed to process environment variables", err)
+	}
+
+	j, _ := json.MarshalIndent(envConfig, "", " ")
+	log.Println(string(j))
+
+	client, err := sarama.NewClient([]string{envConfig.BootstrapServers}, adminConfig)
+	if err != nil {
+		log.Fatal("Failed to create sarama client", err)
+	}
+	consumerGroupLagProvider := kafka.NewConsumerGroupLagProvider(client, sarama.NewClusterAdminFromClient, sarama.OffsetOldest)
+	defer consumerGroupLagProvider.Close()
+
+	lag, err := consumerGroupLagProvider.GetLag(envConfig.Topic, envConfig.Group)
+	if err != nil {
+		log.Fatal("Failed to get lag", err)
+	}
+
+	if lag.Total() != 0 {
+		log.Fatal("Lag is not 0", lag)
+	}
+}
diff --git a/test/test_images/committed-offset/pod.yaml b/test/test_images/committed-offset/pod.yaml
new file mode 100644
index 000000000..f18e0f706
--- /dev/null
+++ b/test/test_images/committed-offset/pod.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: committed-offset
+spec:
+  containers:
+    - name: committed-offset
+      image: ko://knative.dev/eventing-kafka-broker/test/test_images/committed-offset
diff --git a/test/test_images/consumer-group-lag-provider-test/admin.go b/test/test_images/consumer-group-lag-provider-test/admin.go
new file mode 100644
index 000000000..428c8acf6
--- /dev/null
+++ b/test/test_images/consumer-group-lag-provider-test/admin.go
@@ -0,0 +1,212 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/Shopify/sarama"
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg/testing"
+)
+
+func main() {
+
+	defer func() {
+		if err := recover(); err != nil {
+			log.Fatal("Panic", err)
+		}
+	}()
+
+	const (
+		topic         = "test-consumer-group-lag-provider"
+		consumerGroup = topic
+		key           = "msg-key"
+		value         = "msg-value"
+	)
+
+	ctx := context.Background()
+
+	client, err := sarama.NewClient(testingpkg.BootstrapServersPlaintextArr, sarama.NewConfig())
+	mustBeNil(err)
+
+	admin, err := sarama.NewClusterAdminFromClient(client)
+	mustBeNil(err)
+	// Do not close admin since it closes the underlying client
+
+	topicDetail := &sarama.TopicDetail{NumPartitions: 2, ReplicationFactor: 2}
+	err = admin.CreateTopic(topic, topicDetail, false)
+	mustBeNil(err)
+
+	producerConfig := sarama.NewConfig()
+	// Disable buffering
+	producerConfig.Producer.Flush.Bytes = 1
+	producerConfig.Producer.Flush.MaxMessages = 1
+	producerConfig.Producer.Flush.Messages = 1
+	// Sync producer config
+	producerConfig.Producer.Return.Successes = true
+	producerConfig.Producer.Return.Errors = true
+
+	producer, err := sarama.NewSyncProducer(testingpkg.BootstrapServersPlaintextArr, producerConfig)
+	mustBeNil(err)
+	defer producer.Close()
+
+	// Produce messages with the same key.
+	n := 3
+	producedMsgPartition := int32(-1)
+	lastOffset := int64(-1)
+
+	log.Println("Sending events to topic", topic)
+
+	for i := 0; i < n; i++ {
+		msg := &sarama.ProducerMessage{
+			Topic: topic,
+			Key:   sarama.StringEncoder(key),
+			Value: sarama.StringEncoder(value),
+		}
+		// Send message might fail with:
+		// "kafka server: Request was for a topic or partition that does not exist on this broker."
+		err := wait.PollImmediateUntil(time.Minute, func() (done bool, err error) {
+			partition, offset, err := producer.SendMessage(msg)
+			if err != nil {
+				return false, nil
+			}
+			log.Printf("Event sent partition %d offset %d", partition, offset)
+			if producedMsgPartition == -1 {
+				producedMsgPartition = partition
+			}
+			lastOffset = offset
+			return true, nil
+		}, ctx.Done())
+		mustBeNil(err)
+	}
+	if int64(n) != lastOffset+1 { // Consistency check
+		log.Fatalf("Expected last offset to be equal to %d got %d", n, lastOffset+1)
+	}
+
+	log.Println("Consuming events from topic", topic)
+
+	consumerConfig := sarama.NewConfig()
+	consumerConfig.Consumer.Fetch.Min = 1
+	consumerConfig.Consumer.Offsets.Initial = sarama.OffsetOldest
+	consumerConfig.Consumer.Offsets.AutoCommit.Enable = false
+
+	consumer, err := sarama.NewConsumerGroup(testingpkg.BootstrapServersPlaintextArr, consumerGroup, consumerConfig)
+	mustBeNil(err)
+
+	msgs := make(chan *sarama.ConsumerMessage, n)
+	handler := &handler{msgs: msgs}
+
+	count := 0
+	func() {
+		go func() {
+			_ = consumer.Consume(ctx, []string{topic}, handler)
+		}()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-msgs:
+				count++
+				log.Println("Count", n)
+				if count == n {
+					return
+				}
+			}
+		}
+	}()
+	err = consumer.Close()
+	mustBeNil(err)
+
+	// Wait for propagation of the committed offset
+	err = wait.PollImmediateUntil(time.Minute, func() (done bool, err error) {
+		log.Println("Starting consumer group lag provider")
+
+		consumerGroupLagProvider := kafka.NewConsumerGroupLagProvider(client, sarama.NewClusterAdminFromClient, sarama.OffsetOldest)
+		defer consumerGroupLagProvider.Close()
+
+		log.Printf("Getting lag for topic %s and consumer group %s\n", topic, consumerGroup)
+		consumerGroupLag, err := consumerGroupLagProvider.GetLag(topic, consumerGroup)
+		mustBeNil(err)
+
+		log.Println("ConsumerGroupLag", consumerGroupLag)
+
+		nNonZeroOffsets := 0
+		for p, l := range consumerGroupLag.ByPartition {
+			log.Printf("Partition %d Lag %s\n", p, l)
+			if l.ConsumerOffset > 0 {
+				nNonZeroOffsets++
+			}
+			if l.ConsumerOffset < 0 {
+				log.Fatal("Consumer offset cannot be negative")
+			}
+			if l.LatestOffset < 0 {
+				log.Fatal("Latest offset cannot be negative")
+			}
+			if l.LatestOffset-l.ConsumerOffset != l.Lag() {
+				log.Fatal("Expected Lag to be equal to LatestOffset - ConsumerOffset got", l)
+			}
+		}
+		if nNonZeroOffsets != 1 { // consistency check
+			log.Println("Only 1 partition is expected to have records since test uses the same key for all of them ", nNonZeroOffsets)
+			return false, nil
+		}
+
+		total := consumerGroupLag.Total()
+		log.Println("Total lag", total)
+		if total != 0 {
+			return false, nil
+		}
+		return true, nil
+	}, ctx.Done())
+	mustBeNil(err)
+}
+
+type handler struct {
+	msgs chan<- *sarama.ConsumerMessage
+}
+
+func (h *handler) Setup(sarama.ConsumerGroupSession) error {
+	log.Println("Consumer started")
+	return nil
+}
+
+func (h *handler) Cleanup(session sarama.ConsumerGroupSession) error {
+	session.Commit()
+	return nil
+}
+
+func (h *handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+
+	for msg := range claim.Messages() {
+		session.MarkMessage(msg, "")
+		log.Println("Message received", string(msg.Key), string(msg.Value))
+		h.msgs <- msg
+	}
+
+	return nil
+}
+
+func mustBeNil(err error) {
+	if err != nil {
+		log.Fatal(err)
+	}
+}
diff --git a/test/test_images/consumer-group-lag-provider-test/pod.yaml b/test/test_images/consumer-group-lag-provider-test/pod.yaml
new file mode 100644
index 000000000..4cbfbf7b1
--- /dev/null
+++ b/test/test_images/consumer-group-lag-provider-test/pod.yaml
@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: consumer-group-lag-provider-test
+spec:
+  containers:
+    - name: consumer-group-lag-provider-test
+      image: ko://knative.dev/eventing-kafka-broker/test/test_images/consumer-group-lag-provider-test
diff --git a/third_party/eventing-latest/eventing-core.yaml b/third_party/eventing-latest/eventing-core.yaml
index 4c260814f..36a110650 100644
--- a/third_party/eventing-latest/eventing-core.yaml
+++ b/third_party/eventing-latest/eventing-core.yaml
@@ -16,7 +16,7 @@ kind: Namespace
 metadata:
   name: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 
 ---
 # Copyright 2018 The Knative Authors
@@ -39,14 +39,14 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: eventing-controller
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -61,7 +61,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -76,7 +76,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-source-observer
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -91,7 +91,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-sources-controller
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -106,7 +106,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-controller-manipulator
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-controller
@@ -137,14 +137,14 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: pingsource-mt-adapter
@@ -175,14 +175,14 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -198,7 +198,7 @@ metadata:
   namespace: knative-eventing
   name: eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -213,7 +213,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -228,7 +228,7 @@ kind: ClusterRoleBinding
 metadata:
   name: eventing-webhook-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 subjects:
   - kind: ServiceAccount
     name: eventing-webhook
@@ -259,7 +259,7 @@ metadata:
   name: config-br-default-channel
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 data:
   channelTemplateSpec: |
     apiVersion: messaging.knative.dev/v1
@@ -286,7 +286,7 @@ metadata:
   name: config-br-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
   default-br-config: |
@@ -318,7 +318,7 @@ metadata:
   name: default-ch-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 data:
   # Configuration for defaulting channels that do not specify CRD implementations.
   default-ch-config: |
@@ -351,7 +351,7 @@ metadata:
   name: config-ping-defaults
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
   annotations:
     knative.dev/example-checksum: "f8e5a744"
 data:
@@ -396,7 +396,7 @@ metadata:
   name: config-leader-election
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
   annotations:
     knative.dev/example-checksum: "96896b00"
 data:
@@ -456,7 +456,7 @@ metadata:
   name: config-logging
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
 data:
@@ -508,7 +508,7 @@ metadata:
   name: config-observability
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -583,7 +583,7 @@ metadata:
   name: config-tracing
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/config-propagation: original
     knative.dev/config-category: eventing
   annotations:
@@ -644,7 +644,7 @@ metadata:
   name: eventing-controller
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/high-availability: "true"
 spec:
   selector:
@@ -654,7 +654,7 @@ spec:
     metadata:
       labels:
         app: eventing-controller
-        eventing.knative.dev/release: "v20210506-2f44af0cd"
+        eventing.knative.dev/release: "v20210512-8e157adaa"
     spec:
       # To avoid node becoming SPOF, spread our replicas to different nodes.
       affinity:
@@ -671,7 +671,7 @@ spec:
       containers:
         - name: eventing-controller
           terminationMessagePolicy: FallbackToLogsOnError
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:9ab644ec8df420fb40df3d5450ae9d2b65008a14f4c1329219ef589b9e07d7ac
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/controller@sha256:964cfab75d7ae12743daae1c18d38e217a48e50d0b2b33d471c546bd4fb2ebd0
           resources:
             requests:
               cpu: 100m
@@ -689,7 +689,7 @@ spec:
               value: knative.dev/eventing
             # APIServerSource
             - name: APISERVER_RA_IMAGE
-              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:b989f95b6790edbc77d123aad39f3de61aa6b0b501cc7531ab8695e73114da4a
+              value: gcr.io/knative-nightly/knative.dev/eventing/cmd/apiserver_receive_adapter@sha256:00b5fc02d7c5469ff8645ccb88162803a5475af546c5be28eb4437742ae14232
             - name: POD_NAME
               valueFrom:
                 fieldRef:
@@ -733,7 +733,7 @@ metadata:
   name: pingsource-mt-adapter
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 spec:
   # when set to 0 (and only 0) will be set to 1 when the first PingSource is created.
   replicas: 0
@@ -746,12 +746,12 @@ spec:
       labels:
         eventing.knative.dev/source: ping-source-controller
         sources.knative.dev/role: adapter
-        eventing.knative.dev/release: "v20210506-2f44af0cd"
+        eventing.knative.dev/release: "v20210512-8e157adaa"
     spec:
       enableServiceLinks: false
       containers:
         - name: dispatcher
-          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:bcf43620036f64f2ae3e94c064a230f10c9b1c9cd96f3b60d2884d1a569be316
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/mtping@sha256:6b9dcd05d389a21c8122d2023cd89ccd1e14bd9b114156f68ea64768fb3b9d33
           env:
             - name: SYSTEM_NAMESPACE
               value: ''
@@ -809,7 +809,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 spec:
   scaleTargetRef:
     apiVersion: apps/v1
@@ -832,7 +832,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 spec:
   minAvailable: 80%
   selector:
@@ -860,7 +860,7 @@ metadata:
   name: eventing-webhook
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 spec:
   selector:
     matchLabels: &labels
@@ -887,7 +887,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:8b3eb56de45ef8ba25886840316a53b921a960b978df4eb1655e2c9c75f3e32c
+          image: gcr.io/knative-nightly/knative.dev/eventing/cmd/webhook@sha256:4fd6dc346a2557a91261cd8e08d65ef2f3bcf317ae31cda2cd137a40f79f783b
           resources:
             requests:
               # taken from serving.
@@ -951,7 +951,7 @@ apiVersion: v1
 kind: Service
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     role: eventing-webhook
   name: eventing-webhook
   namespace: knative-eventing
@@ -983,7 +983,7 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1009,45 +1009,43 @@ spec:
         status: {}
       schema:
         openAPIV3Schema:
-          type: object
           description: 'ApiServerSource is an event source that brings Kubernetes API server events into Knative.'
+          type: object
           properties:
             spec:
               type: object
-              description: 'ApiServerSourceSpec defines the desired state of ApiServerSource (from the client).'
               required:
                 - resources
               properties:
                 ceOverrides:
-                  description: 'CloudEventOverrides defines overrides to control the output format and modifications of the event sent to the sink.'
+                  description: CloudEventOverrides defines overrides to control the output format and modifications of the event sent to the sink.
                   type: object
                   properties:
                     extensions:
-                      description: 'Extensions specify what attribute are added or overridden on the outbound event. Each `Extensions` key-value pair are set on the event as an attribute extension independently.'
+                      description: Extensions specify what attribute are added or overridden on the outbound event. Each `Extensions` key-value pair are set on the event as an attribute extension independently.
                       type: object
-                      additionalProperties:
-                        type: string
+                      x-kubernetes-preserve-unknown-fields: true
                 mode:
-                  description: 'EventMode controls the format of the event. `Reference` sends a dataref event type for the resource under watch. `Resource` send the full resource lifecycle event. Defaults to `Reference`'
+                  description: EventMode controls the format of the event. `Reference` sends a dataref event type for the resource under watch. `Resource` send the full resource lifecycle event. Defaults to `Reference`
                   type: string
                 owner:
-                  description: 'ResourceOwner is an additional filter to only track resources that are owned by a specific resource type. If ResourceOwner matches Resources[n] then Resources[n] is allowed to pass the ResourceOwner filter.'
+                  description: ResourceOwner is an additional filter to only track resources that are owned by a specific resource type. If ResourceOwner matches Resources[n] then Resources[n] is allowed to pass the ResourceOwner filter.
                   type: object
                   properties:
                     apiVersion:
-                      description: 'APIVersion - the API version of the resource to watch.'
+                      description: APIVersion - the API version of the resource to watch.
                       type: string
                     kind:
                       description: 'Kind of the resource to watch. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
                       type: string
                 resources:
-                  description: 'Resource are the resources this source will track and send related lifecycle events from the Kubernetes ApiServer, with an optional label selector to help filter.'
+                  description: Resource are the resources this source will track and send related lifecycle events from the Kubernetes ApiServer, with an optional label selector to help filter.
                   type: array
                   items:
                     type: object
                     properties:
                       apiVersion:
-                        description: 'APIVersion - the API version of the resource to watch.'
+                        description: APIVersion - the API version of the resource to watch.
                         type: string
                       kind:
                         description: 'Kind of the resource to watch. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -1057,39 +1055,39 @@ spec:
                         type: object
                         properties:
                           matchExpressions:
-                            description: 'matchExpressions is a list of label selector requirements. The requirements are ANDed.'
+                            description: matchExpressions is a list of label selector requirements. The requirements are ANDed.
                             type: array
                             items:
                               type: object
                               properties:
                                 key:
-                                  description: 'key is the label key that the selector applies to.'
+                                  description: key is the label key that the selector applies to.
                                   type: string
                                 operator:
-                                  description: 'operator represents a key''s relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist.'
+                                  description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist.
                                   type: string
                                 values:
-                                  description: 'values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch.'
+                                  description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch.
                                   type: array
                                   items:
                                     type: string
                           matchLabels:
-                            description: 'matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed.'
+                            description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed.
                             type: object
                             x-kubernetes-preserve-unknown-fields: true
                 serviceAccountName:
-                  description: 'ServiceAccountName is the name of the ServiceAccount to use to run this source. Defaults to default if not set.'
+                  description: ServiceAccountName is the name of the ServiceAccount to use to run this source. Defaults to default if not set.
                   type: string
                 sink:
-                  description: 'Sink is a reference to an object that will resolve to a uri to use as the sink.'
+                  description: Sink is a reference to an object that will resolve to a uri to use as the sink.
                   type: object
                   properties:
                     ref:
-                      description: 'Ref points to an Addressable.'
+                      description: Ref points to an Addressable.
                       type: object
                       properties:
                         apiVersion:
-                          description: 'API version of the referent.'
+                          description: API version of the referent.
                           type: string
                         kind:
                           description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -1101,30 +1099,29 @@ spec:
                           description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                           type: string
                     uri:
-                      description: 'URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.'
+                      description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
             status:
               type: object
-              description: 'ApiServerSourceStatus defines the observed state of ApiServerSource (from the controller).'
               properties:
                 annotations:
-                  description: 'Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.'
+                  description: Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.
                   type: object
                   x-kubernetes-preserve-unknown-fields: true
                 ceAttributes:
-                  description: 'CloudEventAttributes are the specific attributes that the Source uses as part of its CloudEvents.'
+                  description: CloudEventAttributes are the specific attributes that the Source uses as part of its CloudEvents.
                   type: array
                   items:
                     type: object
                     properties:
                       source:
-                        description: 'Source is the CloudEvents source attribute.'
+                        description: Source is the CloudEvents source attribute.
                         type: string
                       type:
-                        description: 'Type refers to the CloudEvent type attribute.'
+                        description: Type refers to the CloudEvent type attribute.
                         type: string
                 conditions:
-                  description: 'Conditions the latest available observations of a resource''s current state.'
+                  description: Conditions the latest available observations of a resource's current state.
                   type: array
                   items:
                     type: object
@@ -1133,29 +1130,29 @@ spec:
                       - status
                     properties:
                       lastTransitionTime:
-                        description: 'LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).'
+                        description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
                         type: string
                       message:
-                        description: 'A human readable message indicating details about the transition.'
+                        description: A human readable message indicating details about the transition.
                         type: string
                       reason:
-                        description: 'The reason for the condition''s last transition.'
+                        description: The reason for the condition's last transition.
                         type: string
                       severity:
-                        description: 'Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.'
+                        description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
                         type: string
                       status:
-                        description: 'Status of the condition, one of True, False, Unknown.'
+                        description: Status of the condition, one of True, False, Unknown.
                         type: string
                       type:
-                        description: 'Type of condition.'
+                        description: Type of condition.
                         type: string
                 observedGeneration:
-                  description: 'ObservedGeneration is the "Generation" of the Service that was last processed by the controller.'
+                  description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
                   format: int64
                 sinkUri:
-                  description: 'SinkURI is the current active sink URI that has been configured for the Source.'
+                  description: SinkURI is the current active sink URI that has been configured for the Source.
                   type: string
       additionalPrinterColumns:
         - name: Sink
@@ -1179,14 +1176,6 @@ spec:
     plural: apiserversources
     singular: apiserversource
   scope: Namespaced
-  conversion:
-    strategy: Webhook
-    webhook:
-      conversionReviewVersions: ["v1", "v1beta1"]
-      clientConfig:
-        service:
-          name: eventing-webhook
-          namespace: knative-eventing
 
 ---
 # Copyright 2020 The Knative Authors
@@ -1208,7 +1197,7 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1369,7 +1358,7 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -1396,7 +1385,7 @@ spec:
           jsonPath: ".status.conditions[?(@.type==\"Ready\")].reason"
       schema:
         openAPIV3Schema:
-          description: 'Channel represents a generic Channel. It is normally used when we want a Channel, but don''t need a specific Channel implementation.'
+          description: 'Channel represents a generic Channel. It is normally used when we want a Channel, but do not need a specific Channel implementation.'
           type: object
           properties:
             spec:
@@ -1417,8 +1406,8 @@ spec:
                       description: Spec defines the Spec to use for each channel created. Passed in verbatim to the Channel CRD as Spec section.
                       type: object
                       x-kubernetes-preserve-unknown-fields: true
-                delivery: &deliverySpec
-                  description: DeliverySpec contains options controlling the event delivery
+                delivery:
+                  description: DeliverySpec contains the default delivery spec for each subscription to this Channelable. Each subscription delivery spec, if any, overrides this global delivery spec.
                   type: object
                   properties:
                     backoffDelay:
@@ -1434,7 +1423,7 @@ spec:
                         ref:
                           description: Ref points to an Addressable.
                           type: object
-                          properties: &referentProperties
+                          properties:
                             apiVersion:
                               description: API version of the referent.
                               type: string
@@ -1461,7 +1450,42 @@ spec:
                     type: object
                     properties:
                       delivery:
-                        !!merge <<: *deliverySpec
+                        description: DeliverySpec contains options controlling the event delivery
+                        type: object
+                        properties:
+                          backoffDelay:
+                            description: 'BackoffDelay is the delay before retrying. More information on Duration format: - https://www.iso.org/iso-8601-date-and-time-format.html - https://en.wikipedia.org/wiki/ISO_8601  For linear policy, backoff delay is backoffDelay*<numberOfRetries>. For exponential policy, backoff delay is backoffDelay*2^<numberOfRetries>.'
+                            type: string
+                          backoffPolicy:
+                            description: BackoffPolicy is the retry backoff policy (linear, exponential).
+                            type: string
+                          deadLetterSink:
+                            description: DeadLetterSink is the sink receiving event that could not be sent to a destination.
+                            type: object
+                            properties:
+                              ref:
+                                description: Ref points to an Addressable.
+                                type: object
+                                properties:
+                                  apiVersion:
+                                    description: API version of the referent.
+                                    type: string
+                                  kind:
+                                    description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                                    type: string
+                                  name:
+                                    description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                                    type: string
+                                  namespace:
+                                    description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
+                                    type: string
+                              uri:
+                                description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
+                                type: string
+                          retry:
+                            description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
+                            type: integer
+                            format: int32
                       generation:
                         description: Generation of the origin of the subscriber with uid:UID.
                         type: integer
@@ -1492,14 +1516,30 @@ spec:
                   description: Channel is an KReference to the Channel CRD backing this Channel.
                   type: object
                   properties:
-                    !!merge <<: *referentProperties
+                    apiVersion:
+                      description: API version of the referent.
+                      type: string
+                    kind:
+                      description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                      type: string
+                    name:
+                      description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                      type: string
+                    namespace:
+                      description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
+                      type: string
                 conditions:
                   description: Conditions the latest available observations of a resource's current state.
                   type: array
                   items:
                     type: object
-                    x-kubernetes-preserve-unknown-fields: true
+                    required:
+                      - type
+                      - status
                     properties:
+                      lastTransitionTime:
+                        description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
+                        type: string
                       message:
                         description: A human readable message indicating details about the transition.
                         type: string
@@ -1519,7 +1559,18 @@ spec:
                   description: DeadLetterChannel is a KReference and is set by the channel when it supports native error handling via a channel Failed messages are delivered here.
                   type: object
                   properties:
-                    !!merge <<: *referentProperties
+                    apiVersion:
+                      description: API version of the referent.
+                      type: string
+                    kind:
+                      description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                      type: string
+                    name:
+                      description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                      type: string
+                    namespace:
+                      description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
+                      type: string
                 observedGeneration:
                   description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
@@ -1555,14 +1606,6 @@ spec:
     shortNames:
       - ch
   scope: Namespaced
-  conversion:
-    strategy: Webhook
-    webhook:
-      conversionReviewVersions: ["v1", "v1beta1"]
-      clientConfig:
-        service:
-          name: eventing-webhook
-          namespace: knative-eventing
 
 ---
 # Copyright 2020 The Knative Authors
@@ -1583,7 +1626,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1597,32 +1640,31 @@ spec:
       subresources:
         status: {}
       schema:
-        openAPIV3Schema: &openAPIV3Schema
-          type: object
+        openAPIV3Schema:
           description: 'ContainerSource is an event source that starts a container image which generates events under certain situations and sends messages to a sink URI'
+          type: object
           properties:
             spec:
               type: object
-              description: 'ContainerSourceSpec defines the desired state of ContainerSource (from the client).'
               properties:
                 ceOverrides:
-                  description: 'CloudEventOverrides defines overrides to control the output format and modifications of the event sent to the sink.'
+                  description: CloudEventOverrides defines overrides to control the output format and modifications of the event sent to the sink.
                   type: object
                   properties:
                     extensions:
-                      description: 'Extensions specify what attribute are added or overridden on the outbound event. Each `Extensions` key-value pair are set on the event as an attribute extension independently.'
+                      description: Extensions specify what attribute are added or overridden on the outbound event. Each `Extensions` key-value pair are set on the event as an attribute extension independently.
                       type: object
                       x-kubernetes-preserve-unknown-fields: true
                 sink:
-                  description: 'Sink is a reference to an object that will resolve to a uri to use as the sink.'
+                  description: Sink is a reference to an object that will resolve to a uri to use as the sink.
                   type: object
                   properties:
                     ref:
-                      description: 'Ref points to an Addressable.'
+                      description: Ref points to an Addressable.
                       type: object
                       properties:
                         apiVersion:
-                          description: 'API version of the referent.'
+                          description: API version of the referent.
                           type: string
                         kind:
                           description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -1634,134 +1676,34 @@ spec:
                           description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                           type: string
                     uri:
-                      description: 'URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.'
+                      description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
+                # WARNING: the schema tool can not parse PodTemplateSpec, stub here and redirect to Deployment documentation.
                 template:
-                  description: 'Template describes the pods that will be created'
                   type: object
-                  properties:
-                    annotations:
-                      description: 'Annotations is an unstructured key value map stored with a resource that may be set by external tools to store and retrieve arbitrary metadata. They are not queryable and should be preserved when modifying objects. More info: http://kubernetes.io/docs/user-guide/annotations'
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
-                    clusterName:
-                      description: 'The name of the cluster which the object belongs to. This is used to distinguish resources with same name and namespace in different clusters. This field is not set anywhere right now and apiserver is going to ignore it if set in create or update request.'
-                      type: string
-                    creationTimestamp:
-                      description: 'CreationTimestamp is a timestamp representing the server time when this object was created. It is not guaranteed to be set in happens-before order across separate operations. Clients may not set this value. It is represented in RFC3339 form and is in UTC.  Populated by the system. Read-only. Null for lists. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata'
-                      type: string
-                    deletionGracePeriodSeconds:
-                      description: 'Number of seconds allowed for this object to gracefully terminate before it will be removed from the system. Only set when deletionTimestamp is also set. May only be shortened. Read-only.'
-                      type: integer
-                      format: int64
-                    deletionTimestamp:
-                      description: 'DeletionTimestamp is RFC 3339 date and time at which this resource will be deleted. This field is set by the server when a graceful deletion is requested by the user, and is not directly settable by a client. The resource is expected to be deleted (no longer visible from resource lists, and not reachable by name) after the time in this field, once the finalizers list is empty. As long as the finalizers list contains items, deletion is blocked. Once the deletionTimestamp is set, this value may not be unset or be set further into the future, although it may be shortened or the resource may be deleted prior to this time. For example, a user may request that a pod is deleted in 30 seconds. The Kubelet will react by sending a graceful termination signal to the containers in the pod. After that 30 seconds, the Kubelet will send a hard termination signal (SIGKILL) to the container and after cleanup, remove the pod from the API. In the presence of network partitions, this object may still exist after this timestamp, until an administrator or automated process can determine the resource is fully terminated. If not set, graceful deletion of the object has not been requested.  Populated by the system when a graceful deletion is requested. Read-only. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata'
-                      type: string
-                    finalizers:
-                      description: 'Must be empty before the object is deleted from the registry. Each entry is an identifier for the responsible component that will remove the entry from the list. If the deletionTimestamp of the object is non-nil, entries in this list can only be removed. Finalizers may be processed and removed in any order.  Order is NOT enforced because it introduces significant risk of stuck finalizers. finalizers is a shared field, any actor with permission can reorder it. If the finalizer list is processed in order, then this can lead to a situation in which the component responsible for the first finalizer in the list is waiting for a signal (field value, external system, or other) produced by a component responsible for a finalizer later in the list, resulting in a deadlock. Without enforced ordering finalizers are free to order amongst themselves and are not vulnerable to ordering changes in the list.'
-                      type: array
-                      items:
-                        type: string
-                    generateName:
-                      description: 'GenerateName is an optional prefix, used by the server, to generate a unique name ONLY IF the Name field has not been provided. If this field is used, the name returned to the client will be different than the name passed. This value will also be combined with a unique suffix. The provided value has the same validation rules as the Name field, and may be truncated by the length of the suffix required to make the value unique on the server.  If this field is specified and the generated name exists, the server will NOT return a 409 - instead, it will either return 201 Created or 500 with Reason ServerTimeout indicating a unique name could not be found in the time allotted, and the client should retry (optionally after the time indicated in the Retry-After header).  Applied only if Name is not specified. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#idempotency'
-                      type: string
-                    generation:
-                      description: 'A sequence number representing a specific generation of the desired state. Populated by the system. Read-only.'
-                      type: integer
-                      format: int64
-                    labels:
-                      description: 'Map of string keys and values that can be used to organize and categorize (scope and select) objects. May match selectors of replication controllers and services. More info: http://kubernetes.io/docs/user-guide/labels'
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
-                    managedFields:
-                      description: 'ManagedFields maps workflow-id and version to the set of fields that are managed by that workflow. This is mostly for internal housekeeping, and users typically shouldn''t need to set or understand this field. A workflow can be the user''s name, a controller''s name, or the name of a specific apply path like "ci-cd". The set of fields is always in the version that the workflow used when modifying the object. '
-                      type: array
-                      items:
-                        type: object
-                        properties:
-                          apiVersion:
-                            description: 'APIVersion defines the version of this resource that this field set applies to. The format is "group/version" just like the top-level APIVersion field. It is necessary to track the version of a field set because it cannot be automatically converted.'
-                            type: string
-                          fieldsType:
-                            description: 'FieldsType is the discriminator for the different fields format and version. There is currently only one possible value: "FieldsV1"'
-                            type: string
-                          fieldsV1:
-                            description: 'FieldsV1 holds the first JSON version format as described in the "FieldsV1" type.'
-                            type: string
-                          manager:
-                            description: 'Manager is an identifier of the workflow managing these fields.'
-                            type: string
-                          operation:
-                            description: 'Operation is the type of operation which lead to this ManagedFieldsEntry being created. The only valid values for this field are "Apply" and "Update".'
-                            type: string
-                          time:
-                            description: 'Time is timestamp of when these fields were set. It should always be empty if Operation is "Apply"'
-                            type: string
-                    name:
-                      description: 'Name must be unique within a namespace. Is required when creating resources, although some resources may allow a client to request the generation of an appropriate name automatically. Name is primarily intended for creation idempotence and configuration definition. Cannot be updated. More info: http://kubernetes.io/docs/user-guide/identifiers#names'
-                      type: string
-                    namespace:
-                      description: 'Namespace defines the space within each name must be unique. An empty namespace is equivalent to the "default" namespace, but "default" is the canonical representation. Not all objects are required to be scoped to a namespace - the value of this field for those objects will be empty.  Must be a DNS_LABEL. Cannot be updated. More info: http://kubernetes.io/docs/user-guide/namespaces'
-                      type: string
-                    ownerReferences:
-                      description: 'List of objects depended by this object. If ALL objects in the list have been deleted, this object will be garbage collected. If this object is managed by a controller, then an entry in this list will point to this controller, with the controller field set to true. There cannot be more than one managing controller.'
-                      type: array
-                      items:
-                        type: object
-                        properties:
-                          apiVersion:
-                            description: 'API version of the referent.'
-                            type: string
-                          blockOwnerDeletion:
-                            description: 'If true, AND if the owner has the "foregroundDeletion" finalizer, then the owner cannot be deleted from the key-value store until this reference is removed. Defaults to false. To set this field, a user needs "delete" permission of the owner, otherwise 422 (Unprocessable Entity) will be returned.'
-                            type: boolean
-                          controller:
-                            description: 'If true, this reference points to the managing controller.'
-                            type: boolean
-                          kind:
-                            description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
-                            type: string
-                          name:
-                            description: 'Name of the referent. More info: http://kubernetes.io/docs/user-guide/identifiers#names'
-                            type: string
-                          uid:
-                            description: 'UID of the referent. More info: http://kubernetes.io/docs/user-guide/identifiers#uids'
-                            type: string
-                    resourceVersion:
-                      description: 'An opaque value that represents the internal version of this object that can be used by clients to determine when objects have changed. May be used for optimistic concurrency, change detection, and the watch operation on a resource or set of resources. Clients must treat these values as opaque and passed unmodified back to the server. They may only be valid for a particular resource or set of resources.  Populated by the system. Read-only. Value must be treated as opaque by clients and . More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#concurrency-control-and-consistency'
-                      type: string
-                    selfLink:
-                      description: 'SelfLink is a URL representing this object. Populated by the system. Read-only.  DEPRECATED Kubernetes will stop propagating this field in 1.20 release and the field is planned to be removed in 1.21 release.'
-                      type: string
-                    spec:
-                      description: 'Specification of the desired behavior of the pod. More info: Type ''kubectl explain pod.spec''. Also, https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#spec-and-status'
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
-                    uid:
-                      description: 'UID is the unique in time and space value for this object. It is typically generated by the server on successful creation of a resource and is not allowed to change on PUT operations.  Populated by the system. Read-only. More info: http://kubernetes.io/docs/user-guide/identifiers#uids'
-                      type: string
+                  x-kubernetes-preserve-unknown-fields: true
+                  description: 'A template in the shape of `Deployment.spec.template` to be used for this ContainerSource. More info: https://kubernetes.io/docs/concepts/workloads/controllers/deployment/'
             status:
               type: object
-              description: 'ContainerSourceStatus defines the observed state of ContainerSource (from the controller).'
               properties:
                 annotations:
-                  description: 'Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.'
+                  description: Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.
                   type: object
                   x-kubernetes-preserve-unknown-fields: true
                 ceAttributes:
-                  description: 'CloudEventAttributes are the specific attributes that the Source uses as part of its CloudEvents.'
+                  description: CloudEventAttributes are the specific attributes that the Source uses as part of its CloudEvents.
                   type: array
                   items:
                     type: object
                     properties:
                       source:
-                        description: 'Source is the CloudEvents source attribute.'
+                        description: Source is the CloudEvents source attribute.
                         type: string
                       type:
-                        description: 'Type refers to the CloudEvent type attribute.'
+                        description: Type refers to the CloudEvent type attribute.
                         type: string
                 conditions:
-                  description: 'Conditions the latest available observations of a resource''s current state.'
+                  description: Conditions the latest available observations of a resource's current state.
                   type: array
                   items:
                     type: object
@@ -1770,29 +1712,29 @@ spec:
                       - status
                     properties:
                       lastTransitionTime:
-                        description: 'LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).'
+                        description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
                         type: string
                       message:
-                        description: 'A human readable message indicating details about the transition.'
+                        description: A human readable message indicating details about the transition.
                         type: string
                       reason:
-                        description: 'The reason for the condition''s last transition.'
+                        description: The reason for the condition's last transition.
                         type: string
                       severity:
-                        description: 'Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.'
+                        description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
                         type: string
                       status:
-                        description: 'Status of the condition, one of True, False, Unknown.'
+                        description: Status of the condition, one of True, False, Unknown.
                         type: string
                       type:
                         description: Type of condition.
                         type: string
                 observedGeneration:
-                  description: 'ObservedGeneration is the "Generation" of the Service that was last processed by the controller.'
+                  description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
                   format: int64
                 sinkUri:
-                  description: 'SinkURI is the current active sink URI that has been configured for the Source.'
+                  description: SinkURI is the current active sink URI that has been configured for the Source.
                   type: string
       additionalPrinterColumns:
         - name: Sink
@@ -1816,14 +1758,6 @@ spec:
     plural: containersources
     singular: containersource
   scope: Namespaced
-  conversion:
-    strategy: Webhook
-    webhook:
-      conversionReviewVersions: ["v1", "v1beta1"]
-      clientConfig:
-        service:
-          name: eventing-webhook
-          namespace: knative-eventing
 
 ---
 # Copyright 2020 The Knative Authors
@@ -1844,7 +1778,7 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -1951,14 +1885,6 @@ spec:
       - knative
       - eventing
   scope: Namespaced
-  conversion:
-    strategy: Webhook
-    webhook:
-      conversionReviewVersions: ["v1", "v1beta1"]
-      clientConfig:
-        service:
-          name: eventing-webhook
-          namespace: knative-eventing
 
 ---
 # Copyright 2020 The Knative Authors
@@ -1979,7 +1905,7 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -2211,14 +2137,6 @@ spec:
       - knative
       - flows
   scope: Namespaced
-  conversion:
-    strategy: Webhook
-    webhook:
-      conversionReviewVersions: ["v1", "v1beta1"]
-      clientConfig:
-        service:
-          name: eventing-webhook
-          namespace: knative-eventing
 
 ---
 # Copyright 2020 The Knative Authors
@@ -2239,7 +2157,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -2428,7 +2346,7 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -2460,12 +2378,11 @@ spec:
                       type: string
                     spec:
                       description: Spec defines the Spec to use for each channel created. Passed in verbatim to the Channel CRD as Spec section.
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
+                      type: string
                 reply:
                   description: Reply is a Reference to where the result of the last Subscriber gets sent to.
                   type: object
-                  properties: &addressableProperties
+                  properties:
                     ref:
                       description: Ref points to an Addressable.
                       type: object
@@ -2491,7 +2408,6 @@ spec:
                   items:
                     type: object
                     properties:
-                      !!merge <<: *addressableProperties
                       delivery:
                         description: Delivery is the delivery specification for events to the subscriber This includes things like retries, DLQ, etc.
                         type: object
@@ -2506,23 +2422,63 @@ spec:
                             description: DeadLetterSink is the sink receiving event that could not be sent to a destination.
                             type: object
                             properties:
-                              !!merge <<: *addressableProperties
+                              ref:
+                                description: Ref points to an Addressable.
+                                type: object
+                                properties:
+                                  apiVersion:
+                                    description: API version of the referent.
+                                    type: string
+                                  kind:
+                                    description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                                    type: string
+                                  name:
+                                    description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                                    type: string
+                                  namespace:
+                                    description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
+                                    type: string
+                              uri:
+                                description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
+                                type: string
                           retry:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                      ref:
+                        description: Ref points to an Addressable.
+                        type: object
+                        properties:
+                          apiVersion:
+                            description: API version of the referent.
+                            type: string
+                          kind:
+                            description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                            type: string
+                          name:
+                            description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                            type: string
+                          namespace:
+                            description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
+                            type: string
+                      uri:
+                        description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
+                        type: string
             status:
               description: Status represents the current state of the Sequence. This data may be out of date.
               type: object
               properties:
                 address:
                   type: object
+                  required:
+                    - url
                   properties:
                     url:
                       type: string
                 annotations:
                   description: Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.
                   type: object
+                  x-kubernetes-preserve-unknown-fields: true
                 channelStatuses:
                   description: ChannelStatuses is an array of corresponding Channel statuses. Matches the Spec.Steps array in the order.
                   type: array
@@ -2532,7 +2488,7 @@ spec:
                       channel:
                         description: Channel is the reference to the underlying channel.
                         type: object
-                        properties: &referentProperties
+                        properties:
                           apiVersion:
                             description: API version of the referent.
                             type: string
@@ -2557,8 +2513,13 @@ spec:
                       ready:
                         description: ReadyCondition indicates whether the Channel is ready or not.
                         type: object
-                        x-kubernetes-preserve-unknown-fields: true
-                        properties: &readyConditionProperties
+                        required:
+                          - type
+                          - status
+                        properties:
+                          lastTransitionTime:
+                            description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
+                            type: string
                           message:
                             description: A human readable message indicating details about the transition.
                             type: string
@@ -2579,8 +2540,28 @@ spec:
                   type: array
                   items:
                     type: object
+                    required:
+                      - type
+                      - status
                     properties:
-                      !!merge <<: *readyConditionProperties
+                      lastTransitionTime:
+                        description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
+                        type: string
+                      message:
+                        description: A human readable message indicating details about the transition.
+                        type: string
+                      reason:
+                        description: The reason for the condition's last transition.
+                        type: string
+                      severity:
+                        description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
+                        type: string
+                      status:
+                        description: Status of the condition, one of True, False, Unknown.
+                        type: string
+                      type:
+                        description: Type of condition.
+                        type: string
                 observedGeneration:
                   description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
@@ -2594,13 +2575,53 @@ spec:
                       ready:
                         description: ReadyCondition indicates whether the Subscription is ready or not.
                         type: object
+                        required:
+                          - type
+                          - status
                         properties:
-                          !!merge <<: *readyConditionProperties
+                          lastTransitionTime:
+                            description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
+                            type: string
+                          message:
+                            description: A human readable message indicating details about the transition.
+                            type: string
+                          reason:
+                            description: The reason for the condition's last transition.
+                            type: string
+                          severity:
+                            description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
+                            type: string
+                          status:
+                            description: Status of the condition, one of True, False, Unknown.
+                            type: string
+                          type:
+                            description: Type of condition.
+                            type: string
                       subscription:
                         description: Subscription is the reference to the underlying Subscription.
                         type: object
                         properties:
-                          !!merge <<: *referentProperties
+                          apiVersion:
+                            description: API version of the referent.
+                            type: string
+                          fieldPath:
+                            description: 'If referring to a piece of an object instead of an entire object, this string should contain a valid JSON/Go field access statement, such as desiredState.manifest.containers[2]. For example, if the object reference is to a container within a pod, this would take on a value like: "spec.containers{name}" (where "name" refers to the name of the container that triggered the event) or if no container name is specified "spec.containers[2]" (container with index 2 in this pod). This syntax is chosen only to have some well-defined way of referencing a part of an object.'
+                            type: string
+                          kind:
+                            description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                            type: string
+                          name:
+                            description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                            type: string
+                          namespace:
+                            description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/'
+                            type: string
+                          resourceVersion:
+                            description: 'Specific resourceVersion to which this reference is made, if any. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#concurrency-control-and-consistency'
+                            type: string
+                          uid:
+                            description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
+                            type: string
       additionalPrinterColumns:
         - name: URL
           type: string
@@ -2623,14 +2644,6 @@ spec:
       - knative
       - flows
   scope: Namespaced
-  conversion:
-    strategy: Webhook
-    webhook:
-      conversionReviewVersions: ["v1", "v1beta1"]
-      clientConfig:
-        service:
-          name: eventing-webhook
-          namespace: knative-eventing
 
 ---
 # Copyright 2020 The Knative Authors
@@ -2651,7 +2664,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
@@ -2841,7 +2854,7 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
 spec:
   group: messaging.knative.dev
@@ -2853,21 +2866,18 @@ spec:
         status: {}
       schema:
         openAPIV3Schema:
-          type: object
           description: 'Subscription routes events received on a Channel to a DNS name and corresponds to the subscriptions.channels.knative.dev CRD.'
+          type: object
           properties:
             spec:
               type: object
-              description: 'Specifies the Channel for incoming events, a Subscriber target for processing those events and where to put the result of the processing. Only From (where the events are coming from) is always required. You can optionally only Process the events (results in no output events) by leaving out the Result. You can also perform an identity transformation on the incoming events by leaving out the Subscriber and only specifying Result.
-
-                The following are all valid specifications: channel --[subscriber]--> reply Sink, no outgoing events: channel -- subscriber no-op function (identity transformation): channel --> reply'
               properties:
                 channel:
                   description: 'Reference to a channel that will be used to create the subscription You can specify only the following fields of the ObjectReference: - Kind - APIVersion - Name The resource pointed by this ObjectReference must meet the contract to the ChannelableSpec duck type. If the resource does not meet this contract it will be reflected in the Subscription''s status.  This field is immutable. We have no good answer on what happens to the events that are currently in the channel being consumed from and what the semantics there should be. For now, you can always delete the Subscription and recreate it to point to a different channel, giving the user more control over what semantics should be used (drain the channel first, possibly have events dropped, etc.)'
                   type: object
                   properties:
                     apiVersion:
-                      description: 'API version of the referent.'
+                      description: API version of the referent.
                       type: string
                     fieldPath:
                       description: 'If referring to a piece of an object instead of an entire object, this string should contain a valid JSON/Go field access statement, such as desiredState.manifest.containers[2]. For example, if the object reference is to a container within a pod, this would take on a value like: "spec.containers{name}" (where "name" refers to the name of the container that triggered the event) or if no container name is specified "spec.containers[2]" (container with index 2 in this pod). This syntax is chosen only to have some well-defined way of referencing a part of an object.'
@@ -2888,25 +2898,25 @@ spec:
                       description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
                       type: string
                 delivery:
-                  description: 'Delivery configuration'
+                  description: Delivery configuration
                   type: object
                   properties:
                     backoffDelay:
                       description: 'BackoffDelay is the delay before retrying. More information on Duration format: - https://www.iso.org/iso-8601-date-and-time-format.html - https://en.wikipedia.org/wiki/ISO_8601  For linear policy, backoff delay is backoffDelay*<numberOfRetries>. For exponential policy, backoff delay is backoffDelay*2^<numberOfRetries>.'
                       type: string
                     backoffPolicy:
-                      description: 'BackoffPolicy is the retry backoff policy (linear, exponential).'
+                      description: BackoffPolicy is the retry backoff policy (linear, exponential).
                       type: string
                     deadLetterSink:
-                      description: 'DeadLetterSink is the sink receiving event that could not be sent to a destination.'
+                      description: DeadLetterSink is the sink receiving event that could not be sent to a destination.
                       type: object
                       properties:
                         ref:
-                          description: 'Ref points to an Addressable.'
+                          description: Ref points to an Addressable.
                           type: object
                           properties:
                             apiVersion:
-                              description: 'API version of the referent.'
+                              description: API version of the referent.
                               type: string
                             kind:
                               description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -2918,22 +2928,22 @@ spec:
                               description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                               type: string
                         uri:
-                          description: 'URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.'
+                          description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                           type: string
                     retry:
-                      description: 'Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.'
+                      description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
                       format: int32
                 reply:
-                  description: 'Reply specifies (optionally) how to handle events returned from the Subscriber target.'
+                  description: Reply specifies (optionally) how to handle events returned from the Subscriber target.
                   type: object
                   properties:
                     ref:
-                      description: 'Ref points to an Addressable.'
+                      description: Ref points to an Addressable.
                       type: object
                       properties:
                         apiVersion:
-                          description: 'API version of the referent.'
+                          description: API version of the referent.
                           type: string
                         kind:
                           description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -2945,18 +2955,18 @@ spec:
                           description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                           type: string
                     uri:
-                      description: 'URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.'
+                      description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
                 subscriber:
-                  description: 'Subscriber is reference to (optional) function for processing events. Events from the Channel will be delivered here and replies are sent to a Destination as specified by the Reply.'
+                  description: Subscriber is reference to (optional) function for processing events. Events from the Channel will be delivered here and replies are sent to a Destination as specified by the Reply.
                   type: object
                   properties:
                     ref:
-                      description: 'Ref points to an Addressable.'
+                      description: Ref points to an Addressable.
                       type: object
                       properties:
                         apiVersion:
-                          description: 'API version of the referent.'
+                          description: API version of the referent.
                           type: string
                         kind:
                           description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -2968,56 +2978,58 @@ spec:
                           description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                           type: string
                     uri:
-                      description: 'URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.'
+                      description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
             status:
               type: object
-              description: Status (computed) for a subscription
               properties:
                 annotations:
-                  description: 'Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.'
+                  description: Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.
                   type: object
                   x-kubernetes-preserve-unknown-fields: true
                 conditions:
-                  description: 'Conditions the latest available observations of a resource''s current state.'
+                  description: Conditions the latest available observations of a resource's current state.
                   type: array
                   items:
                     type: object
+                    required:
+                      - type
+                      - status
                     properties:
                       lastTransitionTime:
-                        description: 'LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).'
+                        description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
                         type: string
                       message:
-                        description: 'A human readable message indicating details about the transition.'
+                        description: A human readable message indicating details about the transition.
                         type: string
                       reason:
-                        description: 'The reason for the condition''s last transition.'
+                        description: The reason for the condition's last transition.
                         type: string
                       severity:
-                        description: 'Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.'
+                        description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
                         type: string
                       status:
-                        description: 'Status of the condition, one of True, False, Unknown.'
+                        description: Status of the condition, one of True, False, Unknown.
                         type: string
                       type:
-                        description: 'Type of condition.'
+                        description: Type of condition.
                         type: string
                 observedGeneration:
-                  description: 'ObservedGeneration is the ''Generation'' of the Service that was last processed by the controller.'
+                  description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
                   format: int64
                 physicalSubscription:
-                  description: 'PhysicalSubscription is the fully resolved values that this Subscription represents.'
+                  description: PhysicalSubscription is the fully resolved values that this Subscription represents.
                   type: object
                   properties:
                     deadLetterSinkUri:
-                      description: 'ReplyURI is the fully resolved URI for the spec.delivery.deadLetterSink.'
+                      description: ReplyURI is the fully resolved URI for the spec.delivery.deadLetterSink.
                       type: string
                     replyUri:
-                      description: 'ReplyURI is the fully resolved URI for the spec.reply.'
+                      description: ReplyURI is the fully resolved URI for the spec.reply.
                       type: string
                     subscriberUri:
-                      description: 'SubscriberURI is the fully resolved URI for spec.subscriber.'
+                      description: SubscriberURI is the fully resolved URI for spec.subscriber.
                       type: string
       additionalPrinterColumns:
         - name: Age
@@ -3040,14 +3052,6 @@ spec:
     shortNames:
       - sub
   scope: Namespaced
-  conversion:
-    strategy: Webhook
-    webhook:
-      conversionReviewVersions: ["v1", "v1beta1"]
-      clientConfig:
-        service:
-          name: eventing-webhook
-          namespace: knative-eventing
 
 ---
 # Copyright 2020 The Knative Authors
@@ -3069,7 +3073,7 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -3248,7 +3252,7 @@ kind: ClusterRole
 metadata:
   name: addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3260,7 +3264,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: service-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3278,7 +3282,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: serving-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3299,7 +3303,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: channel-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3324,7 +3328,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: broker-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3343,7 +3347,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: messaging-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3364,7 +3368,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: flows-addressable-resolver
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/addressable: "true"
 # Do not use this role directly. These rules will be added to the "addressable-resolver" role.
 rules:
@@ -3400,7 +3404,7 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-filter
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   - apiGroups:
       - ""
@@ -3425,7 +3429,7 @@ kind: ClusterRole
 metadata:
   name: eventing-broker-ingress
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   - apiGroups:
       - ""
@@ -3441,7 +3445,7 @@ kind: ClusterRole
 metadata:
   name: eventing-config-reader
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   - apiGroups:
       - ""
@@ -3473,7 +3477,7 @@ kind: ClusterRole
 metadata:
   name: channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3485,7 +3489,7 @@ kind: ClusterRole
 metadata:
   name: meta-channelable-manipulator
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/channelable: "true"
 # Do not use this role directly. These rules will be added to the "channelable-manipulator" role.
 rules:
@@ -3522,7 +3526,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-eventing-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["eventing.knative.dev"]
@@ -3534,7 +3538,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-messaging-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["messaging.knative.dev"]
@@ -3546,7 +3550,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-flows-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["flows.knative.dev"]
@@ -3558,7 +3562,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-sources-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["sources.knative.dev"]
@@ -3570,7 +3574,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: knative-bindings-namespaced-admin
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     rbac.authorization.k8s.io/aggregate-to-admin: "true"
 rules:
   - apiGroups: ["bindings.knative.dev"]
@@ -3583,7 +3587,7 @@ metadata:
   name: knative-eventing-namespaced-edit
   labels:
     rbac.authorization.k8s.io/aggregate-to-edit: "true"
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
     resources: ["*"]
@@ -3595,7 +3599,7 @@ metadata:
   name: knative-eventing-namespaced-view
   labels:
     rbac.authorization.k8s.io/aggregate-to-view: "true"
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev"]
     resources: ["*"]
@@ -3621,7 +3625,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-controller
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   - apiGroups:
       - ""
@@ -3755,7 +3759,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-pingsource-mt-adapter
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   - apiGroups:
       - ""
@@ -3821,7 +3825,7 @@ kind: ClusterRole
 metadata:
   name: podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3833,7 +3837,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: builtin-podspecable-binding
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/podspecable: "true"
 # Do not use this role directly. These rules will be added to the "podspecable-binding role.
 rules:
@@ -3879,7 +3883,7 @@ kind: ClusterRole
 metadata:
   name: source-observer
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 aggregationRule:
   clusterRoleSelectors:
     - matchLabels:
@@ -3891,7 +3895,7 @@ apiVersion: rbac.authorization.k8s.io/v1
 metadata:
   name: eventing-sources-source-observer
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     duck.knative.dev/source: "true"
 # Do not use this role directly. These rules will be added to the "source-observer" role.
 rules:
@@ -3927,7 +3931,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-sources-controller
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   - apiGroups:
       - ""
@@ -4012,7 +4016,7 @@ kind: ClusterRole
 metadata:
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   # For watching logging configuration and getting certs.
   - apiGroups:
@@ -4103,7 +4107,7 @@ metadata:
   namespace: knative-eventing
   name: knative-eventing-webhook
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 rules:
   # For manipulating certs into secrets.
   - apiGroups:
@@ -4138,7 +4142,7 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: config.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4174,7 +4178,7 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4206,7 +4210,7 @@ kind: ValidatingWebhookConfiguration
 metadata:
   name: validation.webhook.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
@@ -4239,7 +4243,7 @@ metadata:
   name: eventing-webhook-certs
   namespace: knative-eventing
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 # The data is populated at install time.
 
 ---
@@ -4262,7 +4266,7 @@ kind: MutatingWebhookConfiguration
 metadata:
   name: sinkbindings.webhook.sources.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210506-2f44af0cd"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
 webhooks:
   - admissionReviewVersions: ["v1", "v1beta1"]
     clientConfig:
diff --git a/third_party/eventing-latest/eventing-crds.yaml b/third_party/eventing-latest/eventing-crds.yaml
index 4e2fddde2..e100ac213 100644
--- a/third_party/eventing-latest/eventing-crds.yaml
+++ b/third_party/eventing-latest/eventing-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   creationTimestamp: null
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -43,45 +43,43 @@ spec:
         status: {}
       schema:
         openAPIV3Schema:
-          type: object
           description: 'ApiServerSource is an event source that brings Kubernetes API server events into Knative.'
+          type: object
           properties:
             spec:
               type: object
-              description: 'ApiServerSourceSpec defines the desired state of ApiServerSource (from the client).'
               required:
                 - resources
               properties:
                 ceOverrides:
-                  description: 'CloudEventOverrides defines overrides to control the output format and modifications of the event sent to the sink.'
+                  description: CloudEventOverrides defines overrides to control the output format and modifications of the event sent to the sink.
                   type: object
                   properties:
                     extensions:
-                      description: 'Extensions specify what attribute are added or overridden on the outbound event. Each `Extensions` key-value pair are set on the event as an attribute extension independently.'
+                      description: Extensions specify what attribute are added or overridden on the outbound event. Each `Extensions` key-value pair are set on the event as an attribute extension independently.
                       type: object
-                      additionalProperties:
-                        type: string
+                      x-kubernetes-preserve-unknown-fields: true
                 mode:
-                  description: 'EventMode controls the format of the event. `Reference` sends a dataref event type for the resource under watch. `Resource` send the full resource lifecycle event. Defaults to `Reference`'
+                  description: EventMode controls the format of the event. `Reference` sends a dataref event type for the resource under watch. `Resource` send the full resource lifecycle event. Defaults to `Reference`
                   type: string
                 owner:
-                  description: 'ResourceOwner is an additional filter to only track resources that are owned by a specific resource type. If ResourceOwner matches Resources[n] then Resources[n] is allowed to pass the ResourceOwner filter.'
+                  description: ResourceOwner is an additional filter to only track resources that are owned by a specific resource type. If ResourceOwner matches Resources[n] then Resources[n] is allowed to pass the ResourceOwner filter.
                   type: object
                   properties:
                     apiVersion:
-                      description: 'APIVersion - the API version of the resource to watch.'
+                      description: APIVersion - the API version of the resource to watch.
                       type: string
                     kind:
                       description: 'Kind of the resource to watch. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
                       type: string
                 resources:
-                  description: 'Resource are the resources this source will track and send related lifecycle events from the Kubernetes ApiServer, with an optional label selector to help filter.'
+                  description: Resource are the resources this source will track and send related lifecycle events from the Kubernetes ApiServer, with an optional label selector to help filter.
                   type: array
                   items:
                     type: object
                     properties:
                       apiVersion:
-                        description: 'APIVersion - the API version of the resource to watch.'
+                        description: APIVersion - the API version of the resource to watch.
                         type: string
                       kind:
                         description: 'Kind of the resource to watch. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -91,39 +89,39 @@ spec:
                         type: object
                         properties:
                           matchExpressions:
-                            description: 'matchExpressions is a list of label selector requirements. The requirements are ANDed.'
+                            description: matchExpressions is a list of label selector requirements. The requirements are ANDed.
                             type: array
                             items:
                               type: object
                               properties:
                                 key:
-                                  description: 'key is the label key that the selector applies to.'
+                                  description: key is the label key that the selector applies to.
                                   type: string
                                 operator:
-                                  description: 'operator represents a key''s relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist.'
+                                  description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist.
                                   type: string
                                 values:
-                                  description: 'values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch.'
+                                  description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch.
                                   type: array
                                   items:
                                     type: string
                           matchLabels:
-                            description: 'matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed.'
+                            description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed.
                             type: object
                             x-kubernetes-preserve-unknown-fields: true
                 serviceAccountName:
-                  description: 'ServiceAccountName is the name of the ServiceAccount to use to run this source. Defaults to default if not set.'
+                  description: ServiceAccountName is the name of the ServiceAccount to use to run this source. Defaults to default if not set.
                   type: string
                 sink:
-                  description: 'Sink is a reference to an object that will resolve to a uri to use as the sink.'
+                  description: Sink is a reference to an object that will resolve to a uri to use as the sink.
                   type: object
                   properties:
                     ref:
-                      description: 'Ref points to an Addressable.'
+                      description: Ref points to an Addressable.
                       type: object
                       properties:
                         apiVersion:
-                          description: 'API version of the referent.'
+                          description: API version of the referent.
                           type: string
                         kind:
                           description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -135,30 +133,29 @@ spec:
                           description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                           type: string
                     uri:
-                      description: 'URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.'
+                      description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
             status:
               type: object
-              description: 'ApiServerSourceStatus defines the observed state of ApiServerSource (from the controller).'
               properties:
                 annotations:
-                  description: 'Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.'
+                  description: Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.
                   type: object
                   x-kubernetes-preserve-unknown-fields: true
                 ceAttributes:
-                  description: 'CloudEventAttributes are the specific attributes that the Source uses as part of its CloudEvents.'
+                  description: CloudEventAttributes are the specific attributes that the Source uses as part of its CloudEvents.
                   type: array
                   items:
                     type: object
                     properties:
                       source:
-                        description: 'Source is the CloudEvents source attribute.'
+                        description: Source is the CloudEvents source attribute.
                         type: string
                       type:
-                        description: 'Type refers to the CloudEvent type attribute.'
+                        description: Type refers to the CloudEvent type attribute.
                         type: string
                 conditions:
-                  description: 'Conditions the latest available observations of a resource''s current state.'
+                  description: Conditions the latest available observations of a resource's current state.
                   type: array
                   items:
                     type: object
@@ -167,29 +164,29 @@ spec:
                       - status
                     properties:
                       lastTransitionTime:
-                        description: 'LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).'
+                        description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
                         type: string
                       message:
-                        description: 'A human readable message indicating details about the transition.'
+                        description: A human readable message indicating details about the transition.
                         type: string
                       reason:
-                        description: 'The reason for the condition''s last transition.'
+                        description: The reason for the condition's last transition.
                         type: string
                       severity:
-                        description: 'Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.'
+                        description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
                         type: string
                       status:
-                        description: 'Status of the condition, one of True, False, Unknown.'
+                        description: Status of the condition, one of True, False, Unknown.
                         type: string
                       type:
-                        description: 'Type of condition.'
+                        description: Type of condition.
                         type: string
                 observedGeneration:
-                  description: 'ObservedGeneration is the "Generation" of the Service that was last processed by the controller.'
+                  description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
                   format: int64
                 sinkUri:
-                  description: 'SinkURI is the current active sink URI that has been configured for the Source.'
+                  description: SinkURI is the current active sink URI that has been configured for the Source.
                   type: string
       additionalPrinterColumns:
         - name: Sink
@@ -234,7 +231,7 @@ kind: CustomResourceDefinition
 metadata:
   name: brokers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -395,7 +392,7 @@ kind: CustomResourceDefinition
 metadata:
   name: channels.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
@@ -663,7 +660,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -677,32 +674,31 @@ spec:
       subresources:
         status: {}
       schema:
-        openAPIV3Schema: &openAPIV3Schema
-          type: object
+        openAPIV3Schema:
           description: 'ContainerSource is an event source that starts a container image which generates events under certain situations and sends messages to a sink URI'
+          type: object
           properties:
             spec:
               type: object
-              description: 'ContainerSourceSpec defines the desired state of ContainerSource (from the client).'
               properties:
                 ceOverrides:
-                  description: 'CloudEventOverrides defines overrides to control the output format and modifications of the event sent to the sink.'
+                  description: CloudEventOverrides defines overrides to control the output format and modifications of the event sent to the sink.
                   type: object
                   properties:
                     extensions:
-                      description: 'Extensions specify what attribute are added or overridden on the outbound event. Each `Extensions` key-value pair are set on the event as an attribute extension independently.'
+                      description: Extensions specify what attribute are added or overridden on the outbound event. Each `Extensions` key-value pair are set on the event as an attribute extension independently.
                       type: object
                       x-kubernetes-preserve-unknown-fields: true
                 sink:
-                  description: 'Sink is a reference to an object that will resolve to a uri to use as the sink.'
+                  description: Sink is a reference to an object that will resolve to a uri to use as the sink.
                   type: object
                   properties:
                     ref:
-                      description: 'Ref points to an Addressable.'
+                      description: Ref points to an Addressable.
                       type: object
                       properties:
                         apiVersion:
-                          description: 'API version of the referent.'
+                          description: API version of the referent.
                           type: string
                         kind:
                           description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
@@ -714,134 +710,34 @@ spec:
                           description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                           type: string
                     uri:
-                      description: 'URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.'
+                      description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
+                # WARNING: the schema tool can not parse PodTemplateSpec, stub here and redirect to Deployment documentation.
                 template:
-                  description: 'Template describes the pods that will be created'
                   type: object
-                  properties:
-                    annotations:
-                      description: 'Annotations is an unstructured key value map stored with a resource that may be set by external tools to store and retrieve arbitrary metadata. They are not queryable and should be preserved when modifying objects. More info: http://kubernetes.io/docs/user-guide/annotations'
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
-                    clusterName:
-                      description: 'The name of the cluster which the object belongs to. This is used to distinguish resources with same name and namespace in different clusters. This field is not set anywhere right now and apiserver is going to ignore it if set in create or update request.'
-                      type: string
-                    creationTimestamp:
-                      description: 'CreationTimestamp is a timestamp representing the server time when this object was created. It is not guaranteed to be set in happens-before order across separate operations. Clients may not set this value. It is represented in RFC3339 form and is in UTC.  Populated by the system. Read-only. Null for lists. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata'
-                      type: string
-                    deletionGracePeriodSeconds:
-                      description: 'Number of seconds allowed for this object to gracefully terminate before it will be removed from the system. Only set when deletionTimestamp is also set. May only be shortened. Read-only.'
-                      type: integer
-                      format: int64
-                    deletionTimestamp:
-                      description: 'DeletionTimestamp is RFC 3339 date and time at which this resource will be deleted. This field is set by the server when a graceful deletion is requested by the user, and is not directly settable by a client. The resource is expected to be deleted (no longer visible from resource lists, and not reachable by name) after the time in this field, once the finalizers list is empty. As long as the finalizers list contains items, deletion is blocked. Once the deletionTimestamp is set, this value may not be unset or be set further into the future, although it may be shortened or the resource may be deleted prior to this time. For example, a user may request that a pod is deleted in 30 seconds. The Kubelet will react by sending a graceful termination signal to the containers in the pod. After that 30 seconds, the Kubelet will send a hard termination signal (SIGKILL) to the container and after cleanup, remove the pod from the API. In the presence of network partitions, this object may still exist after this timestamp, until an administrator or automated process can determine the resource is fully terminated. If not set, graceful deletion of the object has not been requested.  Populated by the system when a graceful deletion is requested. Read-only. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata'
-                      type: string
-                    finalizers:
-                      description: 'Must be empty before the object is deleted from the registry. Each entry is an identifier for the responsible component that will remove the entry from the list. If the deletionTimestamp of the object is non-nil, entries in this list can only be removed. Finalizers may be processed and removed in any order.  Order is NOT enforced because it introduces significant risk of stuck finalizers. finalizers is a shared field, any actor with permission can reorder it. If the finalizer list is processed in order, then this can lead to a situation in which the component responsible for the first finalizer in the list is waiting for a signal (field value, external system, or other) produced by a component responsible for a finalizer later in the list, resulting in a deadlock. Without enforced ordering finalizers are free to order amongst themselves and are not vulnerable to ordering changes in the list.'
-                      type: array
-                      items:
-                        type: string
-                    generateName:
-                      description: 'GenerateName is an optional prefix, used by the server, to generate a unique name ONLY IF the Name field has not been provided. If this field is used, the name returned to the client will be different than the name passed. This value will also be combined with a unique suffix. The provided value has the same validation rules as the Name field, and may be truncated by the length of the suffix required to make the value unique on the server.  If this field is specified and the generated name exists, the server will NOT return a 409 - instead, it will either return 201 Created or 500 with Reason ServerTimeout indicating a unique name could not be found in the time allotted, and the client should retry (optionally after the time indicated in the Retry-After header).  Applied only if Name is not specified. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#idempotency'
-                      type: string
-                    generation:
-                      description: 'A sequence number representing a specific generation of the desired state. Populated by the system. Read-only.'
-                      type: integer
-                      format: int64
-                    labels:
-                      description: 'Map of string keys and values that can be used to organize and categorize (scope and select) objects. May match selectors of replication controllers and services. More info: http://kubernetes.io/docs/user-guide/labels'
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
-                    managedFields:
-                      description: 'ManagedFields maps workflow-id and version to the set of fields that are managed by that workflow. This is mostly for internal housekeeping, and users typically shouldn''t need to set or understand this field. A workflow can be the user''s name, a controller''s name, or the name of a specific apply path like "ci-cd". The set of fields is always in the version that the workflow used when modifying the object. '
-                      type: array
-                      items:
-                        type: object
-                        properties:
-                          apiVersion:
-                            description: 'APIVersion defines the version of this resource that this field set applies to. The format is "group/version" just like the top-level APIVersion field. It is necessary to track the version of a field set because it cannot be automatically converted.'
-                            type: string
-                          fieldsType:
-                            description: 'FieldsType is the discriminator for the different fields format and version. There is currently only one possible value: "FieldsV1"'
-                            type: string
-                          fieldsV1:
-                            description: 'FieldsV1 holds the first JSON version format as described in the "FieldsV1" type.'
-                            type: string
-                          manager:
-                            description: 'Manager is an identifier of the workflow managing these fields.'
-                            type: string
-                          operation:
-                            description: 'Operation is the type of operation which lead to this ManagedFieldsEntry being created. The only valid values for this field are "Apply" and "Update".'
-                            type: string
-                          time:
-                            description: 'Time is timestamp of when these fields were set. It should always be empty if Operation is "Apply"'
-                            type: string
-                    name:
-                      description: 'Name must be unique within a namespace. Is required when creating resources, although some resources may allow a client to request the generation of an appropriate name automatically. Name is primarily intended for creation idempotence and configuration definition. Cannot be updated. More info: http://kubernetes.io/docs/user-guide/identifiers#names'
-                      type: string
-                    namespace:
-                      description: 'Namespace defines the space within each name must be unique. An empty namespace is equivalent to the "default" namespace, but "default" is the canonical representation. Not all objects are required to be scoped to a namespace - the value of this field for those objects will be empty.  Must be a DNS_LABEL. Cannot be updated. More info: http://kubernetes.io/docs/user-guide/namespaces'
-                      type: string
-                    ownerReferences:
-                      description: 'List of objects depended by this object. If ALL objects in the list have been deleted, this object will be garbage collected. If this object is managed by a controller, then an entry in this list will point to this controller, with the controller field set to true. There cannot be more than one managing controller.'
-                      type: array
-                      items:
-                        type: object
-                        properties:
-                          apiVersion:
-                            description: 'API version of the referent.'
-                            type: string
-                          blockOwnerDeletion:
-                            description: 'If true, AND if the owner has the "foregroundDeletion" finalizer, then the owner cannot be deleted from the key-value store until this reference is removed. Defaults to false. To set this field, a user needs "delete" permission of the owner, otherwise 422 (Unprocessable Entity) will be returned.'
-                            type: boolean
-                          controller:
-                            description: 'If true, this reference points to the managing controller.'
-                            type: boolean
-                          kind:
-                            description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
-                            type: string
-                          name:
-                            description: 'Name of the referent. More info: http://kubernetes.io/docs/user-guide/identifiers#names'
-                            type: string
-                          uid:
-                            description: 'UID of the referent. More info: http://kubernetes.io/docs/user-guide/identifiers#uids'
-                            type: string
-                    resourceVersion:
-                      description: 'An opaque value that represents the internal version of this object that can be used by clients to determine when objects have changed. May be used for optimistic concurrency, change detection, and the watch operation on a resource or set of resources. Clients must treat these values as opaque and passed unmodified back to the server. They may only be valid for a particular resource or set of resources.  Populated by the system. Read-only. Value must be treated as opaque by clients and . More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#concurrency-control-and-consistency'
-                      type: string
-                    selfLink:
-                      description: 'SelfLink is a URL representing this object. Populated by the system. Read-only.  DEPRECATED Kubernetes will stop propagating this field in 1.20 release and the field is planned to be removed in 1.21 release.'
-                      type: string
-                    spec:
-                      description: 'Specification of the desired behavior of the pod. More info: Type ''kubectl explain pod.spec''. Also, https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#spec-and-status'
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
-                    uid:
-                      description: 'UID is the unique in time and space value for this object. It is typically generated by the server on successful creation of a resource and is not allowed to change on PUT operations.  Populated by the system. Read-only. More info: http://kubernetes.io/docs/user-guide/identifiers#uids'
-                      type: string
+                  x-kubernetes-preserve-unknown-fields: true
+                  description: 'A template in the shape of `Deployment.spec.template` to be used for this ContainerSource. More info: https://kubernetes.io/docs/concepts/workloads/controllers/deployment/'
             status:
               type: object
-              description: 'ContainerSourceStatus defines the observed state of ContainerSource (from the controller).'
               properties:
                 annotations:
-                  description: 'Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.'
+                  description: Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.
                   type: object
                   x-kubernetes-preserve-unknown-fields: true
                 ceAttributes:
-                  description: 'CloudEventAttributes are the specific attributes that the Source uses as part of its CloudEvents.'
+                  description: CloudEventAttributes are the specific attributes that the Source uses as part of its CloudEvents.
                   type: array
                   items:
                     type: object
                     properties:
                       source:
-                        description: 'Source is the CloudEvents source attribute.'
+                        description: Source is the CloudEvents source attribute.
                         type: string
                       type:
-                        description: 'Type refers to the CloudEvent type attribute.'
+                        description: Type refers to the CloudEvent type attribute.
                         type: string
                 conditions:
-                  description: 'Conditions the latest available observations of a resource''s current state.'
+                  description: Conditions the latest available observations of a resource's current state.
                   type: array
                   items:
                     type: object
@@ -850,29 +746,29 @@ spec:
                       - status
                     properties:
                       lastTransitionTime:
-                        description: 'LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).'
+                        description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
                         type: string
                       message:
-                        description: 'A human readable message indicating details about the transition.'
+                        description: A human readable message indicating details about the transition.
                         type: string
                       reason:
-                        description: 'The reason for the condition''s last transition.'
+                        description: The reason for the condition's last transition.
                         type: string
                       severity:
-                        description: 'Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.'
+                        description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
                         type: string
                       status:
-                        description: 'Status of the condition, one of True, False, Unknown.'
+                        description: Status of the condition, one of True, False, Unknown.
                         type: string
                       type:
                         description: Type of condition.
                         type: string
                 observedGeneration:
-                  description: 'ObservedGeneration is the "Generation" of the Service that was last processed by the controller.'
+                  description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
                   format: int64
                 sinkUri:
-                  description: 'SinkURI is the current active sink URI that has been configured for the Source.'
+                  description: SinkURI is the current active sink URI that has been configured for the Source.
                   type: string
       additionalPrinterColumns:
         - name: Sink
@@ -916,7 +812,7 @@ kind: CustomResourceDefinition
 metadata:
   name: eventtypes.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
@@ -1043,7 +939,7 @@ kind: CustomResourceDefinition
 metadata:
   name: parallels.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1295,7 +1191,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     knative.dev/crd-install: "true"
@@ -1484,7 +1380,7 @@ kind: CustomResourceDefinition
 metadata:
   name: sequences.flows.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
     duck.knative.dev/addressable: "true"
 spec:
@@ -1516,12 +1412,11 @@ spec:
                       type: string
                     spec:
                       description: Spec defines the Spec to use for each channel created. Passed in verbatim to the Channel CRD as Spec section.
-                      type: object
-                      x-kubernetes-preserve-unknown-fields: true
+                      type: string
                 reply:
                   description: Reply is a Reference to where the result of the last Subscriber gets sent to.
                   type: object
-                  properties: &addressableProperties
+                  properties:
                     ref:
                       description: Ref points to an Addressable.
                       type: object
@@ -1547,7 +1442,6 @@ spec:
                   items:
                     type: object
                     properties:
-                      !!merge <<: *addressableProperties
                       delivery:
                         description: Delivery is the delivery specification for events to the subscriber This includes things like retries, DLQ, etc.
                         type: object
@@ -1562,23 +1456,63 @@ spec:
                             description: DeadLetterSink is the sink receiving event that could not be sent to a destination.
                             type: object
                             properties:
-                              !!merge <<: *addressableProperties
+                              ref:
+                                description: Ref points to an Addressable.
+                                type: object
+                                properties:
+                                  apiVersion:
+                                    description: API version of the referent.
+                                    type: string
+                                  kind:
+                                    description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                                    type: string
+                                  name:
+                                    description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                                    type: string
+                                  namespace:
+                                    description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
+                                    type: string
+                              uri:
+                                description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
+                                type: string
                           retry:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
                             format: int32
+                      ref:
+                        description: Ref points to an Addressable.
+                        type: object
+                        properties:
+                          apiVersion:
+                            description: API version of the referent.
+                            type: string
+                          kind:
+                            description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                            type: string
+                          name:
+                            description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                            type: string
+                          namespace:
+                            description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
+                            type: string
+                      uri:
+                        description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
+                        type: string
             status:
               description: Status represents the current state of the Sequence. This data may be out of date.
               type: object
               properties:
                 address:
                   type: object
+                  required:
+                    - url
                   properties:
                     url:
                       type: string
                 annotations:
                   description: Annotations is additional Status fields for the Resource to save some additional State as well as convey more information to the user. This is roughly akin to Annotations on any k8s resource, just the reconciler conveying richer information outwards.
                   type: object
+                  x-kubernetes-preserve-unknown-fields: true
                 channelStatuses:
                   description: ChannelStatuses is an array of corresponding Channel statuses. Matches the Spec.Steps array in the order.
                   type: array
@@ -1588,7 +1522,7 @@ spec:
                       channel:
                         description: Channel is the reference to the underlying channel.
                         type: object
-                        properties: &referentProperties
+                        properties:
                           apiVersion:
                             description: API version of the referent.
                             type: string
@@ -1613,8 +1547,13 @@ spec:
                       ready:
                         description: ReadyCondition indicates whether the Channel is ready or not.
                         type: object
-                        x-kubernetes-preserve-unknown-fields: true
-                        properties: &readyConditionProperties
+                        required:
+                          - type
+                          - status
+                        properties:
+                          lastTransitionTime:
+                            description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
+                            type: string
                           message:
                             description: A human readable message indicating details about the transition.
                             type: string
@@ -1635,8 +1574,28 @@ spec:
                   type: array
                   items:
                     type: object
+                    required:
+                      - type
+                      - status
                     properties:
-                      !!merge <<: *readyConditionProperties
+                      lastTransitionTime:
+                        description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
+                        type: string
+                      message:
+                        description: A human readable message indicating details about the transition.
+                        type: string
+                      reason:
+                        description: The reason for the condition's last transition.
+                        type: string
+                      severity:
+                        description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
+                        type: string
+                      status:
+                        description: Status of the condition, one of True, False, Unknown.
+                        type: string
+                      type:
+                        description: Type of condition.
+                        type: string
                 observedGeneration:
                   description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
@@ -1650,13 +1609,53 @@ spec:
                       ready:
                         description: ReadyCondition indicates whether the Subscription is ready or not.
                         type: object
+                        required:
+                          - type
+                          - status
                         properties:
-                          !!merge <<: *readyConditionProperties
+                          lastTransitionTime:
+                            description: LastTransitionTime is the last time the condition transitioned from one status to another. We use VolatileTime in place of metav1.Time to exclude this from creating equality.Semantic differences (all other things held constant).
+                            type: string
+                          message:
+                            description: A human readable message indicating details about the transition.
+                            type: string
+                          reason:
+                            description: The reason for the condition's last transition.
+                            type: string
+                          severity:
+                            description: Severity with which to treat failures of this type of condition. When this is not specified, it defaults to Error.
+                            type: string
+                          status:
+                            description: Status of the condition, one of True, False, Unknown.
+                            type: string
+                          type:
+                            description: Type of condition.
+                            type: string
                       subscription:
                         description: Subscription is the reference to the underlying Subscription.
                         type: object
                         properties:
-                          !!merge <<: *referentProperties
+                          apiVersion:
+                            description: API version of the referent.
+                            type: string
+                          fieldPath:
+                            description: 'If referring to a piece of an object instead of an entire object, this string should contain a valid JSON/Go field access statement, such as desiredState.manifest.containers[2]. For example, if the object reference is to a container within a pod, this would take on a value like: "spec.containers{name}" (where "name" refers to the name of the container that triggered the event) or if no container name is specified "spec.containers[2]" (container with index 2 in this pod). This syntax is chosen only to have some well-defined way of referencing a part of an object.'
+                            type: string
+                          kind:
+                            description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                            type: string
+                          name:
+                            description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                            type: string
+                          namespace:
+                            description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/'
+                            type: string
+                          resourceVersion:
+                            description: 'Specific resourceVersion to which this reference is made, if any. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#concurrency-control-and-consistency'
+                            type: string
+                          uid:
+                            description: 'UID of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#uids'
+                            type: string
       additionalPrinterColumns:
         - name: URL
           type: string
@@ -1699,7 +1698,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     eventing.knative.dev/source: "true"
     duck.knative.dev/source: "true"
     duck.knative.dev/binding: "true"
@@ -1889,7 +1888,7 @@ kind: CustomResourceDefinition
 metadata:
   name: subscriptions.messaging.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
 spec:
   group: messaging.knative.dev
@@ -2108,7 +2107,7 @@ kind: CustomResourceDefinition
 metadata:
   name: triggers.eventing.knative.dev
   labels:
-    eventing.knative.dev/release: "v20210507-88d8a3eda"
+    eventing.knative.dev/release: "v20210512-8e157adaa"
     knative.dev/crd-install: "true"
 spec:
   group: eventing.knative.dev
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/listers.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/listers.go
index 5e9bf8d84..cc6d07494 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/listers.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/listers.go
@@ -207,3 +207,11 @@ func (l *Listers) GetConfigMapLister() corev1listers.ConfigMapLister {
 func (l *Listers) GetCustomResourceDefinitionLister() apiextensionsv1listers.CustomResourceDefinitionLister {
 	return apiextensionsv1listers.NewCustomResourceDefinitionLister(l.indexerFor(&apiextensionsv1.CustomResourceDefinition{}))
 }
+
+func (l *Listers) GetNodeLister() corev1listers.NodeLister {
+	return corev1listers.NewNodeLister(l.indexerFor(&corev1.Node{}))
+}
+
+func (l *Listers) GetPodLister() corev1listers.PodLister {
+	return corev1listers.NewPodLister(l.indexerFor(&corev1.Pod{}))
+}
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_chain_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_chain_test_helper.go
deleted file mode 100644
index d46e39a39..000000000
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_chain_test_helper.go
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
-Copyright 2019 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 helpers
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	. "github.com/cloudevents/sdk-go/v2/test"
-	"github.com/google/uuid"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/recordevents"
-	"knative.dev/eventing/test/lib/resources"
-)
-
-// ChannelChainTestHelper is the helper function for channel_chain_test
-func ChannelChainTestHelper(
-	ctx context.Context,
-	t *testing.T,
-	subscriptionVersion SubscriptionVersion,
-	channelTestRunner testlib.ComponentsTestRunner,
-	options ...testlib.SetupClientOption) {
-	const (
-		senderName          = "e2e-channelchain-sender"
-		recordEventsPodName = "e2e-channelchain-recordevents-pod"
-	)
-	channelNames := []string{"e2e-channelchain1", "e2e-channelchain2"}
-	// subscriptionNames1 corresponds to Subscriptions on channelNames[0]
-	subscriptionNames1 := []string{"e2e-channelchain-subs11", "e2e-channelchain-subs12"}
-	// subscriptionNames2 corresponds to Subscriptions on channelNames[1]
-	subscriptionNames2 := []string{"e2e-channelchain-subs21"}
-	eventSource := fmt.Sprintf("http://%s.svc/", senderName)
-
-	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(st *testing.T, channel metav1.TypeMeta) {
-		client := testlib.Setup(st, true, options...)
-		defer testlib.TearDown(client)
-
-		// create channels
-		client.CreateChannelsOrFail(channelNames, &channel)
-		client.WaitForResourcesReadyOrFail(&channel)
-
-		// create loggerPod and expose it as a service
-		eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, recordEventsPodName)
-		// create subscription to subscribe the channel, and forward the received events to the logger service
-		switch subscriptionVersion {
-		case SubscriptionV1:
-			// create subscriptions that subscribe the first channel, and reply events directly to the second channel
-			client.CreateSubscriptionsOrFail(
-				subscriptionNames1,
-				channelNames[0],
-				&channel,
-				resources.WithReplyForSubscription(channelNames[1], &channel),
-			)
-			// create subscriptions that subscribe the second channel, and call the logging service
-			client.CreateSubscriptionsOrFail(
-				subscriptionNames2,
-				channelNames[1],
-				&channel,
-				resources.WithSubscriberForSubscription(recordEventsPodName),
-			)
-		default:
-			t.Fatalf("Invalid subscription version")
-		}
-		// wait for all test resources to be ready, so that we can start sending events
-		client.WaitForAllTestResourcesReadyOrFail(ctx)
-
-		// send CloudEvent to the first channel
-		event := cloudevents.NewEvent()
-		event.SetID("test")
-		event.SetSource(eventSource)
-		event.SetType(testlib.DefaultEventType)
-
-		body := fmt.Sprintf(`{"msg":"TestSingleEvent %s"}`, uuid.New().String())
-		if err := event.SetData(cloudevents.ApplicationJSON, []byte(body)); err != nil {
-			st.Fatalf("Cannot set the payload of the event: %s", err.Error())
-		}
-
-		client.SendEventToAddressable(ctx, senderName, channelNames[0], &channel, event)
-
-		// verify the logger service receives the event
-		eventTracker.AssertAtLeast(len(subscriptionNames1)*len(subscriptionNames2), recordevents.MatchEvent(
-			HasSource(eventSource),
-			HasData([]byte(body)),
-		))
-	})
-}
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_dls_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_dls_test_helper.go
deleted file mode 100644
index 17130e67f..000000000
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_dls_test_helper.go
+++ /dev/null
@@ -1,180 +0,0 @@
-/*
-Copyright 2019 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 helpers
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-	. "github.com/cloudevents/sdk-go/v2/test"
-	"github.com/google/uuid"
-	"k8s.io/apimachinery/pkg/api/equality"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/utils/pointer"
-
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-
-	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
-	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
-	testlib "knative.dev/eventing/test/lib"
-	"knative.dev/eventing/test/lib/recordevents"
-	"knative.dev/eventing/test/lib/resources"
-)
-
-// ChannelDeadLetterSinkTestHelper is the helper function for channel_deadlettersink_test
-func ChannelDeadLetterSinkTestHelper(
-	ctx context.Context,
-	t *testing.T,
-	subscriptionVersion SubscriptionVersion,
-	channelTestRunner testlib.ComponentsTestRunner,
-	options ...testlib.SetupClientOption) {
-	const (
-		senderName          = "e2e-channelchain-sender"
-		recordEventsPodName = "e2e-channel-dls-recordevents-pod"
-	)
-	channelNames := []string{"e2e-channel-dls"}
-	// subscriptionNames corresponds to Subscriptions
-	subscriptionNames := []string{"e2e-channel-dls-subs1"}
-
-	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(st *testing.T, channel metav1.TypeMeta) {
-		client := testlib.Setup(st, true, options...)
-		defer testlib.TearDown(client)
-
-		// create channels
-		client.CreateChannelsOrFail(channelNames, &channel)
-		client.WaitForResourcesReadyOrFail(&channel)
-
-		// create event logger pod and service as the subscriber
-		eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, recordEventsPodName)
-		// create subscriptions that subscribe to a service that does not exist
-		switch subscriptionVersion {
-		case SubscriptionV1:
-			client.CreateSubscriptionsOrFail(
-				subscriptionNames,
-				channelNames[0],
-				&channel,
-				resources.WithSubscriberForSubscription("does-not-exist"),
-				resources.WithDeadLetterSinkForSubscription(recordEventsPodName),
-			)
-		default:
-			t.Fatalf("Invalid subscription version")
-		}
-
-		// wait for all test resources to be ready, so that we can start sending events
-		client.WaitForAllTestResourcesReadyOrFail(ctx)
-
-		// send CloudEvent to the first channel
-		event := cloudevents.NewEvent()
-		event.SetID("test")
-		eventSource := fmt.Sprintf("http://%s.svc/", senderName)
-		event.SetSource(eventSource)
-		event.SetType(testlib.DefaultEventType)
-		body := fmt.Sprintf(`{"msg":"TestChannelDeadLetterSink %s"}`, uuid.New().String())
-		if err := event.SetData(cloudevents.ApplicationJSON, []byte(body)); err != nil {
-			t.Fatal("Cannot set the payload of the event:", err.Error())
-		}
-		client.SendEventToAddressable(ctx, senderName, channelNames[0], &channel, event)
-
-		// check if the logging service receives the correct number of event messages
-		eventTracker.AssertAtLeast(len(subscriptionNames), recordevents.MatchEvent(
-			HasSource(eventSource),
-			HasData([]byte(body)),
-		))
-	})
-}
-
-// ChannelDeadLetterDefaultSinkTestHelper is the helper function for channel_deadlettersink_test, but setting the delivery from the channel spec
-func ChannelDeadLetterSinkDefaultTestHelper(
-	ctx context.Context,
-	t *testing.T,
-	channelTestRunner testlib.ComponentsTestRunner,
-	options ...testlib.SetupClientOption) {
-	const (
-		senderName          = "e2e-channelchain-sender"
-		recordEventsPodName = "e2e-channel-dls-recordevents-pod"
-		channelName         = "e2e-channel-dls"
-	)
-	channelGK := messagingv1.SchemeGroupVersion.WithKind("Channel").GroupKind()
-	// subscriptionNames corresponds to Subscriptions
-	subscriptionNames := []string{"e2e-channel-dls-subs1"}
-
-	channelTestRunner.RunTests(t, testlib.FeatureBasic, func(st *testing.T, channel metav1.TypeMeta) {
-		thisChannelGk := channel.GroupVersionKind().GroupKind()
-		if equality.Semantic.DeepEqual(thisChannelGk, channelGK) {
-			st.Skip("It doesn't make sense to create a messaging.Channel with a backing messaging.Channel")
-			return
-		}
-
-		client := testlib.Setup(st, true, options...)
-		defer testlib.TearDown(client)
-
-		// create channel
-		client.CreateChannelWithDefaultOrFail(&messagingv1.Channel{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      channelName,
-				Namespace: client.Namespace,
-			},
-			Spec: messagingv1.ChannelSpec{
-				ChannelTemplate: &messagingv1.ChannelTemplateSpec{
-					TypeMeta: channel,
-				},
-				ChannelableSpec: eventingduckv1.ChannelableSpec{
-					Delivery: &eventingduckv1.DeliverySpec{
-						DeadLetterSink: &duckv1.Destination{
-							Ref: resources.KnativeRefForService(recordEventsPodName, client.Namespace),
-						},
-						Retry: pointer.Int32Ptr(10),
-					},
-				},
-			},
-		})
-		client.WaitForResourcesReadyOrFail(&channel)
-
-		// create event logger pod and service as the subscriber
-		eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, recordEventsPodName)
-		// create subscriptions that subscribe to a service that does not exist
-		client.CreateSubscriptionsOrFail(
-			subscriptionNames,
-			channelName,
-			&channel,
-			resources.WithSubscriberForSubscription("does-not-exist"),
-		)
-
-		// wait for all test resources to be ready, so that we can start sending events
-		client.WaitForAllTestResourcesReadyOrFail(ctx)
-
-		// send CloudEvent to the first channel
-		event := cloudevents.NewEvent()
-		event.SetID("test")
-		eventSource := fmt.Sprintf("http://%s.svc/", senderName)
-		event.SetSource(eventSource)
-		event.SetType(testlib.DefaultEventType)
-		body := fmt.Sprintf(`{"msg":"TestChannelDeadLetterSink %s"}`, uuid.New().String())
-		if err := event.SetData(cloudevents.ApplicationJSON, []byte(body)); err != nil {
-			t.Fatal("Cannot set the payload of the event:", err.Error())
-		}
-		client.SendEventToAddressable(ctx, senderName, channelName, &channel, event)
-
-		// check if the logging service receives the correct number of event messages
-		eventTracker.AssertAtLeast(len(subscriptionNames), recordevents.MatchEvent(
-			HasSource(eventSource),
-			HasData([]byte(body)),
-		))
-	})
-}