Repository: knative-sandbox/eventing-kafka
Pull request #855 (2021-09-03 22:37:03)

Add Kafka Source defaults, starting with KEDA annotations
 
<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add new config-kafka-source-defaults configmap to be used by the defaulting mutating webhook 
- When autoscaler is enabled in the configmap, the defaulting webhook automatically adds KEDA annotations.
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
- 🎁Autoscaling annotations can now be automatically added to KafkaSource objects. See the documentation for more details. 
```

**Docs**
See https://github.com/knative/docs/issues/4181
<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->

/cc @steven0711dong @matzew @zroubalik 


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

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

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

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

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

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

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

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/apis/sources/config/kafka_defaults.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/855/pull-knative-sandbox-eventing-kafka-go-coverage/1433922014623567872/artifacts/line-cov.html#file0) | Do not exist | 84.2% | 
[pkg/apis/sources/config/store.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/855/pull-knative-sandbox-eventing-kafka-go-coverage/1433922014623567872/artifacts/line-cov.html#file1) | Do not exist | 93.8% | 

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#855](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a295d55) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/75b06f9c186ac6865d3dda89666563d2d7a2d7b5?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (75b06f9) will **decrease** coverage by `0.50%`.
> The diff coverage is `81.08%`.

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

```diff
@@            Coverage Diff             @@
##             main     #855      +/-   ##
==========================================
- Coverage   75.14%   74.64%   -0.51%     
==========================================
  Files         139      143       +4     
  Lines        6442     6633     +191     
