Repository: knative-sandbox/eventing-kafka
Pull request #189 (2020-11-12 21:48:23)

Allow changing resync period via environment variable
## Proposed Changes

- 🎁 Added a new optional RESYNC_PERIOD_MINUTES environment variable to the distributed controller (which is propagated to the receiver and dispatcher deployments) which allows convenient setting of a different resync period than the default of 10 hours.


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/189#" title="Author self-approved">eric-sap</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **lberk** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @lberk` in a comment when ready.

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["lberk"]} -->
knative-prow-robot: Hi @eric-sap. Thanks for your PR.

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

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

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

<details>

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

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189?src=pr&el=h1) Report
> Merging [#189](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189?src=pr&el=desc) (ffb9741) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/d11800b4c38b14b6ceb62517e189a4a6b9891c97?el=desc) (d11800b) will **increase** coverage by `0.08%`.
> The diff coverage is `82.00%`.

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

```diff
@@            Coverage Diff             @@
##           master     #189      +/-   ##
==========================================
+ Coverage   76.00%   76.08%   +0.08%     
==========================================
  Files         112      112              
  Lines        4163     4203      +40     
==========================================
+ Hits         3164     3198      +34     
- Misses        810      816       +6     
  Partials      189      189              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `49.68% <0.00%> (ø)` | |
| [pkg/channel/distributed/common/env/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2Vudi91dGlsLmdv) | `92.30% <0.00%> (-7.70%)` | :arrow_down: |
| [.../channel/distributed/controller/env/environment.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9lbnYvZW52aXJvbm1lbnQuZ28=) | `93.75% <84.61%> (-6.25%)` | :arrow_down: |
| [.../distributed/controller/kafkachannel/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvY29udHJvbGxlci5nbw==) | `82.25% <100.00%> (+2.57%)` | :arrow_up: |
| [.../distributed/controller/kafkachannel/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvZGlzcGF0Y2hlci5nbw==) | `90.09% <100.00%> (+0.09%)` | :arrow_up: |
| [...nel/distributed/controller/kafkasecret/receiver.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYXNlY3JldC9yZWNlaXZlci5nbw==) | `93.33% <100.00%> (+0.07%)` | :arrow_up: |
| [...troller/kafkasecretinformer/kafkasecretinformer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYXNlY3JldGluZm9ybWVyL2thZmthc2VjcmV0aW5mb3JtZXIuZ28=) | `68.42% <100.00%> (+1.75%)` | :arrow_up: |
| [.../channel/distributed/dispatcher/env/environment.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9lbnYvZW52aXJvbm1lbnQuZ28=) | `100.00% <100.00%> (ø)` | |
| [...kg/channel/distributed/receiver/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvcmVjZWl2ZXIvY2hhbm5lbC9jaGFubmVsLmdv) | `60.00% <100.00%> (ø)` | |
| [...kg/channel/distributed/receiver/env/environment.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvcmVjZWl2ZXIvZW52L2Vudmlyb25tZW50Lmdv) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189?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/pull/189?src=pr&el=footer). Last update [d11800b...2974196](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/189?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/common/env/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327012457469710336/artifacts/line-cov.html#file1) | 100.0% | 92.2% | -7.8
[pkg/channel/distributed/controller/env/environment.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327012457469710336/artifacts/line-cov.html#file2) | 100.0% | 96.4% | -3.6
[pkg/channel/distributed/controller/kafkachannel/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327012457469710336/artifacts/line-cov.html#file3) | 81.6% | 83.3% | 1.8

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/common/env/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327016657322774528/artifacts/line-cov.html#file1) | 100.0% | 92.2% | -7.8
[pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327016657322774528/artifacts/line-cov.html#file7) | 76.9% | 76.5% | -0.5

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/common/env/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327040677418110976/artifacts/line-cov.html#file1) | 100.0% | 92.2% | -7.8
[pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327040677418110976/artifacts/line-cov.html#file7) | 76.9% | 76.5% | -0.5

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/common/env/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327042217897889792/artifacts/line-cov.html#file1) | 100.0% | 92.2% | -7.8
[pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/189/pull-knative-sandbox-eventing-kafka-go-coverage/1327042217897889792/artifacts/line-cov.html#file7) | 76.9% | 76.5% | -0.5


Patch:
diff --git a/cmd/channel/distributed/controller/main.go b/cmd/channel/distributed/controller/main.go
index 3c5e2ec4b..e8c386cad 100644
--- a/cmd/channel/distributed/controller/main.go
+++ b/cmd/channel/distributed/controller/main.go
@@ -20,10 +20,15 @@ import (
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	// _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
+	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/env"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/kafkachannel"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/kafkasecret"
+	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/signals"
 )
 
 // Eventing-Kafka Controller Main
@@ -34,5 +39,12 @@ func main() {
 	defer kafkasecret.Shutdown()
 
 	// Create The SharedMain Instance With The Various Controllers
-	sharedmain.Main(constants.ControllerComponentName, kafkachannel.NewController, kafkasecret.NewController)
+	ctx := signals.NewContext()
+	logger := logging.FromContext(ctx).Desugar()
+	environment, err := env.GetEnvironment(logger)
+	if err != nil {
+		logger.Fatal("Failed To Load Environment Variables - Terminating!", zap.Error(err))
+	}
+	ctx = controller.WithResyncPeriod(ctx, environment.ResyncPeriod)
+	sharedmain.MainWithContext(ctx, constants.ControllerComponentName, kafkachannel.NewController, kafkasecret.NewController)
 }
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index e0662f885..e25e80ac1 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -130,7 +130,8 @@ func main() {
 	config.Burst = numControllers * rest.DefaultBurst
 	kafkaClientSet := versioned.NewForConfigOrDie(config)
 	kubeClient := kubernetes.NewForConfigOrDie(config)
-	kafkaInformerFactory := externalversions.NewSharedInformerFactory(kafkaClientSet, kncontroller.DefaultResyncPeriod)
+
+	kafkaInformerFactory := externalversions.NewSharedInformerFactory(kafkaClientSet, environment.ResyncPeriod)
 
 	// Create KafkaChannel Informer
 	kafkaChannelInformer := kafkaInformerFactory.Messaging().V1beta1().KafkaChannels()
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 3fea6929a..5af0cddf7 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -102,7 +102,7 @@ func main() {
 	healthServer.Start(logger)
 
 	// Initialize The KafkaChannel Lister Used To Validate Events
-	err = channel.InitializeKafkaChannelLister(ctx, *serverURL, *kubeconfig, healthServer)
+	err = channel.InitializeKafkaChannelLister(ctx, *serverURL, *kubeconfig, healthServer, environment.ResyncPeriod)
 	if err != nil {
 		logger.Fatal("Failed To Initialize KafkaChannel Lister", zap.Error(err))
 	}
diff --git a/go.mod b/go.mod
index 9b0a88f39..ebb49c0dd 100644
--- a/go.mod
+++ b/go.mod
@@ -31,9 +31,9 @@ require (
 	k8s.io/apimachinery v0.18.8
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
-	knative.dev/eventing v0.19.0
-	knative.dev/hack v0.0.0-20201103151104-3d5abc3a0075
-	knative.dev/pkg v0.0.0-20201103163404-5514ab0c1fdf
+	knative.dev/eventing v0.19.1-0.20201113093000-1616938f9f1c
+	knative.dev/hack v0.0.0-20201112185459-01a34c573bd8
+	knative.dev/pkg v0.0.0-20201112201059-93fedf141385
 )
 
 replace (
diff --git a/go.sum b/go.sum
index b4117ef6c..7a295fe19 100644
--- a/go.sum
+++ b/go.sum
@@ -1220,12 +1220,14 @@ k8s.io/kube-openapi v0.0.0-20200410145947-bcb3869e6f29/go.mod h1:F+5wygcW0wmRTnM
 k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7MpmSnvtrOieolJKoE=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.19.0 h1:9TdB1E4mwGYnuAJxSbNoljVsZRM+HUKEgsWiBXUIZEc=
-knative.dev/eventing v0.19.0/go.mod h1:jwhDgDvoscWE4jWF8cXh7yHfxJcK0mTawVKVfrSjnvg=
+knative.dev/eventing v0.19.1-0.20201113093000-1616938f9f1c h1:9fz1Hudw8xkrGOLpjzPFLhpEtavUH8ocpzwJESkZFZw=
+knative.dev/eventing v0.19.1-0.20201113093000-1616938f9f1c/go.mod h1:I5ysDZ5GyEut5p+jVHfuBKk33agLfqKJaF4kQpwvS2M=
 knative.dev/hack v0.0.0-20201103151104-3d5abc3a0075 h1:YAgWplKIy4O5e3F5vUUECmXAAyZ0M5ymo6fCt1jeZhs=
 knative.dev/hack v0.0.0-20201103151104-3d5abc3a0075/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/pkg v0.0.0-20201103163404-5514ab0c1fdf h1:QwULgRwcv6R3Ya1GZlf/E1atcaGUNw4DKjxSQUfcR6U=
-knative.dev/pkg v0.0.0-20201103163404-5514ab0c1fdf/go.mod h1:cuKOgUvJvnWHIps/apCXX8wZuMlT0dyMZLqRQfsENbQ=
+knative.dev/hack v0.0.0-20201112185459-01a34c573bd8 h1:RNbZsAjhswBPtl4C5C5gEFX5/GfWIOZQxfYD9DhkHdY=
+knative.dev/hack v0.0.0-20201112185459-01a34c573bd8/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/pkg v0.0.0-20201112201059-93fedf141385 h1:DcYWVnDpcvKZR/Rh+8X2Qkq89jbBo6yLNV1qBifovBk=
+knative.dev/pkg v0.0.0-20201112201059-93fedf141385/go.mod h1:5vNHKNtZtzlNeNrcDoUtZIn+dma/8DjmpjfdHiwoQyM=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
index 9824cc640..237da189c 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -113,7 +113,7 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 	if err != nil {
 		return nil, err
 	}
-	containerName, err := env.GetRequiredConfigValue(args.Logger.Desugar(), env.ContainerNameEnvVarKEy)
+	containerName, err := env.GetRequiredConfigValue(args.Logger.Desugar(), env.ContainerNameEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/channel/distributed/common/env/constants.go b/pkg/channel/distributed/common/env/constants.go
index 17484837e..575d9baf3 100644
--- a/pkg/channel/distributed/common/env/constants.go
+++ b/pkg/channel/distributed/common/env/constants.go
@@ -20,12 +20,13 @@ package env
 const (
 
 	// Eventing-Kafka Configuration
-	ServiceAccountEnvVarKey = "SERVICE_ACCOUNT"
-	MetricsPortEnvVarKey    = "METRICS_PORT"
-	MetricsDomainEnvVarKey  = "METRICS_DOMAIN"
-	HealthPortEnvVarKey     = "HEALTH_PORT"
-	PodNameEnvVarKey        = "POD_NAME"
-	ContainerNameEnvVarKEy  = "CONTAINER_NAME"
+	ServiceAccountEnvVarKey      = "SERVICE_ACCOUNT"
+	MetricsPortEnvVarKey         = "METRICS_PORT"
+	MetricsDomainEnvVarKey       = "METRICS_DOMAIN"
+	HealthPortEnvVarKey          = "HEALTH_PORT"
+	PodNameEnvVarKey             = "POD_NAME"
+	ContainerNameEnvVarKey       = "CONTAINER_NAME"
+	ResyncPeriodMinutesEnvVarKey = "RESYNC_PERIOD_MINUTES"
 
 	// Kafka Authorization
 	KafkaBrokerEnvVarKey   = "KAFKA_BROKERS"
diff --git a/pkg/channel/distributed/common/env/util.go b/pkg/channel/distributed/common/env/util.go
index 855fb9e57..4b337118d 100644
--- a/pkg/channel/distributed/common/env/util.go
+++ b/pkg/channel/distributed/common/env/util.go
@@ -138,6 +138,17 @@ func GetOptionalConfigInt64(logger *zap.Logger, envKey string, defaultValue stri
 	return envInt, nil
 }
 
+// Get The Specified Optional Config Value From OS & Log Errors If Not Present Or Not An Int
+func GetOptionalConfigInt(logger *zap.Logger, envKey string, defaultValue string, name string) (int, error) {
+	envString := GetOptionalConfigValue(logger, envKey, defaultValue)
+	envInt, err := strconv.Atoi(envString)
+	if err != nil {
+		logger.Error("Invalid "+name+" (Non Int)", zap.String("Value", envString), zap.Error(err))
+		return 0, fmt.Errorf("invalid (non int) value '%s' for environment variable '%s'", envString, envKey)
+	}
+	return envInt, nil
+}
+
 // Parse Quantity Value
 func GetRequiredQuantityConfigValue(logger *zap.Logger, envVarKey string) (*resource.Quantity, error) {
 
diff --git a/pkg/channel/distributed/controller/env/environment.go b/pkg/channel/distributed/controller/env/environment.go
index 60a64e241..470dea79b 100644
--- a/pkg/channel/distributed/controller/env/environment.go
+++ b/pkg/channel/distributed/controller/env/environment.go
@@ -17,8 +17,12 @@ limitations under the License.
 package env
 
 import (
+	"strconv"
+	"time"
+
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	"knative.dev/pkg/controller"
 )
 
 // Package Constants
@@ -35,9 +39,10 @@ const (
 type Environment struct {
 
 	// Eventing-kafka Configuration
-	ServiceAccount string // Required
-	MetricsPort    int    // Required
-	MetricsDomain  string // Required
+	ServiceAccount string        // Required
+	MetricsPort    int           // Required
+	MetricsDomain  string        // Required
+	ResyncPeriod   time.Duration // Optional
 
 	// Dispatcher Configuration
 	DispatcherImage string // Required
@@ -73,6 +78,17 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 		return nil, err
 	}
 
+	// Get The Optional Resync Period config Value & Convert To Duration
+	resyncMinutes, err := env.GetOptionalConfigInt(
+		logger,
+		env.ResyncPeriodMinutesEnvVarKey,
+		strconv.Itoa(int(controller.DefaultResyncPeriod/time.Minute)),
+		"ResyncPeriodMinutes")
+	if err != nil {
+		return nil, err
+	}
+	environment.ResyncPeriod = time.Duration(resyncMinutes) * time.Minute
+
 	//
 	// Dispatcher Configuration
 	//
diff --git a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
index ab4c4e658..856300321 100644
--- a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
+++ b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"time"
 
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
@@ -484,7 +485,7 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1beta1.KafkaChan
 			},
 		},
 		{
-			Name:  commonenv.ContainerNameEnvVarKEy,
+			Name:  commonenv.ContainerNameEnvVarKey,
 			Value: constants.DispatcherContainerName,
 		},
 		{
@@ -515,6 +516,10 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1beta1.KafkaChan
 			Name:  commonenv.KafkaTopicEnvVarKey,
 			Value: topicName,
 		},
+		{
+			Name:  commonenv.ResyncPeriodMinutesEnvVarKey,
+			Value: strconv.Itoa(int(r.environment.ResyncPeriod / time.Minute)),
+		},
 	}
 
 	// Get The Kafka Secret From The Kafka Admin Client
diff --git a/pkg/channel/distributed/controller/kafkasecret/receiver.go b/pkg/channel/distributed/controller/kafkasecret/receiver.go
index 2dcf32250..65bfec161 100644
--- a/pkg/channel/distributed/controller/kafkasecret/receiver.go
+++ b/pkg/channel/distributed/controller/kafkasecret/receiver.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"time"
 
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
@@ -365,7 +366,7 @@ func (r *Reconciler) receiverDeploymentEnvVars(secret *corev1.Secret) ([]corev1.
 			},
 		},
 		{
-			Name:  commonenv.ContainerNameEnvVarKEy,
+			Name:  commonenv.ContainerNameEnvVarKey,
 			Value: constants.ReceiverContainerName,
 		},
 		{
@@ -388,6 +389,10 @@ func (r *Reconciler) receiverDeploymentEnvVars(secret *corev1.Secret) ([]corev1.
 			Name:  commonenv.HealthPortEnvVarKey,
 			Value: strconv.Itoa(constants.HealthPort),
 		},
+		{
+			Name:  commonenv.ResyncPeriodMinutesEnvVarKey,
+			Value: strconv.Itoa(int(r.environment.ResyncPeriod / time.Minute)),
+		},
 	}
 
 	// Append The Kafka Brokers As Env Var
diff --git a/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer.go b/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer.go
index 81bd9d37f..b7937fe7f 100644
--- a/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer.go
+++ b/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer.go
@@ -20,12 +20,15 @@ import (
 	"context"
 	"fmt"
 
+	"go.uber.org/zap"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/informers"
 	informerscorev1 "k8s.io/client-go/informers/core/v1"
 	listerscorev1 "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	commonconstants "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/env"
 	"knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection"
@@ -68,7 +71,13 @@ func withInformer(ctx context.Context) (context.Context, controller.Informer) {
 	}
 
 	// Create A SharedInformerFactory With The Namespaced / Labelled Options
-	namespacedSharedInformerFactory := informers.NewSharedInformerFactoryWithOptions(client.Get(ctx), controller.DefaultResyncPeriod, sharedInformerOptions...)
+	logger := logging.FromContext(ctx).Desugar()
+	environment, err := env.GetEnvironment(logger)
+	if err != nil {
+		logger.Fatal("Failed To Load Environment Variables - Terminating!", zap.Error(err))
+	}
+	namespacedSharedInformerFactory := informers.NewSharedInformerFactoryWithOptions(
+		client.Get(ctx), environment.ResyncPeriod, sharedInformerOptions...)
 
 	// Create The Custom Kafka SecretInformer
 	kafkaSecretInformer := KafkaSecretInformer{
diff --git a/pkg/channel/distributed/dispatcher/env/environment.go b/pkg/channel/distributed/dispatcher/env/environment.go
index 447fd5464..8b295d3fc 100644
--- a/pkg/channel/distributed/dispatcher/env/environment.go
+++ b/pkg/channel/distributed/dispatcher/env/environment.go
@@ -17,8 +17,12 @@ limitations under the License.
 package env
 
 import (
+	"strconv"
+	"time"
+
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	"knative.dev/pkg/controller"
 )
 
 // Environment Structure
@@ -36,10 +40,11 @@ type Environment struct {
 	HealthPort int // Required
 
 	// Kafka Configuration
-	KafkaBrokers string // Required
-	KafkaTopic   string // Required
-	ChannelKey   string // Required
-	ServiceName  string // Required
+	KafkaBrokers string        // Required
+	KafkaTopic   string        // Required
+	ChannelKey   string        // Required
+	ServiceName  string        // Required
+	ResyncPeriod time.Duration // Optional
 
 	// Kafka Authorization
 	KafkaUsername string // Optional
@@ -74,7 +79,7 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	}
 
 	// Get The Required ContainerName Config Value
-	environment.ContainerName, err = env.GetRequiredConfigValue(logger, env.ContainerNameEnvVarKEy)
+	environment.ContainerName, err = env.GetRequiredConfigValue(logger, env.ContainerNameEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
@@ -109,6 +114,17 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 		return nil, err
 	}
 
+	// Get The Optional Resync Period Config Value & Convert To Duration
+	resyncMinutes, err := env.GetOptionalConfigInt(
+		logger,
+		env.ResyncPeriodMinutesEnvVarKey,
+		strconv.Itoa(int(controller.DefaultResyncPeriod/time.Minute)),
+		"ResyncPeriodMinutes")
+	if err != nil {
+		return nil, err
+	}
+	environment.ResyncPeriod = time.Duration(resyncMinutes) * time.Minute
+
 	// Get The Optional KafkaUsername Config Value
 	environment.KafkaUsername = env.GetOptionalConfigValue(logger, env.KafkaUsernameEnvVarKey, "")
 
diff --git a/pkg/channel/distributed/receiver/channel/channel.go b/pkg/channel/distributed/receiver/channel/channel.go
index 9c03d5a11..9b4058c7f 100644
--- a/pkg/channel/distributed/receiver/channel/channel.go
+++ b/pkg/channel/distributed/receiver/channel/channel.go
@@ -19,6 +19,7 @@ package channel
 import (
 	"context"
 	"errors"
+	"time"
 
 	"go.uber.org/zap"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -28,7 +29,6 @@ import (
 	kafkainformers "knative.dev/eventing-kafka/pkg/client/informers/externalversions"
 	kafkalisters "knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1"
 	eventingChannel "knative.dev/eventing/pkg/channel"
-	knativecontroller "knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 )
 
@@ -54,7 +54,7 @@ var getKafkaClient = func(ctx context.Context, serverUrl string, kubeconfigPath
 }
 
 // Initialize The KafkaChannel Lister Singleton
-func InitializeKafkaChannelLister(ctx context.Context, serverUrl string, kubeconfigPath string, healthServer *health.Server) error {
+func InitializeKafkaChannelLister(ctx context.Context, serverUrl string, kubeconfigPath string, healthServer *health.Server, resyncDuration time.Duration) error {
 
 	// Get The Logger From The Provided Context
 	logger = logging.FromContext(ctx).Desugar()
@@ -67,7 +67,7 @@ func InitializeKafkaChannelLister(ctx context.Context, serverUrl string, kubecon
 	}
 
 	// Create A New KafkaChannel SharedInformerFactory For ALL Namespaces (Default Resync Is 10 Hrs)
-	sharedInformerFactory := kafkainformers.NewSharedInformerFactory(client, knativecontroller.DefaultResyncPeriod)
+	sharedInformerFactory := kafkainformers.NewSharedInformerFactory(client, resyncDuration)
 
 	// Initialize The Stop Channel (Close If Previously Created)
 	Close()
diff --git a/pkg/channel/distributed/receiver/env/environment.go b/pkg/channel/distributed/receiver/env/environment.go
index ecab1a250..b39335118 100644
--- a/pkg/channel/distributed/receiver/env/environment.go
+++ b/pkg/channel/distributed/receiver/env/environment.go
@@ -17,8 +17,12 @@ limitations under the License.
 package env
 
 import (
+	"strconv"
+	"time"
+
 	"go.uber.org/zap"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	"knative.dev/pkg/controller"
 )
 
 // The Environment Struct
@@ -36,8 +40,9 @@ type Environment struct {
 	HealthPort int // Required
 
 	// Kafka Configuration
-	KafkaBrokers string // Required
-	ServiceName  string // Required
+	KafkaBrokers string        // Required
+	ServiceName  string        // Required
+	ResyncPeriod time.Duration // Optional
 
 	// Kafka Authorization
 	KafkaUsername string // Optional
@@ -72,7 +77,7 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	}
 
 	// Get The Required ContainerName Config Value
-	environment.ContainerName, err = env.GetRequiredConfigValue(logger, env.ContainerNameEnvVarKEy)
+	environment.ContainerName, err = env.GetRequiredConfigValue(logger, env.ContainerNameEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
@@ -95,6 +100,17 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 		return nil, err
 	}
 
+	// Get The Optional ResyncPeriod Config Value & Convert To Duration
+	resyncMinutes, err := env.GetOptionalConfigInt(
+		logger,
+		env.ResyncPeriodMinutesEnvVarKey,
+		strconv.Itoa(int(controller.DefaultResyncPeriod/time.Minute)),
+		"ResyncPeriodMinutes")
+	if err != nil {
+		return nil, err
+	}
+	environment.ResyncPeriod = time.Duration(resyncMinutes) * time.Minute
+
 	// Get The Optional KafkaUsername Config Value
 	environment.KafkaUsername = env.GetOptionalConfigValue(logger, env.KafkaUsernameEnvVarKey, "")
 
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/sinkbinding_lifecycle.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/sinkbinding_lifecycle.go
index 399cebbc8..d62830777 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/sinkbinding_lifecycle.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/sinkbinding_lifecycle.go
@@ -124,8 +124,7 @@ func (sb *SinkBinding) Do(ctx context.Context, ps *duckv1.WithPod) {
 		spec.InitContainers[i].Env = append(spec.InitContainers[i].Env, corev1.EnvVar{
 			Name:  "K_SINK",
 			Value: uri.String(),
-		})
-		spec.InitContainers[i].Env = append(spec.InitContainers[i].Env, corev1.EnvVar{
+		}, corev1.EnvVar{
 			Name:  "K_CE_OVERRIDES",
 			Value: ceOverrides,
 		})
@@ -134,8 +133,7 @@ func (sb *SinkBinding) Do(ctx context.Context, ps *duckv1.WithPod) {
 		spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
 			Name:  "K_SINK",
 			Value: uri.String(),
-		})
-		spec.Containers[i].Env = append(spec.Containers[i].Env, corev1.EnvVar{
+		}, corev1.EnvVar{
 			Name:  "K_CE_OVERRIDES",
 			Value: ceOverrides,
 		})
diff --git a/vendor/knative.dev/eventing/pkg/channel/fanout/fanout_message_handler.go b/vendor/knative.dev/eventing/pkg/channel/fanout/fanout_message_handler.go
index 96e61011b..b59b99a7f 100644
--- a/vendor/knative.dev/eventing/pkg/channel/fanout/fanout_message_handler.go
+++ b/vendor/knative.dev/eventing/pkg/channel/fanout/fanout_message_handler.go
@@ -236,10 +236,8 @@ func parseFanoutResultAndReportMetrics(result dispatchResult, reporter channel.S
 	err := result.err
 	if err != nil {
 		channel.ReportEventCountMetricsForDispatchError(err, reporter, &reportArgs)
-	} else {
-		if result.info != nil {
-			_ = reporter.ReportEventCount(&reportArgs, result.info.ResponseCode)
-		}
+	} else if result.info != nil {
+		_ = reporter.ReportEventCount(&reportArgs, result.info.ResponseCode)
 	}
 	return err
 }
diff --git a/vendor/knative.dev/hack/README.md b/vendor/knative.dev/hack/README.md
index af2cbc657..1df3cfe5b 100644
--- a/vendor/knative.dev/hack/README.md
+++ b/vendor/knative.dev/hack/README.md
@@ -220,7 +220,7 @@ benchmarking jobs for each repo. To use it:
    - `SERVICE_ACCOUNT_NAME`: Service account name for controlling GKE clusters
      and interacting with [Mako](https://github.com/google/mako) server. It MUST
      have `Kubernetes Engine Admin` and `Storage Admin` role, and be
-     [whitelisted](https://github.com/google/mako/blob/master/docs/ACCESS.md) by
+     [allowed](https://github.com/google/mako/blob/master/docs/ACCESS.md) by
      Mako admin. Defaults to `mako-job`.
 
 1. [optional] Customize root path of the benchmarks. This root folder should
diff --git a/vendor/knative.dev/hack/library.sh b/vendor/knative.dev/hack/library.sh
index 020eb7206..f37b3f8df 100644
--- a/vendor/knative.dev/hack/library.sh
+++ b/vendor/knative.dev/hack/library.sh
@@ -353,13 +353,10 @@ function report_go_test() {
   report="$(mktemp)"
   local xml
   xml="$(mktemp_with_extension "${ARTIFACTS}"/junit_XXXXXXXX xml)"
-  local json
-  json="$(mktemp_with_extension "${ARTIFACTS}"/json_XXXXXXXX json)"
   echo "Running go test with args: ${go_test_args[*]}"
   # TODO(chizhg): change to `--format testname`?
   capture_output "${report}" gotestsum --format "${GO_TEST_VERBOSITY:-standard-verbose}" \
     --junitfile "${xml}" --junitfile-testsuite-name relative --junitfile-testcase-classname relative \
-    --jsonfile "${json}" \
     -- "${go_test_args[@]}"
   local failed=$?
   echo "Finished run, return code is ${failed}"
@@ -517,7 +514,7 @@ function go_update_deps() {
   echo "=== Update Deps for Golang"
 
   local UPGRADE=0
-  local VERSION="master"
+  local VERSION="v9000.1" # release v9000 is so far in the future, it will always pick the default branch.
   local DOMAIN="knative.dev"
   while [[ $# -ne 0 ]]; do
     parameter=$1
diff --git a/vendor/knative.dev/pkg/logging/config.go b/vendor/knative.dev/pkg/logging/config.go
index 2b5eb696d..e1480033d 100644
--- a/vendor/knative.dev/pkg/logging/config.go
+++ b/vendor/knative.dev/pkg/logging/config.go
@@ -65,7 +65,10 @@ func NewLogger(configJSON string, levelOverride string, opts ...zap.Option) (*za
 	if err2 != nil {
 		panic(err2)
 	}
-	return enrichLoggerWithCommitID(logger.Named(fallbackLoggerName)), loggingCfg.Level
+
+	slogger := enrichLoggerWithCommitID(logger.Named(fallbackLoggerName))
+	slogger.Warnw("Failed to parse logging config - using default zap production config", zap.Error(err))
+	return slogger, loggingCfg.Level
 }
 
 func enrichLoggerWithCommitID(logger *zap.Logger) *zap.SugaredLogger {
diff --git a/vendor/knative.dev/pkg/metrics/config.go b/vendor/knative.dev/pkg/metrics/config.go
index 38ac02e65..d4ae24ad5 100644
--- a/vendor/knative.dev/pkg/metrics/config.go
+++ b/vendor/knative.dev/pkg/metrics/config.go
@@ -60,7 +60,9 @@ const (
 	defaultPrometheusPort = 9090
 	maxPrometheusPort     = 65535
 	minPrometheusPort     = 1024
+	defaultPrometheusHost = "0.0.0.0"
 	prometheusPortEnvName = "METRICS_PROMETHEUS_PORT"
+	prometheusHostEnvName = "METRICS_PROMETHEUS_HOST"
 )
 
 // Metrics backend "enum".
@@ -105,6 +107,10 @@ type metricsConfig struct {
 	// format. It defaults to 9090.
 	prometheusPort int
 
+	// prometheusHost is the host where the metrics are exposed in Prometheus
+	// format. It defaults to "0.0.0.0"
+	prometheusHost string
+
 	// ---- Stackdriver specific below ----
 	// True if backendDestination equals to "stackdriver". Store this in a variable
 	// to reduce string comparison operations.
@@ -240,6 +246,7 @@ func createMetricsConfig(ctx context.Context, ops ExporterOptions) (*metricsConf
 		}
 
 		mc.prometheusPort = pp
+		mc.prometheusHost = prometheusHost()
 	case stackdriver:
 		// If stackdriverClientConfig is not provided for stackdriver backend destination, OpenCensus will try to
 		// use the application default credentials. If that is not available, Opencensus would fail to create the
@@ -341,6 +348,17 @@ func prometheusPort() (int, error) {
 	return int(pp), nil
 }
 
+// prometheusHost returns the host configured via the environment
+// for the Prometheus metrics exporter if it's set, a default value otherwise.
+// No validation is done here.
+func prometheusHost() string {
+	phStr := os.Getenv(prometheusHostEnvName)
+	if phStr == "" {
+		return defaultPrometheusHost
+	}
+	return phStr
+}
+
 // JSONToOptions converts a json string to ExporterOptions.
 func JSONToOptions(jsonOpts string) (*ExporterOptions, error) {
 	var opts ExporterOptions
diff --git a/vendor/knative.dev/pkg/metrics/exporter.go b/vendor/knative.dev/pkg/metrics/exporter.go
index a7411fccb..0c680106e 100644
--- a/vendor/knative.dev/pkg/metrics/exporter.go
+++ b/vendor/knative.dev/pkg/metrics/exporter.go
@@ -66,10 +66,15 @@ type ExporterOptions struct {
 
 	// PrometheusPort is the port to expose metrics if metrics backend is Prometheus.
 	// It should be between maxPrometheusPort and maxPrometheusPort. 0 value means
-	// using the default 9090 value. If is ignored if metrics backend is not
+	// using the default 9090 value. It is ignored if metrics backend is not
 	// Prometheus.
 	PrometheusPort int
 
+	// PrometheusHost is the host to expose metrics on if metrics backend is Prometheus.
+	// The default value is "0.0.0.0". It is ignored if metrics backend is not
+	// Prometheus.
+	PrometheusHost string
+
 	// ConfigMap is the data from config map config-observability. Must be present.
 	// See https://github.com/knative/serving/blob/master/config/config-observability.yaml
 	// for details.
diff --git a/vendor/knative.dev/pkg/metrics/prometheus_exporter.go b/vendor/knative.dev/pkg/metrics/prometheus_exporter.go
index 7f49302cc..784eab2e8 100644
--- a/vendor/knative.dev/pkg/metrics/prometheus_exporter.go
+++ b/vendor/knative.dev/pkg/metrics/prometheus_exporter.go
@@ -17,8 +17,8 @@ limitations under the License.
 package metrics
 
 import (
-	"fmt"
 	"net/http"
+	"strconv"
 	"sync"
 
 	prom "contrib.go.opencensus.io/exporter/prometheus"
@@ -41,6 +41,7 @@ func (emptyPromExporter) ExportView(viewData *view.Data) {
 	// a signal to enrich the internal Meters with Resource information.
 }
 
+//nolint: unparam // False positive of flagging the second result of this function unused.
 func newPrometheusExporter(config *metricsConfig, logger *zap.SugaredLogger) (view.Exporter, ResourceExporterFactory, error) {
 	e, err := prom.NewExporter(prom.Options{Namespace: config.component})
 	if err != nil {
@@ -50,7 +51,7 @@ func newPrometheusExporter(config *metricsConfig, logger *zap.SugaredLogger) (vi
 	logger.Infof("Created Opencensus Prometheus exporter with config: %v. Start the server for Prometheus exporter.", config)
 	// Start the server for Prometheus scraping
 	go func() {
-		srv := startNewPromSrv(e, config.prometheusPort)
+		srv := startNewPromSrv(e, config.prometheusHost, config.prometheusPort)
 		srv.ListenAndServe()
 	}()
 	return e,
@@ -73,7 +74,7 @@ func resetCurPromSrv() {
 	}
 }
 
-func startNewPromSrv(e *prom.Exporter, port int) *http.Server {
+func startNewPromSrv(e *prom.Exporter, host string, port int) *http.Server {
 	sm := http.NewServeMux()
 	sm.Handle("/metrics", e)
 	curPromSrvMux.Lock()
@@ -82,7 +83,7 @@ func startNewPromSrv(e *prom.Exporter, port int) *http.Server {
 		curPromSrv.Close()
 	}
 	curPromSrv = &http.Server{
-		Addr:    fmt.Sprint(":", port),
+		Addr:    host + ":" + strconv.Itoa(port),
 		Handler: sm,
 	}
 	return curPromSrv
diff --git a/vendor/knative.dev/pkg/tracker/enqueue.go b/vendor/knative.dev/pkg/tracker/enqueue.go
index 2e001470c..70fbe0e46 100644
--- a/vendor/knative.dev/pkg/tracker/enqueue.go
+++ b/vendor/knative.dev/pkg/tracker/enqueue.go
@@ -216,9 +216,18 @@ func isExpired(expiry time.Time) bool {
 
 // OnChanged implements Interface.
 func (i *impl) OnChanged(obj interface{}) {
+	observers := i.GetObservers(obj)
+
+	for _, observer := range observers {
+		i.cb(observer)
+	}
+}
+
+// GetObservers implements Interface.
+func (i *impl) GetObservers(obj interface{}) []types.NamespacedName {
 	item, err := kmeta.DeletionHandlingAccessor(obj)
 	if err != nil {
-		return
+		return nil
 	}
 
 	or := kmeta.ObjectReference(item)
@@ -229,14 +238,9 @@ func (i *impl) OnChanged(obj interface{}) {
 		Name:       or.Name,
 	}
 
-	i.m.Lock()
-	// Call the callbacks without the lock held.
 	var keys []types.NamespacedName
-	defer func(cb func(types.NamespacedName)) {
-		for _, key := range keys {
-			cb(key)
-		}
-	}(i.cb) // read i.cb with the lock held
+
+	i.m.Lock()
 	defer i.m.Unlock()
 
 	// Handle exact matches.
@@ -274,6 +278,8 @@ func (i *impl) OnChanged(obj interface{}) {
 			delete(i.exact, ref)
 		}
 	}
+
+	return keys
 }
 
 // OnChanged implements Interface.
diff --git a/vendor/knative.dev/pkg/tracker/interface.go b/vendor/knative.dev/pkg/tracker/interface.go
index c8ab479fa..bfcda9cf2 100644
--- a/vendor/knative.dev/pkg/tracker/interface.go
+++ b/vendor/knative.dev/pkg/tracker/interface.go
@@ -23,6 +23,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/validation"
 
 	"knative.dev/pkg/apis"
@@ -71,6 +72,10 @@ type Interface interface {
 	// so that we are notified for appropriate object changes.
 	OnChanged(obj interface{})
 
+	// GetObservers returns the names of all observers for the given
+	// object.
+	GetObservers(obj interface{}) []types.NamespacedName
+
 	// OnDeletedObserver is a callback to register with the InformerFactory
 	// so that we are notified for deletions of a watching parent to
 	// remove the respective tracking.
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 00aa0047f..f0d415a38 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1067,7 +1067,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.19.0
+# knative.dev/eventing v0.19.1-0.20201113093000-1616938f9f1c
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/util/crstatusevent
@@ -1178,8 +1178,9 @@ 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/logger_vent
-knative.dev/eventing/test/lib/recordevents/observer
+knative.dev/eventing/test/lib/recordevents/receiver
 knative.dev/eventing/test/lib/recordevents/recorder_vent
+knative.dev/eventing/test/lib/recordevents/sender
 knative.dev/eventing/test/lib/resources
 knative.dev/eventing/test/lib/sender
 knative.dev/eventing/test/performance/infra
@@ -1194,10 +1195,10 @@ knative.dev/eventing/test/test_images/heartbeats
 knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
-# knative.dev/hack v0.0.0-20201103151104-3d5abc3a0075
+# knative.dev/hack v0.0.0-20201112185459-01a34c573bd8
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20201103163404-5514ab0c1fdf
+# knative.dev/pkg v0.0.0-20201112201059-93fedf141385
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/pkg/channel/consolidated/utils/util_test.go b/pkg/channel/consolidated/utils/util_test.go
index 4beb51290..8a687f7ec 100644
--- a/pkg/channel/consolidated/utils/util_test.go
+++ b/pkg/channel/consolidated/utils/util_test.go
@@ -89,35 +89,8 @@ func TestGetKafkaConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "partition consumer",
-			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "consumerMode": "partitions"},
-			expected: &KafkaConfig{
-				Brokers:             []string{"kafkabroker.kafka:9092"},
-				MaxIdleConns:        1000,
-				MaxIdleConnsPerHost: 100,
-			},
-		},
-		{
-			name: "default multiplex",
-			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "consumerMode": "multiplex"},
-			expected: &KafkaConfig{
-				Brokers:             []string{"kafkabroker.kafka:9092"},
-				MaxIdleConns:        1000,
-				MaxIdleConnsPerHost: 100,
-			},
-		},
-		{
-			name: "default multiplex from invalid consumerMode",
-			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "consumerMode": "foo"},
-			expected: &KafkaConfig{
-				Brokers:             []string{"kafkabroker.kafka:9092"},
-				MaxIdleConns:        1000,
-				MaxIdleConnsPerHost: 100,
-			},
-		},
-		{
-			name: "default multiplex from invalid consumerMode elevated max idle connections",
-			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "consumerMode": "foo", "maxIdleConns": "9000"},
+			name: "elevated max idle connections",
+			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "maxIdleConns": "9000"},
 			expected: &KafkaConfig{
 				Brokers:             []string{"kafkabroker.kafka:9092"},
 				MaxIdleConns:        9000,
@@ -125,8 +98,8 @@ func TestGetKafkaConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "default multiplex from invalid consumerMode elevated max idle connections per host",
-			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "consumerMode": "foo", "maxIdleConnsPerHost": "900"},
+			name: "elevated max idle connections per host",
+			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "maxIdleConnsPerHost": "900"},
 			expected: &KafkaConfig{
 				Brokers:             []string{"kafkabroker.kafka:9092"},
 				MaxIdleConns:        1000,
@@ -134,8 +107,8 @@ func TestGetKafkaConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "default multiplex from invalid consumerMode elevated max idle values",
-			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "consumerMode": "foo", "maxIdleConns": "9000", "maxIdleConnsPerHost": "600"},
+			name: "elevated max idle values",
+			data: map[string]string{"bootstrapServers": "kafkabroker.kafka:9092", "maxIdleConns": "9000", "maxIdleConnsPerHost": "600"},
 			expected: &KafkaConfig{
 				Brokers:             []string{"kafkabroker.kafka:9092"},
 				MaxIdleConns:        9000,
diff --git a/pkg/channel/distributed/controller/env/environment_test.go b/pkg/channel/distributed/controller/env/environment_test.go
index b4f20cc40..dda4eb1d0 100644
--- a/pkg/channel/distributed/controller/env/environment_test.go
+++ b/pkg/channel/distributed/controller/env/environment_test.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
@@ -30,9 +31,10 @@ import (
 
 // Test Constants
 const (
-	serviceAccount = "TestServiceAccount"
-	metricsPort    = "9999"
-	metricsDomain  = "example.com/kafka-eventing"
+	serviceAccount      = "TestServiceAccount"
+	metricsPort         = "9999"
+	metricsDomain       = "example.com/kafka-eventing"
+	resyncPeriodMinutes = "3600"
 
 	defaultKafkaConsumers = "5"
 
@@ -47,10 +49,12 @@ type TestCase struct {
 	serviceAccount        string
 	metricsPort           string
 	metricsDomain         string
+	resyncPeriodMinutes   string
 	defaultKafkaConsumers string
 	dispatcherImage       string
 	channelImage          string
 	expectedError         error
+	expectedResyncPeriod  string
 }
 
 // Test All Permutations Of The GetEnvironment() Functionality
@@ -94,16 +98,21 @@ func TestGetEnvironment(t *testing.T) {
 	testCase.expectedError = getMissingRequiredEnvironmentVariableError(ReceiverImageEnvVarKey)
 	testCases = append(testCases, testCase)
 
+	testCase = getValidTestCase("Invalid Config - ResyncPeriodMinutes")
+	testCase.resyncPeriodMinutes = "NAN"
+	testCase.expectedError = getInvalidIntEnvironmentVariableError(testCase.resyncPeriodMinutes, env.ResyncPeriodMinutesEnvVarKey)
+	testCases = append(testCases, testCase)
+
+	testCase = getValidTestCase("Valid Config - Default ResyncPeriodMinutes")
+	testCase.resyncPeriodMinutes = ""
+	testCase.expectedResyncPeriod = "600" // 10 hours - default value
+	testCases = append(testCases, testCase)
+
 	// Loop Over All The TestCases
 	for _, testCase := range testCases {
 
 		// (Re)Setup The Environment Variables From TestCase
-		os.Clearenv()
-		assertSetenv(t, env.ServiceAccountEnvVarKey, testCase.serviceAccount)
-		assertSetenv(t, env.MetricsDomainEnvVarKey, testCase.metricsDomain)
-		assertSetenvNonempty(t, env.MetricsPortEnvVarKey, testCase.metricsPort)
-		assertSetenv(t, DispatcherImageEnvVarKey, testCase.dispatcherImage)
-		assertSetenv(t, ReceiverImageEnvVarKey, testCase.channelImage)
+		setupTestEnvironment(t, testCase)
 
 		// Perform The Test
 		environment, err := GetEnvironment(logger)
@@ -117,6 +126,7 @@ func TestGetEnvironment(t *testing.T) {
 			assert.Equal(t, testCase.metricsPort, strconv.Itoa(environment.MetricsPort))
 			assert.Equal(t, testCase.channelImage, environment.ReceiverImage)
 			assert.Equal(t, testCase.dispatcherImage, environment.DispatcherImage)
+			assert.Equal(t, testCase.expectedResyncPeriod, strconv.Itoa(int(environment.ResyncPeriod/time.Minute)))
 
 		} else {
 			assert.Equal(t, testCase.expectedError, err)
@@ -126,6 +136,17 @@ func TestGetEnvironment(t *testing.T) {
 	}
 }
 
+// Add TestCase Variables To The Environment
+func setupTestEnvironment(t *testing.T, testCase TestCase) {
+	os.Clearenv()
+	assertSetenv(t, env.ServiceAccountEnvVarKey, testCase.serviceAccount)
+	assertSetenv(t, env.MetricsDomainEnvVarKey, testCase.metricsDomain)
+	assertSetenvNonempty(t, env.MetricsPortEnvVarKey, testCase.metricsPort)
+	assertSetenv(t, DispatcherImageEnvVarKey, testCase.dispatcherImage)
+	assertSetenv(t, ReceiverImageEnvVarKey, testCase.channelImage)
+	assertSetenvNonempty(t, env.ResyncPeriodMinutesEnvVarKey, testCase.resyncPeriodMinutes)
+}
+
 // Get The Base / Valid Test Case - All Config Specified / No Errors
 func getValidTestCase(name string) TestCase {
 	return TestCase{
@@ -133,10 +154,12 @@ func getValidTestCase(name string) TestCase {
 		serviceAccount:        serviceAccount,
 		metricsPort:           metricsPort,
 		metricsDomain:         metricsDomain,
+		resyncPeriodMinutes:   resyncPeriodMinutes,
 		defaultKafkaConsumers: defaultKafkaConsumers,
 		dispatcherImage:       dispatcherImage,
 		channelImage:          receiverImage,
 		expectedError:         nil,
+		expectedResyncPeriod:  resyncPeriodMinutes,
 	}
 }
 
diff --git a/pkg/channel/distributed/controller/kafkachannel/controller_test.go b/pkg/channel/distributed/controller/kafkachannel/controller_test.go
index f257ca3a0..680da5ce0 100644
--- a/pkg/channel/distributed/controller/kafkachannel/controller_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/controller_test.go
@@ -21,6 +21,7 @@ import (
 	"os"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
@@ -207,6 +208,7 @@ func populateEnvironmentVariables(t *testing.T) {
 	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, strconv.Itoa(controllertesting.MetricsPort)))
 	assert.Nil(t, os.Setenv(controllerenv.DispatcherImageEnvVarKey, controllertesting.DispatcherImage))
 	assert.Nil(t, os.Setenv(controllerenv.ReceiverImageEnvVarKey, controllertesting.ReceiverImage))
+	assert.Nil(t, os.Setenv(commonenv.ResyncPeriodMinutesEnvVarKey, strconv.Itoa(int(controllertesting.ResyncPeriod/time.Minute))))
 }
 
 // Utility Function For Creating A K8S Service With Specified ObjectMeta
diff --git a/pkg/channel/distributed/controller/kafkasecret/controller_test.go b/pkg/channel/distributed/controller/kafkasecret/controller_test.go
index 27b1d410d..0d92cbda1 100644
--- a/pkg/channel/distributed/controller/kafkasecret/controller_test.go
+++ b/pkg/channel/distributed/controller/kafkasecret/controller_test.go
@@ -21,6 +21,7 @@ import (
 	"os"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"k8s.io/client-go/rest"
@@ -85,4 +86,5 @@ func populateEnvironmentVariables(t *testing.T) {
 	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, strconv.Itoa(controllertesting.MetricsPort)))
 	assert.Nil(t, os.Setenv(controllerenv.DispatcherImageEnvVarKey, controllertesting.DispatcherImage))
 	assert.Nil(t, os.Setenv(controllerenv.ReceiverImageEnvVarKey, controllertesting.ReceiverImage))
+	assert.Nil(t, os.Setenv(commonenv.ResyncPeriodMinutesEnvVarKey, strconv.Itoa(int(controllertesting.ResyncPeriod/time.Minute))))
 }
diff --git a/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer_test.go b/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer_test.go
index c71afb88e..394a236c6 100644
--- a/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer_test.go
+++ b/pkg/channel/distributed/controller/kafkasecretinformer/kafkasecretinformer_test.go
@@ -18,10 +18,16 @@ package kafkasecretinformer
 
 import (
 	"context"
+	"os"
+	"strconv"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"k8s.io/client-go/kubernetes/fake"
+	commonenv "knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	controllerenv "knative.dev/eventing-kafka/pkg/channel/distributed/controller/env"
+	controllertesting "knative.dev/eventing-kafka/pkg/channel/distributed/controller/testing"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/logging"
@@ -35,6 +41,7 @@ func TestGet(t *testing.T) {
 	// Create A Context With Test Logger & K8S Client
 	ctx := logging.WithLogger(context.TODO(), logtesting.TestLogger(t))
 	ctx = context.WithValue(ctx, injectionclient.Key{}, fake.NewSimpleClientset())
+	populateEnvironmentVariables(t)
 
 	// Verify The KafkaSecretInformer Was Added To Knative Injection
 	informers := injection.Default.GetInformers()
@@ -50,3 +57,14 @@ func TestGet(t *testing.T) {
 	kafkaSecretInformer := Get(ctx)
 	assert.NotNil(t, kafkaSecretInformer)
 }
+
+// Utility Function For Populating Required Environment Variables For Testing
+func populateEnvironmentVariables(t *testing.T) {
+	// Most of these are not actually used, but they need to exist or the GetEnvironment call will fail
+	assert.Nil(t, os.Setenv(commonenv.ServiceAccountEnvVarKey, controllertesting.ServiceAccount))
+	assert.Nil(t, os.Setenv(commonenv.MetricsDomainEnvVarKey, controllertesting.MetricsDomain))
+	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, strconv.Itoa(controllertesting.MetricsPort)))
+	assert.Nil(t, os.Setenv(controllerenv.DispatcherImageEnvVarKey, controllertesting.DispatcherImage))
+	assert.Nil(t, os.Setenv(controllerenv.ReceiverImageEnvVarKey, controllertesting.ReceiverImage))
+	assert.Nil(t, os.Setenv(commonenv.ResyncPeriodMinutesEnvVarKey, strconv.Itoa(int(controllertesting.ResyncPeriod/time.Minute))))
+}
diff --git a/pkg/channel/distributed/controller/testing/data.go b/pkg/channel/distributed/controller/testing/data.go
index 9be943507..6ec6f4dd1 100644
--- a/pkg/channel/distributed/controller/testing/data.go
+++ b/pkg/channel/distributed/controller/testing/data.go
@@ -19,6 +19,7 @@ package testing
 import (
 	"fmt"
 	"strconv"
+	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -56,6 +57,7 @@ const (
 	MetricsPort              = 9876
 	MetricsDomain            = "eventing-kafka"
 	HealthPort               = 8082
+	ResyncPeriod             = 3600 * time.Minute
 	ReceiverImage            = "TestReceiverImage"
 	ReceiverReplicas         = 1
 	DispatcherImage          = "TestDispatcherImage"
@@ -194,6 +196,7 @@ func NewEnvironment() *env.Environment {
 		MetricsDomain:   MetricsDomain,
 		DispatcherImage: DispatcherImage,
 		ReceiverImage:   ReceiverImage,
+		ResyncPeriod:    ResyncPeriod,
 	}
 }
 
@@ -616,7 +619,7 @@ func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Depl
 									},
 								},
 								{
-									Name:  commonenv.ContainerNameEnvVarKEy,
+									Name:  commonenv.ContainerNameEnvVarKey,
 									Value: constants.ReceiverContainerName,
 								},
 								{
@@ -639,6 +642,10 @@ func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Depl
 									Name:  commonenv.HealthPortEnvVarKey,
 									Value: strconv.Itoa(HealthPort),
 								},
+								{
+									Name:  commonenv.ResyncPeriodMinutesEnvVarKey,
+									Value: strconv.Itoa(int(ResyncPeriod / time.Minute)),
+								},
 								{
 									Name: commonenv.KafkaBrokerEnvVarKey,
 									ValueFrom: &corev1.EnvVarSource{
@@ -822,7 +829,7 @@ func NewKafkaChannelDispatcherDeployment(options ...DeploymentOption) *appsv1.De
 									},
 								},
 								{
-									Name:  commonenv.ContainerNameEnvVarKEy,
+									Name:  commonenv.ContainerNameEnvVarKey,
 									Value: constants.DispatcherContainerName,
 								},
 								{
@@ -853,6 +860,10 @@ func NewKafkaChannelDispatcherDeployment(options ...DeploymentOption) *appsv1.De
 									Name:  commonenv.KafkaTopicEnvVarKey,
 									Value: topicName,
 								},
+								{
+									Name:  commonenv.ResyncPeriodMinutesEnvVarKey,
+									Value: strconv.Itoa(int(ResyncPeriod / time.Minute)),
+								},
 								{
 									Name: commonenv.KafkaBrokerEnvVarKey,
 									ValueFrom: &corev1.EnvVarSource{
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index 8ab00e278..7391845c5 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -17,9 +17,12 @@ limitations under the License.
 package controller
 
 import (
+	"os"
 	"testing"
 	"time"
 
+	commonenv "knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -60,6 +63,7 @@ func TestNewController(t *testing.T) {
 	mockDispatcher := NewMockDispatcher(t)
 	fakeKafkaChannelClientSet := fakeclientset.NewSimpleClientset()
 	fakeK8sClientSet := fake.NewSimpleClientset()
+	populateEnvironmentVariables(t)
 	kafkaInformerFactory := externalversions.NewSharedInformerFactory(fakeKafkaChannelClientSet, kncontroller.DefaultResyncPeriod)
 	kafkaChannelInformer := kafkaInformerFactory.Messaging().V1beta1().KafkaChannels()
 	stopChan := make(chan struct{})
@@ -173,6 +177,21 @@ func TestAllCases(t *testing.T) {
 	time.Sleep(1 * time.Second)
 }
 
+// Utility Function For Populating Required Environment Variables For Testing
+func populateEnvironmentVariables(t *testing.T) {
+	// Most of these are not actually used, but they need to exist or the GetEnvironment call will fail
+	assert.Nil(t, os.Setenv(commonenv.MetricsDomainEnvVarKey, "testMetricsDomain"))
+	assert.Nil(t, os.Setenv(commonenv.MetricsPortEnvVarKey, "6789"))
+	assert.Nil(t, os.Setenv(commonenv.ResyncPeriodMinutesEnvVarKey, "3600"))
+	assert.Nil(t, os.Setenv(commonenv.PodNameEnvVarKey, "testPodName"))
+	assert.Nil(t, os.Setenv(commonenv.ContainerNameEnvVarKey, "testContainerName"))
+	assert.Nil(t, os.Setenv(commonenv.HealthPortEnvVarKey, "5678"))
+	assert.Nil(t, os.Setenv(commonenv.KafkaBrokerEnvVarKey, "testKafkaBroker"))
+	assert.Nil(t, os.Setenv(commonenv.KafkaTopicEnvVarKey, "testKafkaTopic"))
+	assert.Nil(t, os.Setenv(commonenv.ChannelKeyEnvVarKey, "testChannelKey"))
+	assert.Nil(t, os.Setenv(commonenv.ServiceNameEnvVarKey, "testServiceName"))
+}
+
 //
 // Mock Dispatcher Implementation
 //
diff --git a/pkg/channel/distributed/dispatcher/env/environment_test.go b/pkg/channel/distributed/dispatcher/env/environment_test.go
index 14ecc828e..f5fd05eb2 100644
--- a/pkg/channel/distributed/dispatcher/env/environment_test.go
+++ b/pkg/channel/distributed/dispatcher/env/environment_test.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
@@ -33,6 +34,7 @@ const (
 	metricsPort   = "9999"
 	metricsDomain = "kafka-eventing"
 	healthPort    = "1234"
+	resyncPeriod  = "3600"
 	kafkaBrokers  = "TestKafkaBrokers"
 	kafkaTopic    = "TestKafkaTopic"
 	channelKey    = "TestChannelKey"
@@ -45,19 +47,21 @@ const (
 
 // Define The TestCase Struct
 type TestCase struct {
-	name          string
-	metricsPort   string
-	metricsDomain string
-	healthPort    string
-	kafkaBrokers  string
-	kafkaTopic    string
-	channelKey    string
-	serviceName   string
-	kafkaUsername string
-	kafkaPassword string
-	podName       string
-	containerName string
-	expectedError error
+	name                 string
+	metricsPort          string
+	metricsDomain        string
+	healthPort           string
+	resyncPeriodMinutes  string
+	kafkaBrokers         string
+	kafkaTopic           string
+	channelKey           string
+	serviceName          string
+	kafkaUsername        string
+	kafkaPassword        string
+	podName              string
+	containerName        string
+	expectedError        error
+	expectedResyncPeriod string
 }
 
 // Test All Permutations Of The GetEnvironment() Functionality
@@ -123,7 +127,17 @@ func TestGetEnvironment(t *testing.T) {
 
 	testCase = getValidTestCase("Missing Required Config - ContainerName")
 	testCase.containerName = ""
-	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.ContainerNameEnvVarKEy)
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.ContainerNameEnvVarKey)
+	testCases = append(testCases, testCase)
+
+	testCase = getValidTestCase("Invalid Config - ResyncPeriodMinutes")
+	testCase.resyncPeriodMinutes = "NAN"
+	testCase.expectedError = getInvalidIntEnvironmentVariableError(testCase.resyncPeriodMinutes, commonenv.ResyncPeriodMinutesEnvVarKey)
+	testCases = append(testCases, testCase)
+
+	testCase = getValidTestCase("Valid Config - Default ResyncPeriodMinutes")
+	testCase.resyncPeriodMinutes = ""
+	testCase.expectedResyncPeriod = "600" // 10 hours - default value
 	testCases = append(testCases, testCase)
 
 	// Loop Over All The TestCases
@@ -134,6 +148,7 @@ func TestGetEnvironment(t *testing.T) {
 		assertSetenv(t, commonenv.MetricsDomainEnvVarKey, testCase.metricsDomain)
 		assertSetenvNonempty(t, commonenv.MetricsPortEnvVarKey, testCase.metricsPort)
 		assertSetenvNonempty(t, commonenv.HealthPortEnvVarKey, testCase.healthPort)
+		assertSetenvNonempty(t, commonenv.ResyncPeriodMinutesEnvVarKey, testCase.resyncPeriodMinutes)
 		assertSetenv(t, commonenv.KafkaBrokerEnvVarKey, testCase.kafkaBrokers)
 		assertSetenv(t, commonenv.KafkaTopicEnvVarKey, testCase.kafkaTopic)
 		assertSetenv(t, commonenv.ChannelKeyEnvVarKey, testCase.channelKey)
@@ -141,7 +156,7 @@ func TestGetEnvironment(t *testing.T) {
 		assertSetenv(t, commonenv.KafkaUsernameEnvVarKey, testCase.kafkaUsername)
 		assertSetenv(t, commonenv.KafkaPasswordEnvVarKey, testCase.kafkaPassword)
 		assertSetenv(t, commonenv.PodNameEnvVarKey, testCase.podName)
-		assertSetenv(t, commonenv.ContainerNameEnvVarKEy, testCase.containerName)
+		assertSetenv(t, commonenv.ContainerNameEnvVarKey, testCase.containerName)
 
 		// Perform The Test
 		environment, err := GetEnvironment(logger)
@@ -161,6 +176,7 @@ func TestGetEnvironment(t *testing.T) {
 			assert.Equal(t, testCase.kafkaPassword, environment.KafkaPassword)
 			assert.Equal(t, testCase.podName, environment.PodName)
 			assert.Equal(t, testCase.containerName, environment.ContainerName)
+			assert.Equal(t, testCase.expectedResyncPeriod, strconv.Itoa(int(environment.ResyncPeriod/time.Minute)))
 
 		} else {
 			assert.Equal(t, testCase.expectedError, err)
@@ -183,19 +199,21 @@ func assertSetenvNonempty(t *testing.T, envKey string, value string) {
 // Get The Base / Valid Test Case - All Config Specified / No Errors
 func getValidTestCase(name string) TestCase {
 	return TestCase{
-		name:          name,
-		metricsPort:   metricsPort,
-		metricsDomain: metricsDomain,
-		healthPort:    healthPort,
-		kafkaBrokers:  kafkaBrokers,
-		kafkaTopic:    kafkaTopic,
-		channelKey:    channelKey,
-		serviceName:   serviceName,
-		kafkaUsername: kafkaUsername,
-		kafkaPassword: kafkaPassword,
-		podName:       podName,
-		containerName: containerName,
-		expectedError: nil,
+		name:                 name,
+		metricsPort:          metricsPort,
+		metricsDomain:        metricsDomain,
+		healthPort:           healthPort,
+		resyncPeriodMinutes:  resyncPeriod,
+		kafkaBrokers:         kafkaBrokers,
+		kafkaTopic:           kafkaTopic,
+		channelKey:           channelKey,
+		serviceName:          serviceName,
+		kafkaUsername:        kafkaUsername,
+		kafkaPassword:        kafkaPassword,
+		podName:              podName,
+		containerName:        containerName,
+		expectedError:        nil,
+		expectedResyncPeriod: resyncPeriod,
 	}
 }
 
diff --git a/pkg/channel/distributed/receiver/channel/channel_test.go b/pkg/channel/distributed/receiver/channel/channel_test.go
index 90cbe6d4c..d6f4d98aa 100644
--- a/pkg/channel/distributed/receiver/channel/channel_test.go
+++ b/pkg/channel/distributed/receiver/channel/channel_test.go
@@ -19,6 +19,7 @@ package channel
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
@@ -43,7 +44,7 @@ func TestInitializeKafkaChannelLister(t *testing.T) {
 
 	// Perform The Test
 	healthServer := channelhealth.NewChannelHealthServer("12345")
-	err := InitializeKafkaChannelLister(ctx, "", "", healthServer)
+	err := InitializeKafkaChannelLister(ctx, "", "", healthServer, 600*time.Minute)
 
 	// Verify The Results
 	assert.Nil(t, err)
diff --git a/pkg/channel/distributed/receiver/env/environment_test.go b/pkg/channel/distributed/receiver/env/environment_test.go
index 23a7b6e77..ce852e6fa 100644
--- a/pkg/channel/distributed/receiver/env/environment_test.go
+++ b/pkg/channel/distributed/receiver/env/environment_test.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
@@ -30,30 +31,33 @@ import (
 
 // Test Constants
 const (
-	metricsPort   = "9999"
-	metricsDomain = "kafka-eventing"
-	healthPort    = "1234"
-	kafkaBrokers  = "TestKafkaBrokers"
-	serviceName   = "TestServiceName"
-	kafkaUsername = "TestKafkaUsername"
-	kafkaPassword = "TestKafkaPassword"
-	podName       = "TestPod"
-	containerName = "TestContainer"
+	metricsPort         = "9999"
+	metricsDomain       = "kafka-eventing"
+	healthPort          = "1234"
+	resyncPeriodMinutes = "3600"
+	kafkaBrokers        = "TestKafkaBrokers"
+	serviceName         = "TestServiceName"
+	kafkaUsername       = "TestKafkaUsername"
+	kafkaPassword       = "TestKafkaPassword"
+	podName             = "TestPod"
+	containerName       = "TestContainer"
 )
 
 // Define The TestCase Struct
 type TestCase struct {
-	name          string
-	metricsPort   string
-	metricsDomain string
-	healthPort    string
-	kafkaBrokers  string
-	serviceName   string
-	kafkaUsername string
-	kafkaPassword string
-	podName       string
-	containerName string
-	expectedError error
+	name                 string
+	metricsPort          string
+	metricsDomain        string
+	healthPort           string
+	resyncPeriodMinutes  string
+	kafkaBrokers         string
+	serviceName          string
+	kafkaUsername        string
+	kafkaPassword        string
+	podName              string
+	containerName        string
+	expectedError        error
+	expectedResyncPeriod string
 }
 
 // Test All Permutations Of The GetEnvironment() Functionality
@@ -92,6 +96,16 @@ func TestGetEnvironment(t *testing.T) {
 	testCase.expectedError = getInvalidIntegerEnvironmentVariableError(testCase.healthPort, env.HealthPortEnvVarKey)
 	testCases = append(testCases, testCase)
 
+	testCase = getValidTestCase("Invalid Config - ResyncPeriodMinutes")
+	testCase.resyncPeriodMinutes = "NAN"
+	testCase.expectedError = getInvalidIntegerEnvironmentVariableError(testCase.resyncPeriodMinutes, env.ResyncPeriodMinutesEnvVarKey)
+	testCases = append(testCases, testCase)
+
+	testCase = getValidTestCase("Valid Config - Default ResyncPeriodMinutes")
+	testCase.resyncPeriodMinutes = ""
+	testCase.expectedResyncPeriod = "600" // 10 hours - default value
+	testCases = append(testCases, testCase)
+
 	testCase = getValidTestCase("Missing Required Config - KafkaBrokers")
 	testCase.kafkaBrokers = ""
 	testCase.expectedError = getMissingRequiredEnvironmentVariableError(env.KafkaBrokerEnvVarKey)
@@ -109,7 +123,7 @@ func TestGetEnvironment(t *testing.T) {
 
 	testCase = getValidTestCase("Missing Required Config - ContainerName")
 	testCase.containerName = ""
-	testCase.expectedError = getMissingRequiredEnvironmentVariableError(env.ContainerNameEnvVarKEy)
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(env.ContainerNameEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	// Loop Over All The TestCases
@@ -125,7 +139,8 @@ func TestGetEnvironment(t *testing.T) {
 		assertSetenv(t, env.KafkaUsernameEnvVarKey, testCase.kafkaUsername)
 		assertSetenv(t, env.KafkaPasswordEnvVarKey, testCase.kafkaPassword)
 		assertSetenv(t, env.PodNameEnvVarKey, testCase.podName)
-		assertSetenv(t, env.ContainerNameEnvVarKEy, testCase.containerName)
+		assertSetenv(t, env.ContainerNameEnvVarKey, testCase.containerName)
+		assertSetenvNonempty(t, env.ResyncPeriodMinutesEnvVarKey, testCase.resyncPeriodMinutes)
 
 		// Perform The Test
 		environment, err := GetEnvironment(logger)
@@ -143,6 +158,7 @@ func TestGetEnvironment(t *testing.T) {
 			assert.Equal(t, testCase.kafkaPassword, environment.KafkaPassword)
 			assert.Equal(t, testCase.podName, environment.PodName)
 			assert.Equal(t, testCase.containerName, environment.ContainerName)
+			assert.Equal(t, testCase.expectedResyncPeriod, strconv.Itoa(int(environment.ResyncPeriod/time.Minute)))
 
 		} else {
 			assert.Equal(t, testCase.expectedError, err)
@@ -165,17 +181,19 @@ func assertSetenvNonempty(t *testing.T, envKey string, value string) {
 // Get The Base / Valid Test Case - All Config Specified / No Errors
 func getValidTestCase(name string) TestCase {
 	return TestCase{
-		name:          name,
-		metricsPort:   metricsPort,
-		metricsDomain: metricsDomain,
-		healthPort:    healthPort,
-		kafkaBrokers:  kafkaBrokers,
-		serviceName:   serviceName,
-		kafkaUsername: kafkaUsername,
-		kafkaPassword: kafkaPassword,
-		podName:       podName,
-		containerName: containerName,
-		expectedError: nil,
+		name:                 name,
+		metricsPort:          metricsPort,
+		metricsDomain:        metricsDomain,
+		healthPort:           healthPort,
+		resyncPeriodMinutes:  resyncPeriodMinutes,
+		kafkaBrokers:         kafkaBrokers,
+		serviceName:          serviceName,
+		kafkaUsername:        kafkaUsername,
+		kafkaPassword:        kafkaPassword,
+		podName:              podName,
+		containerName:        containerName,
+		expectedResyncPeriod: resyncPeriodMinutes,
+		expectedError:        nil,
 	}
 }
 
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index 2991d38d1..af242829d 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -162,7 +162,6 @@ function knative_setup() {
     ko apply -f "${EVENTING_IN_MEMORY_CHANNEL_CONFIG}"
     popd
   fi
-  kubectl apply -f "${HOME}/yaml/config-br.yaml"   ## EDV HACK ##
   wait_until_pods_running knative-eventing || fail_test "Knative Eventing did not come up"
 
   install_zipkin
diff --git a/vendor/knative.dev/eventing/test/e2e-common.sh b/vendor/knative.dev/eventing/test/e2e-common.sh
index 8d7fd4fa2..e5a0615a2 100644
--- a/vendor/knative.dev/eventing/test/e2e-common.sh
+++ b/vendor/knative.dev/eventing/test/e2e-common.sh
@@ -50,9 +50,6 @@ readonly CONFIG_TRACING_CONFIG="test/config/config-tracing.yaml"
 # Installs Zipkin for tracing tests.
 readonly KNATIVE_EVENTING_MONITORING_YAML="test/config/monitoring.yaml"
 
-# PreInstall script for v0.18
-readonly PRE_INSTALL_V018="config/pre-install/v0.18.0"
-
 # The number of controlplane replicas to run.
 readonly REPLICAS=3
 
@@ -190,15 +187,6 @@ function install_latest_release() {
     fail_test "Knative latest release installation failed"
 }
 
-function run_preinstall_V018() {
-  local TMP_PRE_INSTALL_V018=${TMP_DIR}/pre_install
-  mkdir -p ${TMP_PRE_INSTALL_V018}
-  cp -r ${PRE_INSTALL_V018}/* ${TMP_PRE_INSTALL_V018}
-  find ${TMP_PRE_INSTALL_V018} -type f -name "*.yaml" -exec sed -i "s/namespace: ${KNATIVE_DEFAULT_NAMESPACE}/namespace: ${SYSTEM_NAMESPACE}/g" {} +
-  ko apply --strict -f "${TMP_PRE_INSTALL_V018}" || return 1
-  wait_until_batch_job_complete ${SYSTEM_NAMESPACE} || return 1
-}
-
 function install_mt_broker() {
   local TMP_MT_CHANNEL_BASED_BROKER_DEFAULT_CONFIG=${TMP_DIR}/${MT_CHANNEL_BASED_BROKER_DEFAULT_CONFIG##*/}
   sed "s/namespace: ${KNATIVE_DEFAULT_NAMESPACE}/namespace: ${SYSTEM_NAMESPACE}/g" ${MT_CHANNEL_BASED_BROKER_DEFAULT_CONFIG} > ${TMP_MT_CHANNEL_BASED_BROKER_DEFAULT_CONFIG}
diff --git a/vendor/knative.dev/eventing/test/e2e-upgrade-tests.sh b/vendor/knative.dev/eventing/test/e2e-upgrade-tests.sh
index abf94eeea..f74bb68c1 100644
--- a/vendor/knative.dev/eventing/test/e2e-upgrade-tests.sh
+++ b/vendor/knative.dev/eventing/test/e2e-upgrade-tests.sh
@@ -58,7 +58,6 @@ echo "Prober PID is ${PROBER_PID}"
 wait_for_file ${PROBER_READY_FILE} || fail_test
 
 header "Performing upgrade to HEAD"
-run_preinstall_V018 || fail_test 'Running preinstall 0.18 failed'
 install_head || fail_test 'Installing HEAD version of eventing failed'
 install_channel_crds || fail_test 'Installing HEAD channel CRDs failed'
 install_mt_broker || fail_test 'Installing HEAD Broker failed'
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/broker_redelivery_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/broker_redelivery_helper.go
index 11a1cf6e5..c69cda68c 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/broker_redelivery_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/broker_redelivery_helper.go
@@ -114,7 +114,7 @@ func brokerRedelivery(ctx context.Context, t *testing.T, creator BrokerCreatorWi
 
 	client.SendEventToAddressable(ctx, senderName, brokerName, testlib.BrokerTypeMeta, eventToSend)
 
-	allEventTracker.AssertAtLeast(1, recordevents.MatchEvent(AllOf(
+	allEventTracker.AssertAtLeast(1, recordevents.MatchKind(recordevents.EventReceived), recordevents.MatchEvent(AllOf(
 		HasSource(eventSource),
 		HasType(eventType),
 		HasData([]byte(eventBody)),
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_chain_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_chain_test_helper.go
index 572ed8667..9ad08fbb1 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_chain_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/channel_chain_test_helper.go
@@ -100,7 +100,7 @@ func ChannelChainTestHelper(
 
 		// send CloudEvent to the first channel
 		event := cloudevents.NewEvent()
-		event.SetID("dummy")
+		event.SetID("test")
 		event.SetSource(eventSource)
 		event.SetType(testlib.DefaultEventType)
 
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_defaulter_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_defaulter_test_helper.go
index 97b3f48ae..5bdc07901 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_defaulter_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/channel_defaulter_test_helper.go
@@ -145,7 +145,7 @@ func defaultChannelTestHelper(ctx context.Context, t *testing.T, client *testlib
 
 	// send CloudEvent to the channel
 	event := cloudevents.NewEvent()
-	event.SetID("dummy")
+	event.SetID("test")
 	eventSource := fmt.Sprintf("http://%s.svc/", senderName)
 	event.SetSource(eventSource)
 	event.SetType(testlib.DefaultEventType)
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_dls_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_dls_test_helper.go
index 383b78670..ca2742ad1 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_dls_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/channel_dls_test_helper.go
@@ -83,7 +83,7 @@ func ChannelDeadLetterSinkTestHelper(
 
 		// send CloudEvent to the first channel
 		event := cloudevents.NewEvent()
-		event.SetID("dummy")
+		event.SetID("test")
 		eventSource := fmt.Sprintf("http://%s.svc/", senderName)
 		event.SetSource(eventSource)
 		event.SetType(testlib.DefaultEventType)
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_event_tranformation_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_event_tranformation_test_helper.go
index 0e857bf7f..ca1718460 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_event_tranformation_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/channel_event_tranformation_test_helper.go
@@ -58,7 +58,7 @@ func EventTransformationForSubscriptionTestHelper(
 
 		// create transformation pod and service
 		eventAfterTransformation := cloudevents.NewEvent()
-		eventAfterTransformation.SetID("dummy-transformed")
+		eventAfterTransformation.SetID("test-transformed")
 		eventAfterTransformation.SetSource(eventSource)
 		eventAfterTransformation.SetType(testlib.DefaultEventType)
 		transformedEventBody := fmt.Sprintf(`{"msg":"eventBody %s"}`, uuid.New().String())
@@ -121,7 +121,7 @@ func EventTransformationForSubscriptionTestHelper(
 
 		// send  CloudEvent to the first channel
 		eventToSend := cloudevents.NewEvent()
-		eventToSend.SetID("dummy")
+		eventToSend.SetID("test")
 		eventToSend.SetSource(eventSource)
 		eventToSend.SetType(testlib.DefaultEventType)
 		eventBody := fmt.Sprintf(`{"msg":"TestEventTransformation %s"}`, uuid.New().String())
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/channel_single_event_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/channel_single_event_helper.go
index ed1d2ec2a..517c1aa2c 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/channel_single_event_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/channel_single_event_helper.go
@@ -30,7 +30,6 @@ import (
 	testlib "knative.dev/eventing/test/lib"
 	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/resources"
-	"knative.dev/eventing/test/lib/sender"
 )
 
 type SubscriptionVersion string
@@ -98,7 +97,7 @@ func SingleEventForChannelTestHelper(
 
 		// send CloudEvent to the channel
 		event := cloudevents.NewEvent()
-		event.SetID("dummy")
+		event.SetID("test")
 
 		eventSource := fmt.Sprintf("http://%s.svc/", senderName)
 		event.SetSource(eventSource)
@@ -109,14 +108,20 @@ func SingleEventForChannelTestHelper(
 			st.Fatalf("Cannot set the payload of the event: %s", err.Error())
 		}
 
-		client.SendEventToAddressable(
+		//TODO(slinkydeveloper) this is temporary, it will be replaced with SendEventToAddressable
+		uri, err := client.GetAddressableURI(channelName, &channel)
+		if err != nil {
+			client.T.Fatalf("Failed to get the URI for %+v-%s", &channel, channelName)
+		}
+
+		recordevents.DeployEventSenderOrFail(
 			ctx,
+			client,
 			senderName,
-			channelName,
-			&channel,
-			event,
-			sender.WithEncoding(encoding),
-			sender.EnableIncrementalId(),
+			uri,
+			recordevents.InputEvent(event),
+			recordevents.InputEncoding(encoding),
+			recordevents.EnableIncrementalId(),
 		)
 
 		// verify the logger service receives the event
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/parallel_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/parallel_test_helper.go
index 2ccd4d576..744bf87f1 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/parallel_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/parallel_test_helper.go
@@ -131,7 +131,7 @@ func ParallelTestHelper(
 
 			// send CloudEvent to the Parallel
 			event := cloudevents.NewEvent()
-			event.SetID("dummy")
+			event.SetID("test")
 
 			eventSource := fmt.Sprintf("http://%s.svc/", senderPodName)
 			event.SetSource(eventSource)
@@ -247,7 +247,7 @@ func ParallelV1TestHelper(
 
 			// send CloudEvent to the Parallel
 			event := cloudevents.NewEvent()
-			event.SetID("dummy")
+			event.SetID("test")
 
 			eventSource := fmt.Sprintf("http://%s.svc/", senderPodName)
 			event.SetSource(eventSource)
diff --git a/vendor/knative.dev/eventing/test/e2e/helpers/sequence_test_helper.go b/vendor/knative.dev/eventing/test/e2e/helpers/sequence_test_helper.go
index 5f0c44dc0..c3ac44d3f 100644
--- a/vendor/knative.dev/eventing/test/e2e/helpers/sequence_test_helper.go
+++ b/vendor/knative.dev/eventing/test/e2e/helpers/sequence_test_helper.go
@@ -128,7 +128,7 @@ func SequenceTestHelper(
 
 		// send CloudEvent to the Sequence
 		event := cloudevents.NewEvent()
-		event.SetID("dummy")
+		event.SetID("test")
 		eventSource := fmt.Sprintf("http://%s.svc/", senderPodName)
 		event.SetSource(eventSource)
 		event.SetType(testlib.DefaultEventType)
@@ -246,7 +246,7 @@ func SequenceV1TestHelper(
 
 		// send CloudEvent to the Sequence
 		event := cloudevents.NewEvent()
-		event.SetID("dummy")
+		event.SetID("test")
 		eventSource := fmt.Sprintf("http://%s.svc/", senderPodName)
 		event.SetSource(eventSource)
 		event.SetType(testlib.DefaultEventType)
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 e8e721fc4..69d16025a 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_info.go
@@ -29,48 +29,74 @@ const (
 	CloudEventObservedReason = "CloudEventObserved"
 )
 
+type EventKind string
+
+const (
+	EventReceived EventKind = "Received"
+	EventRejected EventKind = "Rejected"
+
+	EventSent     EventKind = "Sent"
+	EventResponse EventKind = "Response"
+)
+
 // Structure to hold information about an event seen by recordevents pod.
 type EventInfo struct {
+	Kind EventKind `json:"kind"`
+
 	// Set if the http request received by the pod couldn't be decoded or
 	// didn't pass validation
 	Error string `json:"error,omitempty"`
 	// Event received if the cloudevent received by the pod passed validation
 	Event *cloudevents.Event `json:"event,omitempty"`
-	// HTTPHeaders of the connection that delivered the event
+	// In case there is a valid event in this instance, this contains only non CE headers.
+	// Otherwise, it contains all the headers
 	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"`
-	Dropped     bool                `json:"dropped"`
+	// In case there is a valid event in this instance, this field is not filled
+	Body []byte `json:"body,omitempty"`
+
+	StatusCode int `json:"statusCode,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.
 func (ei *EventInfo) String() string {
 	var sb strings.Builder
 	sb.WriteString("-- EventInfo --\n")
+	sb.WriteString(fmt.Sprintf("--- Kind: %v ---\n", ei.Kind))
 	if ei.Event != nil {
 		sb.WriteString("--- Event ---\n")
 		sb.WriteString(ei.Event.String())
 		sb.WriteRune('\n')
-		sb.WriteRune('\n')
 	}
 	if ei.Error != "" {
 		sb.WriteString("--- Error ---\n")
 		sb.WriteString(ei.Error)
 		sb.WriteRune('\n')
+	}
+	if len(ei.HTTPHeaders) != 0 {
+		sb.WriteString("--- HTTP headers ---\n")
+		for k, v := range ei.HTTPHeaders {
+			sb.WriteString("  " + k + ": " + v[0] + "\n")
+		}
+		sb.WriteRune('\n')
+	}
+	if ei.Body != nil {
+		sb.WriteString("--- Body ---\n")
+		sb.Write(ei.Body)
 		sb.WriteRune('\n')
 	}
-	sb.WriteString("--- HTTP headers ---\n")
-	for k, v := range ei.HTTPHeaders {
-		sb.WriteString("  " + k + ": " + v[0] + "\n")
+	if ei.StatusCode != 0 {
+		sb.WriteString(fmt.Sprintf("--- Status Code: %d ---\n", ei.StatusCode))
 	}
-	sb.WriteRune('\n')
 	sb.WriteString("--- Origin: '" + ei.Origin + "' ---\n")
 	sb.WriteString("--- Observer: '" + ei.Observer + "' ---\n")
 	sb.WriteString("--- Time: " + ei.Time.String() + " ---\n")
 	sb.WriteString(fmt.Sprintf("--- Sequence: %d ---\n", ei.Sequence))
-	sb.WriteString(fmt.Sprintf("--- Dropped: %v ---\n", ei.Dropped))
+	sb.WriteString("--------------------\n")
 	return sb.String()
 }
 
@@ -86,7 +112,7 @@ type SearchedInfo struct {
 // Pretty print the SearchedInfor for error messages
 func (s *SearchedInfo) String() string {
 	var sb strings.Builder
-	sb.WriteString(fmt.Sprintf("%d events seen, last %d events (total events seen %d, events ignored %d):",
+	sb.WriteString(fmt.Sprintf("%d events seen, last %d events (total events seen %d, events ignored %d):\n",
 		s.TotalEvent, len(s.LastNEvent), s.storeEventsSeen, s.storeEventsNotMine))
 	for _, ei := range s.LastNEvent {
 		sb.WriteString(ei.String())
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/event_info_matchers.go b/vendor/knative.dev/eventing/test/lib/recordevents/event_info_matchers.go
index 199033d9c..1110d1a83 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/event_info_matchers.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_info_matchers.go
@@ -84,6 +84,16 @@ func HasAdditionalHeader(key, value string) EventInfoMatcher {
 	}
 }
 
+// MatchKind matches the kind of EventInfo
+func MatchKind(kind EventKind) EventInfoMatcher {
+	return func(info EventInfo) error {
+		if kind != info.Kind {
+			return fmt.Errorf("event kind don't match. Expected: '%s', Actual: '%s'", kind, info.Kind)
+		}
+		return nil
+	}
+}
+
 // MatchHeartBeatsImageMessage matches that the data field of the event, in the format of the heartbeats image, contains the following msg field
 func MatchHeartBeatsImageMessage(expectedMsg string) cetest.EventMatcher {
 	return cetest.AllOf(
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 4a4d16050..8c9fad009 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
@@ -81,6 +81,10 @@ func NewEventInfoStore(client *testlib.Client, podName string, podNamespace stri
 	return store, nil
 }
 
+func (ei *EventInfoStore) getDebugInfo() string {
+	return fmt.Sprintf("Pod '%s' in namespace '%s'", ei.podName, ei.podNamespace)
+}
+
 func (ei *EventInfoStore) getEventInfo() []EventInfo {
 	ei.lock.Lock()
 	defer ei.lock.Unlock()
@@ -222,9 +226,10 @@ func (ei *EventInfoStore) waitAtLeastNMatch(f EventInfoMatcher, min int) ([]Even
 		count := len(allMatch)
 		if count < min {
 			internalErr = fmt.Errorf(
-				"FAIL MATCHING: saw %d/%d matching events.\nRecent events: \n%s\nMatch errors: \n%s\n",
+				"FAIL MATCHING: saw %d/%d matching events.\n- EventInfoStore-\n%s\n- Recent events -\n%s\n- Match errors -\n%s\n",
 				count,
 				min,
+				ei.getDebugInfo(),
 				&sInfo,
 				formatErrors(matchErrs),
 			)
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go b/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go
index 62aba6113..92eb2d275 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/event_log.go
@@ -16,18 +16,44 @@ limitations under the License.
 
 package recordevents
 
+import "sync"
+
 // EventLog is the contract for an event logger to vent an event.
 type EventLog interface {
 	Vent(observed EventInfo) error
 }
 
-type EventLogs []EventLog
+// EventLogs is a struct to hold different EventLog and run them, guarded by a lock
+type EventLogs struct {
+	mutex sync.Mutex
+	logs  []EventLog
+}
+
+func NewEventLogs(logs ...EventLog) *EventLogs {
+	return &EventLogs{logs: logs}
+}
 
-func (e EventLogs) Vent(observed EventInfo) error {
-	for _, el := range e {
+func (e *EventLogs) Vent(observed EventInfo) error {
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
+	for _, el := range e.logs {
 		if err := el.Vent(observed); err != nil {
 			return err
 		}
 	}
 	return nil
 }
+
+type EventGeneratorType string
+
+const (
+	ReceiverEventGenerator EventGeneratorType = "receiver"
+	SenderEventGenerator   EventGeneratorType = "sender"
+)
+
+type EventLogType string
+
+const (
+	RecorderEventLog EventLogType = "recorder"
+	LoggerEventLog   EventLogType = "logger"
+)
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go b/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
similarity index 77%
rename from vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go
rename to vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
index 58e40a1c5..99ad8d18b 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/observer/observer.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
@@ -14,10 +14,11 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package observer
+package receiver
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"strings"
 	"sync/atomic"
@@ -34,23 +35,23 @@ import (
 	"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.
+// Receiver is the entry point for sinking events into the event log.
+type Receiver struct {
+	// Name is the name of this Receiver, used to filter if multiple observers.
 	Name string
-	// EventLogs is the list of EventLog implementors to vent observed events.
-	EventLogs recordevents.EventLogs
+	// EventLogs is the list of EventLogger implementors to vent observed events.
+	EventLogs *recordevents.EventLogs
 
 	ctx       context.Context
 	seq       uint64
+	dropSeq   uint64
 	replyFunc func(context.Context, http.ResponseWriter, recordevents.EventInfo)
 	counter   *dropevents.CounterHandler
 }
 
 type envConfig struct {
-	// ObserverName is used to identify this instance of the observer.
-	ObserverName string `envconfig:"POD_NAME" default:"observer-default" required:"true"`
+	// ReceiverName is used to identify this instance of the receiver.
+	ReceiverName string `envconfig:"POD_NAME" default:"receiver-default" required:"true"`
 
 	// Reply is used to define if the observer should reply back
 	Reply bool `envconfig:"REPLY" default:"false" required:"false"`
@@ -76,20 +77,20 @@ type envConfig struct {
 	SkipCounter uint64 `envconfig:"SKIP_COUNTER" default:"0" required:"false"`
 }
 
-func NewFromEnv(ctx context.Context, eventLogs ...recordevents.EventLog) *Observer {
+func NewFromEnv(ctx context.Context, eventLogs *recordevents.EventLogs) *Receiver {
 	var env envConfig
 	if err := envconfig.Process("", &env); err != nil {
 		logging.FromContext(ctx).Fatal("Failed to process env var", err)
 	}
 
-	logging.FromContext(ctx).Infof("Observer environment configuration: %+v", env)
+	logging.FromContext(ctx).Infof("Receiver environment configuration: %+v", env)
 
 	var replyFunc func(context.Context, http.ResponseWriter, recordevents.EventInfo)
 	if env.Reply {
-		logging.FromContext(ctx).Info("Observer will reply with an event")
+		logging.FromContext(ctx).Info("Receiver will reply with an event")
 		replyFunc = ReplyTransformerFunc(env.ReplyEventType, env.ReplyEventSource, env.ReplyEventData, env.ReplyAppendData)
 	} else {
-		logging.FromContext(ctx).Info("Observer won't reply with an event")
+		logging.FromContext(ctx).Info("Receiver won't reply with an event")
 		replyFunc = NoOpReply
 	}
 	var counter *dropevents.CounterHandler
@@ -105,8 +106,8 @@ func NewFromEnv(ctx context.Context, eventLogs ...recordevents.EventLog) *Observ
 		}
 	}
 
-	return &Observer{
-		Name:      env.ObserverName,
+	return &Receiver{
+		Name:      env.ReceiverName,
 		EventLogs: eventLogs,
 		ctx:       ctx,
 		replyFunc: replyFunc,
@@ -116,7 +117,7 @@ func NewFromEnv(ctx context.Context, eventLogs ...recordevents.EventLog) *Observ
 
 // 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 {
+func (o *Receiver) Start(ctx context.Context, handlerFuncs ...func(handler http.Handler) http.Handler) error {
 	var handler http.Handler = o
 
 	for _, dec := range handlerFuncs {
@@ -125,20 +126,23 @@ func (o *Observer) Start(ctx context.Context, handlerFuncs ...func(handler http.
 
 	server := &http.Server{Addr: ":8080", Handler: handler}
 
+	var err error
 	go func() {
-		if err := server.ListenAndServe(); err != nil {
-			logging.FromContext(ctx).Fatal("Error while starting the HTTP server", err)
-		}
+		err = server.ListenAndServe()
 	}()
 
 	<-ctx.Done()
 
+	if err != nil {
+		return fmt.Errorf("error while starting the HTTP server: %w", err)
+	}
+
 	logging.FromContext(ctx).Info("Closing the HTTP server")
 
 	return server.Close()
 }
 
-func (o *Observer) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+func (o *Receiver) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	m := cloudeventshttp.NewMessageFromHttpRequest(request)
 	defer m.Finish(nil)
 
@@ -160,6 +164,15 @@ func (o *Observer) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 	}
 
 	shouldSkip := o.counter.Skip()
+	var s uint64
+	var kind recordevents.EventKind
+	if shouldSkip {
+		kind = recordevents.EventRejected
+		s = atomic.AddUint64(&o.dropSeq, 1)
+	} else {
+		kind = recordevents.EventReceived
+		s = atomic.AddUint64(&o.seq, 1)
+	}
 
 	eventInfo := recordevents.EventInfo{
 		Error:       eventErrStr,
@@ -168,19 +181,11 @@ func (o *Observer) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 		Origin:      request.RemoteAddr,
 		Observer:    o.Name,
 		Time:        time.Now(),
-		Sequence:    atomic.AddUint64(&o.seq, 1),
-		Dropped:     shouldSkip,
-	}
-
-	// We still want to emit the event to make it easier to see what we had oberved, but
-	// we want to transform it a little bit before emitting so that it does not count
-	// as the real event that we want to emit.
-	if shouldSkip {
-		eventInfo.Event.SetType("dropped-" + eventInfo.Event.Type())
+		Sequence:    s,
+		Kind:        kind,
 	}
 
-	err := o.EventLogs.Vent(eventInfo)
-	if err != nil {
+	if err := o.EventLogs.Vent(eventInfo); err != nil {
 		logging.FromContext(o.ctx).Fatalw("Error while venting the recorded event", zap.Error(err))
 	}
 
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/observer/reply.go b/vendor/knative.dev/eventing/test/lib/recordevents/receiver/reply.go
similarity index 99%
rename from vendor/knative.dev/eventing/test/lib/recordevents/observer/reply.go
rename to vendor/knative.dev/eventing/test/lib/recordevents/receiver/reply.go
index 296655f4a..229ed8c0b 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/observer/reply.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/receiver/reply.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package observer
+package receiver
 
 import (
 	"context"
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
index 50c1acc3f..fb51aaddf 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/constructor.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/constructor.go
@@ -18,7 +18,6 @@ package recorder_vent
 
 import (
 	"context"
-	"log"
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
@@ -48,7 +47,7 @@ const (
 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).Fatal("Failed to process env var: ", err)
 	}
 
 	logging.FromContext(ctx).Infof("Recorder vent environment configuration: %+v", env)
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
index 117ac2df1..b173d64af 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/recorder.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/recorder_vent/recorder.go
@@ -54,7 +54,7 @@ func (r *recorder) Vent(observed recordevents.EventInfo) error {
 	// - https://github.com/kubernetes/kubernetes/blob/master/pkg/apis/core/validation/events.go#L122
 	event := &corev1.Event{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      fmt.Sprintf("%v.%d", r.ref.Name, observed.Sequence),
+			Name:      fmt.Sprintf("%v.%v.%d", r.ref.Name, observed.Kind, observed.Sequence),
 			Namespace: r.namespace,
 		},
 		InvolvedObject: *r.ref,
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/resources.go b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
index 26b7baccb..38bd7f228 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
@@ -18,7 +18,10 @@ package recordevents
 
 import (
 	"context"
+	"encoding/json"
+	"strings"
 
+	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
@@ -32,63 +35,103 @@ import (
 
 type EventRecordOption = func(*corev1.Pod, *testlib.Client) error
 
-// EchoEvent is an option to let the recordevents reply with the received event
-func EchoEvent(pod *corev1.Pod, client *testlib.Client) error {
-	pod.Spec.Containers[0].Env = append(
-		pod.Spec.Containers[0].Env,
-		corev1.EnvVar{Name: "REPLY", Value: "true"},
-	)
+func noop(pod *corev1.Pod, client *testlib.Client) error {
 	return nil
 }
 
-var _ EventRecordOption = EchoEvent
-
-// ReplyWithTransformedEvent is an option to let the recordevents reply with the transformed event
-func ReplyWithTransformedEvent(replyEventType string, replyEventSource string, replyEventData string) EventRecordOption {
+func compose(options ...EventRecordOption) EventRecordOption {
 	return func(pod *corev1.Pod, client *testlib.Client) error {
-		pod.Spec.Containers[0].Env = append(
-			pod.Spec.Containers[0].Env,
-			corev1.EnvVar{Name: "REPLY", Value: "true"},
-		)
-		if replyEventType != "" {
-			pod.Spec.Containers[0].Env = append(
-				pod.Spec.Containers[0].Env,
-				corev1.EnvVar{Name: "REPLY_EVENT_TYPE", Value: replyEventType},
-			)
+		for _, opt := range options {
+			if err := opt(pod, client); err != nil {
+				return err
+			}
 		}
-		if replyEventSource != "" {
-			pod.Spec.Containers[0].Env = append(
-				pod.Spec.Containers[0].Env,
-				corev1.EnvVar{Name: "REPLY_EVENT_SOURCE", Value: replyEventSource},
-			)
-		}
-		if replyEventData != "" {
+		return nil
+	}
+}
+
+func envOptionalOpt(key, value string) EventRecordOption {
+	if value != "" {
+		return func(pod *corev1.Pod, client *testlib.Client) error {
 			pod.Spec.Containers[0].Env = append(
 				pod.Spec.Containers[0].Env,
-				corev1.EnvVar{Name: "REPLY_EVENT_DATA", Value: replyEventData},
+				corev1.EnvVar{Name: key, Value: value},
 			)
+			return nil
 		}
-
-		return nil
+	} else {
+		return noop
 	}
 }
 
-// ReplyWithAppendedData is an option to let the recordevents reply with the transformed event with appended data
-func ReplyWithAppendedData(appendData string) EventRecordOption {
+func envOption(key, value string) EventRecordOption {
 	return func(pod *corev1.Pod, client *testlib.Client) error {
 		pod.Spec.Containers[0].Env = append(
 			pod.Spec.Containers[0].Env,
-			corev1.EnvVar{Name: "REPLY", Value: "true"},
+			corev1.EnvVar{Name: key, Value: value},
 		)
-		if appendData != "" {
-			pod.Spec.Containers[0].Env = append(
-				pod.Spec.Containers[0].Env,
-				corev1.EnvVar{Name: "REPLY_APPEND_DATA", Value: appendData},
-			)
+		return nil
+	}
+}
+
+// EchoEvent is an option to let the recordevents reply with the received event
+var EchoEvent EventRecordOption = envOption("REPLY", "true")
+
+// ReplyWithTransformedEvent is an option to let the recordevents reply with the transformed event
+func ReplyWithTransformedEvent(replyEventType string, replyEventSource string, replyEventData string) EventRecordOption {
+	return compose(
+		envOption("REPLY", "true"),
+		envOptionalOpt("REPLY_EVENT_TYPE", replyEventType),
+		envOptionalOpt("REPLY_EVENT_SOURCE", replyEventSource),
+		envOptionalOpt("REPLY_EVENT_DATA", replyEventData),
+	)
+}
+
+// ReplyWithAppendedData is an option to let the recordevents reply with the transformed event with appended data
+func ReplyWithAppendedData(appendData string) EventRecordOption {
+	return compose(
+		envOption("REPLY", "true"),
+		envOptionalOpt("REPLY_APPEND_DATA", appendData),
+	)
+}
+
+// InputEvent is an option to provide the event to send when deploying the event sender
+func InputEvent(event cloudevents.Event) EventRecordOption {
+	encodedEvent, err := json.Marshal(event)
+	if err != nil {
+		return func(pod *corev1.Pod, client *testlib.Client) error {
+			return err
 		}
+	}
+	return envOption("INPUT_EVENT", string(encodedEvent))
+}
 
-		return nil
+// AddTracing adds tracing headers when sending events.
+func AddTracing() EventRecordOption {
+	return envOption("ADD_TRACING", "true")
+}
+
+// EnableIncrementalId creates a new incremental id for each sent event.
+func EnableIncrementalId() EventRecordOption {
+	return envOption("INCREMENTAL_ID", "true")
+}
+
+// InputEncoding forces the encoding of the event for each sent event.
+func InputEncoding(encoding cloudevents.Encoding) EventRecordOption {
+	return envOption("EVENT_ENCODING", encoding.String())
+}
+
+// InputHeaders adds the following headers to the sent requests.
+func InputHeaders(headers map[string]string) EventRecordOption {
+	return envOption("INPUT_HEADERS", serializeHeaders(headers))
+}
+
+func serializeHeaders(headers map[string]string) string {
+	kv := make([]string, 0, len(headers))
+	for k, v := range headers {
+		kv = append(kv, k+":"+v)
 	}
+	return strings.Join(kv, ",")
 }
 
 // DeployEventRecordOrFail deploys the recordevents image with necessary sa, roles, rb to execute the image
@@ -108,8 +151,14 @@ func DeployEventRecordOrFail(ctx context.Context, client *testlib.Client, name s
 	))
 	client.CreateRoleBindingOrFail(name, "Role", name, name, client.Namespace)
 
+	options = append(
+		options,
+		testlib.WithService(name),
+		envOption("EVENT_GENERATORS", "receiver"),
+	)
+
 	eventRecordPod := recordEventsPod("recordevents", name, name)
-	client.CreatePodOrFail(eventRecordPod, append(options, testlib.WithService(name))...)
+	client.CreatePodOrFail(eventRecordPod, options...)
 	err := pkgtest.WaitForPodRunning(ctx, client.Kube, name, client.Namespace)
 	if err != nil {
 		client.T.Fatalf("Failed to start the recordevent pod '%s': %v", name, errors.WithStack(err))
@@ -118,6 +167,38 @@ func DeployEventRecordOrFail(ctx context.Context, client *testlib.Client, name s
 	return eventRecordPod
 }
 
+// DeployEventSenderOrFail deploys the recordevents image with necessary sa, roles, rb to execute the image
+func DeployEventSenderOrFail(ctx context.Context, client *testlib.Client, name string, sink 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)
+
+	options = append(
+		options,
+		envOption("EVENT_GENERATORS", "sender"),
+		envOption("SINK", sink),
+	)
+
+	eventRecordPod := recordEventsPod("recordevents", name, name)
+	client.CreatePodOrFail(eventRecordPod, options...)
+	err := pkgtest.WaitForPodRunning(ctx, client.Kube, name, client.Namespace)
+	if err != nil {
+		client.T.Fatalf("Failed to start the recordevent pod '%s': %v", name, errors.WithStack(err))
+	}
+	return eventRecordPod
+}
+
 func recordEventsPod(imageName string, name string, serviceAccountName string) *corev1.Pod {
 	return &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -139,6 +220,9 @@ func recordEventsPod(imageName string, name string, serviceAccountName string) *
 					ValueFrom: &corev1.EnvVarSource{
 						FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"},
 					},
+				}, {
+					Name:  "EVENT_LOGS",
+					Value: "recorder,logger",
 				}},
 			}},
 			ServiceAccountName: serviceAccountName,
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/sender/sender.go b/vendor/knative.dev/eventing/test/lib/recordevents/sender/sender.go
new file mode 100644
index 000000000..88b9b8106
--- /dev/null
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/sender/sender.go
@@ -0,0 +1,258 @@
+/*
+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 sender
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	nethttp "net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/cloudevents/sdk-go/v2/binding"
+	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
+	"github.com/kelseyhightower/envconfig"
+	"go.opencensus.io/plugin/ochttp"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/tracing/propagation/tracecontextb3"
+
+	"knative.dev/eventing/test/lib/recordevents"
+)
+
+type envConfig struct {
+	SenderName string `envconfig:"POD_NAME" default:"sender-default" required:"true"`
+
+	// Sink url for the message destination
+	Sink string `envconfig:"SINK" required:"true"`
+
+	// InputEvent json encoded
+	InputEvent string `envconfig:"INPUT_EVENT" required:"false"`
+
+	// InputHeaders to send (this overrides any event provided input)
+	InputHeaders map[string]string `envconfig:"INPUT_HEADERS" required:"false"`
+
+	// InputBody to send (this overrides any event provided input)
+	InputBody string `envconfig:"INPUT_BODY" required:"false"`
+
+	// The encoding of the cloud event: [binary, structured].
+	EventEncoding string `envconfig:"EVENT_ENCODING" default:"binary" required:"false"`
+
+	// The number of seconds between messages.
+	Period int `envconfig:"PERIOD" default:"5" required:"false"`
+
+	// The number of seconds to wait before starting sending the first message
+	Delay int `envconfig:"DELAY" default:"5" required:"false"`
+
+	// The number of messages to attempt to send. 0 for unlimited.
+	MaxMessages int `envconfig:"MAX_MESSAGES" default:"1" required:"false"`
+
+	// Should tracing be added to events sent.
+	AddTracing bool `envconfig:"ADD_TRACING" default:"false" required:"false"`
+
+	// Should add extension 'sequence' identifying the sequence number.
+	AddSequence bool `envconfig:"ADD_SEQUENCE" default:"false" required:"false"`
+
+	// Override the event id with an incremental id.
+	IncrementalId bool `envconfig:"INCREMENTAL_ID" default:"false" required:"false"`
+}
+
+func Start(ctx context.Context, logs *recordevents.EventLogs) error {
+	var env envConfig
+	if err := envconfig.Process("", &env); err != nil {
+		return fmt.Errorf("failed to process env var. %w", err)
+	}
+
+	logging.FromContext(ctx).Infof("Sender environment configuration: %+v", env)
+
+	if env.InputEvent == "" && env.InputBody == "" && len(env.InputHeaders) == 0 {
+		return fmt.Errorf("input values not provided")
+	}
+
+	flag.Parse()
+	period := time.Duration(env.Period) * time.Second
+	delay := time.Duration(env.Delay) * time.Second
+
+	if delay > 0 {
+		logging.FromContext(ctx).Info("will sleep for ", delay)
+		time.Sleep(delay)
+		logging.FromContext(ctx).Info("awake, continuing")
+	}
+
+	switch env.EventEncoding {
+	case "binary":
+		ctx = cloudevents.WithEncodingBinary(ctx)
+	case "structured":
+		ctx = cloudevents.WithEncodingStructured(ctx)
+	default:
+		return fmt.Errorf("unsupported encoding option: %q", env.EventEncoding)
+	}
+
+	httpClient := &nethttp.Client{}
+	if env.AddTracing {
+		httpClient.Transport = &ochttp.Transport{
+			Base:        nethttp.DefaultTransport,
+			Propagation: tracecontextb3.TraceContextEgress,
+		}
+	}
+
+	var baseEvent *cloudevents.Event
+	if env.InputEvent != "" {
+		if err := json.Unmarshal([]byte(env.InputEvent), &baseEvent); err != nil {
+			return fmt.Errorf("unable to unmarshal the event from json: %w", err)
+		}
+	}
+
+	sequence := 0
+
+	ticker := time.NewTicker(period)
+	for {
+		req, err := nethttp.NewRequest(nethttp.MethodPost, env.Sink, nil)
+		if err != nil {
+			logging.FromContext(ctx).Error("Cannot create the request: ", err)
+			return err
+		}
+
+		var event *cloudevents.Event
+		if baseEvent != nil {
+			e := baseEvent.Clone()
+			event = &e
+
+			sequence++
+			if env.AddSequence {
+				event.SetExtension("sequence", sequence)
+			}
+			if env.IncrementalId {
+				event.SetID(strconv.Itoa(sequence))
+			}
+
+			logging.FromContext(ctx).Info("I'm going to send\n", event)
+
+			err := cehttp.WriteRequest(ctx, binding.ToMessage(event), req)
+			if err != nil {
+				logging.FromContext(ctx).Error("Cannot write the event: ", err)
+				return err
+			}
+		}
+
+		if len(env.InputHeaders) != 0 {
+			for k, v := range env.InputHeaders {
+				req.Header.Add(k, v)
+			}
+		}
+
+		if env.InputBody != "" {
+			req.Body = ioutil.NopCloser(bytes.NewReader([]byte(env.InputBody)))
+		}
+
+		res, err := httpClient.Do(req)
+
+		if err != nil {
+			// Publish error
+			if err := logs.Vent(recordevents.EventInfo{
+				Kind:     recordevents.EventSent,
+				Error:    err.Error(),
+				Origin:   env.SenderName,
+				Observer: env.SenderName,
+				Time:     time.Now(),
+				Sequence: uint64(sequence),
+			}); err != nil {
+				return fmt.Errorf("cannot forward event info: %w", err)
+			}
+		} else {
+			sentEventInfo := recordevents.EventInfo{
+				Kind:     recordevents.EventSent,
+				Event:    event,
+				Origin:   env.SenderName,
+				Observer: env.SenderName,
+				Time:     time.Now(),
+				Sequence: uint64(sequence),
+			}
+
+			sentHeaders := make(nethttp.Header)
+			for k, v := range req.Header {
+				if !strings.HasPrefix(k, "Ce-") {
+					sentHeaders[k] = v
+				}
+			}
+			sentEventInfo.HTTPHeaders = sentHeaders
+
+			if env.InputBody != "" {
+				sentEventInfo.Body = []byte(env.InputBody)
+			}
+
+			// Publish sent event info
+			if err := logs.Vent(sentEventInfo); err != nil {
+				return fmt.Errorf("cannot forward event info: %w", err)
+			}
+
+			// Now let's figure out what's inside the response
+			responseMessage := cehttp.NewMessageFromHttpResponse(res)
+
+			responseInfo := recordevents.EventInfo{
+				Kind:        recordevents.EventResponse,
+				HTTPHeaders: res.Header,
+				Origin:      env.Sink,
+				Observer:    env.SenderName,
+				Time:        time.Now(),
+				Sequence:    uint64(sequence),
+				StatusCode:  res.StatusCode,
+			}
+			if responseMessage.ReadEncoding() == binding.EncodingUnknown {
+				body, err := ioutil.ReadAll(res.Body)
+
+				if err != nil {
+					responseInfo.Error = err.Error()
+				} else {
+					responseInfo.Body = body
+				}
+			} else {
+				responseEvent, err := binding.ToEvent(ctx, responseMessage)
+				if err != nil {
+					responseInfo.Error = err.Error()
+				} else {
+					responseInfo.Event = responseEvent
+				}
+			}
+
+			// Vent the response info
+			if err := logs.Vent(responseInfo); err != nil {
+				return fmt.Errorf("cannot forward event info: %w", err)
+			}
+		}
+
+		// Wait for next tick
+		<-ticker.C
+		// Only send a limited number of messages.
+		if env.MaxMessages != 0 && env.MaxMessages == sequence {
+			return nil
+		}
+
+		// Check if ctx is done before the next loop
+		select {
+		case <-ctx.Done():
+			logging.FromContext(ctx).Infof("Canceled sending messages because context was closed")
+			return nil
+		default:
+		}
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/lib/sender/resources.go b/vendor/knative.dev/eventing/test/lib/sender/resources.go
index 8ca13aa3d..ff573c589 100644
--- a/vendor/knative.dev/eventing/test/lib/sender/resources.go
+++ b/vendor/knative.dev/eventing/test/lib/sender/resources.go
@@ -27,6 +27,7 @@ import (
 )
 
 // EnableTracing enables tracing in sender pod
+// Deprecated: Now you should use recordevents.DeployEventSenderOrFail to send events
 func EnableTracing() func(*corev1.Pod) {
 	return func(pod *corev1.Pod) {
 		pod.Spec.Containers[0].Args = append(
@@ -38,6 +39,7 @@ func EnableTracing() func(*corev1.Pod) {
 }
 
 // EnableIncrementalId creates a new incremental id for each event sent from the sender pod. Supported only by event-sender
+// Deprecated: Now you should use recordevents.DeployEventSenderOrFail to send events
 func EnableIncrementalId() func(*corev1.Pod) {
 	return func(pod *corev1.Pod) {
 		pod.Spec.Containers[0].Args = append(
@@ -49,6 +51,7 @@ func EnableIncrementalId() func(*corev1.Pod) {
 }
 
 // WithEncoding forces the encoding of the event to send from the sender pod. Supported only by event-sender
+// Deprecated: Now you should use recordevents.DeployEventSenderOrFail to send events
 func WithEncoding(encoding cloudevents.Encoding) func(*corev1.Pod) {
 	return func(pod *corev1.Pod) {
 		pod.Spec.Containers[0].Args = append(
@@ -60,6 +63,7 @@ func WithEncoding(encoding cloudevents.Encoding) func(*corev1.Pod) {
 }
 
 // WithResponseSink sends the response information as CloudEvent to another sink
+// Deprecated: Now you should use recordevents.DeployEventSenderOrFail to send events and start an EventInfoStore on it to assert the responses
 func WithResponseSink(responseSink string) func(*corev1.Pod) {
 	return func(pod *corev1.Pod) {
 		pod.Spec.Containers[0].Args = append(
@@ -71,6 +75,7 @@ func WithResponseSink(responseSink string) func(*corev1.Pod) {
 }
 
 // WithEncoding forces the encoding of the event to send from the sender pod. Supported only by event-sender
+// Deprecated: Now you should use recordevents.DeployEventSenderOrFail to send events
 func WithAdditionalHeaders(headers map[string]string) func(*corev1.Pod) {
 	return func(pod *corev1.Pod) {
 		pod.Spec.Containers[0].Args = append(
@@ -82,6 +87,7 @@ func WithAdditionalHeaders(headers map[string]string) func(*corev1.Pod) {
 }
 
 // EventSenderPod creates a Pod that sends events to the given address.
+// Deprecated: Now you should use recordevents.DeployEventSenderOrFail to send events
 func EventSenderPod(imageName string, name string, sink string, event cloudevents.Event, options ...func(*corev1.Pod)) (*corev1.Pod, error) {
 	encodedEvent, err := json.Marshal(event)
 	if err != nil {
@@ -119,6 +125,7 @@ func EventSenderPod(imageName string, name string, sink string, event cloudevent
 }
 
 // WithMethod configures the method used to send the http request. Supported only by request-sender
+// Deprecated: Now you should use recordevents.DeployEventSenderOrFail to send requests
 func WithMethod(method string) func(*corev1.Pod) {
 	return func(pod *corev1.Pod) {
 		pod.Spec.Containers[0].Args = append(
@@ -130,6 +137,7 @@ func WithMethod(method string) func(*corev1.Pod) {
 }
 
 // EventSenderPod creates a Pod that sends http requests to the given address.
+// Deprecated: Now you should use recordevents.DeployEventSenderOrFail to send requests
 func RequestSenderPod(imageName string, name string, sink string, headers map[string]string, body string, options ...func(*corev1.Pod)) (*corev1.Pod, error) {
 	args := []string{
 		"-sink",
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 cd95e9563..ca718ddc2 100644
--- a/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
+++ b/vendor/knative.dev/eventing/test/test_images/recordevents/main.go
@@ -17,20 +17,30 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"log"
 
+	"github.com/kelseyhightower/envconfig"
+	"golang.org/x/sync/errgroup"
 	"k8s.io/client-go/rest"
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/logging"
 	_ "knative.dev/pkg/system/testing"
 
 	"knative.dev/eventing/pkg/kncloudevents"
+	"knative.dev/eventing/test/lib/recordevents"
 	"knative.dev/eventing/test/lib/recordevents/logger_vent"
-	"knative.dev/eventing/test/lib/recordevents/observer"
+	"knative.dev/eventing/test/lib/recordevents/receiver"
 	"knative.dev/eventing/test/lib/recordevents/recorder_vent"
+	"knative.dev/eventing/test/lib/recordevents/sender"
 	"knative.dev/eventing/test/test_images"
 )
 
+type envConfig struct {
+	EventGenerators []string `envconfig:"EVENT_GENERATORS" required:"true"`
+	EventLogs       []string `envconfig:"EVENT_LOGS" required:"true"`
+}
+
 func main() {
 	cfg, err := rest.InClusterConfig()
 	if err != nil {
@@ -44,16 +54,51 @@ func main() {
 		logging.FromContext(ctx).Fatal("Unable to setup trace publishing", err)
 	}
 
-	obs := observer.NewFromEnv(ctx,
-		logger_vent.Logger(logging.FromContext(ctx).Infof),
-		recorder_vent.NewFromEnv(ctx),
-	)
+	var env envConfig
+	if err := envconfig.Process("", &env); err != nil {
+		logging.FromContext(ctx).Fatal("Failed to process env var", err)
+	}
 
-	err = obs.Start(ctx, kncloudevents.CreateHandler)
+	eventLogs := createEventLogs(ctx, env.EventLogs)
+	err = startEventGenerators(ctx, env.EventGenerators, eventLogs)
 
 	if err != nil {
-		logging.FromContext(ctx).Fatal("Error during start", err)
+		logging.FromContext(ctx).Fatal("Error during start: ", err)
 	}
 
 	logging.FromContext(ctx).Info("Closing the recordevents process")
 }
+
+func createEventLogs(ctx context.Context, logTypes []string) *recordevents.EventLogs {
+	var l []recordevents.EventLog
+	for _, logType := range logTypes {
+		switch recordevents.EventLogType(logType) {
+		case recordevents.RecorderEventLog:
+			l = append(l, recorder_vent.NewFromEnv(ctx))
+		case recordevents.LoggerEventLog:
+			l = append(l, logger_vent.Logger(logging.FromContext(ctx).Named("event logger").Infof))
+		default:
+			logging.FromContext(ctx).Fatal("Cannot recognize event log type: ", logType)
+		}
+	}
+	return recordevents.NewEventLogs(l...)
+}
+
+func startEventGenerators(ctx context.Context, genTypes []string, eventLogs *recordevents.EventLogs) error {
+	errs, _ := errgroup.WithContext(ctx)
+	for _, genType := range genTypes {
+		switch recordevents.EventGeneratorType(genType) {
+		case recordevents.ReceiverEventGenerator:
+			errs.Go(func() error {
+				return receiver.NewFromEnv(ctx, eventLogs).Start(ctx, kncloudevents.CreateHandler)
+			})
+		case recordevents.SenderEventGenerator:
+			errs.Go(func() error {
+				return sender.Start(ctx, eventLogs)
+			})
+		default:
+			logging.FromContext(ctx).Fatal("Cannot recognize event generator type: ", genType)
+		}
+	}
+	return errs.Wait()
+}
diff --git a/vendor/knative.dev/pkg/reconciler/testing/tracker.go b/vendor/knative.dev/pkg/reconciler/testing/tracker.go
index 35a15f896..e586097d9 100644
--- a/vendor/knative.dev/pkg/reconciler/testing/tracker.go
+++ b/vendor/knative.dev/pkg/reconciler/testing/tracker.go
@@ -41,6 +41,31 @@ var _ tracker.Interface = (*FakeTracker)(nil)
 // OnChanged implements OnChanged.
 func (*FakeTracker) OnChanged(interface{}) {}
 
+// GetObservers implements GetObservers.
+func (n *FakeTracker) GetObservers(obj interface{}) []types.NamespacedName {
+	item, err := kmeta.DeletionHandlingAccessor(obj)
+	if err != nil {
+		return nil
+	}
+
+	or := kmeta.ObjectReference(item)
+	ref := tracker.Reference{
+		APIVersion: or.APIVersion,
+		Kind:       or.Kind,
+		Namespace:  or.Namespace,
+		Name:       or.Name,
+	}
+
+	n.Lock()
+	defer n.Unlock()
+
+	keys := make([]types.NamespacedName, 0, len(n.references[ref]))
+	for key := range n.references[ref] {
+		keys = append(keys, key)
+	}
+	return keys
+}
+
 // OnDeletedObserver implements OnDeletedObserver.
 func (n *FakeTracker) OnDeletedObserver(obj interface{}) {
 	item, err := kmeta.DeletionHandlingAccessor(obj)
diff --git a/vendor/knative.dev/pkg/test/logging/.gitattributes b/vendor/knative.dev/pkg/test/logging/.gitattributes
deleted file mode 100644
index 19c29c12f..000000000
--- a/vendor/knative.dev/pkg/test/logging/.gitattributes
+++ /dev/null
@@ -1,1 +0,0 @@
-memory_encoder.go coverage-excluded=true
diff --git a/vendor/knative.dev/pkg/test/logging/doc.go b/vendor/knative.dev/pkg/test/logging/doc.go
deleted file mode 100644
index e76b9cbe9..000000000
--- a/vendor/knative.dev/pkg/test/logging/doc.go
+++ /dev/null
@@ -1,49 +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
-
-    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 logging assists setting up test logging and using leveled logging in tests.
-
-The TLogger is designed to assist the test writer in creating more useful tests and
-collecting log data in multiple streams, optimizing for human readability in one and
-machine readability in another. It's designed to mimic the testing.T object rather closely and
-use Zap logging semantics, both things already in use in Knative, to minimize the time developers
-need to spend learning the tool.
-
-Inspired by and uses go-logr.
-
-Advantages
-
-The TLogger enhances test design through subtle nudges and affordances:
-
-* It encourages only logging with .V(), giving the writer a nudge to think about how important it is,
-but without requiring them to fit it in a narrowly-defined category.
-
-* Reduces boilerplate of carrying around context for errors in several different variables,
-using .WithValues(), which results in more consistent and reusable code across the tests.
-
-Porting
-
-To port code from using testing.T to logging.TLogger, the interfaces knative.dev/pkg/test.T and
-knative.dev/pkg/test.TLegacy have been created. All library functions should be refactored to use
-one interface and all .Log() calls rewritten to use structured format, which works with testing and
-TLogger. If a library function needs test functions not available even in test.TLegacy,
-it's probably badly written.
-
-Then any test can be incrementally rewritten to use TLogger, as it coexists with testing.T without issue.
-
-*/
-package logging
diff --git a/vendor/knative.dev/pkg/test/logging/error.go b/vendor/knative.dev/pkg/test/logging/error.go
deleted file mode 100644
index 774a60612..000000000
--- a/vendor/knative.dev/pkg/test/logging/error.go
+++ /dev/null
@@ -1,90 +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
-
-    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 logging
-
-import (
-	"fmt"
-
-	"github.com/davecgh/go-spew/spew"
-)
-
-// StructuredError is an error which can hold arbitrary key-value arguments.
-//
-// TODO(coryrc): The Structured Error is experimental and likely to be removed, but is currently in use in a refactored test.
-type StructuredError interface {
-	error
-	GetValues() []interface{}
-	WithValues(...interface{}) StructuredError
-	DisableValuePrinting()
-	EnableValuePrinting()
-}
-
-type structuredError struct {
-	msg           string
-	keysAndValues []interface{}
-	print         bool
-}
-
-func keysAndValuesToSpewedMap(args ...interface{}) map[string]string {
-	m := make(map[string]string, len(args)/2)
-	for i := 0; i < len(args); i += 2 {
-		key, val := args[i], args[i+1]
-		if keyStr, ok := key.(string); ok {
-			m[keyStr] = spew.Sdump(val)
-		}
-	}
-	return m
-}
-
-// Error implements `error` interface
-func (e structuredError) Error() string {
-	// TODO(coryrc): accept zap.Field entries?
-	if e.print {
-		// %v for fmt.Sprintf does print keys sorted
-		return fmt.Sprintf("Error: %s\nContext:\n%v", e.msg, keysAndValuesToSpewedMap(e.keysAndValues...))
-	}
-	return e.msg
-}
-
-// GetValues gives you the structured key values in a plist
-func (e structuredError) GetValues() []interface{} {
-	return e.keysAndValues
-}
-
-// DisableValuePrinting disables printing out the keys and values from the Error() method
-func (e *structuredError) DisableValuePrinting() {
-	e.print = false
-}
-
-// EnableValuePrinting enables printing out the keys and values from the Error() method
-func (e *structuredError) EnableValuePrinting() {
-	e.print = true
-}
-
-// Create a StructuredError. Gives a little better logging when given to a TLogger.
-// This may prove to not be useful if users use the logger's WithValues() better.
-func Error(msg string, keysAndValues ...interface{}) error {
-	return &structuredError{msg, keysAndValues, true}
-}
-
-// WithValues operates just like TLogger's WithValues but stores them in the error object.
-func (e *structuredError) WithValues(keysAndValues ...interface{}) StructuredError {
-	newKAV := make([]interface{}, 0, len(keysAndValues)+len(e.keysAndValues))
-	newKAV = append(newKAV, e.keysAndValues...)
-	newKAV = append(newKAV, keysAndValues...)
-	return &structuredError{e.msg, newKAV, e.print}
-}
diff --git a/vendor/knative.dev/pkg/test/logging/logger.go b/vendor/knative.dev/pkg/test/logging/logger.go
deleted file mode 100644
index b8f3e4a71..000000000
--- a/vendor/knative.dev/pkg/test/logging/logger.go
+++ /dev/null
@@ -1,76 +0,0 @@
-// Copyright 2020 The Knative Authors
-// Copyright (c) 2017 Uber Technologies, Inc.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-//
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-
-// Copying testingWriter from zaptest and allowing it to be disabled
-
-package logging
-
-import (
-	"bytes"
-	"errors"
-	"testing"
-)
-
-// testingWriter is a WriteSyncer that writes to the given testing.TB.
-type testingWriter struct {
-	t *testing.T
-
-	// If true, the test will be marked as failed if this testingWriter is
-	// ever used.
-	markFailed bool
-}
-
-func newTestingWriter(t *testing.T) testingWriter {
-	return testingWriter{t: t}
-}
-
-// WithMarkFailed returns a copy of this testingWriter with markFailed set to
-// the provided value.
-func (w testingWriter) WithMarkFailed(v bool) testingWriter {
-	w.markFailed = v
-	return w
-}
-
-func (w testingWriter) Write(p []byte) (n int, err error) {
-	if w.t == nil {
-		return 0, errors.New("Write to buffer after test function completed")
-	}
-	n = len(p)
-
-	// Strip trailing newline because t.Log always adds one.
-	p = bytes.TrimRight(p, "\n")
-
-	// Note: t.Log is safe for concurrent use.
-	w.t.Logf("%s", p)
-	if w.markFailed {
-		w.t.Fail()
-	}
-
-	return n, nil
-}
-
-func (w testingWriter) Sync() error {
-	return nil
-}
-
-func (w *testingWriter) Disable() {
-	w.t = nil
-}
diff --git a/vendor/knative.dev/pkg/test/logging/memory_encoder.go b/vendor/knative.dev/pkg/test/logging/memory_encoder.go
deleted file mode 100644
index 248046b62..000000000
--- a/vendor/knative.dev/pkg/test/logging/memory_encoder.go
+++ /dev/null
@@ -1,74 +0,0 @@
-// Copyright 2020 The Knative Authors
-// Copyright (c) 2016 Uber Technologies, Inc.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-//
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-
-package logging
-
-import (
-	"time"
-
-	"go.uber.org/zap/zapcore"
-)
-
-// sliceArrayEncoder is an ArrayEncoder backed by a simple []interface{}. Like
-// the MapObjectEncoder, it's not designed for production use.
-type sliceArrayEncoder struct {
-	elems []interface{}
-}
-
-func (s *sliceArrayEncoder) AppendArray(v zapcore.ArrayMarshaler) error {
-	enc := &sliceArrayEncoder{}
-	err := v.MarshalLogArray(enc)
-	s.elems = append(s.elems, enc.elems)
-	return err
-}
-
-func (s *sliceArrayEncoder) AppendObject(v zapcore.ObjectMarshaler) error {
-	m := zapcore.NewMapObjectEncoder()
-	err := v.MarshalLogObject(m)
-	s.elems = append(s.elems, m.Fields)
-	return err
-}
-
-func (s *sliceArrayEncoder) AppendReflected(v interface{}) error {
-	s.elems = append(s.elems, v)
-	return nil
-}
-
-func (s *sliceArrayEncoder) AppendBool(v bool)              { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendByteString(v []byte)      { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendComplex128(v complex128)  { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendComplex64(v complex64)    { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendDuration(v time.Duration) { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendFloat64(v float64)        { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendFloat32(v float32)        { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendInt(v int)                { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendInt64(v int64)            { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendInt32(v int32)            { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendInt16(v int16)            { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendInt8(v int8)              { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendString(v string)          { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendTime(v time.Time)         { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendUint(v uint)              { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendUint64(v uint64)          { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendUint32(v uint32)          { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendUint16(v uint16)          { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendUint8(v uint8)            { s.elems = append(s.elems, v) }
-func (s *sliceArrayEncoder) AppendUintptr(v uintptr)        { s.elems = append(s.elems, v) }
diff --git a/vendor/knative.dev/pkg/test/logging/spew_encoder.go b/vendor/knative.dev/pkg/test/logging/spew_encoder.go
deleted file mode 100644
index 38c83c854..000000000
--- a/vendor/knative.dev/pkg/test/logging/spew_encoder.go
+++ /dev/null
@@ -1,196 +0,0 @@
-// Copyright 2020 The Knative Authors
-// Copyright (c) 2016 Uber Technologies, Inc.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-//
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-
-package logging
-
-import (
-	"fmt"
-	"sort"
-	"strings"
-	"sync"
-
-	"go.uber.org/zap"
-	"go.uber.org/zap/buffer"
-	"go.uber.org/zap/zapcore"
-)
-
-var (
-	_pool             = buffer.NewPool()
-	_sliceEncoderPool = sync.Pool{
-		New: func() interface{} {
-			return &sliceArrayEncoder{elems: make([]interface{}, 0, 2)}
-		},
-	}
-)
-
-func init() {
-	zap.RegisterEncoder("spew", func(encoderConfig zapcore.EncoderConfig) (zapcore.Encoder, error) {
-		return NewSpewEncoder(encoderConfig), nil
-	})
-}
-
-// NewSpewEncoder encodes logs using the spew library.
-//
-// The JSON encoder (also used by the console encoder) included in Zap can only print objects that
-// can be serialized to JSON and doesn't print them in the most readable way. This spew encoder is
-// designed to make human-readable log only and get the most information to the user on any data type.
-//
-// Code is mostly from console_encoder.go in zapcore.
-func NewSpewEncoder(cfg zapcore.EncoderConfig) *SpewEncoder {
-	enc := SpewEncoder{}
-	enc.MapObjectEncoder = zapcore.NewMapObjectEncoder()
-	enc.EncoderConfig = &cfg
-	return &enc
-}
-
-// SpewEncoder implements zapcore.Encoder interface
-type SpewEncoder struct {
-	*zapcore.MapObjectEncoder
-	*zapcore.EncoderConfig
-}
-
-// Implements zapcore.Encoder interface
-func (enc *SpewEncoder) Clone() zapcore.Encoder {
-	n := NewSpewEncoder(*(enc.EncoderConfig))
-	for k, v := range enc.Fields {
-		n.Fields[k] = v
-	}
-	return n
-}
-
-func getSliceEncoder() *sliceArrayEncoder {
-	return _sliceEncoderPool.Get().(*sliceArrayEncoder)
-}
-
-func putSliceEncoder(e *sliceArrayEncoder) {
-	e.elems = e.elems[:0]
-	_sliceEncoderPool.Put(e)
-}
-
-// Implements zapcore.Encoder interface.
-func (enc *SpewEncoder) EncodeEntry(ent zapcore.Entry, fields []zapcore.Field) (*buffer.Buffer, error) {
-	line := _pool.Get()
-
-	// Could probably rewrite this portion and remove the copied
-	// memory_encoder.go from this folder
-	arr := getSliceEncoder()
-	defer putSliceEncoder(arr)
-
-	if enc.TimeKey != "" && enc.EncodeTime != nil {
-		enc.EncodeTime(ent.Time, arr)
-	}
-	if enc.LevelKey != "" && enc.EncodeLevel != nil {
-		enc.EncodeLevel(ent.Level, arr)
-	}
-
-	if ent.LoggerName != "" && enc.NameKey != "" {
-		nameEncoder := enc.EncodeName
-
-		if nameEncoder == nil {
-			// Fall back to FullNameEncoder for backward compatibility.
-			nameEncoder = zapcore.FullNameEncoder
-		}
-
-		nameEncoder(ent.LoggerName, arr)
-	}
-	if ent.Caller.Defined && enc.CallerKey != "" && enc.EncodeCaller != nil {
-		enc.EncodeCaller(ent.Caller, arr)
-	}
-	for i := range arr.elems {
-		if i > 0 {
-			line.AppendByte('\t')
-		}
-		fmt.Fprint(line, arr.elems[i])
-	}
-
-	// Add the message itself.
-	if enc.MessageKey != "" {
-		enc.addTabIfNecessary(line)
-		line.AppendString(ent.Message)
-	}
-
-	// Add any structured context.
-	enc.writeContext(line, fields)
-
-	// If there's no stacktrace key, honor that; this allows users to force
-	// single-line output.
-	if ent.Stack != "" && enc.StacktraceKey != "" {
-		line.AppendByte('\n')
-		line.AppendString(ent.Stack)
-	}
-
-	if enc.LineEnding != "" {
-		line.AppendString(enc.LineEnding)
-	} else {
-		line.AppendString(zapcore.DefaultLineEnding)
-	}
-	return line, nil
-}
-
-func (enc *SpewEncoder) writeContext(line *buffer.Buffer, extra []zapcore.Field) {
-	if len(extra) == 0 && len(enc.Fields) == 0 {
-		return
-	}
-
-	// This could probably be more efficient, but .AddTo() is convenient
-
-	context := zapcore.NewMapObjectEncoder()
-	for k, v := range enc.Fields {
-		context.Fields[k] = v
-	}
-	for i := range extra {
-		extra[i].AddTo(context)
-	}
-
-	enc.addTabIfNecessary(line)
-	line.AppendString("\nContext:\n")
-	keys := make([]string, 0, len(context.Fields))
-	for k := range context.Fields {
-		keys = append(keys, k)
-	}
-	sort.Strings(keys)
-	for _, k := range keys {
-		line.AppendString(k)
-		line.AppendString(": ")
-		line.AppendString(stringify(context.Fields[k]))
-		line.TrimNewline()
-		line.AppendString("\n")
-	}
-}
-
-func stringify(a interface{}) string {
-	s, ok := a.(string)
-	if !ok {
-		s = strings.TrimSuffix(spewConfig.Sdump(a), "\n")
-	}
-	ret := strings.ReplaceAll(s, "\n", "\n ")
-	hasNewlines := s != ret
-	if hasNewlines {
-		return "\n " + ret
-	}
-	return s
-}
-
-func (enc *SpewEncoder) addTabIfNecessary(line *buffer.Buffer) {
-	if line.Len() > 0 {
-		line.AppendByte('\t')
-	}
-}
diff --git a/vendor/knative.dev/pkg/test/logging/sugar.go b/vendor/knative.dev/pkg/test/logging/sugar.go
deleted file mode 100644
index 01fc6b29f..000000000
--- a/vendor/knative.dev/pkg/test/logging/sugar.go
+++ /dev/null
@@ -1,114 +0,0 @@
-// Copyright 2020 Knative Authors
-// Copyright (c) 2016 Uber Technologies, Inc.
-//
-// Permission is hereby granted, free of charge, to any person obtaining a copy
-// of this software and associated documentation files (the "Software"), to deal
-// in the Software without restriction, including without limitation the rights
-// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-// copies of the Software, and to permit persons to whom the Software is
-// furnished to do so, subject to the following conditions:
-//
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
-//
-// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
-
-package logging
-
-import (
-	"github.com/davecgh/go-spew/spew"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
-
-	"go.uber.org/multierr"
-)
-
-const (
-	_oddNumberErrMsg    = "Ignored key without a value."
-	_nonStringKeyErrMsg = "Ignored key-value pairs with non-string keys."
-)
-
-var spewConfig *spew.ConfigState
-
-func init() {
-	spewConfig = spew.NewDefaultConfig()
-	spewConfig.DisableCapacities = true
-	spewConfig.SortKeys = true
-	spewConfig.SpewKeys = true
-	spewConfig.ContinueOnMethod = true
-}
-
-func (o *TLogger) handleFields(args []interface{}) []zap.Field {
-	if len(args) == 0 {
-		return nil
-	}
-	s := o.l.Sugar()
-
-	// Allocate enough space for the worst case; if users pass only structured
-	// fields, we shouldn't penalize them with extra allocations.
-	fields := make([]zap.Field, 0, len(args))
-	var invalid invalidPairs
-
-	for i := 0; i < len(args); {
-		// This is a strongly-typed field. Consume it and move on.
-		if f, ok := args[i].(zap.Field); ok {
-			fields = append(fields, f)
-			i++
-			continue
-		}
-
-		// Make sure this element isn't a dangling key.
-		if i == len(args)-1 {
-			s.DPanic(_oddNumberErrMsg, zap.Any("ignored", args[i]))
-			break
-		}
-
-		// Consume this value and the next, treating them as a key-value pair. If the
-		// key isn't a string, add this pair to the slice of invalid pairs.
-		key, val := args[i], args[i+1]
-		if keyStr, ok := key.(string); !ok {
-			// Subsequent errors are likely, so allocate once up front.
-			if cap(invalid) == 0 {
-				invalid = make(invalidPairs, 0, len(args)/2)
-			}
-			invalid = append(invalid, invalidPair{i, key, val})
-		} else {
-			fields = append(fields, zap.Any(keyStr, val))
-		}
-		i += 2
-	}
-
-	// If we encountered any invalid key-value pairs, log an error.
-	if len(invalid) > 0 {
-		s.DPanic(_nonStringKeyErrMsg, zap.Array("invalid", invalid), zap.String("all_input", spew.Sprintf("%#+v", args)))
-	}
-	return fields
-}
-
-type invalidPair struct {
-	position   int
-	key, value interface{}
-}
-
-func (p invalidPair) MarshalLogObject(enc zapcore.ObjectEncoder) error {
-	enc.AddInt64("position", int64(p.position))
-	zap.Any("key", p.key).AddTo(enc)
-	zap.Any("value", p.value).AddTo(enc)
-	return nil
-}
-
-type invalidPairs []invalidPair
-
-func (ps invalidPairs) MarshalLogArray(enc zapcore.ArrayEncoder) error {
-	var err error
-	for i := range ps {
-		err = multierr.Append(err, enc.AppendObject(ps[i]))
-	}
-	return err
-}
diff --git a/vendor/knative.dev/pkg/test/logging/tlogger.go b/vendor/knative.dev/pkg/test/logging/tlogger.go
deleted file mode 100644
index f6e05c5ce..000000000
--- a/vendor/knative.dev/pkg/test/logging/tlogger.go
+++ /dev/null
@@ -1,372 +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
-
-    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 logging
-
-import (
-	"errors"
-	"fmt"
-	"testing"
-
-	"github.com/go-logr/logr"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
-)
-
-// TLogger is TLogger
-type TLogger struct {
-	l        *zap.Logger
-	level    int
-	t        *testing.T
-	errs     map[string][]interface{} // For Collect()
-	dontFail bool
-}
-
-// V() returns an InfoLogger from go-logr.
-//
-// This should be the main way your tests log.
-// Most frequent usage is used directly:
-//   t.V(2).Info("Something at regular level")
-// But if something computationally difficult is to be done, can do:
-//   if l := t.V(8); l.Enabled() {
-//     x := somethingExpensive()
-//     l.Info("logging it", "expensiveThing", x)
-//   }
-//
-// Elsewhere in this documentation refers to a hypothetical .V(errorLevel) to simplify explanations.
-// The V() function cannot write to the error level; the Error, ErrorIfErr, Fatal, and
-// FatalIfErr methods are the only way to write to the error level.
-func (o *TLogger) V(level int) logr.InfoLogger {
-	// Consider adding || (level <= logrZapDebugLevel && o.l.Core().Enabled(zapLevelFromLogrLevel(level)))
-	// Reason to add it is even if you ask for verbosity=1, in case of error you'll get up to verbosity=3 in the debug output
-	// but since zapTest uses Debug, you always get V(<=3) even when verbosity < 3
-	// Probable solution is to write to t.Log at Info level?
-	if level <= o.level {
-		return &infoLogger{
-			logrLevel: o.level,
-			t:         o,
-		}
-	}
-	return disabledInfoLogger
-}
-
-// WithValues() acts like Zap's With() method.
-// Consistent with logr.Logger.WithValues()
-// Whenever anything is logged with the returned TLogger,
-// it will act as if these keys and values were passed into every logging call.
-func (o *TLogger) WithValues(keysAndValues ...interface{}) *TLogger {
-	return o.cloneWithNewLogger(o.l.With(o.handleFields(keysAndValues)...))
-}
-
-// WithName() acts like Zap's Named() method.
-// Consistent with logr.Logger.WithName()
-// Appends the name onto the current logger
-func (o *TLogger) WithName(name string) *TLogger {
-	return o.cloneWithNewLogger(o.l.Named(name))
-}
-
-// Custom additions:
-
-// ErrorIfErr fails the current test if the err != nil.
-// Remaining arguments function as if passed to .V(errorLevel).Info() (were that a thing)
-// Same signature as logr.Logger.Error() method, but as this is a test, it functions slightly differently.
-func (o *TLogger) ErrorIfErr(err error, msg string, keysAndValues ...interface{}) {
-	if err != nil {
-		o.error(err, msg, keysAndValues)
-		o.fail()
-	}
-}
-
-// FatalIfErr is just like ErrorIfErr() but test execution stops immediately
-func (o *TLogger) FatalIfErr(err error, msg string, keysAndValues ...interface{}) {
-	if err != nil {
-		o.error(err, msg, keysAndValues)
-		o.failNow()
-	}
-}
-
-// Error is essentially a .V(errorLevel).Info() followed by failing the test.
-// Intended usage is Error(msg string, key-value alternating arguments)
-// Same effect as testing.T.Error
-// Generic definition for compatibility with test.T interface
-// Implements test.T
-func (o *TLogger) Error(stringThenKeysAndValues ...interface{}) {
-	// Using o.error to have consistent call depth for Error, FatalIfErr, Info, etc
-	o.error(o.errorWithRuntimeCheck(stringThenKeysAndValues...))
-	o.fail()
-}
-
-// Fatal is essentially a .V(errorLevel).Info() followed by failing and immediately stopping the test.
-// Intended usage is Fatal(msg string, key-value alternating arguments)
-// Same effect as testing.T.Fatal
-// Generic definition for compatibility with test.TLegacy interface
-// Implements test.TLegacy
-func (o *TLogger) Fatal(stringThenKeysAndValues ...interface{}) {
-	o.error(o.errorWithRuntimeCheck(stringThenKeysAndValues...))
-	o.failNow()
-}
-
-func (o *TLogger) fail() {
-	if o.t != nil && !o.dontFail {
-		o.t.Fail()
-	}
-}
-
-func (o *TLogger) failNow() {
-	if o.t != nil && !o.dontFail {
-		o.t.FailNow()
-	}
-}
-
-func validateKeysAndValues(keysAndValues ...interface{}) bool {
-	length := len(keysAndValues)
-	for i := 0; i < length; {
-		_, isField := keysAndValues[i].(zapcore.Field)
-		_, isString := keysAndValues[i].(string)
-		if isField {
-			i++
-		} else if isString {
-			if i == length-1 {
-				return false
-			}
-			i += 2
-		} else {
-			return false
-		}
-	}
-	return true
-}
-
-func (o *TLogger) interfacesToFields(things ...interface{}) []interface{} {
-	o.V(5).Info("DEPRECATED Error/Fatal usage", zap.Stack("callstack"))
-	fields := make([]interface{}, 2*len(things))
-	for i, d := range things {
-		fields[i*2] = fmt.Sprintf("arg %d", i)
-		fields[i*2+1] = d
-	}
-	return fields
-}
-
-func (o *TLogger) errorWithRuntimeCheck(stringThenKeysAndValues ...interface{}) (error, string, []interface{}) { //nolint // Returning the error first is okay and expected here.
-	if len(stringThenKeysAndValues) == 0 {
-		return nil, "", nil
-	}
-	s, isString := stringThenKeysAndValues[0].(string)
-	if isString {
-		// Desired case (hopefully)
-		remainder := stringThenKeysAndValues[1:]
-		if !validateKeysAndValues(remainder...) {
-			remainder = o.interfacesToFields(remainder...)
-		}
-		return nil, s, remainder
-	}
-	e, isError := stringThenKeysAndValues[0].(error)
-	if isError && len(stringThenKeysAndValues) == 1 {
-		return e, "", nil
-	}
-	return nil, "unstructured error", o.interfacesToFields(stringThenKeysAndValues...)
-}
-
-// Cleanup registers a cleanup callback.
-func (o *TLogger) Cleanup(c func()) {
-	o.t.Cleanup(c)
-}
-
-// Run a subtest. Just like testing.T.Run but creates a TLogger.
-func (o *TLogger) Run(name string, f func(t *TLogger)) {
-	tfunc := func(ts *testing.T) {
-		tl, cancel := newTLogger(ts, o.level, o.dontFail)
-		defer cancel()
-		f(tl)
-	}
-	o.t.Run(name, tfunc)
-}
-
-// Name is just like testing.T.Name()
-// Implements test.T
-func (o *TLogger) Name() string {
-	return o.t.Name()
-}
-
-// Helper cannot work as an indirect call, so just do nothing :(
-// Implements test.T
-func (o *TLogger) Helper() {
-}
-
-// SkipNow immediately stops test execution
-// Implements test.T
-func (o *TLogger) SkipNow() {
-	o.t.SkipNow()
-}
-
-// Log is deprecated: only existing for test.T compatibility
-// Please use leveled logging via .V().Info()
-// Will panic if given data incompatible with Info() function
-// Implements test.T
-func (o *TLogger) Log(args ...interface{}) {
-	// This is complicated to ensure exactly 2 levels of indirection
-	i := o.V(2)
-	iL, ok := i.(*infoLogger)
-	if ok {
-		iL.indirectWrite(args[0].(string), args[1:]...)
-	}
-}
-
-// Parallel allows tests or subtests to run in parallel
-// Just calls the testing.T.Parallel() under the hood
-func (o *TLogger) Parallel() {
-	o.t.Parallel()
-}
-
-// Logf is deprecated: only existing for test.TLegacy compatibility
-// Please use leveled logging via .V().Info()
-// Implements test.TLegacy
-func (o *TLogger) Logf(fmtS string, args ...interface{}) {
-	// This is complicated to ensure exactly 2 levels of indirection
-	iL, ok := o.V(2).(*infoLogger)
-	if ok {
-		iL.indirectWrite(fmt.Sprintf(fmtS, args...))
-	}
-}
-
-func (o *TLogger) error(err error, msg string, keysAndValues []interface{}) {
-	var newKAV []interface{}
-	var serr StructuredError
-	if errors.As(err, &serr) {
-		serr.DisableValuePrinting()
-		defer serr.EnableValuePrinting()
-		newLen := len(keysAndValues) + len(serr.GetValues())
-		newKAV = make([]interface{}, 0, newLen+2)
-		newKAV = append(newKAV, keysAndValues...)
-		newKAV = append(newKAV, serr.GetValues()...)
-	}
-	if err != nil {
-		if msg == "" { // This is used if just the error is given to .Error() or .Fatal()
-			msg = err.Error()
-		} else {
-			if newKAV == nil {
-				newKAV = make([]interface{}, 0, len(keysAndValues)+1)
-				newKAV = append(newKAV, keysAndValues...)
-			}
-			newKAV = append(newKAV, zap.Error(err))
-		}
-	}
-	if newKAV != nil {
-		keysAndValues = newKAV
-	}
-	if checkedEntry := o.l.Check(zap.ErrorLevel, msg); checkedEntry != nil {
-		checkedEntry.Write(o.handleFields(keysAndValues)...)
-	}
-}
-
-// Creation and Teardown
-
-// Create a TLogger object using the global Zap logger and the current testing.T
-// `defer` a call to second return value immediately after.
-func NewTLogger(t *testing.T) (*TLogger, func()) {
-	return newTLogger(t, verbosity, false)
-}
-
-func newTLogger(t *testing.T, verbosity int, dontFail bool) (*TLogger, func()) {
-	testOptions := []zap.Option{
-		zap.AddCaller(),
-		zap.AddCallerSkip(2),
-		zap.Development(),
-	}
-	writer := newTestingWriter(t)
-	// Based off zap.NewDevelopmentEncoderConfig()
-	cfg := zapcore.EncoderConfig{
-		// Wanted keys can be anything except the empty string.
-		TimeKey:        "",
-		LevelKey:       "",
-		NameKey:        "",
-		CallerKey:      "C",
-		MessageKey:     "M",
-		StacktraceKey:  "S",
-		LineEnding:     zapcore.DefaultLineEnding,
-		EncodeLevel:    zapcore.CapitalLevelEncoder,
-		EncodeTime:     zapcore.ISO8601TimeEncoder,
-		EncodeDuration: zapcore.StringDurationEncoder,
-		EncodeCaller:   zapcore.ShortCallerEncoder,
-	}
-	core := zapcore.NewCore(
-		NewSpewEncoder(cfg),
-		writer,
-		zapcore.DebugLevel,
-	)
-	if zapCore != nil {
-		core = zapcore.NewTee(
-			zapCore,
-			core,
-			// TODO(coryrc): Open new file (maybe creating JUnit!?) with test output?
-		)
-	}
-	log := zap.New(core, testOptions...).Named(t.Name())
-	tlogger := TLogger{
-		l:        log,
-		level:    verbosity,
-		t:        t,
-		errs:     make(map[string][]interface{}),
-		dontFail: dontFail,
-	}
-	return &tlogger, func() {
-		tlogger.handleCollectedErrors()
-		// Sometimes goroutines exist after a test and they cause panics if they attempt to call t.Log().
-		// Prevent this panic by disabling writes to the testing.T (we'll still get them everywhere else).
-		writer.Disable()
-	}
-}
-
-func (o *TLogger) cloneWithNewLogger(l *zap.Logger) *TLogger {
-	t := TLogger{
-		l:        l,
-		level:    o.level,
-		t:        o.t,
-		errs:     o.errs,
-		dontFail: o.dontFail,
-	}
-	return &t
-}
-
-// Collect allows you to commingle multiple validations during one test execution.
-// Under the hood, it creates a sub-test during cleanup and iterates through the collected values, printing them.
-// If any are errors, it fails the subtest.
-// Currently experimental and likely to be removed
-func (o *TLogger) Collect(key string, value interface{}) {
-	list, hasKey := o.errs[key]
-	if hasKey {
-		list = append(list, value)
-	} else {
-		list = make([]interface{}, 1)
-		list[0] = value
-	}
-	o.errs[key] = list
-}
-
-func (o *TLogger) handleCollectedErrors() {
-	for name, list := range o.errs {
-		o.Run(name, func(t *TLogger) {
-			for _, item := range list {
-				_, isError := item.(error)
-				if isError {
-					t.Error(item)
-				} else {
-					t.V(3).Info(spewConfig.Sprint(item))
-				}
-			}
-		})
-	}
-}
diff --git a/vendor/knative.dev/pkg/test/logging/zapr.go b/vendor/knative.dev/pkg/test/logging/zapr.go
deleted file mode 100644
index c0b9f5858..000000000
--- a/vendor/knative.dev/pkg/test/logging/zapr.go
+++ /dev/null
@@ -1,47 +0,0 @@
-// Copyright 2020 Knative Authors
-// Copyright 2018 Solly Ross
-//
-// 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.
-
-// The useful parts of this file have been preserved
-// from their origin at https://github.com/go-logr/zapr/tree/8f2487342d52a33a1793e50e3ca04bc1767aa65c
-
-package logging
-
-// noopInfoLogger is a logr.InfoLogger that's always disabled, and does nothing.
-type noopInfoLogger struct{}
-
-func (l *noopInfoLogger) Enabled() bool                   { return false }
-func (l *noopInfoLogger) Info(_ string, _ ...interface{}) {}
-
-var disabledInfoLogger = &noopInfoLogger{}
-
-// infoLogger is a logr.InfoLogger that uses Zap to log at a particular
-// level.
-type infoLogger struct {
-	logrLevel int
-	t         *TLogger
-}
-
-func (i *infoLogger) Enabled() bool { return true }
-func (i *infoLogger) Info(msg string, keysAndVals ...interface{}) {
-	i.indirectWrite(msg, keysAndVals...)
-}
-
-// This function just exists to have consistent 2-level call depth for Zap proxying
-func (i *infoLogger) indirectWrite(msg string, keysAndVals ...interface{}) {
-	lvl := zapLevelFromLogrLevel(i.logrLevel)
-	if checkedEntry := i.t.l.Check(lvl, msg); checkedEntry != nil {
-		checkedEntry.Write(i.t.handleFields(keysAndVals)...)
-	}
-}