Repository: knative-sandbox/eventing-kafka-broker
Pull request #263 (2020-10-07 09:23:11)

Add Retry Config to EgressConfig during broker reconciliation
Part of #116

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add Retry Config to EgressConfig during broker reconciler

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
- 🧽 Update or clean up current behavior
The control plane sends retry configurations to the data plane.
```

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

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263?src=pr&el=h1) Report
> Merging [#263](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263?src=pr&el=desc) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/77aab310e11868e19133d743440fb517e649e35c?el=desc) will **decrease** coverage by `0.25%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263?src=pr&el=tree)

```diff
@@             Coverage Diff              @@
##             master     #263      +/-   ##
============================================
- Coverage     76.30%   76.05%   -0.26%     
  Complexity      170      170              
============================================
  Files            49       27      -22     
  Lines          1435      714     -721     
  Branches         46       46              
============================================
- Hits           1095      543     -552     
+ Misses          271      149     -122     
+ Partials         69       22      -47     
```

| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| #java-unittests | `76.05% <ø> (ø)` | `170.00 <ø> (ø)` | |
| #unittests | `?` | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | | |
| [...ntrol-plane/pkg/apis/eventing/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9yZWdpc3Rlci5nbw==) | | | |
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | | | |
| [control-plane/pkg/reconciler/kafka/broker\_class.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9icm9rZXJfY2xhc3MuZ28=) | | | |
| [control-plane/pkg/reconciler/kafka/admin.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9hZG1pbi5nbw==) | | | |
| [control-plane/pkg/reconciler/sink/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2NvbnRyb2xsZXIuZ28=) | | | |
| [control-plane/pkg/reconciler/trigger/trigger.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXIuZ28=) | | | |
| [...-plane/pkg/reconciler/trigger/trigger\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXJfbGlmZWN5Y2xlLmdv) | | | |
| [control-plane/pkg/core/config/resource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvcmVzb3VyY2UuZ28=) | | | |
| [...kg/apis/eventing/v1alpha1/kafka\_sink\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3ZhbGlkYXRpb24uZ28=) | | | |
| ... and [9 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263?src=pr&el=footer). Last update [77aab31...158f77e](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/263?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/config/env_config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313771980062724096/artifacts/line-cov.html#file0) | 80.0% | 90.0% | 10.0
[control-plane/pkg/core/config/utils.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313771980062724096/artifacts/line-cov.html#file1) | 75.0% | 69.2% | -5.8
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313771980062724096/artifacts/line-cov.html#file2) | 88.8% | 89.6% | 0.8

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/config/env_config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313776887662645248/artifacts/line-cov.html#file0) | 80.0% | 90.0% | 10.0
[control-plane/pkg/core/config/utils.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313776887662645248/artifacts/line-cov.html#file1) | 75.0% | 71.4% | -3.6
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313776887662645248/artifacts/line-cov.html#file2) | 88.8% | 89.6% | 0.8

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/config/env_config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313781794637615107/artifacts/line-cov.html#file0) | 80.0% | 90.0% | 10.0
[control-plane/pkg/core/config/utils.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313781794637615107/artifacts/line-cov.html#file1) | 75.0% | 92.9% | 17.9
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313781794637615107/artifacts/line-cov.html#file2) | 88.8% | 89.6% | 0.8

pierDipi: /cc slinkydeveloper 
knative-prow-robot: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 875d036a2883446af3563cf74c5a04d74e1cc104 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1313781794637615104) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
pierDipi: Maven error
```
[ERROR] Failed to execute goal org.jacoco:jacoco-maven-plugin:0.8.6:prepare-agent (default) on project data-plane: Execution default of goal org.jacoco:jacoco-maven-plugin:0.8.6:prepare-agent failed: Plugin org.jacoco:jacoco-maven-plugin:0.8.6 or one of its dependencies could not be resolved: Could not transfer artifact org.apache.maven.shared:file-management:jar:1.2.1 from/to central (https://repo.maven.apache.org/maven2): Transfer failed for https://repo.maven.apache.org/maven2/org/apache/maven/shared/file-management/1.2.1/file-management-1.2.1.jar: Connection reset -> [Help 1]
```
/retest
knative-prow-robot: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 875d036a2883446af3563cf74c5a04d74e1cc104 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1313786073389731840) | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/config/env_config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313826087066669057/artifacts/line-cov.html#file0) | 80.0% | 90.0% | 10.0
[control-plane/pkg/core/config/utils.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313826087066669057/artifacts/line-cov.html#file1) | 75.0% | 92.9% | 17.9
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/263/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1313826087066669057/artifacts/line-cov.html#file2) | 88.8% | 89.6% | 0.8

pierDipi: I can't update only the control plane :facepalm: 
/hold

Patch:
diff --git a/control-plane/config/500-controller.yaml b/control-plane/config/500-controller.yaml
index 3eedb95ad..62a1040ad 100644
--- a/control-plane/config/500-controller.yaml
+++ b/control-plane/config/500-controller.yaml
@@ -86,6 +86,8 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
+            - name: BROKER_DEFAULT_BACKOFF_DELAY
+              value: "PT1S"
             - name: SYSTEM_NAMESPACE
               valueFrom:
                 fieldRef:
diff --git a/control-plane/pkg/config/env_config.go b/control-plane/pkg/config/env_config.go
index 19d8cb0b8..95c25c283 100644
--- a/control-plane/pkg/config/env_config.go
+++ b/control-plane/pkg/config/env_config.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/kelseyhightower/envconfig"
+	"github.com/rickb777/date/period"
 )
 
 type Env struct {
@@ -29,6 +30,7 @@ type Env struct {
 	IngressName                 string `required:"true" split_words:"true"`
 	SystemNamespace             string `required:"true" split_words:"true"`
 	DataPlaneConfigFormat       string `required:"true" split_words:"true"`
+	DefaultBackoffDelay         string `required:"false" split_words:"true"`
 }
 
 func GetEnvConfig(prefix string) (*Env, error) {
@@ -38,9 +40,20 @@ func GetEnvConfig(prefix string) (*Env, error) {
 		return nil, fmt.Errorf("failed to process env config: %w", err)
 	}
 
+	if env.DefaultBackoffDelay != "" {
+		if isNotValidBackoffDelay(env.DefaultBackoffDelay) {
+			return nil, fmt.Errorf("invalid backoff delay: %s", env.DefaultBackoffDelay)
+		}
+	}
+
 	return env, nil
 }
 
 func (c *Env) DataPlaneConfigMapAsString() string {
 	return fmt.Sprintf("%s/%s", c.DataPlaneConfigMapNamespace, c.DataPlaneConfigMapName)
 }
+
+func isNotValidBackoffDelay(delay string) bool {
+	_, err := period.Parse(delay)
+	return err != nil
+}
diff --git a/control-plane/pkg/core/config/utils.go b/control-plane/pkg/core/config/utils.go
index 18bb8eec8..25e9c65bf 100644
--- a/control-plane/pkg/core/config/utils.go
+++ b/control-plane/pkg/core/config/utils.go
@@ -19,6 +19,8 @@ package config
 import (
 	"fmt"
 
+	duck "knative.dev/eventing/pkg/apis/duck/v1"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
@@ -39,3 +41,33 @@ func ContentModeFromString(mode string) contract.ContentMode {
 		))
 	}
 }
+
+// BackoffPolicyFromString returns the BackoffPolicy from the given string.
+//
+// Default value is contract.BackoffPolicy_Exponential.
+func BackoffPolicyFromString(backoffPolicy *duck.BackoffPolicyType) contract.BackoffPolicy {
+	if backoffPolicy == nil {
+		return contract.BackoffPolicy_Exponential
+	}
+
+	bp := *backoffPolicy
+	switch bp {
+	case duck.BackoffPolicyLinear:
+		return contract.BackoffPolicy_Linear
+	case duck.BackoffPolicyExponential: // The linter complains for missing case in switch
+		return contract.BackoffPolicy_Exponential
+	default:
+		return contract.BackoffPolicy_Exponential
+	}
+}
+
+// BackoffDelayFromString returns the BackoffDelay from the given string.
+//
+// Default value is the specified defaultDelay.
+func BackoffDelayFromString(backoffDelay *string, defaultDelay string) string {
+	if backoffDelay == nil {
+		return defaultDelay
+	}
+
+	return *backoffDelay
+}
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index ea04057ae..28765aca0 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -317,17 +317,37 @@ func (r *Reconciler) getBrokerResource(ctx context.Context, topic string, broker
 		BootstrapServers: config.GetBootstrapServers(),
 	}
 
-	if broker.Spec.Delivery != nil && broker.Spec.Delivery.DeadLetterSink != nil {
-		deadLetterSinkURL, err := r.Resolver.URIFromDestinationV1(ctx, *broker.Spec.Delivery.DeadLetterSink, broker)
-		if err != nil {
-			return nil, fmt.Errorf("failed to resolve broker.Spec.Deliver.DeadLetterSink: %w", err)
+	delivery := broker.Spec.Delivery
+	if delivery != nil {
+
+		if delivery.DeadLetterSink != nil {
+
+			deadLetterSinkURL, err := r.Resolver.URIFromDestinationV1(ctx, *delivery.DeadLetterSink, broker)
+			if err != nil {
+				return nil, fmt.Errorf("failed to resolve broker.Spec.Deliver.DeadLetterSink: %w", err)
+			}
+
+			ensureEgressConfig(res)
+			res.EgressConfig.DeadLetter = deadLetterSinkURL.String()
+		}
+
+		if delivery.Retry != nil {
+			ensureEgressConfig(res)
+			res.EgressConfig.Retry = uint32(*delivery.Retry)
+			res.EgressConfig.BackoffDelay = coreconfig.BackoffDelayFromString(delivery.BackoffDelay, "PT1S")
+			res.EgressConfig.BackoffPolicy = coreconfig.BackoffPolicyFromString(delivery.BackoffPolicy)
 		}
-		res.EgressConfig = &contract.EgressConfig{DeadLetter: deadLetterSinkURL.String()}
 	}
 
 	return res, nil
 }
 
+func ensureEgressConfig(res *contract.Resource) {
+	if res.EgressConfig == nil {
+		res.EgressConfig = &contract.EgressConfig{}
+	}
+}
+
 func (r *Reconciler) ConfigMapUpdated(ctx context.Context) func(configMap *corev1.ConfigMap) {
 
 	logger := logging.FromContext(ctx).Desugar()
diff --git a/data-plane/config/template/500-dispatcher.yaml b/data-plane/config/template/500-dispatcher.yaml
index 636724afe..cbbb6eba1 100644
--- a/data-plane/config/template/500-dispatcher.yaml
+++ b/data-plane/config/template/500-dispatcher.yaml
@@ -87,6 +87,26 @@ spec:
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
+          livenessProbe:
+            failureThreshold: 3
+            httpGet:
+              port: 9090
+              path: /metrics
+              scheme: HTTP
+            initialDelaySeconds: 5
+            periodSeconds: 3
+            successThreshold: 1
+            timeoutSeconds: 1
+          readinessProbe:
+            failureThreshold: 3
+            httpGet:
+              port: 9090
+              path: /metrics
+              scheme: HTTP
+            initialDelaySeconds: 5
+            periodSeconds: 3
+            successThreshold: 1
+            timeoutSeconds: 1
           terminationMessagePolicy: FallbackToLogsOnError
           terminationMessagePath: /dev/temination-log
           securityContext:
@@ -101,7 +121,7 @@ spec:
           configMap:
             name: kafka-broker-brokers-triggers
         - name: cache
-          emptyDir: {}
+          emptyDir: { }
         - name: kafka-broker-config-logging
           configMap:
             name: kafka-config-logging
diff --git a/go.mod b/go.mod
index ed033a477..16bb11795 100644
--- a/go.mod
+++ b/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/google/go-cmp v0.5.2
 	github.com/google/uuid v1.1.1
 	github.com/kelseyhightower/envconfig v1.4.0
+	github.com/rickb777/date v1.13.0
 	github.com/stretchr/testify v1.6.0
 	go.uber.org/zap v1.15.0
 	google.golang.org/protobuf v1.25.0
@@ -20,9 +21,9 @@ require (
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/kubernetes v1.14.7
 	k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
-	knative.dev/eventing v0.18.1-0.20201006012020-59269224fe65
-	knative.dev/pkg v0.0.0-20201005175134-21cb070a8b05
-	knative.dev/test-infra v0.0.0-20201005191434-b298834c10f1
+	knative.dev/eventing v0.0.0-20201007101620-0aa53c127c61
+	knative.dev/pkg v0.0.0-20201006235820-46761ba7c3dc
+	knative.dev/test-infra v0.0.0-20201006161322-f94f2bcbb2ee
 )
 
 replace (
diff --git a/go.sum b/go.sum
index eeae8b763..6c65571d3 100644
--- a/go.sum
+++ b/go.sum
@@ -1811,10 +1811,11 @@ gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20190709130402-674ba3eaed22/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0-20191026110619-0b21df46bc1d/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200601152816-913338de1bd2 h1:VEmvx0P+GVTgkNu2EdTN988YCZPcD3lo9AoczZpucwc=
 gopkg.in/yaml.v3 v3.0.0-20200601152816-913338de1bd2/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 h1:tQIYjPdBoyREyB9XMu+nnTclpTYkz2zFM+lzLJFO4gQ=
+gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 helm.sh/helm/v3 v3.1.1/go.mod h1:WYsFJuMASa/4XUqLyv54s0U/f3mlAaRErGmyy4z921g=
@@ -1897,8 +1898,8 @@ k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7Mpm
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/caching v0.0.0-20190719140829-2032732871ff/go.mod h1:dHXFU6CGlLlbzaWc32g80cR92iuBSpsslDNBWI8C7eg=
 knative.dev/caching v0.0.0-20200116200605-67bca2c83dfa/go.mod h1:dHXFU6CGlLlbzaWc32g80cR92iuBSpsslDNBWI8C7eg=
-knative.dev/eventing v0.18.1-0.20201006012020-59269224fe65 h1:tHWf5VAkbIXtqXd1NP9BhypaEcbuShHAe4zBcs+gIEs=
-knative.dev/eventing v0.18.1-0.20201006012020-59269224fe65/go.mod h1:gPb7tc5qIHsO6DHNPcRK4CNbQLeYzA2n8YlIW9eSVc0=
+knative.dev/eventing v0.0.0-20201007101620-0aa53c127c61 h1:WYxL+ZewOEWW2BSFnJ7RkuCVuf0aGWoPoRFGsLmQDNw=
+knative.dev/eventing v0.0.0-20201007101620-0aa53c127c61/go.mod h1:5w75BxdFlwi/uJJapeOx25AUEABp0wnY1HB74phJUso=
 knative.dev/eventing-contrib v0.6.1-0.20190723221543-5ce18048c08b/go.mod h1:SnXZgSGgMSMLNFTwTnpaOH7hXDzTFtw0J8OmHflNx3g=
 knative.dev/eventing-contrib v0.11.2/go.mod h1:SnXZgSGgMSMLNFTwTnpaOH7hXDzTFtw0J8OmHflNx3g=
 knative.dev/pkg v0.0.0-20191101194912-56c2594e4f11/go.mod h1:pgODObA1dTyhNoFxPZTTjNWfx6F0aKsKzn+vaT9XO/Q=
@@ -1909,21 +1910,17 @@ knative.dev/pkg v0.0.0-20200505191044-3da93ebb24c2/go.mod h1:Q6sL35DdGs8hIQZKdaC
 knative.dev/pkg v0.0.0-20200515002500-16d7b963416f/go.mod h1:tMOHGbxtRz8zYFGEGpV/bpoTEM1o89MwYFC4YJXl3GY=
 knative.dev/pkg v0.0.0-20200528142800-1c6815d7e4c9/go.mod h1:QgNZTxnwpB/oSpNcfnLVlw+WpEwwyKAvJlvR3hgeltA=
 knative.dev/pkg v0.0.0-20200711004937-22502028e31a/go.mod h1:AqAJV6rYi8IGikDjJ/9ZQd9qKdkXVlesVnVjwx62YB8=
-knative.dev/pkg v0.0.0-20201003175733-ea7374e81105 h1:dx4o9pDAyyYAqpDiCQiIVbcBMjVYGAV+lzgJbVijZtg=
-knative.dev/pkg v0.0.0-20201003175733-ea7374e81105/go.mod h1:o9w/M7OnPc0ou07PIUORHUBSZgszeZebAYZkxQ8GdFw=
-knative.dev/pkg v0.0.0-20201005175134-21cb070a8b05 h1:EKh6IFZisRV8qaT2KqUYsX8VovGJfO6978z3xuQvUuE=
-knative.dev/pkg v0.0.0-20201005175134-21cb070a8b05/go.mod h1:xpPYLHYXh306wQISWkRC/5t3+CD76nOcEHrCQ+5pxBM=
+knative.dev/pkg v0.0.0-20201006235820-46761ba7c3dc h1:oXTD9Cx0RVuCJERAREuI/C7r7QCLXmzHwqAjQdqksqg=
+knative.dev/pkg v0.0.0-20201006235820-46761ba7c3dc/go.mod h1:xpPYLHYXh306wQISWkRC/5t3+CD76nOcEHrCQ+5pxBM=
 knative.dev/test-infra v0.0.0-20200407185800-1b88cb3b45a5/go.mod h1:xcdUkMJrLlBswIZqL5zCuBFOC22WIPMQoVX1L35i0vQ=
 knative.dev/test-infra v0.0.0-20200505052144-5ea2f705bb55/go.mod h1:WqF1Azka+FxPZ20keR2zCNtiQA1MP9ZB4BH4HuI+SIU=
 knative.dev/test-infra v0.0.0-20200513011557-d03429a76034/go.mod h1:aMif0KXL4g19YCYwsy4Ocjjz5xgPlseYV+B95Oo4JGE=
 knative.dev/test-infra v0.0.0-20200519015156-82551620b0a9/go.mod h1:A5b2OAXTOeHT3hHhVQm3dmtbuWvIDP7qzgtqxA3/2pE=
 knative.dev/test-infra v0.0.0-20200707183444-aed09e56ddc7/go.mod h1:RjYAhXnZqeHw9+B0zsbqSPlae0lCvjekO/nw5ZMpLCs=
-knative.dev/test-infra v0.0.0-20201001200229-a6988e3b3b38 h1:g2LdlRp7jB6QHz3D69fnvn2OfX5C6bydIzubyLpFNl0=
-knative.dev/test-infra v0.0.0-20201001200229-a6988e3b3b38/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
 knative.dev/test-infra v0.0.0-20201002164834-8c07ff018549 h1:8UObRJ0lnWvU9FuD8ubmkwJo14+aZtboq1ncBWbs2uI=
 knative.dev/test-infra v0.0.0-20201002164834-8c07ff018549/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
-knative.dev/test-infra v0.0.0-20201005191434-b298834c10f1 h1:BCeUtO6TsMQyvH1GpxRViTOY+KANqnfk+n5DdTo4nio=
-knative.dev/test-infra v0.0.0-20201005191434-b298834c10f1/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
+knative.dev/test-infra v0.0.0-20201006161322-f94f2bcbb2ee h1:sMtWVGq7OCEMNaEJc54JK5dHFS/+KhI/R/SSFSAZKlU=
+knative.dev/test-infra v0.0.0-20201006161322-f94f2bcbb2ee/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
 mvdan.cc/interfacer v0.0.0-20180901003855-c20040233aed/go.mod h1:Xkxe497xwlCKkIaQYRfC7CSLworTXY9RMqwhhCm+8Nc=
 mvdan.cc/lint v0.0.0-20170908181259-adc824a0674b/go.mod h1:2odslEg/xrtNQqCYg2/jCoyKnw3vv5biOc3JnIcYfL4=
 mvdan.cc/unparam v0.0.0-20190720180237-d51796306d8f/go.mod h1:4G1h5nDURzA3bwVMZIVpwbkw+04kSxk3rAtzlimaUJw=
diff --git a/vendor/gopkg.in/yaml.v3/.travis.yml b/vendor/gopkg.in/yaml.v3/.travis.yml
index 04d4dae09..a130fe883 100644
--- a/vendor/gopkg.in/yaml.v3/.travis.yml
+++ b/vendor/gopkg.in/yaml.v3/.travis.yml
@@ -11,6 +11,7 @@ go:
     - "1.11.x"
     - "1.12.x"
     - "1.13.x"
+    - "1.14.x"
     - "tip"
 
 go_import_path: gopkg.in/yaml.v3
diff --git a/vendor/gopkg.in/yaml.v3/decode.go b/vendor/gopkg.in/yaml.v3/decode.go
index d50c61aae..21c0dacfd 100644
--- a/vendor/gopkg.in/yaml.v3/decode.go
+++ b/vendor/gopkg.in/yaml.v3/decode.go
@@ -35,6 +35,7 @@ type parser struct {
 	doc      *Node
 	anchors  map[string]*Node
 	doneInit bool
+	textless bool
 }
 
 func newParser(b []byte) *parser {
@@ -173,17 +174,20 @@ func (p *parser) node(kind Kind, defaultTag, tag, value string) *Node {
 	} else if kind == ScalarNode {
 		tag, _ = resolve("", value)
 	}
-	return &Node{
-		Kind:        kind,
-		Tag:         tag,
-		Value:       value,
-		Style:       style,
-		Line:        p.event.start_mark.line + 1,
-		Column:      p.event.start_mark.column + 1,
-		HeadComment: string(p.event.head_comment),
-		LineComment: string(p.event.line_comment),
-		FootComment: string(p.event.foot_comment),
+	n := &Node{
+		Kind:  kind,
+		Tag:   tag,
+		Value: value,
+		Style: style,
+	}
+	if !p.textless {
+		n.Line = p.event.start_mark.line + 1
+		n.Column = p.event.start_mark.column + 1
+		n.HeadComment = string(p.event.head_comment)
+		n.LineComment = string(p.event.line_comment)
+		n.FootComment = string(p.event.foot_comment)
 	}
+	return n
 }
 
 func (p *parser) parseChild(parent *Node) *Node {
@@ -395,7 +399,7 @@ func (d *decoder) callObsoleteUnmarshaler(n *Node, u obsoleteUnmarshaler) (good
 //
 // If n holds a null value, prepare returns before doing anything.
 func (d *decoder) prepare(n *Node, out reflect.Value) (newout reflect.Value, unmarshaled, good bool) {
-	if n.ShortTag() == nullTag {
+	if n.ShortTag() == nullTag || n.Kind == 0 && n.IsZero() {
 		return out, false, false
 	}
 	again := true
@@ -501,8 +505,13 @@ func (d *decoder) unmarshal(n *Node, out reflect.Value) (good bool) {
 		good = d.mapping(n, out)
 	case SequenceNode:
 		good = d.sequence(n, out)
+	case 0:
+		if n.IsZero() {
+			return d.null(out)
+		}
+		fallthrough
 	default:
-		panic("internal error: unknown node kind: " + strconv.Itoa(int(n.Kind)))
+		failf("cannot decode node with unknown kind %d", n.Kind)
 	}
 	return good
 }
@@ -537,6 +546,17 @@ func resetMap(out reflect.Value) {
 	}
 }
 
+func (d *decoder) null(out reflect.Value) bool {
+	if out.CanAddr() {
+		switch out.Kind() {
+		case reflect.Interface, reflect.Ptr, reflect.Map, reflect.Slice:
+			out.Set(reflect.Zero(out.Type()))
+			return true
+		}
+	}
+	return false
+}
+
 func (d *decoder) scalar(n *Node, out reflect.Value) bool {
 	var tag string
 	var resolved interface{}
@@ -554,14 +574,7 @@ func (d *decoder) scalar(n *Node, out reflect.Value) bool {
 		}
 	}
 	if resolved == nil {
-		if out.CanAddr() {
-			switch out.Kind() {
-			case reflect.Interface, reflect.Ptr, reflect.Map, reflect.Slice:
-				out.Set(reflect.Zero(out.Type()))
-				return true
-			}
-		}
-		return false
+		return d.null(out)
 	}
 	if resolvedv := reflect.ValueOf(resolved); out.Type() == resolvedv.Type() {
 		// We've resolved to exactly the type we want, so use that.
diff --git a/vendor/gopkg.in/yaml.v3/emitterc.go b/vendor/gopkg.in/yaml.v3/emitterc.go
index c7bd5d2a3..c29217ef5 100644
--- a/vendor/gopkg.in/yaml.v3/emitterc.go
+++ b/vendor/gopkg.in/yaml.v3/emitterc.go
@@ -235,10 +235,13 @@ func yaml_emitter_increase_indent(emitter *yaml_emitter_t, flow, indentless bool
 			emitter.indent = 0
 		}
 	} else if !indentless {
-		emitter.indent += emitter.best_indent
-		// [Go] If inside a block sequence item, discount the space taken by the indicator.
-		if emitter.best_indent > 2 && emitter.states[len(emitter.states)-1] == yaml_EMIT_BLOCK_SEQUENCE_ITEM_STATE {
-			emitter.indent -= 2
+		// [Go] This was changed so that indentations are more regular.
+		if emitter.states[len(emitter.states)-1] == yaml_EMIT_BLOCK_SEQUENCE_ITEM_STATE {
+			// The first indent inside a sequence will just skip the "- " indicator.
+			emitter.indent += 2
+		} else {
+			// Everything else aligns to the chosen indentation.
+			emitter.indent = emitter.best_indent*((emitter.indent+emitter.best_indent)/emitter.best_indent)
 		}
 	}
 	return true
@@ -725,16 +728,9 @@ func yaml_emitter_emit_flow_mapping_value(emitter *yaml_emitter_t, event *yaml_e
 // Expect a block item node.
 func yaml_emitter_emit_block_sequence_item(emitter *yaml_emitter_t, event *yaml_event_t, first bool) bool {
 	if first {
-		// [Go] The original logic here would not indent the sequence when
-		//      inside a mapping. In Go we always indent it.
-		indentless := false
-		original := emitter.indent
-		if !yaml_emitter_increase_indent(emitter, false, indentless) {
+		if !yaml_emitter_increase_indent(emitter, false, false) {
 			return false
 		}
-		if emitter.indent > original+2 {
-			emitter.indent -= 2
-		}
 	}
 	if event.typ == yaml_SEQUENCE_END_EVENT {
 		emitter.indent = emitter.indents[len(emitter.indents)-1]
@@ -785,6 +781,13 @@ func yaml_emitter_emit_block_mapping_key(emitter *yaml_emitter_t, event *yaml_ev
 	if !yaml_emitter_write_indent(emitter) {
 		return false
 	}
+	if len(emitter.line_comment) > 0 {
+		// [Go] A line comment was provided for the key. That's unusual as the
+		//      scanner associates line comments with the value. Either way,
+		//      save the line comment and render it appropriately later.
+		emitter.key_line_comment = emitter.line_comment
+		emitter.line_comment = nil
+	}
 	if yaml_emitter_check_simple_key(emitter) {
 		emitter.states = append(emitter.states, yaml_EMIT_BLOCK_MAPPING_SIMPLE_VALUE_STATE)
 		return yaml_emitter_emit_node(emitter, event, false, false, true, true)
@@ -810,6 +813,29 @@ func yaml_emitter_emit_block_mapping_value(emitter *yaml_emitter_t, event *yaml_
 			return false
 		}
 	}
+	if len(emitter.key_line_comment) > 0 {
+		// [Go] A line comment was previously provided for the key. Handle it before
+		//      the value so the inline comments are placed correctly.
+		if yaml_emitter_silent_nil_event(emitter, event) && len(emitter.line_comment) == 0 {
+			// Nothing other than the line comment will be written on the line.
+			emitter.line_comment = emitter.key_line_comment
+			emitter.key_line_comment = nil
+		} else {
+			// An actual value is coming, so emit the comment line.
+			emitter.line_comment, emitter.key_line_comment = emitter.key_line_comment, emitter.line_comment
+			if !yaml_emitter_process_line_comment(emitter) {
+				return false
+			}
+			emitter.line_comment, emitter.key_line_comment = emitter.key_line_comment, emitter.line_comment
+			// Indent in unless it's a block that will reindent anyway.
+			if event.sequence_style() == yaml_FLOW_SEQUENCE_STYLE || (event.typ != yaml_MAPPING_START_EVENT && event.typ != yaml_SEQUENCE_START_EVENT) {
+				emitter.indent = emitter.best_indent*((emitter.indent+emitter.best_indent)/emitter.best_indent)
+				if !yaml_emitter_write_indent(emitter) {
+					return false
+				}
+			}
+		}
+	}
 	emitter.states = append(emitter.states, yaml_EMIT_BLOCK_MAPPING_KEY_STATE)
 	if !yaml_emitter_emit_node(emitter, event, false, false, true, false) {
 		return false
@@ -823,6 +849,10 @@ func yaml_emitter_emit_block_mapping_value(emitter *yaml_emitter_t, event *yaml_
 	return true
 }
 
+func yaml_emitter_silent_nil_event(emitter *yaml_emitter_t, event *yaml_event_t) bool {
+	return event.typ == yaml_SCALAR_EVENT && event.implicit && !emitter.canonical && len(emitter.scalar_data.value) == 0
+}
+
 // Expect a node.
 func yaml_emitter_emit_node(emitter *yaml_emitter_t, event *yaml_event_t,
 	root bool, sequence bool, mapping bool, simple_key bool) bool {
diff --git a/vendor/gopkg.in/yaml.v3/encode.go b/vendor/gopkg.in/yaml.v3/encode.go
index e998db941..45e8d1e1b 100644
--- a/vendor/gopkg.in/yaml.v3/encode.go
+++ b/vendor/gopkg.in/yaml.v3/encode.go
@@ -425,18 +425,23 @@ func (e *encoder) nodev(in reflect.Value) {
 }
 
 func (e *encoder) node(node *Node, tail string) {
+	// Zero nodes behave as nil.
+	if node.Kind == 0 && node.IsZero() {
+		e.nilv()
+		return
+	}
+
 	// If the tag was not explicitly requested, and dropping it won't change the
 	// implicit tag of the value, don't include it in the presentation.
 	var tag = node.Tag
 	var stag = shortTag(tag)
-	var rtag string
 	var forceQuoting bool
 	if tag != "" && node.Style&TaggedStyle == 0 {
 		if node.Kind == ScalarNode {
 			if stag == strTag && node.Style&(SingleQuotedStyle|DoubleQuotedStyle|LiteralStyle|FoldedStyle) != 0 {
 				tag = ""
 			} else {
-				rtag, _ = resolve("", node.Value)
+				rtag, _ := resolve("", node.Value)
 				if rtag == stag {
 					tag = ""
 				} else if stag == strTag {
@@ -445,6 +450,7 @@ func (e *encoder) node(node *Node, tail string) {
 				}
 			}
 		} else {
+			var rtag string
 			switch node.Kind {
 			case MappingNode:
 				rtag = mapTag
@@ -474,7 +480,7 @@ func (e *encoder) node(node *Node, tail string) {
 		if node.Style&FlowStyle != 0 {
 			style = yaml_FLOW_SEQUENCE_STYLE
 		}
-		e.must(yaml_sequence_start_event_initialize(&e.event, []byte(node.Anchor), []byte(tag), tag == "", style))
+		e.must(yaml_sequence_start_event_initialize(&e.event, []byte(node.Anchor), []byte(longTag(tag)), tag == "", style))
 		e.event.head_comment = []byte(node.HeadComment)
 		e.emit()
 		for _, node := range node.Content {
@@ -490,7 +496,7 @@ func (e *encoder) node(node *Node, tail string) {
 		if node.Style&FlowStyle != 0 {
 			style = yaml_FLOW_MAPPING_STYLE
 		}
-		yaml_mapping_start_event_initialize(&e.event, []byte(node.Anchor), []byte(tag), tag == "", style)
+		yaml_mapping_start_event_initialize(&e.event, []byte(node.Anchor), []byte(longTag(tag)), tag == "", style)
 		e.event.tail_comment = []byte(tail)
 		e.event.head_comment = []byte(node.HeadComment)
 		e.emit()
@@ -531,11 +537,11 @@ func (e *encoder) node(node *Node, tail string) {
 	case ScalarNode:
 		value := node.Value
 		if !utf8.ValidString(value) {
-			if tag == binaryTag {
+			if stag == binaryTag {
 				failf("explicitly tagged !!binary data must be base64-encoded")
 			}
-			if tag != "" {
-				failf("cannot marshal invalid UTF-8 data as %s", shortTag(tag))
+			if stag != "" {
+				failf("cannot marshal invalid UTF-8 data as %s", stag)
 			}
 			// It can't be encoded directly as YAML so use a binary tag
 			// and encode it as base64.
@@ -560,5 +566,7 @@ func (e *encoder) node(node *Node, tail string) {
 		}
 
 		e.emitScalar(value, node.Anchor, tag, style, []byte(node.HeadComment), []byte(node.LineComment), []byte(node.FootComment), []byte(tail))
+	default:
+		failf("cannot encode node with unknown kind %d", node.Kind)
 	}
 }
diff --git a/vendor/gopkg.in/yaml.v3/parserc.go b/vendor/gopkg.in/yaml.v3/parserc.go
index aea9050b8..ac66fccc0 100644
--- a/vendor/gopkg.in/yaml.v3/parserc.go
+++ b/vendor/gopkg.in/yaml.v3/parserc.go
@@ -648,6 +648,10 @@ func yaml_parser_parse_node(parser *yaml_parser_t, event *yaml_event_t, block, i
 			implicit:   implicit,
 			style:      yaml_style_t(yaml_BLOCK_MAPPING_STYLE),
 		}
+		if parser.stem_comment != nil {
+			event.head_comment = parser.stem_comment
+			parser.stem_comment = nil
+		}
 		return true
 	}
 	if len(anchor) > 0 || len(tag) > 0 {
@@ -694,25 +698,13 @@ func yaml_parser_parse_block_sequence_entry(parser *yaml_parser_t, event *yaml_e
 
 	if token.typ == yaml_BLOCK_ENTRY_TOKEN {
 		mark := token.end_mark
-		prior_head := len(parser.head_comment)
+		prior_head_len := len(parser.head_comment)
 		skip_token(parser)
+		yaml_parser_split_stem_comment(parser, prior_head_len)
 		token = peek_token(parser)
 		if token == nil {
 			return false
 		}
-		if prior_head > 0 && token.typ == yaml_BLOCK_SEQUENCE_START_TOKEN {
-			// [Go] It's a sequence under a sequence entry, so the former head comment
-			//      is for the list itself, not the first list item under it.
-			parser.stem_comment = parser.head_comment[:prior_head]
-			if len(parser.head_comment) == prior_head {
-				parser.head_comment = nil
-			} else {
-				// Copy suffix to prevent very strange bugs if someone ever appends
-				// further bytes to the prefix in the stem_comment slice above.
-				parser.head_comment = append([]byte(nil), parser.head_comment[prior_head+1:]...)
-			}
-
-		}
 		if token.typ != yaml_BLOCK_ENTRY_TOKEN && token.typ != yaml_BLOCK_END_TOKEN {
 			parser.states = append(parser.states, yaml_PARSE_BLOCK_SEQUENCE_ENTRY_STATE)
 			return yaml_parser_parse_node(parser, event, true, false)
@@ -754,7 +746,9 @@ func yaml_parser_parse_indentless_sequence_entry(parser *yaml_parser_t, event *y
 
 	if token.typ == yaml_BLOCK_ENTRY_TOKEN {
 		mark := token.end_mark
+		prior_head_len := len(parser.head_comment)
 		skip_token(parser)
+		yaml_parser_split_stem_comment(parser, prior_head_len)
 		token = peek_token(parser)
 		if token == nil {
 			return false
@@ -780,6 +774,32 @@ func yaml_parser_parse_indentless_sequence_entry(parser *yaml_parser_t, event *y
 	return true
 }
 
+// Split stem comment from head comment.
+//
+// When a sequence or map is found under a sequence entry, the former head comment
+// is assigned to the underlying sequence or map as a whole, not the individual
+// sequence or map entry as would be expected otherwise. To handle this case the
+// previous head comment is moved aside as the stem comment.
+func yaml_parser_split_stem_comment(parser *yaml_parser_t, stem_len int) {
+	if stem_len == 0 {
+		return
+	}
+
+	token := peek_token(parser)
+	if token.typ != yaml_BLOCK_SEQUENCE_START_TOKEN && token.typ != yaml_BLOCK_MAPPING_START_TOKEN {
+		return
+	}
+
+	parser.stem_comment = parser.head_comment[:stem_len]
+	if len(parser.head_comment) == stem_len {
+		parser.head_comment = nil
+	} else {
+		// Copy suffix to prevent very strange bugs if someone ever appends
+		// further bytes to the prefix in the stem_comment slice above.
+		parser.head_comment = append([]byte(nil), parser.head_comment[stem_len+1:]...)
+	}
+}
+
 // Parse the productions:
 // block_mapping        ::= BLOCK-MAPPING_START
 //                          *******************
diff --git a/vendor/gopkg.in/yaml.v3/scannerc.go b/vendor/gopkg.in/yaml.v3/scannerc.go
index 2e9cbf97f..d9a539c39 100644
--- a/vendor/gopkg.in/yaml.v3/scannerc.go
+++ b/vendor/gopkg.in/yaml.v3/scannerc.go
@@ -749,6 +749,11 @@ func yaml_parser_fetch_next_token(parser *yaml_parser_t) (ok bool) {
 		if !ok {
 			return
 		}
+		if len(parser.tokens) > 0 && parser.tokens[len(parser.tokens)-1].typ == yaml_BLOCK_ENTRY_TOKEN {
+			// Sequence indicators alone have no line comments. It becomes
+			// a head comment for whatever follows.
+			return
+		}
 		if !yaml_parser_scan_line_comment(parser, comment_mark) {
 			ok = false
 			return
diff --git a/vendor/gopkg.in/yaml.v3/yaml.go b/vendor/gopkg.in/yaml.v3/yaml.go
index b5d35a50d..56e8a8490 100644
--- a/vendor/gopkg.in/yaml.v3/yaml.go
+++ b/vendor/gopkg.in/yaml.v3/yaml.go
@@ -89,7 +89,7 @@ func Unmarshal(in []byte, out interface{}) (err error) {
 	return unmarshal(in, out, false)
 }
 
-// A Decorder reads and decodes YAML values from an input stream.
+// A Decoder reads and decodes YAML values from an input stream.
 type Decoder struct {
 	parser      *parser
 	knownFields bool
@@ -194,7 +194,7 @@ func unmarshal(in []byte, out interface{}, strict bool) (err error) {
 //                  Zero valued structs will be omitted if all their public
 //                  fields are zero, unless they implement an IsZero
 //                  method (see the IsZeroer interface type), in which
-//                  case the field will be included if that method returns true.
+//                  case the field will be excluded if IsZero returns true.
 //
 //     flow         Marshal using a flow style (useful for structs,
 //                  sequences and maps).
@@ -252,6 +252,24 @@ func (e *Encoder) Encode(v interface{}) (err error) {
 	return nil
 }
 
+// Encode encodes value v and stores its representation in n.
+//
+// See the documentation for Marshal for details about the
+// conversion of Go values into YAML.
+func (n *Node) Encode(v interface{}) (err error) {
+	defer handleErr(&err)
+	e := newEncoder()
+	defer e.destroy()
+	e.marshalDoc("", reflect.ValueOf(v))
+	e.finish()
+	p := newParser(e.out)
+	p.textless = true
+	defer p.destroy()
+	doc := p.parse()
+	*n = *doc.Content[0]
+	return nil
+}
+
 // SetIndent changes the used indentation used when encoding.
 func (e *Encoder) SetIndent(spaces int) {
 	if spaces < 0 {
@@ -328,6 +346,12 @@ const (
 // and maps, Node is an intermediate representation that allows detailed
 // control over the content being decoded or encoded.
 //
+// It's worth noting that although Node offers access into details such as
+// line numbers, colums, and comments, the content when re-encoded will not
+// have its original textual representation preserved. An effort is made to
+// render the data plesantly, and to preserve comments near the data they
+// describe, though.
+//
 // Values that make use of the Node type interact with the yaml package in the
 // same way any other type would do, by encoding and decoding yaml data
 // directly or indirectly into them.
@@ -391,6 +415,13 @@ type Node struct {
 	Column int
 }
 
+// IsZero returns whether the node has all of its fields unset.
+func (n *Node) IsZero() bool {
+	return n.Kind == 0 && n.Style == 0 && n.Tag == "" && n.Value == "" && n.Anchor == "" && n.Alias == nil && n.Content == nil &&
+		n.HeadComment == "" && n.LineComment == "" && n.FootComment == "" && n.Line == 0 && n.Column == 0
+}
+
+
 // LongTag returns the long form of the tag that indicates the data type for
 // the node. If the Tag field isn't explicitly defined, one will be computed
 // based on the node properties.
diff --git a/vendor/gopkg.in/yaml.v3/yamlh.go b/vendor/gopkg.in/yaml.v3/yamlh.go
index 2719cfbb0..7c6d00770 100644
--- a/vendor/gopkg.in/yaml.v3/yamlh.go
+++ b/vendor/gopkg.in/yaml.v3/yamlh.go
@@ -787,6 +787,8 @@ type yaml_emitter_t struct {
 	foot_comment []byte
 	tail_comment []byte
 
+	key_line_comment []byte
+
 	// Dumper stuff
 
 	opened bool // If the stream was already opened?
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/constants.go b/vendor/knative.dev/eventing/pkg/apis/sources/constants.go
index 731372f9b..730116918 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/constants.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/constants.go
@@ -32,12 +32,16 @@ const (
 	ApiServerSourceDeleteRefEventType = "dev.knative.apiserver.ref.delete"
 )
 
-// ApiServerSourceEventTypes is the list of CloudEvent types the ApiServerSource emits.
-var ApiServerSourceEventTypes = []string{
-	ApiServerSourceAddEventType,
-	ApiServerSourceDeleteEventType,
-	ApiServerSourceUpdateEventType,
+// ApiServerSourceEventReferenceModeTypes is the list of CloudEvent types the ApiServerSource with EventMode of ReferenceMode emits.
+var ApiServerSourceEventReferenceModeTypes = []string{
 	ApiServerSourceAddRefEventType,
 	ApiServerSourceDeleteRefEventType,
 	ApiServerSourceUpdateRefEventType,
 }
+
+// ApiServerSourceEventResourceModeTypes is the list of CloudEvent types the ApiServerSource with EventMode of ResourceMode emits.
+var ApiServerSourceEventResourceModeTypes = []string{
+	ApiServerSourceAddEventType,
+	ApiServerSourceDeleteEventType,
+	ApiServerSourceUpdateEventType,
+}
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/binding_types.go b/vendor/knative.dev/pkg/apis/duck/v1/binding_types.go
new file mode 100644
index 000000000..71b224cf5
--- /dev/null
+++ b/vendor/knative.dev/pkg/apis/duck/v1/binding_types.go
@@ -0,0 +1,92 @@
+/*
+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 v1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/apis/duck"
+	"knative.dev/pkg/tracker"
+)
+
+// +genduck
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// Binding is a duck type that specifies the partial schema to which all
+// Binding implementations should adhere.
+type Binding struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec BindingSpec `json:"spec"`
+}
+
+// Verify that Binding implements the appropriate interfaces.
+var (
+	_ duck.Implementable = (*Binding)(nil)
+	_ duck.Populatable   = (*Binding)(nil)
+	_ apis.Listable      = (*Binding)(nil)
+)
+
+// BindingSpec specifies the spec portion of the Binding partial-schema.
+type BindingSpec struct {
+	// Subject references the resource(s) whose "runtime contract" should be
+	// augmented by Binding implementations.
+	Subject tracker.Reference `json:"subject"`
+}
+
+// GetFullType implements duck.Implementable
+func (*Binding) GetFullType() duck.Populatable {
+	return &Binding{}
+}
+
+// Populate implements duck.Populatable
+func (t *Binding) Populate() {
+	t.Spec = BindingSpec{
+		Subject: tracker.Reference{
+			APIVersion: "apps/v1",
+			Kind:       "Deployment",
+			Namespace:  "default",
+			// Name and Selector are mutually exclusive,
+			// but we fill them both in for this test.
+			Name: "bazinga",
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					"foo": "bar",
+					"baz": "blah",
+				},
+			},
+		},
+	}
+}
+
+// GetListType implements apis.Listable
+func (*Binding) GetListType() runtime.Object {
+	return &BindingList{}
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+
+// BindingList is a list of Binding resources
+type BindingList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata"`
+
+	Items []Binding `json:"items"`
+}
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/register.go b/vendor/knative.dev/pkg/apis/duck/v1/register.go
index 69bb11578..67418b3e9 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1/register.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1/register.go
@@ -54,6 +54,8 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 		(&Source{}).GetListType(),
 		&WithPod{},
 		(&WithPod{}).GetListType(),
+		&Binding{},
+		(&Binding{}).GetListType(),
 	)
 	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
 	return nil
diff --git a/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go b/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
index 190d773e7..f4da66047 100644
--- a/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/pkg/apis/duck/v1/zz_generated.deepcopy.go
@@ -127,6 +127,83 @@ func (in *AddressableTypeList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Binding) DeepCopyInto(out *Binding) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	in.Spec.DeepCopyInto(&out.Spec)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Binding.
+func (in *Binding) DeepCopy() *Binding {
+	if in == nil {
+		return nil
+	}
+	out := new(Binding)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *Binding) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BindingList) DeepCopyInto(out *BindingList) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	in.ListMeta.DeepCopyInto(&out.ListMeta)
+	if in.Items != nil {
+		in, out := &in.Items, &out.Items
+		*out = make([]Binding, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BindingList.
+func (in *BindingList) DeepCopy() *BindingList {
+	if in == nil {
+		return nil
+	}
+	out := new(BindingList)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *BindingList) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *BindingSpec) DeepCopyInto(out *BindingSpec) {
+	*out = *in
+	in.Subject.DeepCopyInto(&out.Subject)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new BindingSpec.
+func (in *BindingSpec) DeepCopy() *BindingSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(BindingSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CloudEventAttributes) DeepCopyInto(out *CloudEventAttributes) {
 	*out = *in
diff --git a/vendor/knative.dev/pkg/apis/field_error.go b/vendor/knative.dev/pkg/apis/field_error.go
index 3997b5944..e807ec781 100644
--- a/vendor/knative.dev/pkg/apis/field_error.go
+++ b/vendor/knative.dev/pkg/apis/field_error.go
@@ -327,7 +327,7 @@ func ErrInvalidArrayValue(value interface{}, field string, index int) *FieldErro
 // invalid value.
 func ErrInvalidValue(value interface{}, fieldPath string) *FieldError {
 	return &FieldError{
-		Message: fmt.Sprintf("invalid value: %v", value),
+		Message: fmt.Sprint("invalid value: ", value),
 		Paths:   []string{fieldPath},
 	}
 }
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
index 9345cdea0..d1fa91e09 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/generators/packages.go
@@ -36,7 +36,7 @@ import (
 func Packages(context *generator.Context, arguments *args.GeneratorArgs) generator.Packages {
 	boilerplate, err := arguments.LoadGoBoilerplate()
 	if err != nil {
-		klog.Fatalf("Failed loading boilerplate: %v", err)
+		klog.Fatal("Failed loading boilerplate: ", err)
 	}
 
 	customArgs, ok := arguments.CustomArgs.(*informergenargs.CustomArgs)
diff --git a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/main.go b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/main.go
index 26410cc99..30f20a0dd 100644
--- a/vendor/knative.dev/pkg/codegen/cmd/injection-gen/main.go
+++ b/vendor/knative.dev/pkg/codegen/cmd/injection-gen/main.go
@@ -44,7 +44,7 @@ func main() {
 	pflag.Parse()
 
 	if err := generatorargs.Validate(genericArgs); err != nil {
-		klog.Fatalf("Error: %v", err)
+		klog.Fatal("Error: ", err)
 	}
 
 	// Run it.
@@ -53,7 +53,7 @@ func main() {
 		generators.DefaultNameSystem(),
 		generators.Packages,
 	); err != nil {
-		klog.Fatalf("Error: %v", err)
+		klog.Fatal("Error: ", err)
 	}
 	klog.V(2).Info("Completed successfully.")
 }
diff --git a/vendor/knative.dev/pkg/injection/config.go b/vendor/knative.dev/pkg/injection/config.go
new file mode 100644
index 000000000..01a5962b7
--- /dev/null
+++ b/vendor/knative.dev/pkg/injection/config.go
@@ -0,0 +1,85 @@
+/*
+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 injection
+
+import (
+	"errors"
+	"flag"
+	"log"
+	"os"
+	"os/user"
+	"path/filepath"
+
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/klog"
+)
+
+// ParseAndGetRESTConfigOrDie parses the rest config flags and creates a client or
+// dies by calling log.Fatalf.
+func ParseAndGetRESTConfigOrDie() *rest.Config {
+	var (
+		serverURL = flag.String("server", "",
+			"The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
+		kubeconfig = flag.String("kubeconfig", "",
+			"Path to a kubeconfig. Only required if out-of-cluster.")
+	)
+	klog.InitFlags(flag.CommandLine)
+	flag.Parse()
+
+	cfg, err := GetRESTConfig(*serverURL, *kubeconfig)
+	if err != nil {
+		log.Fatal("Error building kubeconfig: ", err)
+	}
+
+	return cfg
+}
+
+// GetRESTConfig returns a rest.Config to be used for kubernetes client creation.
+// It does so in the following order:
+//   1. Use the passed kubeconfig/serverURL.
+//   2. Fallback to the KUBECONFIG environment variable.
+//   3. Fallback to in-cluster config.
+//   4. Fallback to the ~/.kube/config.
+func GetRESTConfig(serverURL, kubeconfig string) (*rest.Config, error) {
+	if kubeconfig == "" {
+		kubeconfig = os.Getenv("KUBECONFIG")
+	}
+
+	// If we have an explicit indication of where the kubernetes config lives, read that.
+	if kubeconfig != "" {
+		c, err := clientcmd.BuildConfigFromFlags(serverURL, kubeconfig)
+		if err != nil {
+			return nil, err
+		}
+		return c, nil
+	}
+
+	// If not, try the in-cluster config.
+	if c, err := rest.InClusterConfig(); err == nil {
+		return c, nil
+	}
+
+	// If no in-cluster config, try the default location in the user's home directory.
+	if usr, err := user.Current(); err == nil {
+		if c, err := clientcmd.BuildConfigFromFlags("", filepath.Join(usr.HomeDir, ".kube", "config")); err == nil {
+			return c, nil
+		}
+	}
+
+	return nil, errors.New("could not create a valid kubeconfig")
+}
diff --git a/vendor/knative.dev/pkg/injection/injection.go b/vendor/knative.dev/pkg/injection/injection.go
new file mode 100644
index 000000000..d8cefc9fb
--- /dev/null
+++ b/vendor/knative.dev/pkg/injection/injection.go
@@ -0,0 +1,66 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package injection
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+	"k8s.io/client-go/rest"
+
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/signals"
+)
+
+// EnableInjectionOrDie enables Knative Client Injection, and provides a
+// callback to start the informers. Both Context and Config are optional.
+// Returns context with rest config set and a callback to start the informers
+// after watches have been set.
+//
+// Typical integration:
+// ```go
+//   ctx, startInformers := injection.EnableInjectionOrDie(signals.NewContext(), nil)
+//   ... start watches with informers, if required ...
+//   startInformers()
+// ```
+func EnableInjectionOrDie(ctx context.Context, cfg *rest.Config) (context.Context, func()) {
+	if ctx == nil {
+		ctx = signals.NewContext()
+	}
+	if cfg == nil {
+		cfg = ParseAndGetRESTConfigOrDie()
+	}
+
+	// Respect user provided settings, but if omitted customize the default behavior.
+	if cfg.QPS == 0 {
+		cfg.QPS = rest.DefaultQPS
+	}
+	if cfg.Burst == 0 {
+		cfg.Burst = rest.DefaultBurst
+	}
+	ctx = WithConfig(ctx, cfg)
+
+	ctx, informers := Default.SetupInformers(ctx, cfg)
+
+	return ctx, func() {
+		logging.FromContext(ctx).Info("Starting informers...")
+		if err := controller.StartInformers(ctx.Done(), informers...); err != nil {
+			logging.FromContext(ctx).Fatalw("Failed to start informers", zap.Error(err))
+		}
+	}
+}
diff --git a/vendor/knative.dev/pkg/injection/sharedmain/main.go b/vendor/knative.dev/pkg/injection/sharedmain/main.go
index 0fceb6983..c9dd034de 100644
--- a/vendor/knative.dev/pkg/injection/sharedmain/main.go
+++ b/vendor/knative.dev/pkg/injection/sharedmain/main.go
@@ -18,16 +18,15 @@ package sharedmain
 
 import (
 	"context"
-	"errors"
 	"flag"
 	"log"
 	"net/http"
 	"os"
-	"os/user"
-	"path/filepath"
 	"time"
 
 	"go.opencensus.io/stats/view"
+	_ "go.uber.org/automaxprocs" // automatically set GOMAXPROCS based on cgroups
+	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -35,11 +34,6 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/clientcmd"
-	"k8s.io/klog"
-
-	_ "go.uber.org/automaxprocs" // automatically set GOMAXPROCS based on cgroups
-	"go.uber.org/zap"
 
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
@@ -63,37 +57,9 @@ import (
 //   2. Fallback to the KUBECONFIG environment variable.
 //   3. Fallback to in-cluster config.
 //   4. Fallback to the ~/.kube/config.
+// Deprecated: use injection.GetRESTConfig
 func GetConfig(serverURL, kubeconfig string) (*rest.Config, error) {
-	if kubeconfig == "" {
-		kubeconfig = os.Getenv("KUBECONFIG")
-	}
-
-	// We produce configs a bunch of ways, this gives us a single place
-	// to "decorate" them with common useful things (e.g. for debugging)
-	decorate := func(cfg *rest.Config) *rest.Config {
-		return cfg
-	}
-
-	// If we have an explicit indication of where the kubernetes config lives, read that.
-	if kubeconfig != "" {
-		c, err := clientcmd.BuildConfigFromFlags(serverURL, kubeconfig)
-		if err != nil {
-			return nil, err
-		}
-		return decorate(c), nil
-	}
-	// If not, try the in-cluster config.
-	if c, err := rest.InClusterConfig(); err == nil {
-		return decorate(c), nil
-	}
-	// If no in-cluster config, try the default location in the user's home directory.
-	if usr, err := user.Current(); err == nil {
-		if c, err := clientcmd.BuildConfigFromFlags("", filepath.Join(usr.HomeDir, ".kube", "config")); err == nil {
-			return decorate(c), nil
-		}
-	}
-
-	return nil, errors.New("could not create a valid kubeconfig")
+	return injection.GetRESTConfig(serverURL, kubeconfig)
 }
 
 // GetLoggingConfig gets the logging config from either the file system if present
@@ -129,40 +95,16 @@ func GetLeaderElectionConfig(ctx context.Context) (*leaderelection.Config, error
 
 // EnableInjectionOrDie enables Knative Injection and starts the informers.
 // Both Context and Config are optional.
+// Deprecated: use injection.EnableInjectionOrDie
 func EnableInjectionOrDie(ctx context.Context, cfg *rest.Config) context.Context {
-	if ctx == nil {
-		ctx = signals.NewContext()
-	}
-	if cfg == nil {
-		cfg = ParseAndGetConfigOrDie()
-	}
-
-	// Respect user provided settings, but if omitted customize the default behavior.
-	if cfg.QPS == 0 {
-		cfg.QPS = rest.DefaultQPS
-	}
-	if cfg.Burst == 0 {
-		cfg.Burst = rest.DefaultBurst
-	}
-	ctx = injection.WithConfig(ctx, cfg)
-
-	ctx, informers := injection.Default.SetupInformers(ctx, cfg)
-
-	// Start the injection clients and informers.
-	logging.FromContext(ctx).Info("Starting informers...")
-	go func(ctx context.Context) {
-		if err := controller.StartInformers(ctx.Done(), informers...); err != nil {
-			logging.FromContext(ctx).Fatalw("Failed to start informers", zap.Error(err))
-		}
-		<-ctx.Done()
-	}(ctx)
-
+	ctx, startInformers := injection.EnableInjectionOrDie(ctx, cfg)
+	go startInformers()
 	return ctx
 }
 
 // Main runs the generic main flow with a new context.
-// If any of the contructed controllers are AdmissionControllers or Conversion webhooks,
-// then a webhook is started to serve them.
+// If any of the constructed controllers are AdmissionControllers or Conversion
+// webhooks, then a webhook is started to serve them.
 func Main(component string, ctors ...injection.ControllerConstructor) {
 	// Set up signals so we handle the first shutdown signal gracefully.
 	MainWithContext(signals.NewContext(), component, ctors...)
@@ -185,7 +127,7 @@ func MainWithContext(ctx context.Context, component string, ctors ...injection.C
 			"issue upstream!")
 
 	// HACK: This parses flags, so the above should be set once this runs.
-	cfg := ParseAndGetConfigOrDie()
+	cfg := injection.ParseAndGetRESTConfigOrDie()
 
 	if *disableHighAvailability {
 		ctx = WithHADisabled(ctx)
@@ -225,16 +167,7 @@ func MainWithConfig(ctx context.Context, component string, cfg *rest.Config, cto
 		cfg.Burst = len(ctors) * rest.DefaultBurst
 	}
 
-	// Respect user provided settings, but if omitted customize the default behavior.
-	if cfg.QPS == 0 {
-		cfg.QPS = rest.DefaultQPS
-	}
-	if cfg.Burst == 0 {
-		cfg.Burst = rest.DefaultBurst
-	}
-	ctx = injection.WithConfig(ctx, cfg)
-
-	ctx, informers := injection.Default.SetupInformers(ctx, cfg)
+	ctx, startInformers := injection.EnableInjectionOrDie(ctx, cfg)
 
 	logger, atomicLevel := SetupLoggerOrDie(ctx, component)
 	defer flush(logger)
@@ -287,11 +220,10 @@ func MainWithConfig(ctx context.Context, component string, cfg *rest.Config, cto
 			return wh.Run(ctx.Done())
 		})
 	}
+
 	// Start the injection clients and informers.
-	logging.FromContext(ctx).Info("Starting informers...")
-	if err := controller.StartInformers(ctx.Done(), informers...); err != nil {
-		logging.FromContext(ctx).Fatalw("Failed to start informers", zap.Error(err))
-	}
+	startInformers()
+
 	// Wait for webhook informers to sync.
 	if wh != nil {
 		wh.InformersHaveSynced()
@@ -317,22 +249,9 @@ func flush(logger *zap.SugaredLogger) {
 
 // ParseAndGetConfigOrDie parses the rest config flags and creates a client or
 // dies by calling log.Fatalf.
+// Deprecated: use injeciton.ParseAndGetRESTConfigOrDie
 func ParseAndGetConfigOrDie() *rest.Config {
-	var (
-		serverURL = flag.String("server", "",
-			"The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
-		kubeconfig = flag.String("kubeconfig", "",
-			"Path to a kubeconfig. Only required if out-of-cluster.")
-	)
-	klog.InitFlags(flag.CommandLine)
-	flag.Parse()
-
-	cfg, err := GetConfig(*serverURL, *kubeconfig)
-	if err != nil {
-		log.Fatalf("Error building kubeconfig: %v", err)
-	}
-
-	return cfg
+	return injection.ParseAndGetRESTConfigOrDie()
 }
 
 // MemStatsOrDie sets up reporting on Go memory usage every 30 seconds or dies
@@ -342,7 +261,7 @@ func MemStatsOrDie(ctx context.Context) {
 	msp.Start(ctx, 30*time.Second)
 
 	if err := view.Register(msp.DefaultViews()...); err != nil {
-		log.Fatalf("Error exporting go memstats view: %v", err)
+		log.Fatal("Error exporting go memstats view: ", err)
 	}
 }
 
@@ -351,7 +270,7 @@ func MemStatsOrDie(ctx context.Context) {
 func SetupLoggerOrDie(ctx context.Context, component string) (*zap.SugaredLogger, zap.AtomicLevel) {
 	loggingConfig, err := GetLoggingConfig(ctx)
 	if err != nil {
-		log.Fatalf("Error reading/parsing logging configuration: %v", err)
+		log.Fatal("Error reading/parsing logging configuration: ", err)
 	}
 	l, level := logging.NewLoggerFromConfig(loggingConfig, component)
 
diff --git a/vendor/knative.dev/pkg/leaderelection/chaosduck/main.go b/vendor/knative.dev/pkg/leaderelection/chaosduck/main.go
index fb3e5eb9b..2ad685fe3 100644
--- a/vendor/knative.dev/pkg/leaderelection/chaosduck/main.go
+++ b/vendor/knative.dev/pkg/leaderelection/chaosduck/main.go
@@ -27,13 +27,14 @@ import (
 	"strings"
 	"time"
 
+	"knative.dev/pkg/injection"
+
 	"golang.org/x/sync/errgroup"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
-	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/kflag"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/system"
@@ -116,8 +117,7 @@ func quack(ctx context.Context, kc kubernetes.Interface, component string, leade
 }
 
 func main() {
-	ctx := signals.NewContext()
-	ctx = sharedmain.EnableInjectionOrDie(ctx, nil)
+	ctx, _ := injection.EnableInjectionOrDie(signals.NewContext(), nil)
 
 	kc := kubeclient.Get(ctx)
 
@@ -126,7 +126,7 @@ func main() {
 	wait.JitterUntilWithContext(ctx, func(ctx context.Context) {
 		components, err := buildComponents(ctx, kc)
 		if err != nil {
-			log.Printf("Error building components: %v", err)
+			log.Print("Error building components: ", err)
 		}
 		log.Printf("Got components: %#v", components)
 
@@ -141,7 +141,7 @@ func main() {
 			})
 		}
 		if err := eg.Wait(); err != nil {
-			log.Printf("Ended iteration with err: %v", err)
+			log.Print("Ended iteration with err: ", err)
 		}
 	}, tributePeriod, tributeFactor, true /* sliding: do not include the runtime of the above in the interval */)
 }
diff --git a/vendor/knative.dev/pkg/metrics/prometheus_exporter.go b/vendor/knative.dev/pkg/metrics/prometheus_exporter.go
index d60fe8f07..7f49302cc 100644
--- a/vendor/knative.dev/pkg/metrics/prometheus_exporter.go
+++ b/vendor/knative.dev/pkg/metrics/prometheus_exporter.go
@@ -82,7 +82,7 @@ func startNewPromSrv(e *prom.Exporter, port int) *http.Server {
 		curPromSrv.Close()
 	}
 	curPromSrv = &http.Server{
-		Addr:    fmt.Sprintf(":%v", port),
+		Addr:    fmt.Sprint(":", port),
 		Handler: sm,
 	}
 	return curPromSrv
diff --git a/vendor/knative.dev/pkg/profiling/server.go b/vendor/knative.dev/pkg/profiling/server.go
index 0dc1eebc0..9ac2c5f05 100644
--- a/vendor/knative.dev/pkg/profiling/server.go
+++ b/vendor/knative.dev/pkg/profiling/server.go
@@ -94,7 +94,7 @@ func (h *Handler) UpdateFromConfigMap(configMap *corev1.ConfigMap) {
 	}
 
 	if h.enabled.Swap(enabled) != enabled {
-		h.log.Infof("Profiling enabled: %t", enabled)
+		h.log.Info("Profiling enabled: ", enabled)
 	}
 }
 
diff --git a/vendor/knative.dev/pkg/webhook/webhook.go b/vendor/knative.dev/pkg/webhook/webhook.go
index aff5cf8b4..55726c5f5 100644
--- a/vendor/knative.dev/pkg/webhook/webhook.go
+++ b/vendor/knative.dev/pkg/webhook/webhook.go
@@ -181,7 +181,7 @@ func (wh *Webhook) Run(stop <-chan struct{}) error {
 
 	server := &http.Server{
 		Handler: drainer,
-		Addr:    fmt.Sprintf(":%d", wh.Options.Port),
+		Addr:    fmt.Sprint(":", wh.Options.Port),
 		TLSConfig: &tls.Config{
 			MinVersion: tls.VersionTLS12,
 			GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 94c3c63d4..cd396cee5 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -281,6 +281,7 @@ github.com/prometheus/statsd_exporter/pkg/mapper/fsm
 # github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0
 github.com/rcrowley/go-metrics
 # github.com/rickb777/date v1.13.0
+## explicit
 github.com/rickb777/date/period
 # github.com/rickb777/plural v1.2.1
 github.com/rickb777/plural
@@ -617,7 +618,7 @@ gopkg.in/jcmturner/rpc.v1/mstypes
 gopkg.in/jcmturner/rpc.v1/ndr
 # gopkg.in/yaml.v2 v2.3.0
 gopkg.in/yaml.v2
-# gopkg.in/yaml.v3 v3.0.0-20200601152816-913338de1bd2
+# gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776
 gopkg.in/yaml.v3
 # honnef.co/go/tools v0.0.1-2020.1.4
 honnef.co/go/tools/arg
@@ -1039,7 +1040,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.18.1-0.20201006012020-59269224fe65
+# knative.dev/eventing v0.0.0-20201007101620-0aa53c127c61
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/configs
@@ -1149,6 +1150,8 @@ knative.dev/eventing/test/lib/dropevents/dropeventsfibonacci
 knative.dev/eventing/test/lib/dropevents/dropeventsfirst
 knative.dev/eventing/test/lib/duck
 knative.dev/eventing/test/lib/recordevents
+knative.dev/eventing/test/lib/recordevents/observer
+knative.dev/eventing/test/lib/recordevents/recorder_vent
 knative.dev/eventing/test/lib/resources
 knative.dev/eventing/test/lib/sender
 knative.dev/eventing/test/performance/infra
@@ -1167,7 +1170,7 @@ knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
 knative.dev/eventing/test/test_images/sequencestepper
 knative.dev/eventing/test/test_images/transformevents
-# knative.dev/pkg v0.0.0-20201005175134-21cb070a8b05
+# knative.dev/pkg v0.0.0-20201006235820-46761ba7c3dc
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1253,7 +1256,7 @@ knative.dev/pkg/webhook/certificates/resources
 knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/test-infra v0.0.0-20201005191434-b298834c10f1
+# knative.dev/test-infra v0.0.0-20201006161322-f94f2bcbb2ee
 ## explicit
 knative.dev/test-infra/scripts
 # sigs.k8s.io/structured-merge-diff/v3 v3.0.1-0.20200706213357-43c19bbb7fba


Test patch:
diff --git a/control-plane/pkg/config/env_config_test.go b/control-plane/pkg/config/env_config_test.go
index 83923bcf1..7df164dce 100644
--- a/control-plane/pkg/config/env_config_test.go
+++ b/control-plane/pkg/config/env_config_test.go
@@ -78,6 +78,22 @@ func TestGetEnvConfig(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: "invalid backoff delay",
+			args: args{
+				prefix: "BROKER",
+			},
+			setEnv: func() {
+				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAMESPACE", "knative-eventing")
+				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAME", "kafka-brokers-triggers")
+				_ = os.Setenv("BROKER_GENERAL_CONFIG_MAP_NAME", "kafka-config")
+				_ = os.Setenv("BROKER_INGRESS_NAME", "kafka-broker-ingress")
+				_ = os.Setenv("BROKER_SYSTEM_NAMESPACE", "knative-eventing")
+				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_FORMAT", "json")
+				_ = os.Setenv("BROKER_DEFAULT_BACKOFF_DELAY", "PTT")
+			},
+			wantErr: true,
+		},
 	}
 
 	for _, tt := range tests {
diff --git a/control-plane/pkg/core/config/utils_test.go b/control-plane/pkg/core/config/utils_test.go
index 74c280150..0e90edeeb 100644
--- a/control-plane/pkg/core/config/utils_test.go
+++ b/control-plane/pkg/core/config/utils_test.go
@@ -19,6 +19,9 @@ package config
 import (
 	"testing"
 
+	"k8s.io/utils/pointer"
+	duck "knative.dev/eventing/pkg/apis/duck/v1"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
@@ -56,3 +59,71 @@ func TestContentModeFromString(t *testing.T) {
 		})
 	}
 }
+
+func TestBackoffPolicyFromString(t *testing.T) {
+	linerar := duck.BackoffPolicyLinear
+	exponential := duck.BackoffPolicyExponential
+	wrong := duck.BackoffPolicyType("default")
+	tests := []struct {
+		name          string
+		backoffPolicy *duck.BackoffPolicyType
+		want          contract.BackoffPolicy
+	}{
+		{
+			name:          "nil",
+			backoffPolicy: nil,
+			want:          contract.BackoffPolicy_Exponential,
+		},
+		{
+			name:          "exponential",
+			backoffPolicy: &exponential,
+			want:          contract.BackoffPolicy_Exponential,
+		},
+		{
+			name:          "linear",
+			backoffPolicy: &linerar,
+			want:          contract.BackoffPolicy_Linear,
+		},
+		{
+			name:          "default",
+			backoffPolicy: &wrong,
+			want:          contract.BackoffPolicy_Exponential,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := BackoffPolicyFromString(tt.backoffPolicy); got != tt.want {
+				t.Errorf("BackoffPolicyFromString() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestBackoffDelayFromString(t *testing.T) {
+
+	tests := []struct {
+		name         string
+		backoffDelay *string
+		defaultDelay string
+		want         string
+	}{
+		{
+			name:         "happy case",
+			backoffDelay: pointer.StringPtr("PT2S"),
+			defaultDelay: "PT1S",
+			want:         "PT2S",
+		},
+		{
+			name:         "default",
+			defaultDelay: "PT1S",
+			want:         "PT1S",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := BackoffDelayFromString(tt.backoffDelay, tt.defaultDelay); got != tt.want {
+				t.Errorf("BackoffDelayFromString() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index d8c32f55d..4b59bd384 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -24,6 +24,8 @@ import (
 	"testing"
 	"time"
 
+	"k8s.io/utils/pointer"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"github.com/Shopify/sarama"
@@ -76,6 +78,9 @@ var (
 
 	createTopicError = fmt.Errorf("failed to create topic")
 	deleteTopicError = fmt.Errorf("failed to delete topic")
+
+	linear      = eventingduck.BackoffPolicyLinear
+	exponential = eventingduck.BackoffPolicyExponential
 )
 
 func TestBrokerReconciler(t *testing.T) {
@@ -1105,6 +1110,259 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - with retry config - exponential",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					WithRetry(pointer.Int32Ptr(10), &exponential, pointer.StringPtr("PT2S")),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig: &contract.EgressConfig{
+								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
+								Retry:         10,
+								BackoffPolicy: contract.BackoffPolicy_Exponential,
+								BackoffDelay:  "PT2S",
+							},
+						},
+					},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						WithRetry(pointer.Int32Ptr(10), &exponential, pointer.StringPtr("PT2S")),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
+		{
+			Name: "Reconciled normal - with retry config - linear",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					WithRetry(pointer.Int32Ptr(10), &linear, pointer.StringPtr("PT2S")),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig: &contract.EgressConfig{
+								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
+								Retry:         10,
+								BackoffPolicy: contract.BackoffPolicy_Linear,
+								BackoffDelay:  "PT2S",
+							},
+						},
+					},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						WithRetry(pointer.Int32Ptr(10), &linear, pointer.StringPtr("PT2S")),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
+		{
+			Name: "Reconciled normal - with no retry num",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					WithRetry(nil, &linear, pointer.StringPtr("PT2S")),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig: &contract.EgressConfig{
+								DeadLetter: "http://test-service.test-service-namespace.svc.cluster.local/",
+							},
+						},
+					},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						WithRetry(nil, &linear, pointer.StringPtr("PT2S")),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
+		{
+			Name: "Reconciled normal - with retry config - no retry delay - use default delay",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					WithRetry(pointer.Int32Ptr(10), &linear, nil),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig: &contract.EgressConfig{
+								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
+								Retry:         10,
+								BackoffPolicy: contract.BackoffPolicy_Linear,
+								BackoffDelay:  configs.DefaultBackoffDelay,
+							},
+						},
+					},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						WithRetry(pointer.Int32Ptr(10), &linear, nil),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
 	}
 
 	for i := range table {
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index a7970ac99..296c028d3 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -50,11 +50,12 @@ const (
 var DefaultConfigs = &broker.Configs{
 
 	Env: config.Env{
-		DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
 		DataPlaneConfigMapNamespace: "knative-eventing",
+		DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
 		IngressName:                 "kafka-broker-receiver",
 		SystemNamespace:             "knative-eventing",
 		DataPlaneConfigFormat:       base.Json,
+		DefaultBackoffDelay:         "PT1S",
 	},
 
 	BootstrapServers: "",
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 913376d3a..8bf4a203e 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -80,19 +80,31 @@ func WithDelivery() func(*eventing.Broker) {
 	service := NewService()
 
 	return func(broker *eventing.Broker) {
-		broker.Spec.Delivery = &eventingduck.DeliverySpec{
-			DeadLetterSink: &duckv1.Destination{
-				Ref: &duckv1.KReference{
-					Kind:       service.Kind,
-					Namespace:  service.Namespace,
-					Name:       service.Name,
-					APIVersion: service.APIVersion,
-				},
+		if broker.Spec.Delivery == nil {
+			broker.Spec.Delivery = &eventingduck.DeliverySpec{}
+		}
+		broker.Spec.Delivery.DeadLetterSink = &duckv1.Destination{
+			Ref: &duckv1.KReference{
+				Kind:       service.Kind,
+				Namespace:  service.Namespace,
+				Name:       service.Name,
+				APIVersion: service.APIVersion,
 			},
 		}
 	}
 }
 
+func WithRetry(retry *int32, policy *eventingduck.BackoffPolicyType, delay *string) func(*eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		if broker.Spec.Delivery == nil {
+			broker.Spec.Delivery = &eventingduck.DeliverySpec{}
+		}
+		broker.Spec.Delivery.Retry = retry
+		broker.Spec.Delivery.BackoffPolicy = policy
+		broker.Spec.Delivery.BackoffDelay = delay
+	}
+}
+
 func WithBrokerConfig(reference *duckv1.KReference) func(*eventing.Broker) {
 	return func(broker *eventing.Broker) {
 		broker.Spec.Config = reference
diff --git a/test/e2e/broker_trigger_test.go b/test/e2e/broker_trigger_test.go
index 93d7ac735..1061f32d9 100644
--- a/test/e2e/broker_trigger_test.go
+++ b/test/e2e/broker_trigger_test.go
@@ -23,8 +23,9 @@ import (
 	"fmt"
 	"testing"
 
-	cloudevents "github.com/cloudevents/sdk-go/v2"
 	. "github.com/cloudevents/sdk-go/v2/test"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/google/uuid"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
 	testlib "knative.dev/eventing/test/lib"
@@ -72,7 +73,6 @@ func TestBrokerTrigger(t *testing.T) {
 		)
 
 		eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, subscriber)
-		defer eventTracker.Cleanup()
 
 		client.CreateTriggerOrFailV1Beta1(
 			triggerName,
@@ -204,7 +204,6 @@ func TestBrokerWithConfig(t *testing.T) {
 		)
 
 		eventTracker, _ := recordevents.StartEventRecordOrFail(ctx, client, subscriber)
-		defer eventTracker.Cleanup()
 
 		client.CreateTriggerOrFailV1Beta1(
 			triggerName,
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/apiserversouce.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/apiserversouce.go
index 57eb09a1b..f78ab6a83 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/apiserversouce.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/apiserversouce.go
@@ -80,10 +80,23 @@ func WithApiServerSourceDeployed(s *v1.ApiServerSource) {
 	s.Status.PropagateDeploymentAvailability(testing.NewDeployment("any", "any", testing.WithDeploymentAvailable()))
 }
 
-func WithApiServerSourceEventTypes(source string) ApiServerSourceOption {
+func WithApiServerSourceReferenceModeEventTypes(source string) ApiServerSourceOption {
 	return func(s *v1.ApiServerSource) {
-		ceAttributes := make([]duckv1.CloudEventAttributes, 0, len(apisources.ApiServerSourceEventTypes))
-		for _, apiServerSourceType := range apisources.ApiServerSourceEventTypes {
+		ceAttributes := make([]duckv1.CloudEventAttributes, 0, len(apisources.ApiServerSourceEventReferenceModeTypes))
+		for _, apiServerSourceType := range apisources.ApiServerSourceEventReferenceModeTypes {
+			ceAttributes = append(ceAttributes, duckv1.CloudEventAttributes{
+				Type:   apiServerSourceType,
+				Source: source,
+			})
+		}
+		s.Status.CloudEventAttributes = ceAttributes
+	}
+}
+
+func WithApiServerSourceResourceModeEventTypes(source string) ApiServerSourceOption {
+	return func(s *v1.ApiServerSource) {
+		ceAttributes := make([]duckv1.CloudEventAttributes, 0, len(apisources.ApiServerSourceEventResourceModeTypes))
+		for _, apiServerSourceType := range apisources.ApiServerSourceEventResourceModeTypes {
 			ceAttributes = append(ceAttributes, duckv1.CloudEventAttributes{
 				Type:   apiServerSourceType,
 				Source: source,
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
index d82789a57..0121f0471 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
@@ -25,6 +25,7 @@ import (
 	"knative.dev/pkg/reconciler"
 
 	eventingv1beta1 "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	"knative.dev/eventing/test/lib/recordevents"
 
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/duck"
@@ -82,8 +83,7 @@ func triggerV1Beta1BeforeBrokerHelper(triggerName string, client *testlib.Client
 	const etLogger = "logger"
 	const loggerPodName = "logger-pod"
 
-	logPod := resources.EventRecordPod(loggerPodName)
-	client.CreatePodOrFail(logPod, testlib.WithService(loggerPodName))
+	_ = recordevents.DeployEventRecordOrFail(context.TODO(), client, loggerPodName)
 	client.WaitForAllTestResourcesReadyOrFail(context.Background()) // Can't do this for the trigger because it's not 'ready' yet
 	client.CreateTriggerOrFailV1Beta1(triggerName,
 		resources.WithAttributesTriggerFilterV1Beta1(eventingv1beta1.TriggerAnyFilter, etLogger, map[string]interface{}{}),
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/broker_tracing_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/broker_tracing_test_helper.go
index 41db7080d..9c09109c9 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/broker_tracing_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/broker_tracing_test_helper.go
@@ -31,6 +31,7 @@ import (
 	"knative.dev/eventing/pkg/apis/eventing/v1beta1"
 	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 	"knative.dev/eventing/test/lib/sender"
 )
@@ -82,8 +83,7 @@ func setupBrokerTracing(ctx context.Context, brokerClass string) SetupTracingTes
 		)
 
 		// Create a logger (EventRecord) Pod and a K8s Service that points to it.
-		logPod := resources.EventRecordPod(loggerPodName)
-		client.CreatePodOrFail(logPod, testlib.WithService(loggerPodName))
+		_ = recordevents.DeployEventRecordOrFail(ctx, client, loggerPodName)
 
 		// Create a Trigger that receives events (type=bar) and sends them to the logger Pod.
 		loggerTrigger := client.CreateTriggerOrFailV1Beta1(
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
index d148e26d6..4a0c2e798 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/channel_status_subscriber_test_helper.go
@@ -24,6 +24,7 @@ import (
 	duckv1beta1 "knative.dev/eventing/pkg/apis/duck/v1beta1"
 	eventingv1beta1 "knative.dev/eventing/pkg/apis/messaging/v1beta1"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 
 	corev1 "k8s.io/api/core/v1"
@@ -59,8 +60,7 @@ func channelHasRequiredSubscriberStatus(ctx context.Context, st *testing.T, clie
 	client.CreateChannelOrFail(channelName, &channel)
 	client.WaitForResourceReadyOrFail(channelName, &channel)
 
-	pod := resources.EventRecordPod(subscriberServiceName + "-pod")
-	client.CreatePodOrFail(pod, testlib.WithService(subscriberServiceName))
+	_ = recordevents.DeployEventRecordOrFail(context.TODO(), client, subscriberServiceName+"-pod")
 
 	subscription := client.CreateSubscriptionOrFail(
 		subscriberServiceName,
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/channel_tracing_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/channel_tracing_test_helper.go
index 38afa62cf..c8d95f567 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/channel_tracing_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/channel_tracing_test_helper.go
@@ -29,6 +29,7 @@ import (
 
 	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 	"knative.dev/eventing/test/lib/sender"
 )
@@ -69,8 +70,7 @@ func setupChannelTracingWithReply(
 	client.CreateChannelOrFail(replyChannelName, channel)
 
 	// Create the 'sink', a LogEvents Pod and a K8s Service that points to it.
-	recordEventsPod := resources.EventRecordPod(recordEventsPodName)
-	client.CreatePodOrFail(recordEventsPod, testlib.WithService(recordEventsPodName))
+	recordEventsPod := recordevents.DeployEventRecordOrFail(ctx, client, recordEventsPodName)
 
 	// Create the subscriber, a Pod that mutates the event.
 	transformerPod := resources.EventTransformationPod(
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/tracing_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/tracing_test_helper.go
index 1b2f9af79..f3e476f9f 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/tracing_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/tracing_test_helper.go
@@ -67,7 +67,7 @@ func tracingTest(
 	expectedTestSpan, eventMatcher := setupInfrastructure(ctx, t, &channel, client, recordEventsPodName, true)
 
 	// Start the event info store and assert the event was received correctly
-	targetTracker, err := recordevents.NewEventInfoStore(client, recordEventsPodName)
+	targetTracker, err := recordevents.NewEventInfoStore(client, recordEventsPodName, client.Namespace)
 	if err != nil {
 		t.Fatal("Pod tracker failed:", err)
 	}
@@ -107,6 +107,6 @@ func getTraceIDHeader(t *testing.T, evInfos []recordevents.EventInfo) string {
 			}
 		}
 	}
-	t.Fatalf("FAIL: No traceid in %d messages: (%s)", len(evInfos), evInfos)
+	t.Fatalf("FAIL: No traceid in %d messages: (%v)", len(evInfos), evInfos)
 	return ""
 }
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/trigger_no_broker_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/trigger_no_broker_test_helper.go
index ce8a4d9c4..0303be7e8 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/trigger_no_broker_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/trigger_no_broker_test_helper.go
@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	testlib "knative.dev/eventing/test/lib"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
 )
 
@@ -41,8 +42,8 @@ func TestTriggerNoBroker(ctx context.Context, t *testing.T, channel string, brok
 	brokerName := strings.ToLower(channel)
 
 	subscriberName := "dumper-empty"
-	eventRecordPod := resources.EventRecordPod(subscriberName)
-	client.CreatePodOrFail(eventRecordPod, testlib.WithService(subscriberName))
+	recordevents.DeployEventRecordOrFail(context.TODO(), client, subscriberName)
+
 	client.CreateTriggerOrFailV1Beta1("testtrigger",
 		resources.WithSubscriberServiceRefForTriggerV1Beta1(subscriberName),
 		resources.WithBrokerV1Beta1(brokerName),
diff --git a/vendor/knative.dev/eventing/test/lib/client.go b/vendor/knative.dev/eventing/test/lib/client.go
index 1eaef6ece..d1c5782cc 100644
--- a/vendor/knative.dev/eventing/test/lib/client.go
+++ b/vendor/knative.dev/eventing/test/lib/client.go
@@ -48,6 +48,8 @@ type Client struct {
 	Dynamic       dynamic.Interface
 	Config        *rest.Config
 
+	EventListener *EventListener
+
 	Namespace string
 	T         *testing.T
 	Tracker   *Tracker
@@ -93,6 +95,10 @@ func NewClient(configPath string, clusterName string, namespace string, t *testi
 	client.T = t
 	client.Tracker = NewTracker(t, client.Dynamic)
 
+	// Start informer
+	client.EventListener = NewEventListener(client.Kube.Kube, client.Namespace, client.T.Logf)
+	client.Cleanup(client.EventListener.Stop)
+
 	client.tracingEnv, err = getTracingConfig(client.Kube.Kube)
 	if err != nil {
 		return nil, err
diff --git a/vendor/knative.dev/eventing/test/lib/duck/resource_inspectors.go b/vendor/knative.dev/eventing/test/lib/duck/resource_inspectors.go
index af5f2209f..fa29ec282 100644
--- a/vendor/knative.dev/eventing/test/lib/duck/resource_inspectors.go
+++ b/vendor/knative.dev/eventing/test/lib/duck/resource_inspectors.go
@@ -25,29 +25,30 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/dynamic"
-	duckv1alpha1 "knative.dev/pkg/apis/duck/v1alpha1"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"knative.dev/eventing/test/lib/resources"
 )
 
 // GetAddressableURI returns the uri for the given resource that implements Addressable duck-type.
 func GetAddressableURI(dynamicClient dynamic.Interface, obj *resources.MetaResource) (url.URL, error) {
-	untyped, err := GetGenericObject(dynamicClient, obj, &duckv1alpha1.AddressableType{})
+	untyped, err := GetGenericObject(dynamicClient, obj, &duckv1.AddressableType{})
 	if err != nil {
 		return url.URL{}, err
 	}
 
-	at := untyped.(*duckv1alpha1.AddressableType)
+	at := untyped.(*duckv1.AddressableType)
 
 	if at.Status.Address == nil {
 		return url.URL{}, fmt.Errorf("addressable does not have an Address: %+v", at)
 	}
-
-	au := at.Status.Address.GetURL()
-	if au.Host == "" {
+	if at.Status.Address.URL == nil {
+		return url.URL{}, fmt.Errorf("addressable does not have a URL: %+v", at)
+	}
+	if at.Status.Address.URL.Host == "" {
 		return url.URL{}, fmt.Errorf("addressable's URL does not have a Host: %+v", at)
 	}
-	return url.URL(au), nil
+	return url.URL(*at.Status.Address.URL), nil
 }
 
 func GetServiceHostname(svc *corev1.Service) string {
diff --git a/vendor/knative.dev/eventing/test/lib/k8s_events.go b/vendor/knative.dev/eventing/test/lib/k8s_events.go
new file mode 100644
index 000000000..1188d5c23
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/k8s_events.go
@@ -0,0 +1,84 @@
+/*
+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 lib
+
+import (
+	"context"
+	"sync"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+// EventHandler is the callback type for the EventListener
+type EventHandler func(event *corev1.Event)
+
+// EventListener is a type that broadcasts new k8s events
+type EventListener struct {
+	cancel context.CancelFunc
+
+	lock     sync.Mutex
+	handlers []EventHandler
+}
+
+// NewEventListener creates a new event listener
+func NewEventListener(client kubernetes.Interface, namespace string, logf func(string, ...interface{})) *EventListener {
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(
+		client,
+		0,
+		informers.WithNamespace(namespace),
+	)
+	eventsInformer := informerFactory.Core().V1().Events().Informer()
+
+	el := EventListener{
+		cancel: cancelCtx,
+	}
+
+	eventsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			el.handle(obj.(*corev1.Event))
+		},
+	})
+
+	go func() {
+		eventsInformer.Run(ctx.Done())
+		logf("EventListener stopped")
+	}()
+
+	return &el
+}
+
+func (el *EventListener) handle(event *corev1.Event) {
+	el.lock.Lock()
+	defer el.lock.Unlock()
+	for _, handler := range el.handlers {
+		handler(event)
+	}
+}
+
+func (el *EventListener) AddHandler(handler EventHandler) {
+	el.lock.Lock()
+	defer el.lock.Unlock()
+	el.handlers = append(el.handlers, handler)
+}
+
+func (el *EventListener) Stop() {
+	el.cancel()
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go b/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go
index dc0ffd1f0..d9dbbdae8 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go
@@ -17,56 +17,31 @@ limitations under the License.
 package recordevents
 
 import (
-	"context"
-	"encoding/json"
 	"fmt"
-	"io/ioutil"
-	"math/rand"
-	"net/http"
 	"strings"
 	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
-	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
-	"knative.dev/pkg/test/monitoring"
-
-	"knative.dev/pkg/test/logging"
-
-	testlib "knative.dev/eventing/test/lib"
 )
 
-// Port for the recordevents pod REST listener
-const RecordEventsPort = 8392
-
-// HTTP path for the GetMinMax REST call
-const GetMinMaxPath = "/minmax"
-
-// HTTP path for the GetEntry REST call
-const GetEntryPath = "/entry/"
-
-// HTTP path for the TrimThrough REST call
-const TrimThroughPath = "/trimthrough/"
-
-// On-wire json rest api format for recordevents GetMinMax calls
-// sennt to the recordevents pod.
-type MinMaxResponse struct {
-	MinAvail int
-	MaxSeen  int
-}
+const (
+	// EventReason is the Kubernetes event reason used for observed events.
+	CloudEventObservedReason = "CloudEventObserved"
+)
 
 // Structure to hold information about an event seen by recordevents pod.
 type EventInfo struct {
 	// Set if the http request received by the pod couldn't be decoded or
 	// didn't pass validation
-	Error string
+	Error string `json:"error,omitempty"`
 	// Event received if the cloudevent received by the pod passed validation
-	Event *cloudevents.Event
+	Event *cloudevents.Event `json:"event,omitempty"`
 	// HTTPHeaders of the connection that delivered the event
-	HTTPHeaders map[string][]string
+	HTTPHeaders map[string][]string `json:"httpHeaders,omitempty"`
+	Origin      string              `json:"origin,omitempty"`
+	Observer    string              `json:"observer,omitempty"`
+	Time        time.Time           `json:"time,omitempty"`
+	Sequence    uint64              `json:"sequence"`
 }
 
 // Pretty print the event. Meant for debugging.  This formats the validation error
@@ -95,189 +70,3 @@ func (s *SearchedInfo) String() string {
 	}
 	return sb.String()
 }
-
-// Connection state for a REST connection to a pod
-type eventGetter struct {
-	podName       string
-	podNamespace  string
-	podPort       int
-	kubeClientset kubernetes.Interface
-	logf          logging.FormatLogger
-
-	host       string
-	port       int
-	forwardPID int
-}
-
-// Creates a forwarded port to the specified recordevents pod and waits until
-// it can successfully talk to the REST API.  Times out after timeoutEvRetry
-func newEventGetter(podName string, client *testlib.Client, logf logging.FormatLogger) (eventGetterInterface, error) {
-	egi := &eventGetter{podName: podName, podNamespace: client.Namespace,
-		kubeClientset: client.Kube.Kube, podPort: RecordEventsPort, logf: logf}
-	err := egi.forwardPort()
-	if err != nil {
-		return nil, err
-	}
-
-	err = egi.waitTillUp()
-	if err != nil {
-		return nil, err
-	}
-	return egi, nil
-}
-
-// Get information about the provided podName.  Uses list (rather than get) and
-// returns a pod list for compatibility with the monitoring.PortForward
-// interface
-func (eg *eventGetter) getRunningPodInfo(podName, namespace string) (*v1.PodList, error) {
-	pods, err := eg.kubeClientset.CoreV1().Pods(namespace).List(context.Background(),
-		metav1.ListOptions{FieldSelector: fmt.Sprintf("metadata.name=%s", podName)})
-	if err == nil && len(pods.Items) != 1 {
-		err = fmt.Errorf("no %s Pod found on the cluster", podName)
-	} else if pods.Items[0].Status.Phase != corev1.PodRunning {
-		err = fmt.Errorf("pod %s in state %s, wanted Running", podName,
-			pods.Items[0].Status.Phase)
-	}
-
-	return pods, err
-}
-
-// Try to forward the pod port to a local port somewhere in the range 30000-60000.
-// keeps retrying with random ports in that range, timing out after timeoutEvRetry
-func (eg *eventGetter) forwardPort() error {
-	portRand := rand.New(rand.NewSource(time.Now().UnixNano()))
-	portMin := 30000
-	portMax := 60000
-	var internalErr error
-
-	wait.PollImmediate(minEvRetryInterval, timeoutEvRetry, func() (bool, error) {
-		localPort := portMin + portRand.Intn(portMax-portMin)
-		if err := monitoring.CheckPortAvailability(localPort); err != nil {
-			internalErr = err
-			return false, nil
-		}
-		pods, err := eg.getRunningPodInfo(eg.podName, eg.podNamespace)
-		if err != nil {
-			internalErr = err
-			return false, nil
-		}
-
-		pid, err := monitoring.PortForward(eg.logf, pods, localPort, eg.podPort, eg.podNamespace)
-		if err != nil {
-			internalErr = err
-			return false, nil
-		}
-		internalErr = nil
-
-		eg.forwardPID = pid
-		eg.port = localPort
-		eg.host = "localhost"
-		return true, nil
-	})
-	if internalErr != nil {
-		return fmt.Errorf("timeout forwarding port: %v", internalErr)
-	}
-	return nil
-}
-
-// Return the min available, max seen by the recordevents pod.
-// maxRet is the largest event that has ever been seen (whether it's been trimmed
-// or not).  minRet is the smallest event still available via Get, or 1+maxRet if
-// no events are available.  maxRet starts at 0 when no events have been seen.
-func (eg *eventGetter) getMinMax() (minRet int, maxRet int, errRet error) {
-	resp, err := http.Get(fmt.Sprintf("http://%s:%d%s", eg.host, eg.port, GetMinMaxPath))
-	if err != nil {
-		return -1, -1, fmt.Errorf("http get error: %v", err)
-	}
-	defer resp.Body.Close()
-	bodyContents, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return -1, -1, fmt.Errorf("error reading response body %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return -1, -1, fmt.Errorf("error %d reading GetMinMax response", resp.StatusCode)
-	}
-	minMaxResponse := MinMaxResponse{}
-	err = json.Unmarshal(bodyContents, &minMaxResponse)
-	if err != nil {
-		return -1, -1, fmt.Errorf("error unmarshalling response %w", err)
-	}
-	if minMaxResponse.MinAvail == 0 {
-		return -1, -1, fmt.Errorf("invalid decoded json: %+v", minMaxResponse)
-	}
-
-	return minMaxResponse.MinAvail, minMaxResponse.MaxSeen, nil
-}
-
-// Return the event with the provided sequence number.  Returns the appropriate
-// EventInfo or an error if no such event is known, or the event has already
-// been trimmed.
-func (eg *eventGetter) getEntry(seqno int) (EventInfo, error) {
-	resp, err := http.Get(fmt.Sprintf("http://%s:%d%s/%d", eg.host, eg.port, GetEntryPath, seqno))
-	if err != nil {
-		return EventInfo{}, fmt.Errorf("http get err %v", err)
-	}
-	defer resp.Body.Close()
-	bodyContents, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return EventInfo{}, fmt.Errorf("error reading response body %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return EventInfo{}, fmt.Errorf("error %d reading GetEntry response", resp.StatusCode)
-	}
-	entryResponse := EventInfo{}
-	err = json.Unmarshal(bodyContents, &entryResponse)
-	if err != nil {
-		return EventInfo{}, fmt.Errorf("error unmarshalling response %w", err)
-	}
-	if len(entryResponse.Error) == 0 && entryResponse.Event == nil {
-		return EventInfo{}, fmt.Errorf("invalid decoded json: %+v", entryResponse)
-	}
-
-	return entryResponse, nil
-}
-
-// Trim the events up to and including seqno from the recordevents pod.
-// Returns an error if a nonsensical seqno is passed in, but does not return
-// error for trimming already trimmed regions.
-func (eg *eventGetter) trimThrough(seqno int) error {
-	resp, err := http.Post(fmt.Sprintf("http://%s:%d%s/%d", eg.host, eg.port, TrimThroughPath, seqno), "", nil)
-	if err != nil {
-		return fmt.Errorf("http post err %v", err)
-	}
-	defer resp.Body.Close()
-	body, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return fmt.Errorf("error reading response body %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("error %d reading TrimThrough response: %s", resp.StatusCode, string(body))
-	}
-
-	return nil
-}
-
-// Clean up the getter by tearing down the port forward.
-func (eg *eventGetter) cleanup() {
-	pid := eg.forwardPID
-	eg.forwardPID = 0
-	if pid != 0 {
-		monitoring.Cleanup(pid)
-	}
-}
-
-// Wait (up to timeoutEvRetry) for the pod to RestAPI to answer request.
-func (eg *eventGetter) waitTillUp() error {
-	var internalErr error
-	wait.PollImmediate(minEvRetryInterval, timeoutEvRetry, func() (bool, error) {
-		_, _, internalErr = eg.getMinMax()
-		if internalErr != nil {
-			return false, nil
-		}
-		return true, nil
-	})
-	if internalErr != nil {
-		return fmt.Errorf("timeout waiting for recordevents pod to come up: %v", internalErr)
-	}
-	return nil
-}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/event_info_store.go b/vendor/knative.dev/eventing/test/lib/recordevents/event_info_store.go
index 52fcbae3c..007a4c149 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/event_info_store.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_info_store.go
@@ -18,6 +18,7 @@ package recordevents
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strconv"
 	"strings"
@@ -27,6 +28,7 @@ import (
 
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	pkgTest "knative.dev/pkg/test"
 
@@ -36,169 +38,106 @@ import (
 
 const (
 	// The interval and timeout used for checking events
-	minEvRetryInterval = 4 * time.Second
-	timeoutEvRetry     = 4 * time.Minute
+	retryInterval = 4 * time.Second
+	retryTimeout  = 4 * time.Minute
 )
 
-// Stateful store of events received by the recordevents pod it is pointed at.
-// This pulls events from the pod during any Find or Wait call, storing them
-// locally and triming them from the remote pod store.
-type EventInfoStore struct {
-	tb testing.TB
-
-	podName string
-	getter  eventGetterInterface
-
-	lock          sync.Mutex
-	allEvents     []EventInfo
-	firstID       int
-	closeCh       chan struct{}
-	doRefresh     chan chan error
-	timeout       time.Duration
-	retryInterval time.Duration
-}
-
-// Functions used for getting data from the REST api of the recordevents pod.
-// The interface exists for use with unit tests of this module.
-type eventGetterInterface interface {
-	getMinMax() (minRet int, maxRet int, errRet error)
-	getEntry(seqno int) (EventInfo, error)
-	trimThrough(seqno int) error
-	cleanup()
-}
-
-// Internal function to create an event store.  This is called directly by unit tests of
-// this module.
-func newTestableEventInfoStore(egi eventGetterInterface, retryInterval time.Duration,
-	timeout time.Duration) *EventInfoStore {
-	if timeout == -1 {
-		timeout = timeoutEvRetry
-	}
-	if retryInterval == -1 {
-		retryInterval = minEvRetryInterval
-	}
-	ei := &EventInfoStore{getter: egi, firstID: 1, timeout: timeout, retryInterval: retryInterval}
-	ei.start()
-	return ei
-}
+type EventRecordOption = func(*corev1.Pod, *testlib.Client) error
 
-// Creates an EventInfoStore that is used to iteratively download events recorded by the
-// recordevents pod.  Calling this forwards the recordevents port to the local machine
-// and blocks waiting to connect to that pod.  Fails if it cannot connect within
-// the expected timeout (4 minutes currently)
-func NewEventInfoStore(client *testlib.Client, podName string) (*EventInfoStore, error) {
-	egi, err := newEventGetter(podName, client, client.T.Logf)
+func DeployEventRecordOrFail(ctx context.Context, client *testlib.Client, name string, options ...EventRecordOption) *corev1.Pod {
+	client.CreateServiceAccountOrFail(name)
+	client.CreateRoleOrFail(resources.Role(name,
+		resources.WithRuleForRole(&rbacv1.PolicyRule{
+			APIGroups: []string{""},
+			Resources: []string{"pods"},
+			Verbs:     []string{"get"},
+		}),
+		resources.WithRuleForRole(&rbacv1.PolicyRule{
+			APIGroups: []string{""},
+			Resources: []string{"events"},
+			Verbs:     []string{rbacv1.VerbAll},
+		}),
+	))
+	client.CreateRoleBindingOrFail(name, "Role", name, name, client.Namespace)
+
+	eventRecordPod := EventRecordPod(name, name)
+	client.CreatePodOrFail(eventRecordPod, append(options, testlib.WithService(name))...)
+	err := pkgTest.WaitForPodRunning(ctx, client.Kube, name, client.Namespace)
 	if err != nil {
-		return nil, err
+		client.T.Fatalf("Failed to start the recordevent pod '%s': %v", name, errors.WithStack(err))
 	}
-	ei := newTestableEventInfoStore(egi, -1, -1)
-	ei.podName = podName
-	ei.tb = client.T
-	client.Cleanup(ei.cleanup)
-	return ei, nil
+	client.WaitForServiceEndpointsOrFail(ctx, name, 1)
+	return eventRecordPod
 }
 
-type EventRecordOption = func(*corev1.Pod, *testlib.Client) error
-
 // Deploys a new recordevents pod and start the associated EventInfoStore
 func StartEventRecordOrFail(ctx context.Context, client *testlib.Client, podName string, options ...EventRecordOption) (*EventInfoStore, *corev1.Pod) {
-	eventRecordPod := resources.EventRecordPod(podName)
-	client.CreatePodOrFail(eventRecordPod, append(options, testlib.WithService(podName))...)
-	err := pkgTest.WaitForPodRunning(ctx, client.Kube, podName, client.Namespace)
-	if err != nil {
-		client.T.Fatalf("Failed to start the recordevent pod '%s': %v", podName, errors.WithStack(err))
-	}
-	client.WaitForServiceEndpointsOrFail(ctx, podName, 1)
+	eventRecordPod := DeployEventRecordOrFail(ctx, client, podName, options...)
 
-	eventTracker, err := NewEventInfoStore(client, podName)
+	eventTracker, err := NewEventInfoStore(client, podName, client.Namespace)
 	if err != nil {
 		client.T.Fatalf("Failed to start the EventInfoStore associated to pod '%s': %v", podName, err)
 	}
 	return eventTracker, eventRecordPod
 }
 
-// Starts the single threaded background goroutine used to update local state
-// from the remote REST API.
-func (ei *EventInfoStore) start() {
-	ei.closeCh = make(chan struct{})
-	ei.doRefresh = make(chan chan error)
-	go func() {
-		for {
-			select {
-			case <-ei.closeCh:
-				ei.getter.cleanup()
-				return
-			case replyCh := <-ei.doRefresh:
-				replyCh <- ei.doRetrieveData()
-			}
-		}
-	}()
+// Stateful store of events received by the recordevents pod it is pointed at.
+// This pulls events from the pod during any Find or Wait call, storing them
+// locally and triming them from the remote pod store.
+type EventInfoStore struct {
+	tb testing.TB
+
+	podName      string
+	podNamespace string
+
+	lock      sync.Mutex
+	collected []EventInfo
 }
 
-// The data update thread used by the single threaded background goroutine
-// for updating data from the REST api.
-func (ei *EventInfoStore) doRetrieveData() error {
-	min, max, err := ei.getter.getMinMax()
-	if err != nil {
-		return fmt.Errorf("error getting MinMax %v", err)
-	}
-	ei.lock.Lock()
-	curMin := ei.firstID
-	curMax := curMin + len(ei.allEvents) - 1
-	ei.lock.Unlock()
-	if min == max+1 {
-		// Nothing to read or trim
-		return nil
-	} else {
-		if min > curMax+1 {
-			return fmt.Errorf("mismatched stored max/available min: %d, %d", curMax, min)
-		}
-		min = curMax + 1
-		// We may have data to read, definitely have data to trim.
+// Creates an EventInfoStore that is used to iteratively download events recorded by the
+// recordevents pod.
+func NewEventInfoStore(client *testlib.Client, podName string, podNamespace string) (*EventInfoStore, error) {
+	store := &EventInfoStore{
+		tb:           client.T,
+		podName:      podName,
+		podNamespace: podNamespace,
 	}
-	var newEvents []EventInfo
-	for i := min; i <= max; i++ {
-		e, err := ei.getter.getEntry(i)
-		if err != nil {
-			return fmt.Errorf("error calling getEntry of %d %v", i, err)
-		}
 
-		newEvents = append(newEvents, e)
-	}
-	ei.lock.Lock()
-	ei.allEvents = append(ei.allEvents, newEvents...)
-	ei.lock.Unlock()
-	err = ei.getter.trimThrough(max)
-	return err
+	client.EventListener.AddHandler(store.handle)
 
+	return store, nil
 }
 
-// Clean up any background resources used by the store.  Must be called exactly once after
-// the last use.
-func (ei *EventInfoStore) cleanup() {
-	close(ei.closeCh)
+func (ei *EventInfoStore) getEventInfo() []EventInfo {
+	ei.lock.Lock()
+	defer ei.lock.Unlock()
+	return ei.collected
 }
 
-//TODO remove it, this is not useful anymore
-// Deprecated: you can remove the manual cleanup of the event getter, since now it's done at test tear down automatically
-func (ei *EventInfoStore) Cleanup() {}
-
-// Called internally by functions wanting the current list of all
-// known events.  This calls for an update from the REST server and
-// returns the summary of all locally and remotely known events.
-// Returns an error in case of a connection or protocol error.
-func (ei *EventInfoStore) refreshData() ([]EventInfo, error) {
-	var allEvents []EventInfo
-	replyCh := make(chan error)
-	ei.doRefresh <- replyCh
-	err := <-replyCh
+func (ei *EventInfoStore) handle(event *corev1.Event) {
+	// Filter events
+	if !ei.isMyEvent(event) {
+		return
+	}
+
+	eventInfo := EventInfo{}
+	err := json.Unmarshal([]byte(event.Message), &eventInfo)
 	if err != nil {
-		return nil, err
+		ei.tb.Errorf("Received EventInfo that cannot be unmarshalled! %+v", err)
+		return
 	}
+
 	ei.lock.Lock()
-	allEvents = append(allEvents, ei.allEvents...)
-	ei.lock.Unlock()
-	return allEvents, nil
+	defer ei.lock.Unlock()
+	ei.collected = append(ei.collected, eventInfo)
+}
+
+func (ei *EventInfoStore) isMyEvent(event *corev1.Event) bool {
+	return event.Type == corev1.EventTypeNormal &&
+		event.Reason == CloudEventObservedReason &&
+		event.InvolvedObject.Kind == "Pod" &&
+		event.InvolvedObject.Name == ei.podName &&
+		event.InvolvedObject.Namespace == ei.podNamespace
 }
 
 // Find all events received by the recordevents pod that match the provided matchers,
@@ -216,10 +155,7 @@ func (ei *EventInfoStore) Find(matchers ...EventInfoMatcher) ([]EventInfo, Searc
 	lastEvents := []EventInfo{}
 	var nonMatchingErrors []error
 
-	allEvents, err := ei.refreshData()
-	if err != nil {
-		return nil, sInfo, nonMatchingErrors, fmt.Errorf("error getting events %v", err)
-	}
+	allEvents := ei.getEventInfo()
 	for i := range allEvents {
 		if err := f(allEvents[i]); err == nil {
 			allMatch = append(allMatch, allEvents[i])
@@ -241,27 +177,31 @@ func (ei *EventInfoStore) Find(matchers ...EventInfoMatcher) ([]EventInfo, Searc
 // Assert that there are at least min number of match for the provided matchers.
 // This method fails the test if the assert is not fulfilled.
 func (ei *EventInfoStore) AssertAtLeast(min int, matchers ...EventInfoMatcher) []EventInfo {
+	ei.tb.Helper()
 	events, err := ei.waitAtLeastNMatch(AllOf(matchers...), min)
 	if err != nil {
 		ei.tb.Fatalf("Timeout waiting for at least %d matches.\nError: %+v", min, errors.WithStack(err))
 	}
+	ei.tb.Logf("Assert passed")
 	return events
 }
 
 // Assert that there are at least min number of matches and at most max number of matches for the provided matchers.
 // This method fails the test if the assert is not fulfilled.
 func (ei *EventInfoStore) AssertInRange(min int, max int, matchers ...EventInfoMatcher) []EventInfo {
+	ei.tb.Helper()
 	events := ei.AssertAtLeast(min, matchers...)
 	if max > 0 && len(events) > max {
 		ei.tb.Fatalf("Assert in range failed: %+v", errors.WithStack(fmt.Errorf("expected <= %d events, saw %d", max, len(events))))
 	}
-
+	ei.tb.Logf("Assert passed")
 	return events
 }
 
 // Assert that there aren't any matches for the provided matchers.
 // This method fails the test if the assert is not fulfilled.
 func (ei *EventInfoStore) AssertNot(matchers ...EventInfoMatcher) []EventInfo {
+	ei.tb.Helper()
 	res, recentEvents, _, err := ei.Find(matchers...)
 	if err != nil {
 		ei.tb.Fatalf("Unexpected error during find on recordevents '%s': %+v", ei.podName, errors.WithStack(err))
@@ -272,14 +212,17 @@ func (ei *EventInfoStore) AssertNot(matchers ...EventInfoMatcher) []EventInfo {
 			fmt.Errorf("Unexpected matches on recordevents '%s', found: %v. %s", ei.podName, res, &recentEvents)),
 		)
 	}
-
+	ei.tb.Logf("Assert passed")
 	return res
 }
 
 // Assert that there are exactly n matches for the provided matchers.
 // This method fails the test if the assert is not fulfilled.
 func (ei *EventInfoStore) AssertExact(n int, matchers ...EventInfoMatcher) []EventInfo {
-	return ei.AssertInRange(n, n, matchers...)
+	ei.tb.Helper()
+	events := ei.AssertInRange(n, n, matchers...)
+	ei.tb.Logf("Assert passed")
+	return events
 }
 
 // Wait a long time (currently 4 minutes) until the provided function matches at least
@@ -290,7 +233,7 @@ func (ei *EventInfoStore) waitAtLeastNMatch(f EventInfoMatcher, min int) ([]Even
 	var matchRet []EventInfo
 	var internalErr error
 
-	wait.PollImmediate(ei.retryInterval, ei.timeout, func() (bool, error) {
+	wait.PollImmediate(retryInterval, retryTimeout, func() (bool, error) {
 		allMatch, sInfo, matchErrs, err := ei.Find(f)
 		if err != nil {
 			internalErr = fmt.Errorf("FAIL MATCHING: unexpected error during find: %v", err)
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go b/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go
new file mode 100644
index 000000000..62aba6113
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go
@@ -0,0 +1,33 @@
+/*
+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 recordevents
+
+// EventLog is the contract for an event logger to vent an event.
+type EventLog interface {
+	Vent(observed EventInfo) error
+}
+
+type EventLogs []EventLog
+
+func (e EventLogs) Vent(observed EventInfo) error {
+	for _, el := range e {
+		if err := el.Vent(observed); err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go b/vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go
new file mode 100644
index 000000000..e9bd4224a
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go
@@ -0,0 +1,116 @@
+/*
+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 observer
+
+import (
+	"context"
+	"net/http"
+	"sync/atomic"
+	"time"
+
+	cloudeventsbindings "github.com/cloudevents/sdk-go/v2/binding"
+	cloudeventshttp "github.com/cloudevents/sdk-go/v2/protocol/http"
+	"github.com/kelseyhightower/envconfig"
+	"github.com/prometheus/common/log"
+	"knative.dev/pkg/logging"
+
+	"knative.dev/eventing/test/lib/recordevents"
+)
+
+// Observer is the entry point for sinking events into the event log.
+type Observer struct {
+	// Name is the name of this Observer, used to filter if multiple observers.
+	Name string
+	// EventLogs is the list of EventLog implementors to vent observed events.
+	EventLogs recordevents.EventLogs
+
+	seq uint64
+}
+
+// New returns an observer that will vent observations to the list of provided
+// EventLog instances. It will listen on :8080.
+func New(name string, eventLogs ...recordevents.EventLog) *Observer {
+	return &Observer{
+		Name:      name,
+		EventLogs: eventLogs,
+	}
+}
+
+type envConfig struct {
+	// ObserverName is used to identify this instance of the observer.
+	ObserverName string `envconfig:"OBSERVER_NAME" default:"observer-default" required:"true"`
+}
+
+func NewFromEnv(eventLogs ...recordevents.EventLog) *Observer {
+	var env envConfig
+	if err := envconfig.Process("", &env); err != nil {
+		log.Fatal("Failed to process env var", err)
+	}
+
+	return New(env.ObserverName, eventLogs...)
+}
+
+// Start will create the CloudEvents client and start listening for inbound
+// HTTP requests. This is a is a blocking call.
+func (o *Observer) Start(ctx context.Context, handlerFuncs ...func(handler http.Handler) http.Handler) error {
+	var handler http.Handler = o
+
+	for _, dec := range handlerFuncs {
+		handler = dec(handler)
+	}
+
+	server := &http.Server{Addr: ":8080", Handler: handler}
+
+	go func() {
+		if err := server.ListenAndServe(); err != nil {
+			logging.FromContext(ctx).Fatal("Error while starting the HTTP server", err)
+		}
+	}()
+
+	<-ctx.Done()
+
+	logging.FromContext(ctx).Info("Closing the HTTP server")
+
+	return server.Close()
+}
+
+func (o *Observer) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	m := cloudeventshttp.NewMessageFromHttpRequest(request)
+	defer m.Finish(nil)
+
+	event, eventErr := cloudeventsbindings.ToEvent(context.TODO(), m)
+	header := request.Header
+
+	eventErrStr := ""
+	if eventErr != nil {
+		eventErrStr = eventErr.Error()
+	}
+	err := o.EventLogs.Vent(recordevents.EventInfo{
+		Error:       eventErrStr,
+		Event:       event,
+		HTTPHeaders: header,
+		Origin:      request.RemoteAddr,
+		Observer:    o.Name,
+		Time:        time.Now(),
+		Sequence:    atomic.AddUint64(&o.seq, 1),
+	})
+	if err != nil {
+		log.Warn("Error while venting the recorded event", err)
+	}
+
+	writer.WriteHeader(http.StatusAccepted)
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/constructor.go b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/constructor.go
new file mode 100644
index 000000000..cc73a9671
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/constructor.go
@@ -0,0 +1,108 @@
+/*
+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 recorder_vent
+
+import (
+	"context"
+	"log"
+	"strings"
+
+	"github.com/kelseyhightower/envconfig"
+	"knative.dev/pkg/system"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes/scheme"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
+
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+
+	"knative.dev/eventing/test/lib/recordevents"
+)
+
+type envConfig struct {
+	AgentName string `envconfig:"AGENT_NAME" default:"observer-default" required:"true"`
+	PodName   string `envconfig:"POD_NAME" required:"true"`
+	Port      int    `envconfig:"PORT" default:"8080" required:"true"`
+}
+
+func NewFromEnv(ctx context.Context) recordevents.EventLog {
+	var env envConfig
+	if err := envconfig.Process("", &env); err != nil {
+		log.Fatal("Failed to process env var", err)
+	}
+
+	logging.FromContext(ctx).Infof("Environment configuration: %+v", env)
+
+	return NewEventLog(ctx, env.AgentName, env.PodName)
+}
+
+func NewEventLog(ctx context.Context, agentName string, podName string) recordevents.EventLog {
+	on, err := kubeclient.Get(ctx).CoreV1().Pods(system.Namespace()).Get(ctx, podName, metav1.GetOptions{})
+	if err != nil {
+		logging.FromContext(ctx).Fatal("Error while trying to retrieve the pod", err)
+	}
+
+	logging.FromContext(ctx).Infof("Going to send events to pod '%s' in namespace '%s'", on.Name, on.Namespace)
+
+	return &recorder{out: createRecorder(ctx, agentName), on: on}
+}
+
+func createRecorder(ctx context.Context, agentName string) record.EventRecorder {
+	logger := logging.FromContext(ctx)
+
+	recorder := controller.GetEventRecorder(ctx)
+	if recorder == nil {
+		// Create event broadcaster
+		logger.Debug("Creating event broadcaster")
+		eventBroadcaster := record.NewBroadcasterWithCorrelatorOptions(record.CorrelatorOptions{
+			KeyFunc: func(event *corev1.Event) (aggregateKey string, localKey string) {
+				return strings.Join([]string{
+					event.Source.Component,
+					event.Source.Host,
+					event.InvolvedObject.Kind,
+					event.InvolvedObject.Namespace,
+					event.InvolvedObject.Name,
+					string(event.InvolvedObject.UID),
+					event.InvolvedObject.APIVersion,
+					event.Type,
+					event.Reason,
+				}, ""), string(event.UID)
+			},
+		})
+		watches := []watch.Interface{
+			eventBroadcaster.StartLogging(logger.Named("event-broadcaster").Infof),
+			eventBroadcaster.StartRecordingToSink(
+				&v1.EventSinkImpl{Interface: kubeclient.Get(ctx).CoreV1().Events(system.Namespace())},
+			),
+		}
+		recorder = eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: agentName})
+		go func() {
+			<-ctx.Done()
+			for _, w := range watches {
+				w.Stop()
+			}
+			logging.FromContext(ctx).Debug("Closed event-broadcaster")
+		}()
+	}
+
+	return recorder
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/doc.go b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/doc.go
new file mode 100644
index 000000000..1a4bcc290
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/doc.go
@@ -0,0 +1,19 @@
+/*
+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 recorder_vent implements an recordevents.EventLog backed by Kubernetes
+// Events using an event recorder.
+package recorder_vent
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/recorder.go b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/recorder.go
new file mode 100644
index 000000000..9669741f9
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/recorder.go
@@ -0,0 +1,43 @@
+/*
+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 recorder_vent
+
+import (
+	"encoding/json"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/record"
+
+	"knative.dev/eventing/test/lib/recordevents"
+)
+
+type recorder struct {
+	out record.EventRecorder
+	on  runtime.Object
+}
+
+func (r *recorder) Vent(observed recordevents.EventInfo) error {
+	b, err := json.Marshal(observed)
+	if err != nil {
+		return err
+	}
+
+	r.out.Eventf(r.on, corev1.EventTypeNormal, recordevents.CloudEventObservedReason, "%s", string(b))
+
+	return nil
+}
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/resources.go b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
new file mode 100644
index 000000000..54726bedf
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
@@ -0,0 +1,61 @@
+/*
+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 recordevents
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"knative.dev/pkg/test"
+)
+
+// EventRecordPod creates a Pod that stores received events for test retrieval.
+func EventRecordPod(name string, serviceAccountName string) *corev1.Pod {
+	return recordEventsPod("recordevents", name, serviceAccountName)
+}
+
+func recordEventsPod(imageName string, name string, serviceAccountName string) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   name,
+			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{
+				Name:            imageName,
+				Image:           test.ImagePath(imageName),
+				ImagePullPolicy: corev1.PullAlways,
+				Env: []corev1.EnvVar{{
+					Name: "SYSTEM_NAMESPACE",
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.namespace"},
+					},
+				}, {
+					Name:  "OBSERVER",
+					Value: "recorder-" + name,
+				}, {
+					Name: "POD_NAME",
+					ValueFrom: &corev1.EnvVarSource{
+						FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"},
+					},
+				}},
+			}},
+			ServiceAccountName: serviceAccountName,
+			RestartPolicy:      corev1.RestartPolicyAlways,
+		},
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/lib/resources/kube.go b/vendor/knative.dev/eventing/test/lib/resources/kube.go
index 2b154fe2b..bc415d108 100644
--- a/vendor/knative.dev/eventing/test/lib/resources/kube.go
+++ b/vendor/knative.dev/eventing/test/lib/resources/kube.go
@@ -37,28 +37,6 @@ type PodOption func(*corev1.Pod)
 // Option enables further configuration of a Role.
 type RoleOption func(*rbacv1.Role)
 
-// EventRecordPod creates a Pod that stores received events for test retrieval.
-func EventRecordPod(name string) *corev1.Pod {
-	return eventLoggerPod("recordevents", name)
-}
-
-func eventLoggerPod(imageName string, name string) *corev1.Pod {
-	return &corev1.Pod{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:   name,
-			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
-		},
-		Spec: corev1.PodSpec{
-			Containers: []corev1.Container{{
-				Name:            imageName,
-				Image:           pkgTest.ImagePath(imageName),
-				ImagePullPolicy: corev1.PullAlways,
-			}},
-			RestartPolicy: corev1.RestartPolicyAlways,
-		},
-	}
-}
-
 // EventTransformationPod creates a Pod that transforms events received receiving as arg a cloudevents sdk2 Event
 func EventTransformationPod(name string, newEventType string, newEventSource string, newEventData []byte) *corev1.Pod {
 	const imageName = "transformevents"
diff --git a/vendor/knative.dev/eventing/test/performance/infra/sender/http_load_generator.go b/vendor/knative.dev/eventing/test/performance/infra/sender/http_load_generator.go
index c45999c68..7b68ed0cc 100644
--- a/vendor/knative.dev/eventing/test/performance/infra/sender/http_load_generator.go
+++ b/vendor/knative.dev/eventing/test/performance/infra/sender/http_load_generator.go
@@ -107,7 +107,7 @@ func (cet CloudEventsTargeter) VegetaTargeter() vegeta.Targeter {
 	}
 }
 
-type HttpLoadGenerator struct {
+type httpLoadGenerator struct {
 	eventSource string
 	sinkUrl     string
 
@@ -119,13 +119,13 @@ type HttpLoadGenerator struct {
 	ceClient       cloudevents.Client
 }
 
-func NewHttpLoadGeneratorFactory(sinkUrl string, minWorkers uint64) LoadGeneratorFactory {
+func NewHTTPLoadGeneratorFactory(sinkUrl string, minWorkers uint64) LoadGeneratorFactory {
 	return func(eventSource string, sentCh chan common.EventTimestamp, acceptedCh chan common.EventTimestamp) (generator LoadGenerator, e error) {
 		if sinkUrl == "" {
 			panic("Missing --sink flag")
 		}
 
-		loadGen := &HttpLoadGenerator{
+		loadGen := &httpLoadGenerator{
 			eventSource: eventSource,
 			sinkUrl:     sinkUrl,
 
@@ -190,27 +190,27 @@ func newCloudEventsClient(sinkUrl string) (cloudevents.Client, error) {
 		cloudevents.WithTarget(sinkUrl),
 	)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create transport: %v", err)
+		return nil, fmt.Errorf("failed to create transport: %w", err)
 	}
 
 	return cloudevents.NewClient(t)
 }
 
-func (h HttpLoadGenerator) Warmup(pace common.PaceSpec, msgSize uint, fixedBody bool) {
+func (h httpLoadGenerator) Warmup(pace common.PaceSpec, msgSize uint, fixedBody bool) {
 	targeter := NewCloudEventsTargeter(h.sinkUrl, msgSize, common.WarmupEventType, defaultEventSource, fixedBody).VegetaTargeter()
 	vegetaResults := h.warmupAttacker.Attack(targeter, vegeta.ConstantPacer{Freq: pace.Rps, Per: time.Second}, pace.Duration, common.WarmupEventType+"-attack")
 	for range vegetaResults {
 	}
 }
 
-func (h HttpLoadGenerator) RunPace(i int, pace common.PaceSpec, msgSize uint, fixedBody bool) {
+func (h httpLoadGenerator) RunPace(i int, pace common.PaceSpec, msgSize uint, fixedBody bool) {
 	targeter := NewCloudEventsTargeter(h.sinkUrl, msgSize, common.MeasureEventType, eventsSource(), fixedBody).VegetaTargeter()
 	res := h.paceAttacker.Attack(targeter, vegeta.ConstantPacer{Freq: pace.Rps, Per: time.Second}, pace.Duration, fmt.Sprintf("%s-attack-%d", h.eventSource, i))
 	for range res {
 	}
 }
 
-func (h HttpLoadGenerator) SendGCEvent() {
+func (h httpLoadGenerator) SendGCEvent() {
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetID(uuid.New().String())
 	event.SetDataContentType(cloudevents.ApplicationJSON)
@@ -220,7 +220,7 @@ func (h HttpLoadGenerator) SendGCEvent() {
 	_ = h.ceClient.Send(context.TODO(), event)
 }
 
-func (h HttpLoadGenerator) SendEndEvent() {
+func (h httpLoadGenerator) SendEndEvent() {
 	event := cloudevents.NewEvent(cloudevents.VersionV1)
 	event.SetID(uuid.New().String())
 	event.SetDataContentType(cloudevents.ApplicationJSON)
diff --git a/vendor/knative.dev/eventing/test/test_images/performance/main.go b/vendor/knative.dev/eventing/test/test_images/performance/main.go
index b1f1312a6..8ef09d27b 100644
--- a/vendor/knative.dev/eventing/test/test_images/performance/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/performance/main.go
@@ -38,5 +38,5 @@ func init() {
 func main() {
 	flag.Parse()
 
-	infra.StartPerformanceImage(sender.NewHttpLoadGeneratorFactory(sinkURL, minWorkers), receiver.EventTypeExtractor, receiver.EventIdExtractor)
+	infra.StartPerformanceImage(sender.NewHTTPLoadGeneratorFactory(sinkURL, minWorkers), receiver.EventTypeExtractor, receiver.EventIdExtractor)
 }
diff --git a/vendor/knative.dev/eventing/test/test_images/recordevents/eventstore.go b/vendor/knative.dev/eventing/test/test_images/recordevents/eventstore.go
deleted file mode 100644
index 470e5adc2..000000000
--- a/vendor/knative.dev/eventing/test/test_images/recordevents/eventstore.go
+++ /dev/null
@@ -1,191 +0,0 @@
-/*
-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
-
-	https://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"
-	"fmt"
-	"sync"
-
-	cloudevents "github.com/cloudevents/sdk-go/v2"
-
-	"knative.dev/eventing/test/lib/recordevents"
-)
-
-// Number of EventInfo per block
-const evBlockSize = 100
-
-// Block of stored EventInfo
-type eventBlock struct {
-	// seqno of [0] evInfoBytes entry
-	firstIndex int
-	// offset inside block for newly appended event
-	firstOffsetFree int
-	// offset inside block of first non-trimmed event
-	firstValid int
-	// serialized EventInfo structures for each seqno.  We enforce
-	// that there is always at least one block.
-	evInfoBytes [evBlockSize][]byte
-}
-
-// All events currently seen
-type eventStore struct {
-	// Blocks of events in increasing sequency number order
-	evBlocks     []*eventBlock
-	evBlocksLock sync.Mutex
-}
-
-// Create a new event store.
-func newEventStore() *eventStore {
-	es := &eventStore{}
-	es.evBlocks = []*eventBlock{{}}
-
-	// One block with no entries starting at sequence number 1
-	es.evBlocks[0].firstIndex = 1
-	es.evBlocks[0].firstOffsetFree = 0
-	es.evBlocks[0].firstValid = 0
-	return es
-}
-
-// See if there's enough room to append to the current last block.  If not,
-// append an extra block.
-func (es *eventStore) checkAppendBlock() {
-	if es.evBlocks[len(es.evBlocks)-1].firstOffsetFree == evBlockSize {
-		newEVBlock := &eventBlock{
-			firstIndex: es.evBlocks[len(es.evBlocks)-1].firstIndex + evBlockSize,
-		}
-		es.evBlocks = append(es.evBlocks, newEVBlock)
-	}
-}
-
-// Store the specified event.
-func (es *eventStore) StoreEvent(event *cloudevents.Event, evErr error, httpHeaders map[string][]string) {
-	var evInfo recordevents.EventInfo
-	var err error
-	var evInfoBytes []byte
-	if evErr != nil {
-		evInfo.HTTPHeaders = httpHeaders
-		evInfo.Error = evErr.Error()
-		if evInfo.Error == "" {
-			evInfo.Error = "Unknown Incoming Error"
-		}
-		evInfoBytes, err = json.Marshal(&evInfo)
-		if err != nil {
-			panic(fmt.Errorf("unexpected marshal error (%v) (%+v)", err, evInfo))
-		}
-	} else {
-		evInfo.Event = event
-		evInfo.HTTPHeaders = httpHeaders
-		evInfoBytes, err = json.Marshal(&evInfo)
-
-		if err != nil {
-			evInfo.Event = nil
-			evInfo.Error = err.Error()
-			if evInfo.Error == "" {
-				evInfo.Error = "Unknown Error"
-			}
-			evInfoBytes, err = json.Marshal(&evInfo)
-			if err != nil {
-				panic(fmt.Errorf("unexpected marshal error (%v) (%+v)", err, evInfo))
-			}
-		}
-	}
-
-	es.evBlocksLock.Lock()
-	// Add a new block if we're out of space
-	es.checkAppendBlock()
-
-	evBlock := es.evBlocks[len(es.evBlocks)-1]
-	if evBlock.firstOffsetFree < evBlockSize {
-		evBlock.evInfoBytes[evBlock.firstOffsetFree] = evInfoBytes
-		evBlock.firstOffsetFree++
-	}
-
-	es.evBlocksLock.Unlock()
-}
-
-// Logically trim all events up to and include the provided
-// sequence number.  Returns error for patently incorrect
-// values (negative sequence number or sequence number larger
-// than the largest event seen).  Trimming already trimmed
-// regions is legal.
-func (es *eventStore) TrimThrough(through int) error {
-	es.evBlocksLock.Lock()
-	defer es.evBlocksLock.Unlock()
-	minAvail, maxSeen := es.minMaxUnlocked()
-	if through > maxSeen {
-		return fmt.Errorf("invalid trim through %d, maxSeen %d", through, maxSeen)
-	} else if through < 0 {
-		return fmt.Errorf("invalid trim less than zero %d", through)
-	} else if through < minAvail {
-		return nil
-	}
-	// Completely remove blocks if they are full and all events in them are less than
-	// the specified value.
-	for len(es.evBlocks) > 1 && (es.evBlocks[0].firstIndex+evBlockSize-1) <= through {
-		es.evBlocks = es.evBlocks[1:]
-	}
-	// Logically trim the block split by through.
-	es.evBlocks[0].firstValid = through - es.evBlocks[0].firstIndex + 1
-	return nil
-}
-
-// return min/max untrimmed value when already holding the lock
-func (es *eventStore) minMaxUnlocked() (minAvail int, maxSeen int) {
-	minBlock := es.evBlocks[0]
-	minAvail = minBlock.firstIndex + (minBlock.firstValid)
-
-	maxBlock := es.evBlocks[len(es.evBlocks)-1]
-	maxSeen = maxBlock.firstIndex + maxBlock.firstOffsetFree - 1
-	return minAvail, maxSeen
-}
-
-// Returns min available value and max seen value for the store.
-// min is the minimum value that can be retrieved via GetEntry, or
-// if no values can be retrieved, min == max+1.  Max starts at 0
-// when no events have been seen.
-func (es *eventStore) MinMax() (minAvail int, maxSeen int) {
-	es.evBlocksLock.Lock()
-	minAvail, maxSeen = es.minMaxUnlocked()
-
-	es.evBlocksLock.Unlock()
-	return minAvail, maxSeen
-}
-
-// Get the already serialized EventInfo structure for the provided sequence
-// number.
-func (es *eventStore) GetEventInfoBytes(seq int) ([]byte, error) {
-	var evInfoBytes []byte
-	found := false
-
-	es.evBlocksLock.Lock()
-	for _, block := range es.evBlocks {
-		if seq < block.firstIndex+block.firstValid {
-			break
-		}
-		if seq < block.firstIndex+block.firstOffsetFree {
-			found = true
-			evInfoBytes = block.evInfoBytes[seq-block.firstIndex]
-			break
-		}
-	}
-	es.evBlocksLock.Unlock()
-	if !found {
-		return evInfoBytes, fmt.Errorf("Invalid sequence number %d", seq)
-	}
-	return evInfoBytes, nil
-}
diff --git a/vendor/knative.dev/eventing/test/test_images/recordevents/main.go b/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
index 8dd741163..51fde078e 100644
--- a/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
@@ -17,164 +17,60 @@ limitations under the License.
 package main
 
 import (
-	"context"
-	"encoding/json"
-	"fmt"
 	"log"
 	"net/http"
 	"os"
-	"strconv"
-	"strings"
 
-	cloudeventsbindings "github.com/cloudevents/sdk-go/v2/binding"
-	cloudeventshttp "github.com/cloudevents/sdk-go/v2/protocol/http"
-	"go.uber.org/zap"
+	"k8s.io/client-go/rest"
+	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/pkg/logging"
+	_ "knative.dev/pkg/system/testing"
 
 	"knative.dev/eventing/pkg/kncloudevents"
-	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/dropevents"
-	"knative.dev/eventing/test/lib/recordevents"
+	"knative.dev/eventing/test/lib/recordevents/observer"
+	"knative.dev/eventing/test/lib/recordevents/recorder_vent"
 	"knative.dev/eventing/test/test_images"
 )
 
-type eventRecorder struct {
-	es *eventStore
-}
-
-func newEventRecorder() *eventRecorder {
-	return &eventRecorder{es: newEventStore()}
-}
-
-// Start the recordevents REST api server
-func (er *eventRecorder) StartServer(port int) {
-	http.HandleFunc(recordevents.GetMinMaxPath, er.handleMinMax)
-	http.HandleFunc(recordevents.GetEntryPath, er.handleGetEntry)
-	http.HandleFunc(recordevents.TrimThroughPath, er.handleTrim)
-	go http.ListenAndServe(fmt.Sprintf(":%d", port), nil)
-}
-
-// HTTP handler for GetMinMax requests
-func (er *eventRecorder) handleMinMax(w http.ResponseWriter, r *http.Request) {
-	minAvail, maxSeen := er.es.MinMax()
-	minMax := recordevents.MinMaxResponse{
-		MinAvail: minAvail,
-		MaxSeen:  maxSeen,
-	}
-	respBytes, err := json.Marshal(minMax)
-	if err != nil {
-		log.Panicf("Internal error: json marshal shouldn't fail: (%v) (%+v)", err, minMax)
-	}
-
-	w.Header().Set("Content-Type", "text/json")
-	w.WriteHeader(http.StatusOK)
-	w.Write(respBytes)
-}
-
-// HTTP handler for TrimThrough requests
-func (er *eventRecorder) handleTrim(w http.ResponseWriter, r *http.Request) {
-	// If we extend this much at all we should vendor a better mux(gorilla, etc)
-	path := strings.TrimLeft(r.URL.Path, "/")
-	getPrefix := strings.TrimLeft(recordevents.TrimThroughPath, "/")
-	suffix := strings.TrimLeft(strings.TrimPrefix(path, getPrefix), "/")
-
-	seqNum, err := strconv.ParseInt(suffix, 10, 32)
-	if err != nil {
-		http.Error(w, "Can't parse event sequence number in request", http.StatusBadRequest)
-		return
-	}
-
-	err = er.es.TrimThrough(int(seqNum))
-	if err != nil {
-		http.Error(w, "Invalid sequence number in request to trim", http.StatusNotFound)
-		return
-	}
-
-	w.Header().Set("Content-Type", "text/json")
-	w.WriteHeader(http.StatusOK)
-}
-
-// HTTP handler for GetEntry requests
-func (er *eventRecorder) handleGetEntry(w http.ResponseWriter, r *http.Request) {
-	// If we extend this much at all we should vendor a better mux(gorilla, etc)
-	path := strings.TrimLeft(r.URL.Path, "/")
-	getPrefix := strings.TrimLeft(recordevents.GetEntryPath, "/")
-	suffix := strings.TrimLeft(strings.TrimPrefix(path, getPrefix), "/")
-
-	seqNum, err := strconv.ParseInt(suffix, 10, 32)
+func main() {
+	cfg, err := rest.InClusterConfig()
 	if err != nil {
-		http.Error(w, "Can't parse event sequence number in request", http.StatusBadRequest)
-		return
+		log.Fatal("Error while reading the cfg", err)
 	}
+	//nolint // nil ctx is fine here, look at the code of EnableInjectionOrDie
+	ctx := sharedmain.EnableInjectionOrDie(nil, cfg)
 
-	entryBytes, err := er.es.GetEventInfoBytes(int(seqNum))
-	if err != nil {
-		http.Error(w, "Couldn't find requested event", http.StatusNotFound)
-		return
+	if err := test_images.ConfigureTracing(logging.FromContext(ctx), ""); err != nil {
+		logging.FromContext(ctx).Fatal("Unable to setup trace publishing", err)
 	}
 
-	w.Header().Set("Content-Type", "text/json")
-	w.WriteHeader(http.StatusOK)
-	w.Write(entryBytes)
-}
-
-func (er *eventRecorder) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
-	m := cloudeventshttp.NewMessageFromHttpRequest(request)
-	defer m.Finish(nil)
-
-	event, eventErr := cloudeventsbindings.ToEvent(context.TODO(), m)
-	header := request.Header
-
-	er.es.StoreEvent(event, eventErr, map[string][]string(header))
-
-	headerNameList := testlib.InterestingHeaders()
-	for _, headerName := range headerNameList {
-		if headerValue := header.Get(headerName); headerValue != "" {
-			log.Printf("Header %s: %s\n", headerName, headerValue)
-		}
-	}
-
-	if eventErr != nil {
-		log.Printf("error receiving the event: %v", eventErr)
-	} else {
-		valErr := event.Validate()
-		if valErr == nil {
-			log.Printf("eventdetails:\n%s", event.String())
-		} else {
-			log.Printf("error validating the event: %v", valErr)
-		}
-	}
-
-	writer.WriteHeader(http.StatusAccepted)
-}
-
-func main() {
-	er := newEventRecorder()
-	er.StartServer(recordevents.RecordEventsPort)
-
-	logger, _ := zap.NewDevelopment()
-	if err := test_images.ConfigureTracing(logger.Sugar(), ""); err != nil {
-		log.Fatalf("Unable to setup trace publishing: %v", err)
-	}
+	obs := observer.NewFromEnv(
+		recorder_vent.NewFromEnv(ctx),
+	)
 
 	algorithm, ok := os.LookupEnv(dropevents.SkipAlgorithmKey)
-	handler := kncloudevents.CreateHandler(er)
 	if ok {
 		skipper := dropevents.SkipperAlgorithm(algorithm)
 		counter := dropevents.CounterHandler{
 			Skipper: skipper,
 		}
-		next := handler
-		handler = http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
-			if counter.Skip() {
-				writer.WriteHeader(http.StatusConflict)
-				return
-			}
-			next.ServeHTTP(writer, request)
+		err = obs.Start(ctx, kncloudevents.CreateHandler, func(handler http.Handler) http.Handler {
+			return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+				if counter.Skip() {
+					writer.WriteHeader(http.StatusConflict)
+					return
+				}
+				handler.ServeHTTP(writer, request)
+			})
 		})
+	} else {
+		err = obs.Start(ctx, kncloudevents.CreateHandler)
 	}
 
-	err := http.ListenAndServe(":8080", handler)
 	if err != nil {
-		panic(err)
+		logging.FromContext(ctx).Fatal("Error during start", err)
 	}
+
+	logging.FromContext(ctx).Info("Closing the recordevents process")
 }
diff --git a/vendor/knative.dev/pkg/test/helpers/dryrun.go b/vendor/knative.dev/pkg/test/helpers/dryrun.go
index 68f799444..c19881556 100644
--- a/vendor/knative.dev/pkg/test/helpers/dryrun.go
+++ b/vendor/knative.dev/pkg/test/helpers/dryrun.go
@@ -23,7 +23,7 @@ import (
 // Run can run functions that needs dryrun support.
 func Run(message string, call func() error, dryrun bool) error {
 	if dryrun {
-		log.Printf("[dry run] %s", message)
+		log.Print("[dry run] ", message)
 		return nil
 	}
 	log.Print(message)
diff --git a/vendor/knative.dev/pkg/test/helpers/name.go b/vendor/knative.dev/pkg/test/helpers/name.go
index 9cb0d529f..ef5ac4dbd 100644
--- a/vendor/knative.dev/pkg/test/helpers/name.go
+++ b/vendor/knative.dev/pkg/test/helpers/name.go
@@ -71,8 +71,8 @@ func RandomString() string {
 }
 
 // For the same prefix more specific should come first.
-// Note: we expect GRPC.
-var knownNames = []string{"HTTPS", "HTTP2", "HTTP", "GRPC", "TLS", "WS", "H2C"}
+// Note: we expect GRPC vs gRPC.
+var knownNames = []string{"GRPC", "H2C", "HTTPS", "HTTP2", "HTTP", "REST", "TLS", "WS"}
 
 // MakeK8sNamePrefix converts each chunk of non-alphanumeric character into a single dash
 // and also convert camelcase tokens into dash-delimited lowercase tokens.
diff --git a/vendor/knative.dev/pkg/test/kube_checks.go b/vendor/knative.dev/pkg/test/kube_checks.go
index a406f211a..cf804a9b1 100644
--- a/vendor/knative.dev/pkg/test/kube_checks.go
+++ b/vendor/knative.dev/pkg/test/kube_checks.go
@@ -65,7 +65,7 @@ func WaitForDeploymentState(ctx context.Context, client *KubeClient, name string
 // that is emitted to track how long it took to get into the state checked by inState.
 func WaitForPodListState(ctx context.Context, client *KubeClient, inState func(p *corev1.PodList) (bool, error), desc string, namespace string) error {
 	p := client.Kube.CoreV1().Pods(namespace)
-	span := logging.GetEmitableSpan(ctx, fmt.Sprintf("WaitForPodListState/%s", desc))
+	span := logging.GetEmitableSpan(ctx, "WaitForPodListState/"+desc)
 	defer span.End()
 
 	return wait.PollImmediate(interval, podTimeout, func() (bool, error) {
diff --git a/vendor/knative.dev/pkg/test/mako/config/benchmark.go b/vendor/knative.dev/pkg/test/mako/config/benchmark.go
index 4a1f2e0bc..41d046865 100644
--- a/vendor/knative.dev/pkg/test/mako/config/benchmark.go
+++ b/vendor/knative.dev/pkg/test/mako/config/benchmark.go
@@ -34,7 +34,7 @@ const configMako = "/etc/config-mako"
 func MustGetBenchmark() *mpb.BenchmarkInfo {
 	bench, err := getBenchmark()
 	if err != nil {
-		log.Fatalf("unable to determine benchmark info: %v", err)
+		log.Fatal("unable to determine benchmark info: ", err)
 	}
 	return bench
 }
diff --git a/vendor/knative.dev/pkg/test/mako/config/environment.go b/vendor/knative.dev/pkg/test/mako/config/environment.go
index 62e72c243..464d6030a 100644
--- a/vendor/knative.dev/pkg/test/mako/config/environment.go
+++ b/vendor/knative.dev/pkg/test/mako/config/environment.go
@@ -51,7 +51,7 @@ func GetRepository() string {
 func MustGetTags() []string {
 	cfg, err := loadConfig()
 	if err != nil {
-		log.Fatalf("unable to load config from the configmap: %v", err)
+		log.Fatal("unable to load config from the configmap: ", err)
 	}
 	return cfg.AdditionalTags
 }
diff --git a/vendor/knative.dev/pkg/test/mako/sidecar.go b/vendor/knative.dev/pkg/test/mako/sidecar.go
index 92377143a..cf801977d 100644
--- a/vendor/knative.dev/pkg/test/mako/sidecar.go
+++ b/vendor/knative.dev/pkg/test/mako/sidecar.go
@@ -123,17 +123,17 @@ func SetupHelper(ctx context.Context, benchmarkKey *string, benchmarkName *strin
 
 	// Decorate GCP metadata as tags (when we're running on GCP).
 	if projectID, err := metadata.ProjectID(); err != nil {
-		log.Printf("GCP project ID is not available: %v", err)
+		log.Print("GCP project ID is not available: ", err)
 	} else {
 		tags = append(tags, "project-id="+EscapeTag(projectID))
 	}
 	if zone, err := metadata.Zone(); err != nil {
-		log.Printf("GCP zone is not available: %v", err)
+		log.Print("GCP zone is not available: ", err)
 	} else {
 		tags = append(tags, "zone="+EscapeTag(zone))
 	}
 	if machineType, err := metadata.Get("instance/machine-type"); err != nil {
-		log.Printf("GCP machine type is not available: %v", err)
+		log.Print("GCP machine type is not available: ", err)
 	} else if parts := strings.Split(machineType, "/"); len(parts) != 4 {
 		tags = append(tags, "instanceType="+EscapeTag(parts[3]))
 	}