==========================================
+ Hits         4841     4951     +110     
- Misses       1369     1435      +66     
- Partials      232      247      +15     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/apis/sources/config/kafka\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvc291cmNlcy9jb25maWcva2Fma2FfZGVmYXVsdHMuZ28=) | `68.42% <68.42%> (ø)` | |
| [pkg/apis/sources/config/store.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvc291cmNlcy9jb25maWcvc3RvcmUuZ28=) | `92.00% <92.00%> (ø)` | |
| [pkg/apis/sources/v1beta1/kafka\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvc291cmNlcy92MWJldGExL2thZmthX2RlZmF1bHRzLmdv) | `100.00% <100.00%> (ø)` | |
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `54.73% <0.00%> (-3.70%)` | :arrow_down: |
| [...rce/reconciler/source/resources/receive\_adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9yZWNvbmNpbGVyL3NvdXJjZS9yZXNvdXJjZXMvcmVjZWl2ZV9hZGFwdGVyLmdv) | `96.51% <0.00%> (-3.49%)` | :arrow_down: |
| [pkg/source/adapter/message.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9hZGFwdGVyL21lc3NhZ2UuZ28=) | `72.72% <0.00%> (-1.60%)` | :arrow_down: |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `46.61% <0.00%> (-1.22%)` | :arrow_down: |
| [cmd/channel/post-install/updater.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y21kL2NoYW5uZWwvcG9zdC1pbnN0YWxsL3VwZGF0ZXIuZ28=) | `68.65% <0.00%> (ø)` | |
| [cmd/channel/post-install/main.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y21kL2NoYW5uZWwvcG9zdC1pbnN0YWxsL21haW4uZ28=) | `0.00% <0.00%> (ø)` | |
| ... and [2 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/855/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

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


Patch:
diff --git a/cmd/source/controller/main.go b/cmd/source/controller/main.go
index cfee368e5..48fc086a1 100644
--- a/cmd/source/controller/main.go
+++ b/cmd/source/controller/main.go
@@ -20,28 +20,29 @@ import (
 	"context"
 	"os"
 
-	"knative.dev/eventing-kafka/pkg/apis/bindings"
-	"knative.dev/eventing-kafka/pkg/apis/sources"
-	"knative.dev/pkg/webhook/resourcesemantics/conversion"
-
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
-	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
-	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
-
-	"knative.dev/eventing-kafka/pkg/source/reconciler/binding"
-	"knative.dev/eventing-kafka/pkg/source/reconciler/source"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/pkg/logging"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/webhook"
 	"knative.dev/pkg/webhook/certificates"
 	"knative.dev/pkg/webhook/psbinding"
 	"knative.dev/pkg/webhook/resourcesemantics"
+	"knative.dev/pkg/webhook/resourcesemantics/conversion"
 	"knative.dev/pkg/webhook/resourcesemantics/defaulting"
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
+
+	"knative.dev/eventing-kafka/pkg/apis/bindings"
+	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	"knative.dev/eventing-kafka/pkg/apis/sources"
+	kafkasourcedefaultconfig "knative.dev/eventing-kafka/pkg/apis/sources/config"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/eventing-kafka/pkg/source/reconciler/binding"
+	"knative.dev/eventing-kafka/pkg/source/reconciler/source"
 )
 
 const (
@@ -57,6 +58,15 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 var callbacks = map[schema.GroupVersionKind]validation.Callback{}
 
 func NewDefaultingAdmissionController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+	// Decorate contexts with the current state of the config.
+	kafkaStore := kafkasourcedefaultconfig.NewStore(logging.FromContext(ctx).Named("kafka-source-config-store"))
+	kafkaStore.WatchConfigs(cmw)
+
+	// Decorate contexts with the current state of the config.
+	ctxFunc := func(ctx context.Context) context.Context {
+		return kafkaStore.ToContext(ctx)
+	}
+
 	return defaulting.NewAdmissionController(ctx,
 
 		// Name of the resource webhook.
@@ -69,11 +79,7 @@ func NewDefaultingAdmissionController(ctx context.Context, cmw configmap.Watcher
 		types,
 
 		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
-		func(ctx context.Context) context.Context {
-			// Here is where you would infuse the context with state
-			// (e.g. attach a store with configmap data)
-			return ctx
-		},
+		ctxFunc,
 
 		// Whether to disallow unknown fields.
 		true,
diff --git a/cmd/source/mtcontroller/main.go b/cmd/source/mtcontroller/main.go
index 7a5481ca2..ee2fb4453 100644
--- a/cmd/source/mtcontroller/main.go
+++ b/cmd/source/mtcontroller/main.go
@@ -20,21 +20,16 @@ import (
 	"context"
 	"os"
 
-	"knative.dev/eventing-kafka/pkg/apis/bindings"
-	"knative.dev/eventing-kafka/pkg/apis/sources"
 	"knative.dev/pkg/webhook/resourcesemantics/conversion"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
-	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
-	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
-
-	"knative.dev/eventing-kafka/pkg/source/reconciler/binding"
 	source "knative.dev/eventing-kafka/pkg/source/reconciler/mtsource"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/pkg/logging"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/webhook"
 	"knative.dev/pkg/webhook/certificates"
@@ -42,6 +37,14 @@ import (
 	"knative.dev/pkg/webhook/resourcesemantics"
 	"knative.dev/pkg/webhook/resourcesemantics/defaulting"
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
+
+	"knative.dev/eventing-kafka/pkg/apis/bindings"
+	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	"knative.dev/eventing-kafka/pkg/apis/sources"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/eventing-kafka/pkg/source/reconciler/binding"
+
+	kafkasourcedefaultconfig "knative.dev/eventing-kafka/pkg/apis/sources/config"
 )
 
 const (
@@ -57,6 +60,15 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 var callbacks = map[schema.GroupVersionKind]validation.Callback{}
 
 func NewDefaultingAdmissionController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+	// Decorate contexts with the current state of the config.
+	kafkaStore := kafkasourcedefaultconfig.NewStore(logging.FromContext(ctx).Named("kafka-source-config-store"))
+	kafkaStore.WatchConfigs(cmw)
+
+	// Decorate contexts with the current state of the config.
+	ctxFunc := func(ctx context.Context) context.Context {
+		return kafkaStore.ToContext(ctx)
+	}
+
 	return defaulting.NewAdmissionController(ctx,
 
 		// Name of the resource webhook.
@@ -69,11 +81,7 @@ func NewDefaultingAdmissionController(ctx context.Context, cmw configmap.Watcher
 		types,
 
 		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
-		func(ctx context.Context) context.Context {
-			// Here is where you would infuse the context with state
-			// (e.g. attach a store with configmap data)
-			return ctx
-		},
+		ctxFunc,
 
 		// Whether to disallow unknown fields.
 		true,
@@ -92,6 +100,7 @@ func NewValidationAdmissionController(ctx context.Context, cmw configmap.Watcher
 		// The resources to validate.
 		types,
 
+		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
 		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
 		func(ctx context.Context) context.Context {
 			// Here is where you would infuse the context with state
diff --git a/config/channel/resources/kafkachannel-crd.yaml b/config/channel/resources/kafkachannel-crd.yaml
index 77eb1fce8..83600b6a9 100644
--- a/config/channel/resources/kafkachannel-crd.yaml
+++ b/config/channel/resources/kafkachannel-crd.yaml
@@ -147,8 +147,6 @@ spec:
             status:
               description: Status represents the current state of the KafkaChannel. This data may be out of date.
               type: object
-              required:
-                - address
               properties:
                 address:
                   type: object
diff --git a/config/source/common/configmaps/config-kafka-source-defaults.yaml b/config/source/common/configmaps/config-kafka-source-defaults.yaml
new file mode 100644
index 000000000..a79338e33
--- /dev/null
+++ b/config/source/common/configmaps/config-kafka-source-defaults.yaml
@@ -0,0 +1,58 @@
+# 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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-source-defaults
+  namespace: knative-eventing
+  labels:
+    eventing.knative.dev/release: devel
+  annotations:
+    knative.dev/example-checksum: "b6ed351d"
+data:
+  _example: |
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+
+    # autoscalingClass is the autoscaler class name to use.
+    # valid value: keda.autoscaling.knative.dev
+    # autoscalingClass: ""
+
+    # minScale is the minimum number of replicas to scale down to.
+    # minScale: "1"
+
+    # maxScale is the maximum number of replicas to scale up to.
+    # maxScale: "1"
+
+    # pollingInterval is the interval in seconds KEDA uses to poll metrics.
+    # pollingInterval: "30"
+
+    # cooldownPeriod is the period of time in seconds KEDA waits until it scales down.
+    # cooldownPeriod: "300"
+
+    # kafkaLagThreshold is the lag (ie. number of messages in a partition) threshold for KEDA to scale up sources.
+    # kafkaLagThreshold: "10"
diff --git a/config/source/multi/400-config-kafka-source-defaults.yaml b/config/source/multi/400-config-kafka-source-defaults.yaml
new file mode 120000
index 000000000..f54c6c367
--- /dev/null
+++ b/config/source/multi/400-config-kafka-source-defaults.yaml
@@ -0,0 +1,1 @@
+../common/configmaps/config-kafka-source-defaults.yaml
\ No newline at end of file
diff --git a/config/source/single/400-config-kafka-source-defaults.yaml b/config/source/single/400-config-kafka-source-defaults.yaml
new file mode 120000
index 000000000..f54c6c367
--- /dev/null
+++ b/config/source/single/400-config-kafka-source-defaults.yaml
@@ -0,0 +1,1 @@
+../common/configmaps/config-kafka-source-defaults.yaml
\ No newline at end of file
diff --git a/go.mod b/go.mod
index c40a73eee..05b66e01c 100644
--- a/go.mod
+++ b/go.mod
@@ -34,10 +34,10 @@ require (
 	k8s.io/apimachinery v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
-	knative.dev/control-protocol v0.0.0-20210830193954-f9f8d9068ed7
-	knative.dev/eventing v0.25.1-0.20210903045557-38bfba7d7d21
+	knative.dev/control-protocol v0.0.0-20210903191859-3cb84f4563fd
+	knative.dev/eventing v0.25.1-0.20210906161343-38f38456f045
 	knative.dev/hack v0.0.0-20210806075220-815cd312d65c
-	knative.dev/networking v0.0.0-20210903055657-5f2e230c665f
-	knative.dev/pkg v0.0.0-20210903055557-dec98b426132
+	knative.dev/networking v0.0.0-20210907131619-be5cfdebc4e1
+	knative.dev/pkg v0.0.0-20210907112319-73d4fe679a2e
 	knative.dev/reconciler-test v0.0.0-20210820180205-a25de6a08087
 )
diff --git a/go.sum b/go.sum
index 25ddcc876..11c985fb5 100644
--- a/go.sum
+++ b/go.sum
@@ -1306,22 +1306,21 @@ k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7 h1:vEx13qjvaZ4yfObSSXW7Br
 k8s.io/kube-openapi v0.0.0-20210305001622-591a79e4bda7/go.mod h1:wXW5VT87nVfh/iLV8FpR2uDvrFyomxbtb1KivDbvPTE=
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920 h1:CbnUZsM497iRC5QMVkHwyl8s2tB3g7yaSHkYPkpgelw=
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20210830193954-f9f8d9068ed7 h1:yB3lSCACSBunGVkt72FWTagM8fzfy7jkohLnwmcF9Nc=
-knative.dev/control-protocol v0.0.0-20210830193954-f9f8d9068ed7/go.mod h1:9gYmAq6zVKUA+p1+DXslH91jtga5q2z+YJWcH+z88EE=
-knative.dev/eventing v0.25.1-0.20210903045557-38bfba7d7d21 h1:vMjxQvRMaa/m8AvtjP6UZl2CRtCtB3PhflYg/Sk+/K0=
-knative.dev/eventing v0.25.1-0.20210903045557-38bfba7d7d21/go.mod h1:hMorHr/gyBlXg5/JPHIkSWdbOQFdoCtWuLm7Fkli5F8=
+knative.dev/control-protocol v0.0.0-20210903191859-3cb84f4563fd h1:x5TJqhDImHMepYHOJQGKiZcTV7f27N6wW+CoetKS0sk=
+knative.dev/control-protocol v0.0.0-20210903191859-3cb84f4563fd/go.mod h1:BqxraD86fH70VwQGl0GRq58fZUdQwyHvmt9/0PqAHdY=
+knative.dev/eventing v0.25.1-0.20210906161343-38f38456f045 h1:54qhX/rJC3rv8mmPbKOO6BwM4gykRwWjDuyBCEeL6Gs=
+knative.dev/eventing v0.25.1-0.20210906161343-38f38456f045/go.mod h1:aEq47MCAi5nMgILNF8mTK7FG3ONmKXMWVtBzYt3Vp64=
 knative.dev/hack v0.0.0-20210622141627-e28525d8d260/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c h1:nOXoDWAAItwr4o0dp3nHr6skgpVD4IvME/UX84YNl5k=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210806075220-815cd312d65c/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/networking v0.0.0-20210903055657-5f2e230c665f h1:83X6Ck5L53bXVu2MMsS4cFZInBcjoIzI0hEd6mWOIsU=
-knative.dev/networking v0.0.0-20210903055657-5f2e230c665f/go.mod h1:RU3UlJNSFKwyCWUV0tOGtlcca+WA83M3QxBl1wi/vhw=
+knative.dev/networking v0.0.0-20210907131619-be5cfdebc4e1 h1:8mCbTK+Phy5F3zXI5wRqWb0bIDT2rRX97JS1zXL+EuY=
+knative.dev/networking v0.0.0-20210907131619-be5cfdebc4e1/go.mod h1:/Z4pMgP6ieCFpIn2onYIEXEA5HmI9dB6jlSt/XTsGN0=
 knative.dev/pkg v0.0.0-20210803160015-21eb4c167cc5/go.mod h1:RPk5txNA3apR9X40D4MpUOP9/VqOG8CrtABWfOwGVS4=
-knative.dev/pkg v0.0.0-20210830224055-82f3a9f1c5bc/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
-knative.dev/pkg v0.0.0-20210902202457-7a61343b36ca/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
-knative.dev/pkg v0.0.0-20210903055557-dec98b426132 h1:+iDyXF8cRdFclCu8eUTRBjxQ9ApE3KkleBScMl/3wHU=
 knative.dev/pkg v0.0.0-20210903055557-dec98b426132/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
-knative.dev/reconciler-test v0.0.0-20210803183715-b61cc77c06f6/go.mod h1:+Kovy+G5zXZNcuO/uB+zfo37vFKZzsLIlWezt/nKMz0=
+knative.dev/pkg v0.0.0-20210906105443-bb433c98147d/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
+knative.dev/pkg v0.0.0-20210907112319-73d4fe679a2e h1:EpwOC7a0EreNYej/GpuUjbeC9xf1an+66TNRzzPgH6s=
+knative.dev/pkg v0.0.0-20210907112319-73d4fe679a2e/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
 knative.dev/reconciler-test v0.0.0-20210820180205-a25de6a08087 h1:HVcaI8P+CNHl5VgnqWgAzGoc9vbvf3D421rd2qEo3Wc=
 knative.dev/reconciler-test v0.0.0-20210820180205-a25de6a08087/go.mod h1:+Kovy+G5zXZNcuO/uB+zfo37vFKZzsLIlWezt/nKMz0=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
diff --git a/hack/tools.go b/hack/tools.go
index 1dcec792f..2c1c282f5 100644
--- a/hack/tools.go
+++ b/hack/tools.go
@@ -20,6 +20,7 @@ package tools
 
 import (
 	_ "knative.dev/hack"
+	_ "knative.dev/pkg/configmap/hash-gen"
 	_ "knative.dev/pkg/hack"
 
 	// Test images from eventing
diff --git a/hack/update-checksums.sh b/hack/update-checksums.sh
new file mode 100755
index 000000000..3a23023f9
--- /dev/null
+++ b/hack/update-checksums.sh
@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+
+# 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.
+
+set -o errexit
+set -o nounset
+set -o pipefail
+
+export GO111MODULE=on
+
+if [ -z "${GOPATH:-}" ]; then
+  export GOPATH=$(go env GOPATH)
+fi
+
+source $(dirname $0)/../vendor/knative.dev/hack/library.sh
+
+go run "${REPO_ROOT_DIR}/vendor/knative.dev/pkg/configmap/hash-gen" "${REPO_ROOT_DIR}"/config/source/common/configmaps/*.yaml
diff --git a/pkg/apis/sources/config/doc.go b/pkg/apis/sources/config/doc.go
new file mode 100644
index 000000000..a0d4319ef
--- /dev/null
+++ b/pkg/apis/sources/config/doc.go
@@ -0,0 +1,21 @@
+/*
+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.
+*/
+
+// +k8s:deepcopy-gen=package
+
+// Package config holds the typed objects that define the schemas for
+// ConfigMap objects that pertain to our API objects.
+package config
diff --git a/pkg/apis/sources/config/kafka_defaults.go b/pkg/apis/sources/config/kafka_defaults.go
new file mode 100644
index 000000000..6602f66b7
--- /dev/null
+++ b/pkg/apis/sources/config/kafka_defaults.go
@@ -0,0 +1,143 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package config
+
+import (
+	"fmt"
+	"strconv"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
+const (
+	// KafkaDefaultsConfigName is the name of config map for the default
+	// configs that KafkaSource should use.
+	KafkaDefaultsConfigName = "config-kafka-source-defaults"
+
+	// DefaultAutoscalingClassKey is the name of autoscaler class annotation
+	DefaultAutoscalingClassKey = "autoscalingClass"
+
+	// DefaultMinScaleKey is the name of the key corresponding to the KEDA minScale annotation
+	DefaultMinScaleKey = "minScale"
+
+	// DefaultMaxScaleKey is the name of the KEDA maxScale annotation
+	DefaultMaxScaleKey = "maxScale"
+
+	// DefaultPollingIntervalKey is the name of the KEDA pollingInterval annotation
+	DefaultPollingIntervalKey = "pollingInterval"
+
+	// DefaultCooldownPeriodKey is the name of the KEDA cooldownPeriod annotation
+	DefaultCooldownPeriodKey = "cooldownPeriod"
+
+	// DefaultKafkaLagThresholdKey is the name of the KEDA kafkaLagThreshold annotation
+	DefaultKafkaLagThresholdKey = "kafkaLagThreshold"
+
+	// KedaAutoscalingClass is the class name for KEDA
+	KedaAutoscalingClass = "keda.autoscaling.knative.dev"
+
+	// DefaultMinScaleValue is the default value for DefaultMinScaleKey
+	DefaultMinScaleValue = int64(1)
+
+	// DefaultMaxScaleValue is  the default value for DefaultMaxScaleKey
+	DefaultMaxScaleValue = int64(1)
+
+	// DefaultPollingIntervalValue is the default value for DefaultPollingIntervalKey
+	DefaultPollingIntervalValue = int64(30)
+
+	// DefaultCooldownPeriodValue is the default value for DefaultCooldownPeriodKey
+	DefaultCooldownPeriodValue = int64(300)
+
+	// DefaultKafkaLagThresholdValue is the default value for DefaultKafkaLagThresholdKey
+	DefaultKafkaLagThresholdValue = int64(10)
+)
+
+// NewKafkaDefaultsConfigFromMap creates a KafkaSourceDefaults from the supplied Map
+func NewKafkaDefaultsConfigFromMap(data map[string]string) (*KafkaSourceDefaults, error) {
+	nc := &KafkaSourceDefaults{}
+
+	value, present := data[DefaultAutoscalingClassKey]
+	if !present || value == "" {
+		return nc, nil
+	}
+	if value != "keda.autoscaling.knative.dev" {
+		return nil, fmt.Errorf("invalid value %q for %s. Only keda.autoscaling.knative.dev is allowed", value, DefaultAutoscalingClassKey)
+	}
+	nc.AutoscalingClass = value
+
+	int64Value, err := parseInt64Entry(data, DefaultMinScaleKey, DefaultMinScaleValue)
+	if err != nil {
+		return nil, err
+	}
+	nc.MinScale = int64Value
+
+	int64Value, err = parseInt64Entry(data, DefaultMaxScaleKey, DefaultMaxScaleValue)
+	if err != nil {
+		return nil, err
+	}
+	nc.MaxScale = int64Value
+
+	int64Value, err = parseInt64Entry(data, DefaultPollingIntervalKey, DefaultPollingIntervalValue)
+	if err != nil {
+		return nil, err
+	}
+	nc.PollingInterval = int64Value
+
+	int64Value, err = parseInt64Entry(data, DefaultCooldownPeriodKey, DefaultCooldownPeriodValue)
+	if err != nil {
+		return nil, err
+	}
+	nc.CooldownPeriod = int64Value
+
+	int64Value, err = parseInt64Entry(data, DefaultKafkaLagThresholdKey, DefaultKafkaLagThresholdValue)
+	if err != nil {
+		return nil, err
+	}
+	nc.KafkaLagThreshold = int64Value
+
+	return nc, nil
+}
+
+// NewKafkaDefaultsConfigFromConfigMap creates a KafkaSourceDefaults from the supplied configMap
+func NewKafkaDefaultsConfigFromConfigMap(config *corev1.ConfigMap) (*KafkaSourceDefaults, error) {
+	return NewKafkaDefaultsConfigFromMap(config.Data)
+}
+
+type KafkaSourceDefaults struct {
+	AutoscalingClass  string `json:"autoscalingClass,omitempty"`
+	MinScale          int64  `json:"minScale,omitempty"`
+	MaxScale          int64  `json:"maxScale,omitempty"`
+	PollingInterval   int64  `json:"pollingInterval,omitempty"`
+	CooldownPeriod    int64  `json:"cooldownPeriod,omitempty"`
+	KafkaLagThreshold int64  `json:"kafkaLagThreshold,omitempty"`
+}
+
+func (d *KafkaSourceDefaults) DeepCopy() *KafkaSourceDefaults {
+	if d == nil {
+		return nil
+	}
+	out := new(KafkaSourceDefaults)
+	*out = *d
+	return out
+}
+
+func parseInt64Entry(data map[string]string, key string, defaults int64) (int64, error) {
+	value, present := data[key]
+	if !present {
+		return defaults, nil
+	}
+	return strconv.ParseInt(value, 0, 64)
+}
diff --git a/pkg/apis/sources/config/store.go b/pkg/apis/sources/config/store.go
new file mode 100644
index 000000000..1770dca73
--- /dev/null
+++ b/pkg/apis/sources/config/store.go
@@ -0,0 +1,97 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package config
+
+import (
+	"context"
+
+	"knative.dev/pkg/configmap"
+)
+
+type kafkaCfgKey struct{}
+
+// Config holds the collection of configurations that we attach to contexts.
+// +k8s:deepcopy-gen=false
+type Config struct {
+	KafkaSourceDefaults *KafkaSourceDefaults
+}
+
+// FromContext extracts a Config from the provided context.
+func FromContext(ctx context.Context) *Config {
+	x, ok := ctx.Value(kafkaCfgKey{}).(*Config)
+	if ok {
+		return x
+	}
+	return nil
+}
+
+// FromContextOrDefaults is like FromContext, but when no Config is attached it
+// returns a Config populated with the defaults for each of the Config fields.
+func FromContextOrDefaults(ctx context.Context) *Config {
+	if cfg := FromContext(ctx); cfg != nil {
+		return cfg
+	}
+	kafkaDefaults, err := NewKafkaDefaultsConfigFromMap(map[string]string{})
+	if err != nil || kafkaDefaults == nil {
+		kafkaDefaults = &KafkaSourceDefaults{}
+	}
+	x := &Config{
+		KafkaSourceDefaults: kafkaDefaults,
+	}
+
+	return x
+}
+
+// ToContext attaches the provided Config to the provided context, returning the
+// new context with the Config attached.
+func ToContext(ctx context.Context, c *Config) context.Context {
+	return context.WithValue(ctx, kafkaCfgKey{}, c)
+}
+
+// Store is a typed wrapper around configmap.Untyped store to handle our configmaps.
+// +k8s:deepcopy-gen=false
+type Store struct {
+	*configmap.UntypedStore
+}
+
+//NewStore creates a new store of Configs and optionally calls functions when ConfigMaps are updated.
+func NewStore(logger configmap.Logger, onAfterStore ...func(name string, value interface{})) *Store {
+	store := &Store{
+		UntypedStore: configmap.NewUntypedStore(
+			"kafkadefaults",
+			logger,
+			configmap.Constructors{
+				KafkaDefaultsConfigName: NewKafkaDefaultsConfigFromConfigMap,
+			},
+			onAfterStore...,
+		),
+	}
+
+	return store
+}
+
+// ToContext attaches the current Config state to the provided context.
+func (s *Store) ToContext(ctx context.Context) context.Context {
+	return ToContext(ctx, s.Load())
+}
+
+// Load creates a Config from the current config state of the Store.
+func (s *Store) Load() *Config {
+	return &Config{
+		KafkaSourceDefaults: s.UntypedLoad(KafkaDefaultsConfigName).(*KafkaSourceDefaults).DeepCopy(),
+	}
+}
diff --git a/pkg/apis/sources/v1beta1/kafka_defaults.go b/pkg/apis/sources/v1beta1/kafka_defaults.go
index a09416fd6..241aeca89 100644
--- a/pkg/apis/sources/v1beta1/kafka_defaults.go
+++ b/pkg/apis/sources/v1beta1/kafka_defaults.go
@@ -18,13 +18,23 @@ package v1beta1
 
 import (
 	"context"
+	"strconv"
 
 	"github.com/google/uuid"
 	"k8s.io/utils/pointer"
+
+	"knative.dev/eventing-kafka/pkg/apis/sources/config"
 )
 
 const (
 	uuidPrefix = "knative-kafka-source-"
+
+	classAnnotation             = "autoscaling.knative.dev/class"
+	minScaleAnnotation          = "autoscaling.knative.dev/minScale"
+	maxScaleAnnotation          = "autoscaling.knative.dev/maxScale"
+	pollingIntervalAnnotation   = "keda.autoscaling.knative.dev/pollingInterval"
+	cooldownPeriodAnnotation    = "keda.autoscaling.knative.dev/cooldownPeriod"
+	kafkaLagThresholdAnnotation = "keda.autoscaling.knative.dev/kafkaLagThreshold"
 )
 
 // SetDefaults ensures KafkaSource reflects the default values.
@@ -40,4 +50,20 @@ func (k *KafkaSource) SetDefaults(ctx context.Context) {
 	if k.Spec.InitialOffset == "" {
 		k.Spec.InitialOffset = OffsetLatest
 	}
+
+	kafkaConfig := config.FromContextOrDefaults(ctx)
+	kafkaDefaults := kafkaConfig.KafkaSourceDefaults
+	if kafkaDefaults.AutoscalingClass == config.KedaAutoscalingClass {
+		if k.Annotations == nil {
+			k.Annotations = map[string]string{}
+		}
+		k.Annotations[classAnnotation] = kafkaDefaults.AutoscalingClass
+
+		// Set all annotations regardless of defaults
+		k.Annotations[minScaleAnnotation] = strconv.FormatInt(kafkaDefaults.MinScale, 10)
+		k.Annotations[maxScaleAnnotation] = strconv.FormatInt(kafkaDefaults.MaxScale, 10)
+		k.Annotations[pollingIntervalAnnotation] = strconv.FormatInt(kafkaDefaults.PollingInterval, 10)
+		k.Annotations[cooldownPeriodAnnotation] = strconv.FormatInt(kafkaDefaults.CooldownPeriod, 10)
+		k.Annotations[kafkaLagThresholdAnnotation] = strconv.FormatInt(kafkaDefaults.KafkaLagThreshold, 10)
+	}
 }
diff --git a/vendor/knative.dev/pkg/configmap/hash-gen/main.go b/vendor/knative.dev/pkg/configmap/hash-gen/main.go
new file mode 100644
index 000000000..5846167f9
--- /dev/null
+++ b/vendor/knative.dev/pkg/configmap/hash-gen/main.go
@@ -0,0 +1,141 @@
+/*
+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 main
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"os"
+
+	"gopkg.in/yaml.v3"
+	"knative.dev/pkg/configmap"
+)
+
+func main() {
+	for _, fileName := range os.Args[1:] {
+		if err := processFile(fileName); err != nil {
+			log.Fatal(err)
+		}
+	}
+}
+
+// processFile reads the ConfigMap manifest from a file and adds or updates the label
+// containing the checksum of it's _example data if present.
+func processFile(fileName string) error {
+	in, err := ioutil.ReadFile(fileName)
+	if err != nil {
+		return fmt.Errorf("failed to read file: %w", err)
+	}
+
+	out, err := process(in)
+	if out == nil || err != nil {
+		return err
+	}
+
+	//nolint:gosec // This is not security critical so open permissions are fine.
+	if err := ioutil.WriteFile(fileName, out, 0644); err != nil {
+		return fmt.Errorf("failed to write file: %w", err)
+	}
+	return nil
+}
+
+// process processes a YAML file's bytes and adds or updates the label containing
+// the checksum of it's _example data if present.
+func process(data []byte) ([]byte, error) {
+	var doc yaml.Node
+	if err := yaml.Unmarshal(data, &doc); err != nil {
+		return nil, fmt.Errorf("failed to parse YAML: %w", err)
+	}
+	content := doc.Content[0]
+
+	example := traverse(content, "data", configmap.ExampleKey)
+	if example == nil {
+		return nil, nil
+	}
+
+	metadata := traverse(content, "metadata")
+	if metadata == nil {
+		return nil, errors.New("'metadata' not found")
+	}
+
+	annotations := traverse(metadata, "annotations")
+	if annotations == nil {
+		annotations = &yaml.Node{Kind: yaml.MappingNode}
+		metadata.Content = append(metadata.Content, strNode("annotations"), annotations)
+	}
+
+	checksum := configmap.Checksum(example.Value)
+	existingAnnotation := value(annotations, configmap.ExampleChecksumAnnotation)
+	if existingAnnotation != nil {
+		existingAnnotation.Value = checksum
+	} else {
+		sumNode := strNode(checksum)
+		sumNode.Style = yaml.DoubleQuotedStyle
+		annotations.Content = append(annotations.Content,
+			strNode(configmap.ExampleChecksumAnnotation), sumNode)
+	}
+
+	var buffer bytes.Buffer
+	buffer.Grow(len(data))
+	encoder := yaml.NewEncoder(&buffer)
+	encoder.SetIndent(2)
+	if err := encoder.Encode(&doc); err != nil {
+		return nil, fmt.Errorf("failed to encode YAML: %w", err)
+	}
+	return buffer.Bytes(), nil
+}
+
+// traverse traverses the YAML nodes' children using the given path keys. Returns nil if
+// one of the keys can't be found.
+func traverse(parent *yaml.Node, path ...string) *yaml.Node {
+	if parent == nil || len(path) == 0 {
+		return parent
+	}
+	tail := path[1:]
+	child := value(parent, path[0])
+	return traverse(child, tail...)
+}
+
+// value returns the value of the current node under the given key. Returns nil if not
+// found.
+func value(parent *yaml.Node, key string) *yaml.Node {
+	for i := range parent.Content {
+		if parent.Content[i].Value == key {
+			// The Content array of a yaml.Node is just an array of more nodes. The keys
+			// to each field are therefore a string node right before the relevant field's
+			// value, hence we need to return the next value of the array to retrieve
+			// the value under a key.
+			if len(parent.Content) < i+2 {
+				return nil
+			}
+			return parent.Content[i+1]
+		}
+	}
+	return nil
+}
+
+// strNode generate a node that forces the representation to be a string.
+func strNode(value string) *yaml.Node {
+	return &yaml.Node{
+		Kind:  yaml.ScalarNode,
+		Tag:   "!!str",
+		Value: value,
+	}
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index bfad20c8f..89b6b7f6e 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1102,7 +1102,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/control-protocol v0.0.0-20210830193954-f9f8d9068ed7
+# knative.dev/control-protocol v0.0.0-20210903191859-3cb84f4563fd
 ## explicit
 knative.dev/control-protocol/pkg
 knative.dev/control-protocol/pkg/certificates
@@ -1110,7 +1110,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.25.1-0.20210903045557-38bfba7d7d21
+# knative.dev/eventing v0.25.1-0.20210906161343-38f38456f045
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1237,10 +1237,10 @@ knative.dev/eventing/test/upgrade/prober/wathola/sender
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/networking v0.0.0-20210903055657-5f2e230c665f
+# knative.dev/networking v0.0.0-20210907131619-be5cfdebc4e1
 ## explicit
 knative.dev/networking/pkg/prober
-# knative.dev/pkg v0.0.0-20210903055557-dec98b426132
+# knative.dev/pkg v0.0.0-20210907112319-73d4fe679a2e
 ## explicit
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate
@@ -1283,7 +1283,9 @@ knative.dev/pkg/codegen/cmd/injection-gen
 knative.dev/pkg/codegen/cmd/injection-gen/args
 knative.dev/pkg/codegen/cmd/injection-gen/generators
 knative.dev/pkg/configmap
+knative.dev/pkg/configmap/hash-gen
 knative.dev/pkg/configmap/informer
+knative.dev/pkg/configmap/testing
 knative.dev/pkg/controller
 knative.dev/pkg/environment
 knative.dev/pkg/hack


Test patch:
diff --git a/pkg/apis/sources/config/kafka_defaults_test.go b/pkg/apis/sources/config/kafka_defaults_test.go
new file mode 100644
index 000000000..7436b2ccf
--- /dev/null
+++ b/pkg/apis/sources/config/kafka_defaults_test.go
@@ -0,0 +1,274 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package config
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	. "knative.dev/pkg/configmap/testing"
+	"knative.dev/pkg/system"
+	_ "knative.dev/pkg/system/testing"
+)
+
+func TestNewKafkaDefaultsConfigFromConfigMap(t *testing.T) {
+	_, example := ConfigMapsFromTestFile(t, KafkaDefaultsConfigName)
+	if _, err := NewKafkaDefaultsConfigFromConfigMap(example); err != nil {
+		t.Error("NewKafkaDefaultsConfigFromMap(example) =", err)
+	}
+}
+
+func TestKafkaDefaultsConfiguration(t *testing.T) {
+	testCases := []struct {
+		name                  string
+		wantErr               bool
+		wantDefault           KafkaSourceDefaults
+		wantClass             string
+		wantMinScale          int64
+		wantMaxScale          int64
+		wantPollingInterval   int64
+		wantCooldownPeriod    int64
+		wantKafkaLagThreshold int64
+		config                *corev1.ConfigMap
+	}{{
+		name:    "default config",
+		wantErr: false,
+		wantDefault: KafkaSourceDefaults{
+			AutoscalingClass: "",
+		},
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{},
+		},
+	}, {
+		name:    "example text",
+		wantErr: false,
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"_example": `
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that kubectl edit this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+
+    # autoscalingClass is the autoscaler class name to use.
+    # valid value: keda.autoscaling.knative.dev
+    autoscalingClass: "keda.autoscaling.knative.dev"
+
+    # minScale is the minimum number of replicas to scale down to.
+    # minScale: "1"
+
+    # maxScale is the maximum number of replicas to scale up to.
+    # maxScale: "1"
+
+    # pollingInterval is the interval in seconds KEDA uses to poll metrics.
+    # pollingInterval: "30"
+
+    # cooldownPeriod is the period of time in seconds KEDA waits until it scales down.
+    # cooldownPeriod: "300"
+
+    # kafkaLagThreshold is the lag (ie. number of messages in a partition) threshold for KEDA to scale up sources.
+    # kafkaLagThreshold: "10"
+`,
+			},
+		},
+	}, {
+		name:    "valid autoscaler class",
+		wantErr: false,
+		wantDefault: KafkaSourceDefaults{
+			AutoscalingClass:  "keda.autoscaling.knative.dev",
+			MinScale:          DefaultMinScaleValue,
+			MaxScale:          DefaultMaxScaleValue,
+			PollingInterval:   DefaultPollingIntervalValue,
+			CooldownPeriod:    DefaultCooldownPeriodValue,
+			KafkaLagThreshold: DefaultKafkaLagThresholdValue,
+		},
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"autoscalingClass": "keda.autoscaling.knative.dev",
+			},
+		},
+	}, {
+		name:    "invalid autoscaler class",
+		wantErr: true,
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"autoscalingClass": "invalid",
+			},
+		},
+	}, {
+		name:    "dangling key/value pair",
+		wantErr: true,
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"autoscalingClass": "#nothing to see here",
+			},
+		},
+	}, {
+		name:    "change minScale default",
+		wantErr: false,
+		wantDefault: KafkaSourceDefaults{
+			AutoscalingClass:  "keda.autoscaling.knative.dev",
+			MinScale:          40,
+			MaxScale:          DefaultMaxScaleValue,
+			PollingInterval:   DefaultPollingIntervalValue,
+			CooldownPeriod:    DefaultCooldownPeriodValue,
+			KafkaLagThreshold: DefaultKafkaLagThresholdValue,
+		},
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"autoscalingClass": "keda.autoscaling.knative.dev",
+				"minScale":         "40",
+			},
+		},
+	}, {
+		name:    "change maxScale default",
+		wantErr: false,
+		wantDefault: KafkaSourceDefaults{
+			AutoscalingClass:  "keda.autoscaling.knative.dev",
+			MinScale:          DefaultMinScaleValue,
+			MaxScale:          60,
+			PollingInterval:   DefaultPollingIntervalValue,
+			CooldownPeriod:    DefaultCooldownPeriodValue,
+			KafkaLagThreshold: DefaultKafkaLagThresholdValue,
+		},
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"autoscalingClass": "keda.autoscaling.knative.dev",
+				"maxScale":         "60",
+			},
+		},
+	}, {
+		name:    "change pollingInterval default",
+		wantErr: false,
+		wantDefault: KafkaSourceDefaults{
+			AutoscalingClass:  "keda.autoscaling.knative.dev",
+			MinScale:          DefaultMinScaleValue,
+			MaxScale:          DefaultMaxScaleValue,
+			PollingInterval:   500,
+			CooldownPeriod:    DefaultCooldownPeriodValue,
+			KafkaLagThreshold: DefaultKafkaLagThresholdValue,
+		},
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"autoscalingClass": "keda.autoscaling.knative.dev",
+				"pollingInterval":  "500",
+			},
+		},
+	}, {
+		name:    "change cooldownPeriod default",
+		wantErr: false,
+		wantDefault: KafkaSourceDefaults{
+			AutoscalingClass:  "keda.autoscaling.knative.dev",
+			MinScale:          DefaultMinScaleValue,
+			MaxScale:          DefaultMaxScaleValue,
+			PollingInterval:   DefaultPollingIntervalValue,
+			CooldownPeriod:    900,
+			KafkaLagThreshold: DefaultKafkaLagThresholdValue,
+		},
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"autoscalingClass": "keda.autoscaling.knative.dev",
+				"cooldownPeriod":   "900",
+			},
+		},
+	}, {
+		name:    "change kafkaLagThreshold default",
+		wantErr: false,
+		wantDefault: KafkaSourceDefaults{
+			AutoscalingClass:  "keda.autoscaling.knative.dev",
+			MinScale:          DefaultMinScaleValue,
+			MaxScale:          DefaultMaxScaleValue,
+			PollingInterval:   DefaultPollingIntervalValue,
+			CooldownPeriod:    DefaultCooldownPeriodValue,
+			KafkaLagThreshold: 800,
+		},
+		config: &corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: system.Namespace(),
+				Name:      KafkaDefaultsConfigName,
+			},
+			Data: map[string]string{
+				"autoscalingClass":  "keda.autoscaling.knative.dev",
+				"kafkaLagThreshold": "800",
+			},
+		},
+	}}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			actualDefault, err := NewKafkaDefaultsConfigFromConfigMap(tc.config)
+
+			if (err != nil) != tc.wantErr {
+				t.Fatalf("Test: %q: NewKafkaDefaultsConfigFromMap() error = %v, wantErr %v", tc.name, err, tc.wantErr)
+			}
+			if !tc.wantErr {
+				if diff := cmp.Diff(tc.wantDefault, *actualDefault); diff != "" {
+					t.Error("unexpected value (-want, +got)", diff)
+				}
+			}
+		})
+	}
+}
diff --git a/pkg/apis/sources/config/store_test.go b/pkg/apis/sources/config/store_test.go
new file mode 100644
index 000000000..bb0f8e941
--- /dev/null
+++ b/pkg/apis/sources/config/store_test.go
@@ -0,0 +1,63 @@
+/*
+Copyright 2021 The Knative Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package config
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"k8s.io/apimachinery/pkg/api/resource"
+	logtesting "knative.dev/pkg/logging/testing"
+
+	. "knative.dev/pkg/configmap/testing"
+)
+
+var ignoreStuff = cmp.Options{
+	cmpopts.IgnoreUnexported(resource.Quantity{}),
+}
+
+func TestStoreLoadWithContext(t *testing.T) {
+	store := NewStore(logtesting.TestLogger(t))
+
+	_, defaultsConfig := ConfigMapsFromTestFile(t, KafkaDefaultsConfigName)
+
+	store.OnConfigChanged(defaultsConfig)
+
+	config := FromContextOrDefaults(store.ToContext(context.Background()))
+
+	t.Run("defaults", func(t *testing.T) {
+		expected, _ := NewKafkaDefaultsConfigFromConfigMap(defaultsConfig)
+		if diff := cmp.Diff(expected, config.KafkaSourceDefaults, ignoreStuff...); diff != "" {
+			t.Error("Unexpected defaults config (-want, +got):", diff)
+			t.Fatal("Unexpected defaults config (-want, +got):", diff)
+		}
+	})
+}
+
+func TestStoreLoadWithContextOrDefaults(t *testing.T) {
+	defaultsConfig := ConfigMapFromTestFile(t, KafkaDefaultsConfigName)
+	config := FromContextOrDefaults(context.Background())
+
+	t.Run("defaults", func(t *testing.T) {
+		expected, _ := NewKafkaDefaultsConfigFromConfigMap(defaultsConfig)
+		if diff := cmp.Diff(expected, config.KafkaSourceDefaults, ignoreStuff...); diff != "" {
+			t.Error("Unexpected defaults config (-want, +got):", diff)
+		}
+	})
+}
diff --git a/pkg/apis/sources/config/testdata/config-kafka-source-defaults.yaml b/pkg/apis/sources/config/testdata/config-kafka-source-defaults.yaml
new file mode 100644
index 000000000..72b605fed
--- /dev/null
+++ b/pkg/apis/sources/config/testdata/config-kafka-source-defaults.yaml
@@ -0,0 +1,56 @@
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-kafka-source-defaults
+  namespace: knative-eventing
+data:
+  _example: |
+    ################################
+    #                              #
+    #    EXAMPLE CONFIGURATION     #
+    #                              #
+    ################################
+
+    # This block is not actually functional configuration,
+    # but serves to illustrate the available configuration
+    # options and document them in a way that is accessible
+    # to users that `kubectl edit` this config map.
+    #
+    # These sample configuration options may be copied out of
+    # this example block and unindented to be in the data block
+    # to actually change the configuration.
+
+    # autoscalingClass is the autoscaler class name to use.
+    # valid value: keda.autoscaling.knative.dev
+    autoscalingClass: "keda.autoscaling.knative.dev"
+
+    # minScale is the minimum number of replicas to scale down to.
+    # minScale: "1"
+
+    # maxScale is the maximum number of replicas to scale up to.
+    # maxScale: "1"
+
+    # pollingInterval is the interval in seconds KEDA uses to poll metrics.
+    # pollingInterval: "30"
+
+    # cooldownPeriod is the period of time in seconds KEDA waits until it scales down.
+    # cooldownPeriod: "300"
+
+    # kafkaLagThreshold is the lag (ie. number of messages in a partition) threshold for KEDA to scale up sources.
+    # kafkaLagThreshold: "10"
+
+
diff --git a/pkg/apis/sources/v1beta1/kafka_defaults_test.go b/pkg/apis/sources/v1beta1/kafka_defaults_test.go
index fa36bcb0d..121df51fc 100644
--- a/pkg/apis/sources/v1beta1/kafka_defaults_test.go
+++ b/pkg/apis/sources/v1beta1/kafka_defaults_test.go
@@ -25,81 +25,112 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
 	"k8s.io/utils/pointer"
+	"knative.dev/eventing-kafka/pkg/apis/sources/config"
 )
 
+type assertFnType func(t *testing.T, ks KafkaSource, expected interface{})
+
 type defaultKafkaTestArgs struct {
-	Name       string
-	Initial    KafkaSource
-	Expected   string
-	AssertFunc func(t *testing.T, ks KafkaSource, expected string)
+	Name        string
+	Defaults    config.KafkaSourceDefaults
+	Initial     KafkaSource
+	Expected    interface{}
+	AssertFuncs []assertFnType
 }
 
 func TestSetDefaults(t *testing.T) {
-	assertUUID := func(t *testing.T, ks KafkaSource, expected string) {
+	assertUUID := func(t *testing.T, ks KafkaSource, expected interface{}) {
 		consumerGroup := strings.Split(ks.Spec.ConsumerGroup, uuidPrefix)
 		_, err := uuid.Parse(consumerGroup[len(consumerGroup)-1])
 		if err != nil {
 			t.Fatalf("Error Parsing UUID value: %s", err)
 		}
 	}
-	assertGivenGroup := func(t *testing.T, ks KafkaSource, expected string) {
+	assertGivenGroup := func(t *testing.T, ks KafkaSource, expected interface{}) {
 		if diff := cmp.Diff(ks.Spec.ConsumerGroup, expected); diff != "" {
 			t.Fatalf("Unexpected consumerGroup Set (-want, +got): %s", diff)
 		}
 	}
-	assertConsumers := func(t *testing.T, ks KafkaSource, expected string) {
-		i, _ := strconv.Atoi(expected)
+	assertConsumers := func(t *testing.T, ks KafkaSource, expected interface{}) {
+		i, _ := strconv.Atoi(expected.(string))
 		i32 := int32(i)
 		if diff := cmp.Diff(ks.Spec.Consumers, &i32); diff != "" {
 			t.Fatalf("Unexpected consumers (-want, +got): %s", diff)
 		}
 	}
-	assertOffset := func(t *testing.T, ks KafkaSource, expected string) {
+	assertOffset := func(t *testing.T, ks KafkaSource, expected interface{}) {
 		if diff := cmp.Diff(ks.Spec.InitialOffset, OffsetLatest); diff != "" {
 			t.Fatalf("Unexpected initial offset (-want, +got): %s", diff)
 		}
 	}
+	assertNoAnnotations := func(t *testing.T, ks KafkaSource, expected interface{}) {
+		if len(ks.Annotations) != 0 {
+			t.Fatalf("Unexpected annotations: %v", ks.Annotations)
+		}
+	}
+	assertAnnotations := func(t *testing.T, ks KafkaSource, expected interface{}) {
+		if diff := cmp.Diff(ks.Annotations, expected); diff != "" {
+			t.Fatalf("Unexpected annotations (-want, +got): %s", diff)
+		}
+	}
 	testCases := []defaultKafkaTestArgs{
 		{
-			Name:       "nil spec",
-			Initial:    KafkaSource{},
-			AssertFunc: assertUUID,
-		},
-		{
+			Name:        "nil spec",
+			Initial:     KafkaSource{},
+			AssertFuncs: []assertFnType{assertUUID, assertNoAnnotations},
+		}, {
 			Name: "Set consumerGroup",
 			Initial: KafkaSource{
 				Spec: KafkaSourceSpec{
 					ConsumerGroup: "foo",
 				},
 			},
-			Expected:   "foo",
-			AssertFunc: assertGivenGroup,
-		},
-		{
-			Name:       "consumers not set",
-			Initial:    KafkaSource{},
-			Expected:   "1",
-			AssertFunc: assertConsumers,
-		},
-		{
-			Name:       "consumers set",
-			Initial:    KafkaSource{Spec: KafkaSourceSpec{Consumers: pointer.Int32Ptr(4)}},
-			Expected:   "4",
-			AssertFunc: assertConsumers,
-		},
-		{
-			Name:       "offset set",
-			Initial:    KafkaSource{},
-			Expected:   "",
-			AssertFunc: assertOffset,
+			Expected:    "foo",
+			AssertFuncs: []assertFnType{assertGivenGroup, assertNoAnnotations},
+		}, {
+			Name:        "consumers not set",
+			Initial:     KafkaSource{},
+			Expected:    "1",
+			AssertFuncs: []assertFnType{assertConsumers, assertNoAnnotations},
+		}, {
+			Name:        "consumers set",
+			Initial:     KafkaSource{Spec: KafkaSourceSpec{Consumers: pointer.Int32Ptr(4)}},
+			Expected:    "4",
+			AssertFuncs: []assertFnType{assertConsumers, assertNoAnnotations},
+		}, {
+			Name:        "offset set",
+			Initial:     KafkaSource{},
+			Expected:    "",
+			AssertFuncs: []assertFnType{assertOffset, assertNoAnnotations},
+		}, {
+			Name: "autoscaling config",
+			Defaults: config.KafkaSourceDefaults{
+				AutoscalingClass:  "keda.autoscaling.knative.dev",
+				MinScale:          40,
+				MaxScale:          60,
+				PollingInterval:   500,
+				CooldownPeriod:    4000,
+				KafkaLagThreshold: 100,
+			},
+			Initial: KafkaSource{},
+			Expected: map[string]string{
+				classAnnotation:             "keda.autoscaling.knative.dev",
+				minScaleAnnotation:          "40",
+				maxScaleAnnotation:          "60",
+				pollingIntervalAnnotation:   "500",
+				cooldownPeriodAnnotation:    "4000",
+				kafkaLagThresholdAnnotation: "100",
+			},
+			AssertFuncs: []assertFnType{assertAnnotations},
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.Name, func(t *testing.T) {
-			tc.Initial.SetDefaults(context.TODO())
-			if tc.AssertFunc != nil {
-				tc.AssertFunc(t, tc.Initial, tc.Expected)
+			ctx := config.ToContext(context.Background(), &config.Config{KafkaSourceDefaults: &tc.Defaults})
+			tc.Initial.SetDefaults(ctx)
+			for _, assertFunc := range tc.AssertFuncs {
+				assertFunc(t, tc.Initial, tc.Expected)
 			}
 		})
 	}
diff --git a/vendor/knative.dev/eventing/test/lib/test_runner.go b/vendor/knative.dev/eventing/test/lib/test_runner.go
index ebf08468e..9ee78d0ae 100644
--- a/vendor/knative.dev/eventing/test/lib/test_runner.go
+++ b/vendor/knative.dev/eventing/test/lib/test_runner.go
@@ -48,7 +48,7 @@ import (
 const (
 	podLogsDir         = "pod-logs"
 	testPullSecretName = "kn-eventing-test-pull-secret"
-	MaxNamespaceSkip   = 20
+	MaxNamespaceSkip   = 50
 	MaxRetries         = 5
 	RetrySleepDuration = 2 * time.Second
 )
diff --git a/vendor/knative.dev/pkg/configmap/testing/configmap.go b/vendor/knative.dev/pkg/configmap/testing/configmap.go
new file mode 100644
index 000000000..909036892
--- /dev/null
+++ b/vendor/knative.dev/pkg/configmap/testing/configmap.go
@@ -0,0 +1,109 @@
+/*
+Copyright 2019 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package testing
+
+import (
+	"fmt"
+	"io/ioutil"
+	"strings"
+	"testing"
+	"unicode"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"knative.dev/pkg/configmap"
+	"sigs.k8s.io/yaml"
+)
+
+// ConfigMapFromTestFile creates a v1.ConfigMap from a YAML file
+// It loads the YAML file from the testdata folder.
+func ConfigMapFromTestFile(t testing.TB, name string, allowed ...string) *corev1.ConfigMap {
+	t.Helper()
+
+	cm, _ := ConfigMapsFromTestFile(t, name, allowed...)
+	return cm
+}
+
+// ConfigMapsFromTestFile creates two corev1.ConfigMap resources from the config
+// file read from the testdata directory:
+// 1. The raw configmap read in.
+// 2. A second version of the configmap augmenting `data:` with what's parsed from the value of `_example:`
+func ConfigMapsFromTestFile(t testing.TB, name string, allowed ...string) (*corev1.ConfigMap, *corev1.ConfigMap) {
+	t.Helper()
+
+	b, err := ioutil.ReadFile(fmt.Sprintf("testdata/%s.yaml", name))
+	if err != nil {
+		t.Fatal("ReadFile() =", err)
+	}
+
+	var orig corev1.ConfigMap
+
+	// Use sigs.k8s.io/yaml since it reads json struct
+	// tags so things unmarshal properly.
+	if err := yaml.Unmarshal(b, &orig); err != nil {
+		t.Fatal("yaml.Unmarshal() =", err)
+	}
+
+	// We expect each of the allowed keys, and a key holding an example
+	// configuration for us to validate.
+	allowed = append(allowed, configmap.ExampleKey)
+
+	if len(orig.Data) != len(allowed) {
+		// See here for why we only check in empty ConfigMaps:
+		// https://github.com/knative/serving/issues/2668
+		t.Errorf("Data = %v, wanted %v", orig.Data, allowed)
+	}
+	allow := sets.NewString(allowed...)
+	for key := range orig.Data {
+		if !allow.Has(key) {
+			t.Errorf("Encountered key %q in %q that wasn't on the allowed list", key, name)
+		}
+	}
+	// With the length and membership checks, we know that the keyspace matches.
+
+	exampleBody, hasExampleBody := orig.Data[configmap.ExampleKey]
+	// Check that exampleBody does not have lines that end in a trailing space.
+	for i, line := range strings.Split(exampleBody, "\n") {
+		if strings.TrimRightFunc(line, unicode.IsSpace) != line {
+			t.Errorf("line %d of %q example contains trailing spaces", i, name)
+		}
+	}
+
+	// Check that the hashed exampleBody matches the assigned annotation, if present.
+	gotChecksum, hasExampleChecksumAnnotation := orig.Annotations[configmap.ExampleChecksumAnnotation]
+	if hasExampleBody && hasExampleChecksumAnnotation {
+		wantChecksum := configmap.Checksum(exampleBody)
+		if gotChecksum != wantChecksum {
+			t.Errorf("example checksum annotation = %s, want %s (you may need to re-run ./hack/update-codegen.sh)", gotChecksum, wantChecksum)
+		}
+	}
+
+	// Parse exampleBody into exemplar.Data.
+	exemplar := orig.DeepCopy()
+	if err := yaml.Unmarshal([]byte(exampleBody), &exemplar.Data); err != nil {
+		t.Fatal("yaml.Unmarshal() =", err)
+	}
+	// Augment the sample with actual configuration.
+	for k, v := range orig.Data {
+		if _, ok := exemplar.Data[k]; ok {
+			continue
+		}
+		exemplar.Data[k] = v
+	}
+
+	return &orig, exemplar
+}
diff --git a/vendor/knative.dev/pkg/test/request.go b/vendor/knative.dev/pkg/test/request.go
index 821696f5d..fd495c121 100644
--- a/vendor/knative.dev/pkg/test/request.go
+++ b/vendor/knative.dev/pkg/test/request.go
@@ -21,7 +21,6 @@ package test
 import (
 	"context"
 	"net/url"
-	"strings"
 	"time"
 
 	"k8s.io/client-go/kubernetes"
@@ -38,11 +37,6 @@ type RequestOption = spoof.RequestOption
 // Deprecated: Use the spoof package version
 var WithHeader = spoof.WithHeader
 
-// Retrying modifies a ResponseChecker to retry certain response codes.
-//
-// Deprecated: Use the spoof package version
-var Retrying = spoof.Retrying
-
 // IsOneOfStatusCodes checks that the response code is equal to the given one.
 //
 // Deprecated: Use the spoof package version
@@ -75,15 +69,6 @@ var MatchesBody = spoof.MatchesBody
 // Deprecated: Use the spoof package version
 var MatchesAllOf = spoof.MatchesAllOf
 
-// EventuallyMatchesBody checks that the response body *eventually* matches the expected body.
-// TODO(#1178): Delete me. We don't want to need this; we should be waiting for an appropriate Status instead.
-func EventuallyMatchesBody(expected string) spoof.ResponseChecker {
-	return func(resp *spoof.Response) (bool, error) {
-		// Returning (false, nil) causes SpoofingClient.Poll to retry.
-		return strings.Contains(string(resp.Body), expected), nil
-	}
-}
-
 // WaitForEndpointState will poll an endpoint until inState indicates the state is achieved,
 // or default timeout is reached.
 // If resolvableDomain is false, it will use kubeClientset to look up the ingress and spoof
diff --git a/vendor/knative.dev/pkg/test/spoof/response_checks.go b/vendor/knative.dev/pkg/test/spoof/response_checks.go
index d5e96258f..3e22fc7cc 100644
--- a/vendor/knative.dev/pkg/test/spoof/response_checks.go
+++ b/vendor/knative.dev/pkg/test/spoof/response_checks.go
@@ -103,19 +103,3 @@ func IsOneOfStatusCodes(codes ...int) ResponseChecker {
 		return true, fmt.Errorf("status = %d %s, want one of: %v", resp.StatusCode, resp.Status, codes)
 	}
 }
-
-// Retrying modifies a ResponseChecker to retry certain response codes.
-func Retrying(rc ResponseChecker, codes ...int) ResponseChecker {
-	return func(resp *Response) (bool, error) {
-		for _, code := range codes {
-			if resp.StatusCode == code {
-				// Returning (false, nil) causes SpoofingClient.Poll to retry.
-				// sc.logger.Info("Retrying for code ", resp.StatusCode)
-				return false, nil
-			}
-		}
-
-		// If we didn't match any retryable codes, invoke the ResponseChecker that we wrapped.
-		return rc(resp)
-	}
-}
diff --git a/vendor/knative.dev/pkg/test/spoof/spoof.go b/vendor/knative.dev/pkg/test/spoof/spoof.go
index 442fd3940..f2baa4edf 100644
--- a/vendor/knative.dev/pkg/test/spoof/spoof.go
+++ b/vendor/knative.dev/pkg/test/spoof/spoof.go
@@ -164,7 +164,7 @@ func (sc *SpoofingClient) Do(req *http.Request, errorRetryCheckers ...interface{
 // If no retry checkers are specified `DefaultErrorRetryChecker` will be used.
 func (sc *SpoofingClient) Poll(req *http.Request, inState ResponseChecker, checkers ...interface{}) (*Response, error) {
 	if len(checkers) == 0 {
-		checkers = []interface{}{DefaultErrorRetryChecker}
+		checkers = []interface{}{ErrorRetryChecker(DefaultErrorRetryChecker)}
 	}
 
 	var resp *Response