Repository: knative-sandbox/eventing-kafka-broker
Pull request #874 (2021-05-06 05:15:11)

[data-plane]: Bump jmh.version from 1.29 to 1.30 in /data-plane
Bumps `jmh.version` from 1.29 to 1.30.
Updates `jmh-core` from 1.29 to 1.30

Updates `jmh-generator-annprocess` from 1.29 to 1.30


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

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 assign **vaikas** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @vaikas` in a comment when ready.

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":["vaikas"]} -->
knative-prow-robot: Hi @dependabot[bot]. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details>

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

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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main     #874      +/-   ##
============================================
- Coverage     76.54%   71.67%   -4.87%     
============================================
  Files            77       28      -49     
  Lines          2925     1257    -1668     
  Branches        133        0     -133     
============================================
- Hits           2239      901    -1338     
+ Misses          535      277     -258     
+ Partials        151       79      -72     
```

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

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...roker/dispatcher/http/HttpSinkResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaHR0cC9IdHRwU2lua1Jlc3BvbnNlSGFuZGxlci5qYXZh) | | | |
| [.../eventing/kafka/broker/receiver/RequestMapper.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9SZXF1ZXN0TWFwcGVyLmphdmE=) | | | |
| [...a/broker/receiver/SimpleProbeHandlerDecorator.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9TaW1wbGVQcm9iZUhhbmRsZXJEZWNvcmF0b3IuamF2YQ==) | | | |
| [...r/dispatcher/http/HttpConsumerVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaHR0cC9IdHRwQ29uc3VtZXJWZXJ0aWNsZUZhY3RvcnkuamF2YQ==) | | | |
| [...ting/kafka/broker/core/utils/CollectionsUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29sbGVjdGlvbnNVdGlscy5qYXZh) | | | |
| [...fka/broker/core/eventbus/ContractMessageCodec.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RNZXNzYWdlQ29kZWMuamF2YQ==) | | | |
| [...tcher/consumer/impl/UnorderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvY29uc3VtZXIvaW1wbC9Vbm9yZGVyZWRDb25zdW1lclZlcnRpY2xlLmphdmE=) | | | |
| [...core/tracing/OpenTelemetryVertxTracingFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9PcGVuVGVsZW1ldHJ5VmVydHhUcmFjaW5nRmFjdG9yeS5qYXZh) | | | |
| [...tive/eventing/kafka/broker/core/filter/Filter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsdGVyL0ZpbHRlci5qYXZh) | | | |
| [...enting/kafka/broker/receiver/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9SZWNlaXZlclZlcnRpY2xlLmphdmE=) | | | |
| ... and [39 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [5d4015b...e6053a3](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/874?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

dependabot[bot]: OK, I won't notify you again about this release, but will get in touch when a new version is available.

If you change your mind, just re-open this PR and I'll resolve any conflicts on it.
pierDipi: @dependabot recreate
dependabot[bot]: Looks like this PR is closed. If you re-open it I'll rebase it as long as no-one else has edited it (you can use `@dependabot reopen` if the branch has been deleted).
pierDipi: @dependabot reopen
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"]} -->
google-cla[bot]: We found a Contributor License Agreement for you (the sender of this pull request), but were unable to find agreements for all the commit author(s) or Co-authors.  If you authored these, maybe you used a different email address in the git commits than was used to sign the CLA ([login here](https://cla.developers.google.com/) to double check)?  If these were authored by someone else, then they will need to sign a CLA as well, and confirm that they're okay with these being contributed to Google.
In order to pass this check, please resolve this problem and then comment `@googlebot I fixed it.`. If the bot doesn't comment, it means it doesn't think anything has changed.

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka-broker%2Fpull%2F874) for more info**.

<!-- need_author_cla -->
pierDipi: Dependabot pr update works but the CLA is not happy anymore, should we add to the workflow to comment ".. I consent"?

Patch:
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/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/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 1b015b94c..9484edc14 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)
@@ -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.30 - 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.30 - 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..25b3eb62b 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.30</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..a89c7ccb0 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -47,7 +47,7 @@
     <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>


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/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 78c94c283..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: "v20210506-2f44af0cd"
+    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
@@ -213,14 +210,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
@@ -242,7 +231,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:
@@ -403,7 +392,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"
@@ -430,7 +419,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:
@@ -451,8 +440,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:
@@ -468,7 +457,7 @@ spec:
                         ref:
                           description: Ref points to an Addressable.
                           type: object
-                          properties: &referentProperties
+                          properties:
                             apiVersion:
                               description: API version of the referent.
                               type: string
@@ -495,7 +484,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
@@ -526,14 +550,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
@@ -553,7 +593,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
@@ -589,14 +640,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
@@ -617,7 +660,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"
@@ -631,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'
@@ -668,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
@@ -804,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
@@ -850,14 +792,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
@@ -878,7 +812,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
@@ -985,14 +919,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
@@ -1013,7 +939,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:
@@ -1245,14 +1171,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
@@ -1273,7 +1191,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"
@@ -1462,7 +1380,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:
@@ -1494,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
@@ -1525,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
@@ -1540,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
@@ -1566,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
@@ -1591,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
@@ -1613,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
@@ -1628,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
@@ -1657,14 +1678,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
@@ -1685,7 +1698,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"
@@ -1875,7 +1888,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
@@ -1887,21 +1900,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.'
@@ -1922,25 +1932,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'
@@ -1952,22 +1962,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'
@@ -1979,18 +1989,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'
@@ -2002,56 +2012,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
@@ -2074,14 +2086,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
@@ -2103,7 +2107,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