Repository: knative-sandbox/eventing-kafka
Pull request #828 (2021-08-24 19:20:36)

Add retention time to kafkachannel crd
This is part of the solution for Issue #817.  Original PR for this delivery (#818) was closed due to stuck github actions - all comments from that PR had already been addressed.

## Proposed Changes

- 🗑️   Remove "topic" fields from KafkaChannel ConfigMap
- 🎁  Add RetentionDuration field to KafkaChannel Spec (CRD)
- 🐛  Default RetentionDuration field in Defaulting WebHook & remove broken usage of ConfigMap fields.


**Release Note**

```release-note
KafkaChannel CRD Spec now includes a RetentionDuration field, allowing per-channel control over retention.  A Post-Install Job (config/post-install/retentionupdate) is available to migrate existing KafkaChannels forward by populating this field.  The "topic" configuration defaults which were not working have been removed from the config-kafka ConfigMap.
```

Holding for review...
/hold

/assign @devguyio 

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/828#" title="Author self-approved">travis-minke-sap</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)~~ [travis-minke-sap]

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/channel/consolidated/reconciler/controller/kafkachannel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/828/pull-knative-sandbox-eventing-kafka-go-coverage/1430248707764260864/artifacts/line-cov.html#file4) | 54.9% | 54.8% | -0.1
[pkg/channel/distributed/controller/kafkachannel/topic.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/828/pull-knative-sandbox-eventing-kafka-go-coverage/1430248707764260864/artifacts/line-cov.html#file7) | 83.3% | 81.0% | -2.3

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#828](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (be51a85) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/2c7cc36ca74a15939437f496842cca2059291154?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2c7cc36) will **decrease** coverage by `0.13%`.
> The diff coverage is `76.47%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/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/828?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     #828      +/-   ##
==========================================
- Coverage   75.29%   75.16%   -0.14%     
==========================================
  Files         139      139              
  Lines        6440     6438       -2     
==========================================
- Hits         4849     4839      -10     
- Misses       1363     1368       +5     
- Partials      228      231       +3     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/apis/kafka/v1alpha1/reset\_offset\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X3R5cGVzLmdv) | `100.00% <ø> (ø)` | |
| [pkg/channel/consolidated/utils/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3V0aWxzL3V0aWwuZ28=) | `95.45% <ø> (-0.47%)` | :arrow_down: |
| [...kg/channel/distributed/controller/config/config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9jb25maWcvY29uZmlnLmdv) | `100.00% <ø> (ø)` | |
| [pkg/common/config/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb25maWcvdXRpbC5nbw==) | `100.00% <ø> (ø)` | |
| [test/rekt/resources/kafkachannel/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-dGVzdC9yZWt0L3Jlc291cmNlcy9rYWZrYWNoYW5uZWwva2Fma2FjaGFubmVsLmdv) | `23.52% <0.00%> (-3.14%)` | :arrow_down: |
| [...consolidated/reconciler/controller/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3JlY29uY2lsZXIvY29udHJvbGxlci9rYWZrYWNoYW5uZWwuZ28=) | `49.67% <62.50%> (-0.33%)` | :arrow_down: |
| [...annel/distributed/controller/kafkachannel/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci9rYWZrYWNoYW5uZWwvdG9waWMuZ28=) | `78.94% <62.50%> (-3.00%)` | :arrow_down: |
| [...g/apis/messaging/v1beta1/kafka\_channel\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvbWVzc2FnaW5nL3YxYmV0YTEva2Fma2FfY2hhbm5lbF9kZWZhdWx0cy5nbw==) | `100.00% <100.00%> (ø)` | |
| [pkg/apis/messaging/v1beta1/kafka\_channel\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvbWVzc2FnaW5nL3YxYmV0YTEva2Fma2FfY2hhbm5lbF90eXBlcy5nbw==) | `100.00% <100.00%> (ø)` | |
| [...apis/messaging/v1beta1/kafka\_channel\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvbWVzc2FnaW5nL3YxYmV0YTEva2Fma2FfY2hhbm5lbF92YWxpZGF0aW9uLmdv) | `100.00% <100.00%> (ø)` | |
| ... and [1 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828/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/828?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/828?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [2c7cc36...be51a85](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/828?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/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 5c593bd48..a6f9fa926 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -162,6 +162,7 @@ func main() {
 
 	// Construct The KafkaChannel Controller
 	kcController := controller.NewController(
+		ctx,
 		logger,
 		environment.ChannelKey,
 		dispatcher,
diff --git a/config/channel/consolidated/configmaps/kafka-config.yaml b/config/channel/consolidated/configmaps/kafka-config.yaml
index 3bd531733..5d55f9e4f 100644
--- a/config/channel/consolidated/configmaps/kafka-config.yaml
+++ b/config/channel/consolidated/configmaps/kafka-config.yaml
@@ -22,10 +22,6 @@ data:
   eventing-kafka: |
     kafka:
       brokers: REPLACE_WITH_CLUSTER_URL
-      topic:
-        defaultNumPartitions: 4
-        defaultReplicationFactor: 1 # Cannot exceed the number of Kafka Brokers!
-        defaultRetentionMillis: 604800000  # 1 week
 kind: ConfigMap
 metadata:
   name: config-kafka
diff --git a/config/channel/distributed/300-eventing-kafka-configmap.yaml b/config/channel/distributed/300-eventing-kafka-configmap.yaml
index 11b1a077f..419f62d52 100644
--- a/config/channel/distributed/300-eventing-kafka-configmap.yaml
+++ b/config/channel/distributed/300-eventing-kafka-configmap.yaml
@@ -47,10 +47,6 @@ data:
       authSecretName: kafka-cluster
       authSecretNamespace: knative-eventing
       brokers: REPLACE_WITH_CLUSTER_URL
-      topic:
-        defaultNumPartitions: 4
-        defaultReplicationFactor: 1 # Cannot exceed the number of Kafka Brokers!
-        defaultRetentionMillis: 604800000  # 1 week
     channel:
       adminType: kafka # One of "kafka", "azure", "custom"
       dispatcher:
diff --git a/config/channel/resources/kafkachannel-crd.yaml b/config/channel/resources/kafkachannel-crd.yaml
index ac7f4d6f3..77eb1fce8 100644
--- a/config/channel/resources/kafkachannel-crd.yaml
+++ b/config/channel/resources/kafkachannel-crd.yaml
@@ -48,6 +48,9 @@ spec:
                   type: integer
                   maximum: 32767
                   default: 1
+                retentionDuration:
+                  description: RetentionDuration is the retention time for events in a Kafka Topic represented as an ISO-8601 Duration.  By default it is set to 168 hours, which is the precise form of 7 days.
+                  type: string
                 delivery:
                   description: DeliverySpec contains the default delivery spec for each subscription to this Channelable. Each subscription delivery spec, if any, overrides this global delivery spec.
                   type: object
diff --git a/go.mod b/go.mod
index c20e45680..5387920ba 100644
--- a/go.mod
+++ b/go.mod
@@ -20,6 +20,7 @@ require (
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/mitchellh/mapstructure v1.3.3 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
+	github.com/rickb777/date v1.13.0
 	github.com/slinkydeveloper/loadastic v0.0.0-20201218203601-5c69eea3b7d8
 	github.com/stretchr/testify v1.7.0
 	github.com/xdg/scram v1.0.3
@@ -35,9 +36,9 @@ require (
 	k8s.io/client-go v0.20.7
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	knative.dev/control-protocol v0.0.0-20210823140136-471903b6c44b
-	knative.dev/eventing v0.25.1-0.20210823153835-b2700c2dcf57
+	knative.dev/eventing v0.25.1-0.20210824204225-4f6f3075d2ce
 	knative.dev/hack v0.0.0-20210806075220-815cd312d65c
-	knative.dev/networking v0.0.0-20210823062535-cfb7e2379d08
-	knative.dev/pkg v0.0.0-20210823133635-12d915ede293
+	knative.dev/networking v0.0.0-20210824140523-51512a042e23
+	knative.dev/pkg v0.0.0-20210824120823-a94f5f07b3c3
 	knative.dev/reconciler-test v0.0.0-20210820180205-a25de6a08087
 )
diff --git a/go.sum b/go.sum
index 7fa7f9f8c..c8b7f91fd 100644
--- a/go.sum
+++ b/go.sum
@@ -1274,18 +1274,17 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920 h1:CbnUZsM497iRC5QMVkHwyl8s2tB3g
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210823140136-471903b6c44b h1:1kfFg/0CbbaVyNlT0JCYQ/6IctbUibaDiNFJRUShnKA=
 knative.dev/control-protocol v0.0.0-20210823140136-471903b6c44b/go.mod h1:9gYmAq6zVKUA+p1+DXslH91jtga5q2z+YJWcH+z88EE=
-knative.dev/eventing v0.25.1-0.20210823153835-b2700c2dcf57 h1:dIZgOeO7vWeCFBVE2vM9UCseOmw8YaKTW4rzizO4viU=
-knative.dev/eventing v0.25.1-0.20210823153835-b2700c2dcf57/go.mod h1:qGd3YZqi++xwk0WNaX0VYMjpywq2JWptL6l+uR+Jhh4=
+knative.dev/eventing v0.25.1-0.20210824204225-4f6f3075d2ce h1:OHUkMUIFbsmOjINnoB0+q4GqdmQ79k1VHbHbGH9toek=
+knative.dev/eventing v0.25.1-0.20210824204225-4f6f3075d2ce/go.mod h1:J6C3diGWNurw71eAPLuVeILNR4SSqu+YSMePyovq0rY=
 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-20210823062535-cfb7e2379d08 h1:paj2ByFR/7l0WuqCRHlbr2vl1QUc9qXn2+L2G0lkZXk=
-knative.dev/networking v0.0.0-20210823062535-cfb7e2379d08/go.mod h1:4lsqjd2G5aHfh0zJcA+G9F2gh0r0XSYn9vEygCLXDd4=
+knative.dev/networking v0.0.0-20210824140523-51512a042e23 h1:1V//6gQw95MecpJFICaDZwIYlK8gFflXMjt9ZNhnIss=
+knative.dev/networking v0.0.0-20210824140523-51512a042e23/go.mod h1:UmOQHaS6K3xngvhW5SyEOFy2nFltL9BtvroDjysYXSw=
 knative.dev/pkg v0.0.0-20210803160015-21eb4c167cc5/go.mod h1:RPk5txNA3apR9X40D4MpUOP9/VqOG8CrtABWfOwGVS4=
-knative.dev/pkg v0.0.0-20210822203034-3bed4dcf179c/go.mod h1:t6WzDPywQloi+AbOtiOejfpem1mLfKeuV+sjXGQLGN8=
-knative.dev/pkg v0.0.0-20210823133635-12d915ede293 h1:6GvpU4ESZxfb+trIhvMTZcZSfGGeIf+F1aG7Fv3R9gA=
-knative.dev/pkg v0.0.0-20210823133635-12d915ede293/go.mod h1:67yUYAgAWesv4ehCShUuIxkaxrsROSLQr3kApW4xPO4=
+knative.dev/pkg v0.0.0-20210824120823-a94f5f07b3c3 h1:38SNaSkVBI1Fkc4Lk2AMEjPW03h3cAwTS3+dKK9RV+M=
+knative.dev/pkg v0.0.0-20210824120823-a94f5f07b3c3/go.mod h1:67yUYAgAWesv4ehCShUuIxkaxrsROSLQr3kApW4xPO4=
 knative.dev/reconciler-test v0.0.0-20210803183715-b61cc77c06f6/go.mod h1:+Kovy+G5zXZNcuO/uB+zfo37vFKZzsLIlWezt/nKMz0=
 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=
diff --git a/hack/README.md b/hack/README.md
index 649f9120d..00b4fda5e 100644
--- a/hack/README.md
+++ b/hack/README.md
@@ -1,7 +1,7 @@
 # Assorted scripts for development
 
 This directory contains several scripts useful in the development process of
-Knative Eventing Sources.
+Knative Eventing Kafka.
 
 - `boilerplate-go.txt` Used by kubebuilder to set the boilerplate.
 - `release.sh` Creates a new release.
diff --git a/hack/release.sh b/hack/release.sh
index 167978542..a41b0fcab 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -46,16 +46,22 @@ function build_release() {
   local all_yamls=()
   for yaml in "${!COMPONENTS[@]}"; do
     local config="${COMPONENTS[${yaml}]}"
-    echo "Building Knative Eventing Contrib - ${config}"
+    echo "Building Knative Eventing Kafka - ${config}"
     ko resolve ${KO_FLAGS} -f ${config}/ | "${LABEL_YAML_CMD[@]}" > ${yaml}
     all_yamls+=(${yaml})
   done
 
-  if [ -d "${YAML_REPO_ROOT}/config/post-install" ]; then
-    echo "Resolving post install manifests"
+  if [ -d "${REPO_ROOT_DIR}/config/channel/post-install" ]; then
+    echo "Resolving channel post-install manifests"
+    local yaml="channel-post-install.yaml"
+    ko resolve ${KO_FLAGS} -f config/channel/post-install | "${LABEL_YAML_CMD[@]}" > ${yaml}
+    all_yamls+=(${yaml})
+  fi
 
-    local yaml="post-install.yaml"
-    ko resolve ${KO_FLAGS} -f config/post-install/ | "${LABEL_YAML_CMD[@]}" > ${yaml}
+  if [ -d "${REPO_ROOT_DIR}/config/source/post-install" ]; then
+    echo "Resolving source post-install manifests"
+    local yaml="source-post-install.yaml"
+    ko resolve ${KO_FLAGS} -f config/source/post-install | "${LABEL_YAML_CMD[@]}" > ${yaml}
     all_yamls+=(${yaml})
   fi
 
diff --git a/pkg/apis/kafka/v1alpha1/reset_offset_types.go b/pkg/apis/kafka/v1alpha1/reset_offset_types.go
index b74ede068..53a59898b 100644
--- a/pkg/apis/kafka/v1alpha1/reset_offset_types.go
+++ b/pkg/apis/kafka/v1alpha1/reset_offset_types.go
@@ -90,7 +90,7 @@ type ResetOffsetSpec struct {
 //       in order to support specifying explicit offset (int64) values for each Partition.
 type OffsetSpec struct {
 
-	// Time is an string representing the desired offset position to which all partitions
+	// Time is a string representing the desired offset position to which all partitions
 	// will be reset.  Supported values include "earliest", "latest", or a valid date / time
 	// string in the time.RFC3339 format. The "earliest" and "latest" values indicate the
 	// beginning and end, respectively, of the persistence window of the Topic.  There is no
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go b/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
index b68601142..68a82bea7 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_defaults.go
@@ -19,8 +19,9 @@ package v1beta1
 import (
 	"context"
 
-	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/eventing/pkg/apis/messaging"
+
+	"knative.dev/eventing-kafka/pkg/common/constants"
 )
 
 func (kc *KafkaChannel) SetDefaults(ctx context.Context) {
@@ -47,4 +48,7 @@ func (kcs *KafkaChannelSpec) SetDefaults(ctx context.Context) {
 	if kcs.ReplicationFactor == 0 {
 		kcs.ReplicationFactor = constants.DefaultReplicationFactor
 	}
+	if len(kcs.RetentionDuration) <= 0 {
+		kcs.RetentionDuration = constants.DefaultRetentionISO8601Duration
+	}
 }
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_types.go b/pkg/apis/messaging/v1beta1/kafka_channel_types.go
index 4d6891fc0..f39b6b4b7 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_types.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_types.go
@@ -17,6 +17,9 @@ limitations under the License.
 package v1beta1
 
 import (
+	"time"
+
+	"github.com/rickb777/date/period"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -67,10 +70,28 @@ type KafkaChannelSpec struct {
 	// ReplicationFactor is the replication factor of a Kafka topic. By default, it is set to 1.
 	ReplicationFactor int16 `json:"replicationFactor"`
 
+	// RetentionDuration is the duration for which events will be retained in the Kafka Topic.
+	// By default, it is set to 168 hours, which is the precise form for 7 days.
+	// More information on Duration format:
+	//  - https://www.iso.org/iso-8601-date-and-time-format.html
+	//  - https://en.wikipedia.org/wiki/ISO_8601
+	RetentionDuration string `json:"retentionDuration"`
+
 	// Channel conforms to Duck type Channelable.
 	eventingduck.ChannelableSpec `json:",inline"`
 }
 
+// ParseRetentionDuration returns the parsed Offset Time if valid (RFC3339 format) or an error for invalid content.
+// Note - If the optional RetentionDuration field is not present, or is invalid, a Duration of "-1" will be returned.
+func (kcs *KafkaChannelSpec) ParseRetentionDuration() (time.Duration, error) {
+	retentionPeriod, err := period.Parse(kcs.RetentionDuration)
+	if err != nil {
+		return time.Duration(-1), err
+	}
+	retentionDuration, _ := retentionPeriod.Duration() // Ignore precision flag and accept ISO8601 estimation
+	return retentionDuration, nil
+}
+
 // KafkaChannelStatus represents the current state of a KafkaChannel.
 type KafkaChannelStatus struct {
 	// Channel conforms to Duck type ChannelableStatus.
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_validation.go b/pkg/apis/messaging/v1beta1/kafka_channel_validation.go
index e377b1c2d..6e2070521 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_validation.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_validation.go
@@ -54,6 +54,12 @@ func (kcs *KafkaChannelSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(fe)
 	}
 
+	retentionDuration, err := kcs.ParseRetentionDuration()
+	if retentionDuration < 0 || err != nil {
+		fe := apis.ErrInvalidValue(kcs.RetentionDuration, "retentionDuration")
+		errs = errs.Also(fe)
+	}
+
 	for i, subscriber := range kcs.SubscribableSpec.Subscribers {
 		if subscriber.ReplyURI == nil && subscriber.SubscriberURI == nil {
 			fe := apis.ErrMissingField("replyURI", "subscriberURI")
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index 22c921890..3aa017316 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -532,15 +532,18 @@ func (r *Reconciler) reconcileTopic(ctx context.Context, channel *v1beta1.KafkaC
 	logger.Infow("Creating topic on Kafka cluster", zap.String("topic", topicName),
 		zap.Int32("partitions", channel.Spec.NumPartitions), zap.Int16("replication", channel.Spec.ReplicationFactor))
 
-	// TODO - The eventing-kafka KafkaChannel spec does not include RetentionMillis so we're
-	//        currently just using the default value specified in the ConfigMap.  If/when the
-	//        RetentionMillis is added, any value from channel.Spec.RetentionMillis should
-	//        take precedence.
-	retentionMillisString := strconv.FormatInt(r.kafkaConfig.EventingKafka.Kafka.Topic.DefaultRetentionMillis, 10)
-
-	err := kafkaClusterAdmin.CreateTopic(topicName, &sarama.TopicDetail{
-		ReplicationFactor: commonconfig.ReplicationFactor(channel, r.kafkaConfig.EventingKafka, logger),
-		NumPartitions:     commonconfig.NumPartitions(channel, r.kafkaConfig.EventingKafka, logger),
+	// Parse & Format the RetentionDuration into Sarama retention.ms string
+	retentionDuration, err := channel.Spec.ParseRetentionDuration()
+	if err != nil {
+		// Should never happen with webhook defaulting and validation in place.
+		logger.Errorw("Error parsing RetentionDuration, using default instead", zap.String("RetentionDuration", channel.Spec.RetentionDuration), zap.Error(err))
+		retentionDuration = constants.DefaultRetentionDuration
+	}
+	retentionMillisString := strconv.FormatInt(retentionDuration.Milliseconds(), 10)
+
+	err = kafkaClusterAdmin.CreateTopic(topicName, &sarama.TopicDetail{
+		NumPartitions:     channel.Spec.NumPartitions,
+		ReplicationFactor: channel.Spec.ReplicationFactor,
 		ConfigEntries: map[string]*string{
 			constants.KafkaTopicConfigRetentionMs: &retentionMillisString,
 		},
diff --git a/pkg/channel/consolidated/utils/util.go b/pkg/channel/consolidated/utils/util.go
index 1d602485d..a154cd09f 100644
--- a/pkg/channel/consolidated/utils/util.go
+++ b/pkg/channel/consolidated/utils/util.go
@@ -24,11 +24,11 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/system"
 
 	"knative.dev/eventing-kafka/pkg/common/config"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 )
 
@@ -64,11 +64,6 @@ func GetKafkaConfig(ctx context.Context, clientId string, configMap map[string]s
 		Kafka: config.EKKafkaConfig{
 			AuthSecretName:      sarama.DefaultAuthSecretName,
 			AuthSecretNamespace: system.Namespace(),
-			Topic: config.EKKafkaTopicConfig{
-				DefaultNumPartitions:     sarama.DefaultNumPartitions,
-				DefaultReplicationFactor: sarama.DefaultReplicationFactor,
-				DefaultRetentionMillis:   sarama.DefaultRetentionMillis,
-			},
 		},
 	}
 	var err error
diff --git a/pkg/channel/distributed/controller/config/config.go b/pkg/channel/distributed/controller/config/config.go
index ee796aa3a..a7ba26dbf 100644
--- a/pkg/channel/distributed/controller/config/config.go
+++ b/pkg/channel/distributed/controller/config/config.go
@@ -46,12 +46,6 @@ func VerifyConfiguration(configuration *commonconfig.EventingKafkaConfig) error
 
 	// Verify mandatory configuration settings
 	switch {
-	case configuration.Kafka.Topic.DefaultNumPartitions < 1:
-		return ControllerConfigurationError("Kafka.Topic.DefaultNumPartitions must be > 0")
-	case configuration.Kafka.Topic.DefaultReplicationFactor < 1:
-		return ControllerConfigurationError("Kafka.Topic.DefaultReplicationFactor must be > 0")
-	case configuration.Kafka.Topic.DefaultRetentionMillis < 1:
-		return ControllerConfigurationError("Kafka.Topic.DefaultRetentionMillis must be > 0")
 	case configuration.Channel.Dispatcher.Replicas < 1:
 		return ControllerConfigurationError("Distributed.Dispatcher.Replicas must be > 0")
 	case configuration.Channel.Receiver.Replicas < 1:
diff --git a/pkg/channel/distributed/controller/kafkachannel/topic.go b/pkg/channel/distributed/controller/kafkachannel/topic.go
index 2ec5b1845..c05a1ebc8 100644
--- a/pkg/channel/distributed/controller/kafkachannel/topic.go
+++ b/pkg/channel/distributed/controller/kafkachannel/topic.go
@@ -31,7 +31,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/constants"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/event"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/controller/util"
-	"knative.dev/eventing-kafka/pkg/common/config"
 	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 )
 
@@ -44,18 +43,19 @@ func (r *Reconciler) reconcileKafkaTopic(ctx context.Context, channel *kafkav1be
 	// Get Channel-Specific Logger (From The Context) & Add Topic Name
 	logger := logging.FromContext(ctx).With(zap.String("TopicName", topicName))
 
-	// Get The Topic Configuration (First From Channel With Failover To Environment)
-	numPartitions := config.NumPartitions(channel, r.config, logger)
-	replicationFactor := config.ReplicationFactor(channel, r.config, logger)
-
-	// TODO - The eventing-kafka KafkaChannel spec does not include RetentionMillis so we're
-	//        currently just using the default value specified in the ConfigMap.  If/when the
-	//        RetentionMillis is added, any value from channel.Spec.RetentionMillis should
-	//        take precedence.
-	retentionMillis := r.config.Kafka.Topic.DefaultRetentionMillis
+	// Get The Topic Configuration From The Channel
+	numPartitions := channel.Spec.NumPartitions
+	replicationFactor := channel.Spec.ReplicationFactor
+	retentionDuration, err := channel.Spec.ParseRetentionDuration()
+	if err != nil {
+		// Should never happen with webhook defaulting and validation in place.
+		logger.Error("Failed To Parse RetentionDuration Using Default Value Instead", zap.String("RetentionDuration", channel.Spec.RetentionDuration), zap.Error(err))
+		retentionDuration = commonconstants.DefaultRetentionDuration
+	}
+	retentionMillis := retentionDuration.Milliseconds()
 
 	// Create The Topic (Handles Case Where Already Exists)
-	err := r.createTopic(ctx, topicName, numPartitions, replicationFactor, retentionMillis)
+	err = r.createTopic(ctx, topicName, numPartitions, replicationFactor, retentionMillis)
 
 	// Log Results & Return Status
 	if err != nil {
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
index c8be54863..09f6ef885 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
@@ -72,6 +72,7 @@ var _ controller.Reconciler = Reconciler{}
 // NewController initializes the controller and is called by the generated code.
 // Registers event handlers to enqueue events.
 func NewController(
+	ctx context.Context,
 	logger *zap.Logger,
 	channelKey string,
 	dispatcher dispatcher.Dispatcher,
@@ -90,7 +91,7 @@ func NewController(
 		kafkachannelLister:   kafkachannelInformer.Lister(),
 		kafkaClientSet:       kafkaClientSet,
 	}
-	reconciler.impl = controller.NewImplFull(reconciler, controller.ControllerOptions{
+	reconciler.impl = controller.NewContext(ctx, reconciler, controller.ControllerOptions{
 		WorkQueueName: ReconcilerName,
 		Logger:        logger.Sugar(),
 	})
diff --git a/pkg/common/config/eventingkafkaconfig.go b/pkg/common/config/eventingkafkaconfig.go
index 43ae01693..2bb4271e5 100644
--- a/pkg/common/config/eventingkafkaconfig.go
+++ b/pkg/common/config/eventingkafkaconfig.go
@@ -52,13 +52,6 @@ type EKDispatcherConfig struct {
 	EKKubernetesConfig
 }
 
-// EKKafkaTopicConfig contains some defaults that are only used if not provided by the channel spec
-type EKKafkaTopicConfig struct {
-	DefaultNumPartitions     int32 `json:"defaultNumPartitions,omitempty"`
-	DefaultReplicationFactor int16 `json:"defaultReplicationFactor,omitempty"`
-	DefaultRetentionMillis   int64 `json:"defaultRetentionMillis,omitempty"`
-}
-
 // EKCloudEventConfig contains the values send to the Knative cloudevents' ConfigureConnectionArgs function
 // If they are not provided in the configmap, the DefaultMaxIdleConns and DefaultMaxIdleConnsPerHost constants are used
 type EKCloudEventConfig struct {
@@ -68,10 +61,9 @@ type EKCloudEventConfig struct {
 
 // EKKafkaConfig contains items relevant to Kafka specifically
 type EKKafkaConfig struct {
-	Brokers             string             `json:"brokers,omitempty"`
-	AuthSecretName      string             `json:"authSecretName,omitempty"`
-	AuthSecretNamespace string             `json:"authSecretNamespace,omitempty"`
-	Topic               EKKafkaTopicConfig `json:"topic,omitempty"`
+	Brokers             string `json:"brokers,omitempty"`
+	AuthSecretName      string `json:"authSecretName,omitempty"`
+	AuthSecretNamespace string `json:"authSecretNamespace,omitempty"`
 }
 
 // EKSourceConfig is reserved for configuration fields needed by the Kafka Source component
diff --git a/pkg/common/config/util.go b/pkg/common/config/util.go
index e58e61366..9c74c90d2 100644
--- a/pkg/common/config/util.go
+++ b/pkg/common/config/util.go
@@ -7,12 +7,10 @@ import (
 	"strconv"
 
 	"github.com/Shopify/sarama"
-	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 
-	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/client"
 	"knative.dev/eventing-kafka/pkg/common/constants"
 )
@@ -108,26 +106,6 @@ func GetAuthConfigFromSecret(secret *corev1.Secret) *client.KafkaAuthConfig {
 	return &authConfig
 }
 
-// NumPartitions Gets The NumPartitions - First From Channel Spec And Then From ConfigMap-Provided Settings
-func NumPartitions(channel *kafkav1beta1.KafkaChannel, configuration *EventingKafkaConfig, logger *zap.SugaredLogger) int32 {
-	value := channel.Spec.NumPartitions
-	if value <= 0 && configuration != nil {
-		logger.Debug("Kafka Channel Spec 'NumPartitions' Not Specified - Using Default", zap.Int32("Value", configuration.Kafka.Topic.DefaultNumPartitions))
-		value = configuration.Kafka.Topic.DefaultNumPartitions
-	}
-	return value
-}
-
-// ReplicationFactor Gets The ReplicationFactor - First From Channel Spec And Then From ConfigMap-Provided Settings
-func ReplicationFactor(channel *kafkav1beta1.KafkaChannel, configuration *EventingKafkaConfig, logger *zap.SugaredLogger) int16 {
-	value := channel.Spec.ReplicationFactor
-	if value <= 0 && configuration != nil {
-		logger.Debug("Kafka Channel Spec 'ReplicationFactor' Not Specified - Using Default", zap.Int16("Value", configuration.Kafka.Topic.DefaultReplicationFactor))
-		value = configuration.Kafka.Topic.DefaultReplicationFactor
-	}
-	return value
-}
-
 // JoinStringMaps returns a new map containing the contents of both argument maps, preferring the contents of map1 on conflict.
 func JoinStringMaps(map1 map[string]string, map2 map[string]string) map[string]string {
 	resultMap := make(map[string]string, len(map1))
diff --git a/pkg/common/constants/constants.go b/pkg/common/constants/constants.go
index 98a0bb2b6..615679d8b 100644
--- a/pkg/common/constants/constants.go
+++ b/pkg/common/constants/constants.go
@@ -16,13 +16,22 @@ limitations under the License.
 
 package constants
 
+import "time"
+
 const (
 
 	// DefaultNumPartitions is the KafkaChannel Spec default for the number of partitions
 	DefaultNumPartitions = 1
+
 	// DefaultReplicationFactor is the KafkaChannel Spec default for the replication factor
 	DefaultReplicationFactor = 1
 
+	// DefaultRetentionISO8601Duration is the KafkaChannel Spec default for the retention duration as an ISO-8601 string
+	DefaultRetentionISO8601Duration = "PT168H" // Precise 7 Days
+
+	// DefaultRetentionDuration is the time.Duration equivalent of the DefaultRetentionISO8601Duration
+	DefaultRetentionDuration = 7 * 24 * time.Hour // Precise 7 Days
+
 	// SettingsConfigMapName is the name of the configmap used to hold eventing-kafka settings
 	SettingsConfigMapName = "config-kafka"
 
diff --git a/pkg/common/kafka/sarama/sarama.go b/pkg/common/kafka/sarama/sarama.go
index 29acaf486..1d46dc1aa 100644
--- a/pkg/common/kafka/sarama/sarama.go
+++ b/pkg/common/kafka/sarama/sarama.go
@@ -33,12 +33,7 @@ import (
 	"knative.dev/eventing-kafka/pkg/common/constants"
 )
 
-const (
-	DefaultAuthSecretName    = "kafka-cluster"
-	DefaultNumPartitions     = 4
-	DefaultReplicationFactor = 1
-	DefaultRetentionMillis   = 604800000 // 1 week
-)
+const DefaultAuthSecretName = "kafka-cluster"
 
 // EnableSaramaLogging Is A Utility Function For Enabling Sarama Logging (Debugging)
 func EnableSaramaLogging(enable bool) {
@@ -139,12 +134,11 @@ func upgradeConfig(data map[string]string) *commonconfig.EventingKafkaConfig {
 
 	// Pre-1.0.0 structs
 	type oldKafkaConfig struct {
-		Brokers             string                          `json:"brokers,omitempty"`
-		EnableSaramaLogging bool                            `json:"enableSaramaLogging,omitempty"`
-		Topic               commonconfig.EKKafkaTopicConfig `json:"topic,omitempty"`
-		AdminType           string                          `json:"adminType,omitempty"`
-		AuthSecretName      string                          `json:"authSecretName,omitempty"`
-		AuthSecretNamespace string                          `json:"authSecretNamespace,omitempty"`
+		Brokers             string `json:"brokers,omitempty"`
+		EnableSaramaLogging bool   `json:"enableSaramaLogging,omitempty"`
+		AdminType           string `json:"adminType,omitempty"`
+		AuthSecretName      string `json:"authSecretName,omitempty"`
+		AuthSecretNamespace string `json:"authSecretNamespace,omitempty"`
 	}
 	type oldEventingKafkaConfig struct {
 		Receiver    commonconfig.EKReceiverConfig   `json:"receiver,omitempty"`
@@ -171,7 +165,6 @@ func upgradeConfig(data map[string]string) *commonconfig.EventingKafkaConfig {
 			Brokers:             oldConfig.Kafka.Brokers,
 			AuthSecretName:      oldConfig.Kafka.AuthSecretName,
 			AuthSecretNamespace: oldConfig.Kafka.AuthSecretNamespace,
-			Topic:               oldConfig.Kafka.Topic,
 		},
 		Sarama: commonconfig.EKSaramaConfig{
 			EnableLogging: oldConfig.Kafka.EnableSaramaLogging,
diff --git a/pkg/source/reconciler/binding/controller.go b/pkg/source/reconciler/binding/controller.go
index 5ce75ffdf..330bb4f88 100644
--- a/pkg/source/reconciler/binding/controller.go
+++ b/pkg/source/reconciler/binding/controller.go
@@ -36,7 +36,6 @@ import (
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection/clients/dynamicclient"
 	"knative.dev/pkg/logging"
-	"knative.dev/pkg/tracker"
 	"knative.dev/pkg/webhook/psbinding"
 )
 
@@ -81,14 +80,15 @@ func NewController(
 			scheme.Scheme, corev1.EventSource{Component: controllerAgentName}),
 		NamespaceLister: namespaceInformer.Lister(),
 	}
-	impl := controller.NewImpl(c, logger, "KafkaBindings")
-
-	logger.Info("Setting up event handlers")
+	impl := controller.NewContext(ctx, c, controller.ControllerOptions{
+		Logger:        logger,
+		WorkQueueName: "KafkaBindings",
+	})
 
 	kfkInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 	namespaceInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
-	c.Tracker = tracker.New(impl.EnqueueKey, controller.GetTrackerLease(ctx))
+	c.Tracker = impl.Tracker
 	c.Factory = &duck.CachedInformerFactory{
 		Delegate: &duck.EnqueueInformerFactory{
 			Delegate:     psInformerFactory,
diff --git a/pkg/source/reconciler/mtsource/controller.go b/pkg/source/reconciler/mtsource/controller.go
index 68f170565..79dfbdc5c 100644
--- a/pkg/source/reconciler/mtsource/controller.go
+++ b/pkg/source/reconciler/mtsource/controller.go
@@ -75,7 +75,7 @@ func NewController(
 
 	impl := kafkasource.NewImpl(ctx, c)
 
-	c.sinkResolver = resolver.NewURIResolver(ctx, impl.EnqueueKey)
+	c.sinkResolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
 	// Use a different set of conditions
 	sourcesv1beta1.RegisterAlternateKafkaConditionSet(sourcesv1beta1.KafkaMTSourceCondSet)
diff --git a/pkg/source/reconciler/source/controller.go b/pkg/source/reconciler/source/controller.go
index c28fa50c3..a06f8e52c 100644
--- a/pkg/source/reconciler/source/controller.go
+++ b/pkg/source/reconciler/source/controller.go
@@ -67,16 +67,14 @@ func NewController(
 	}
 
 	impl := kafkasource.NewImpl(ctx, c)
-	c.sinkResolver = resolver.NewURIResolver(ctx, impl.EnqueueKey)
+	c.sinkResolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
 	c.claimsNotificationStore = ctrlreconciler.NewNotificationStore(impl.EnqueueKey, kafkasourcecontrol.ClaimsParser)
 
-	logging.FromContext(ctx).Info("Setting up kafka event handlers")
-
 	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
 	deploymentInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterControllerGK(v1beta1.Kind("KafkaSource")),
+		FilterFunc: controller.FilterController(&v1beta1.KafkaSource{}),
 		Handler:    controller.HandleAll(impl.EnqueueControllerOf),
 	})
 
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
index 1d77996cf..3aa377a52 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
@@ -73,9 +73,12 @@ func (imcs *InMemoryChannelStatus) GetCondition(t apis.ConditionType) *apis.Cond
 	return imcCondSet.Manage(imcs).GetCondition(t)
 }
 
-// IsReady returns true if the resource is ready overall.
-func (imcs *InMemoryChannelStatus) IsReady() bool {
-	return imcCondSet.Manage(imcs).IsHappy()
+// IsReady returns true if the Status condition InMemoryChannelConditionReady
+// is true and the latest spec has been observed.
+func (imc *InMemoryChannel) IsReady() bool {
+	imcs := imc.Status
+	return imcs.ObservedGeneration == imc.Generation &&
+		imcCondSet.Manage(&imcs).IsHappy()
 }
 
 // InitializeConditions sets relevant unset conditions to Unknown state.
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_types.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_types.go
index ca295cb52..5b0c971b5 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_types.go
@@ -71,7 +71,7 @@ type InMemoryChannelSpec struct {
 
 // ChannelStatus represents the current state of a Channel.
 type InMemoryChannelStatus struct {
-	// Channel conforms to Duck type Channelable.
+	// Channel conforms to Duck type ChannelableStatus.
 	eventingduckv1.ChannelableStatus `json:",inline"`
 }
 
diff --git a/vendor/knative.dev/pkg/configmap/parse.go b/vendor/knative.dev/pkg/configmap/parse.go
index 4c280757b..faeacab14 100644
--- a/vendor/knative.dev/pkg/configmap/parse.go
+++ b/vendor/knative.dev/pkg/configmap/parse.go
@@ -141,7 +141,11 @@ func AsDuration(key string, target *time.Duration) ParseFunc {
 func AsStringSet(key string, target *sets.String) ParseFunc {
 	return func(data map[string]string) error {
 		if raw, ok := data[key]; ok {
-			*target = sets.NewString(strings.Split(raw, ",")...)
+			splitted := strings.Split(raw, ",")
+			for i, v := range splitted {
+				splitted[i] = strings.TrimSpace(v)
+			}
+			*target = sets.NewString(splitted...)
 		}
 		return nil
 	}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 07c57633b..f9a7f1dae 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -336,6 +336,7 @@ github.com/prometheus/statsd_exporter/pkg/mapper/fsm
 ## explicit
 github.com/rcrowley/go-metrics
 # github.com/rickb777/date v1.13.0
+## explicit
 github.com/rickb777/date/period
 # github.com/rickb777/plural v1.2.1
 github.com/rickb777/plural
@@ -1037,7 +1038,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.20210823153835-b2700c2dcf57
+# knative.dev/eventing v0.25.1-0.20210824204225-4f6f3075d2ce
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1164,10 +1165,10 @@ knative.dev/eventing/test/upgrade/prober/wathola/sender
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/networking v0.0.0-20210823062535-cfb7e2379d08
+# knative.dev/networking v0.0.0-20210824140523-51512a042e23
 ## explicit
 knative.dev/networking/pkg/prober
-# knative.dev/pkg v0.0.0-20210823133635-12d915ede293
+# knative.dev/pkg v0.0.0-20210824120823-a94f5f07b3c3
 ## explicit
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate


Test patch:
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go b/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
index 9e2a86ae4..9e5092c4c 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_defaults_test.go
@@ -19,12 +19,14 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"knative.dev/eventing-kafka/pkg/common/constants"
 )
 
 const (
 	testNumPartitions     = 10
 	testReplicationFactor = 5
+	testRetentionDuration = "P1D"
 )
 
 func TestKafkaChannelDefaults(t *testing.T) {
@@ -41,6 +43,7 @@ func TestKafkaChannelDefaults(t *testing.T) {
 				Spec: KafkaChannelSpec{
 					NumPartitions:     constants.DefaultNumPartitions,
 					ReplicationFactor: constants.DefaultReplicationFactor,
+					RetentionDuration: constants.DefaultRetentionISO8601Duration,
 				},
 			},
 		},
@@ -48,6 +51,7 @@ func TestKafkaChannelDefaults(t *testing.T) {
 			initial: KafkaChannel{
 				Spec: KafkaChannelSpec{
 					ReplicationFactor: testReplicationFactor,
+					RetentionDuration: testRetentionDuration,
 				},
 			},
 			expected: KafkaChannel{
@@ -57,13 +61,15 @@ func TestKafkaChannelDefaults(t *testing.T) {
 				Spec: KafkaChannelSpec{
 					NumPartitions:     constants.DefaultNumPartitions,
 					ReplicationFactor: testReplicationFactor,
+					RetentionDuration: testRetentionDuration,
 				},
 			},
 		},
 		"replicationFactor not set": {
 			initial: KafkaChannel{
 				Spec: KafkaChannelSpec{
-					NumPartitions: testNumPartitions,
+					NumPartitions:     testNumPartitions,
+					RetentionDuration: testRetentionDuration,
 				},
 			},
 			expected: KafkaChannel{
@@ -73,6 +79,25 @@ func TestKafkaChannelDefaults(t *testing.T) {
 				Spec: KafkaChannelSpec{
 					NumPartitions:     testNumPartitions,
 					ReplicationFactor: constants.DefaultReplicationFactor,
+					RetentionDuration: testRetentionDuration,
+				},
+			},
+		},
+		"retentionDuration not set": {
+			initial: KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     testNumPartitions,
+					ReplicationFactor: testReplicationFactor,
+				},
+			},
+			expected: KafkaChannel{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1"},
+				},
+				Spec: KafkaChannelSpec{
+					NumPartitions:     testNumPartitions,
+					ReplicationFactor: testReplicationFactor,
+					RetentionDuration: constants.DefaultRetentionISO8601Duration,
 				},
 			},
 		},
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go b/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go
index 186251c60..a29982ea9 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_types_test.go
@@ -18,8 +18,10 @@ package v1beta1
 
 import (
 	"testing"
+	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
@@ -47,3 +49,54 @@ func TestKafkaChannelGetStatus(t *testing.T) {
 		t.Errorf("GetStatus did not retrieve status. Got=%v Want=%v", config.GetStatus(), status)
 	}
 }
+
+func TestKafkaChannelSpecParseRetentionDuration(t *testing.T) {
+
+	tests := []struct {
+		name             string
+		includeRetention bool
+		retentionString  string
+		expectDuration   time.Duration
+		expectErr        bool
+	}{
+		{
+			name:             "valid",
+			includeRetention: true,
+			retentionString:  "P14D",
+			expectDuration:   14 * 24 * time.Hour,
+			expectErr:        false,
+		},
+		{
+			name:             "precision",
+			includeRetention: true,
+			retentionString:  "PT720H", // 30 Days Precise
+			expectDuration:   30 * 24 * time.Hour,
+			expectErr:        false,
+		}, {
+			name:             "empty",
+			includeRetention: true,
+			retentionString:  "",
+			expectDuration:   time.Duration(-1),
+			expectErr:        true,
+		},
+		{
+			name:             "invalid",
+			includeRetention: true,
+			retentionString:  "abc123",
+			expectDuration:   time.Duration(-1),
+			expectErr:        true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			kcSpec := &KafkaChannelSpec{}
+			if test.includeRetention {
+				kcSpec.RetentionDuration = test.retentionString
+			}
+			retentionDuration, err := kcSpec.ParseRetentionDuration()
+			assert.Equal(t, test.expectErr, err != nil)
+			assert.Equal(t, test.expectDuration, retentionDuration)
+		})
+	}
+}
diff --git a/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go b/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go
index f1463295b..89198a773 100644
--- a/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go
+++ b/pkg/apis/messaging/v1beta1/kafka_channel_validation_test.go
@@ -35,6 +35,15 @@ func TestKafkaChannelValidation(t *testing.T) {
 		cr   resourcesemantics.GenericCRD
 		want *apis.FieldError
 	}{
+		"valid spec": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					RetentionDuration: "P1D",
+				},
+			},
+		},
 		"empty spec": {
 			cr: &KafkaChannel{
 				Spec: KafkaChannelSpec{},
@@ -45,6 +54,8 @@ func TestKafkaChannelValidation(t *testing.T) {
 				errs = errs.Also(fe)
 				fe = apis.ErrInvalidValue(0, "spec.replicationFactor")
 				errs = errs.Also(fe)
+				fe = apis.ErrInvalidValue("", "spec.retentionDuration")
+				errs = errs.Also(fe)
 				return errs
 			}(),
 		},
@@ -53,6 +64,7 @@ func TestKafkaChannelValidation(t *testing.T) {
 				Spec: KafkaChannelSpec{
 					NumPartitions:     -10,
 					ReplicationFactor: 1,
+					RetentionDuration: "P1D",
 				},
 			},
 			want: func() *apis.FieldError {
@@ -65,6 +77,7 @@ func TestKafkaChannelValidation(t *testing.T) {
 				Spec: KafkaChannelSpec{
 					NumPartitions:     1,
 					ReplicationFactor: -10,
+					RetentionDuration: "P1D",
 				},
 			},
 			want: func() *apis.FieldError {
@@ -72,11 +85,38 @@ func TestKafkaChannelValidation(t *testing.T) {
 				return fe
 			}(),
 		},
+		"negative retentionDuration": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					RetentionDuration: "-P1D",
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrInvalidValue("-P1D", "spec.retentionDuration")
+				return fe
+			}(),
+		},
+		"invalid retentionDuration": {
+			cr: &KafkaChannel{
+				Spec: KafkaChannelSpec{
+					NumPartitions:     1,
+					ReplicationFactor: 1,
+					RetentionDuration: "NotAValidISO8601String",
+				},
+			},
+			want: func() *apis.FieldError {
+				fe := apis.ErrInvalidValue("NotAValidISO8601String", "spec.retentionDuration")
+				return fe
+			}(),
+		},
 		"valid subscribers array": {
 			cr: &KafkaChannel{
 				Spec: KafkaChannelSpec{
 					NumPartitions:     1,
 					ReplicationFactor: 1,
+					RetentionDuration: "P1D",
 					ChannelableSpec: eventingduck.ChannelableSpec{
 						SubscribableSpec: eventingduck.SubscribableSpec{
 							Subscribers: []eventingduck.SubscriberSpec{{
@@ -93,6 +133,7 @@ func TestKafkaChannelValidation(t *testing.T) {
 				Spec: KafkaChannelSpec{
 					NumPartitions:     1,
 					ReplicationFactor: 1,
+					RetentionDuration: "P1D",
 					ChannelableSpec: eventingduck.ChannelableSpec{
 						SubscribableSpec: eventingduck.SubscribableSpec{
 							Subscribers: []eventingduck.SubscriberSpec{{
@@ -113,6 +154,7 @@ func TestKafkaChannelValidation(t *testing.T) {
 				Spec: KafkaChannelSpec{
 					NumPartitions:     1,
 					ReplicationFactor: 1,
+					RetentionDuration: "P1D",
 					ChannelableSpec: eventingduck.ChannelableSpec{
 						SubscribableSpec: eventingduck.SubscribableSpec{
 							Subscribers: []eventingduck.SubscriberSpec{{}, {}},
@@ -141,6 +183,7 @@ func TestKafkaChannelValidation(t *testing.T) {
 				Spec: KafkaChannelSpec{
 					NumPartitions:     1,
 					ReplicationFactor: 1,
+					RetentionDuration: "P1D",
 				},
 			},
 			want: func() *apis.FieldError {
diff --git a/pkg/channel/consolidated/utils/util_test.go b/pkg/channel/consolidated/utils/util_test.go
index a8965b47d..4c6b5703c 100644
--- a/pkg/channel/consolidated/utils/util_test.go
+++ b/pkg/channel/consolidated/utils/util_test.go
@@ -240,11 +240,6 @@ func TestGetKafkaConfig(t *testing.T) {
 			Brokers:             brokers,
 			AuthSecretName:      name,
 			AuthSecretNamespace: namespace,
-			Topic: config.EKKafkaTopicConfig{
-				DefaultNumPartitions:     kafkasarama.DefaultNumPartitions,
-				DefaultReplicationFactor: kafkasarama.DefaultReplicationFactor,
-				DefaultRetentionMillis:   kafkasarama.DefaultRetentionMillis,
-			},
 		}
 	}
 
diff --git a/pkg/channel/distributed/controller/config/config_test.go b/pkg/channel/distributed/controller/config/config_test.go
index 838dcd52a..3c4052745 100644
--- a/pkg/channel/distributed/controller/config/config_test.go
+++ b/pkg/channel/distributed/controller/config/config_test.go
@@ -29,10 +29,6 @@ import (
 const (
 	kafkaAdminType = "custom"
 
-	defaultNumPartitions     = 7
-	defaultReplicationFactor = 2
-	defaultRetentionMillis   = 13579
-
 	dispatcherReplicas      = 1
 	dispatcherMemoryRequest = "20Mi"
 	dispatcherCpuRequest    = "100m"
@@ -51,20 +47,17 @@ type TestCase struct {
 	name string
 
 	// Config settings
-	kafkaTopicDefaultNumPartitions     int32
-	kafkaTopicDefaultReplicationFactor int16
-	kafkaTopicDefaultRetentionMillis   int64
-	kafkaAdminType                     string
-	dispatcherCpuLimit                 resource.Quantity
-	dispatcherCpuRequest               resource.Quantity
-	dispatcherMemoryLimit              resource.Quantity
-	dispatcherMemoryRequest            resource.Quantity
-	dispatcherReplicas                 int
-	receiverCpuLimit                   resource.Quantity
-	receiverCpuRequest                 resource.Quantity
-	receiverMemoryLimit                resource.Quantity
-	receiverMemoryRequest              resource.Quantity
-	receiverReplicas                   int
+	kafkaAdminType          string
+	dispatcherCpuLimit      resource.Quantity
+	dispatcherCpuRequest    resource.Quantity
+	dispatcherMemoryLimit   resource.Quantity
+	dispatcherMemoryRequest resource.Quantity
+	dispatcherReplicas      int
+	receiverCpuLimit        resource.Quantity
+	receiverCpuRequest      resource.Quantity
+	receiverMemoryLimit     resource.Quantity
+	receiverMemoryRequest   resource.Quantity
+	receiverReplicas        int
 
 	expectedError error
 }
@@ -72,22 +65,19 @@ type TestCase struct {
 // Get The Base / Valid Test Case - All Config Specified / No Errors
 func getValidTestCase(name string) TestCase {
 	return TestCase{
-		name:                               name,
-		kafkaTopicDefaultNumPartitions:     defaultNumPartitions,
-		kafkaTopicDefaultReplicationFactor: defaultReplicationFactor,
-		kafkaTopicDefaultRetentionMillis:   defaultRetentionMillis,
-		kafkaAdminType:                     kafkaAdminType,
-		dispatcherCpuLimit:                 resource.MustParse(dispatcherCpuLimit),
-		dispatcherCpuRequest:               resource.MustParse(dispatcherCpuRequest),
-		dispatcherMemoryLimit:              resource.MustParse(dispatcherMemoryLimit),
-		dispatcherMemoryRequest:            resource.MustParse(dispatcherMemoryRequest),
-		dispatcherReplicas:                 dispatcherReplicas,
-		receiverCpuLimit:                   resource.MustParse(receiverCpuLimit),
-		receiverCpuRequest:                 resource.MustParse(receiverCpuRquest),
-		receiverMemoryLimit:                resource.MustParse(receiverMemoryLimit),
-		receiverMemoryRequest:              resource.MustParse(receiverMemoryRequest),
-		receiverReplicas:                   receiverReplicas,
-		expectedError:                      nil,
+		name:                    name,
+		kafkaAdminType:          kafkaAdminType,
+		dispatcherCpuLimit:      resource.MustParse(dispatcherCpuLimit),
+		dispatcherCpuRequest:    resource.MustParse(dispatcherCpuRequest),
+		dispatcherMemoryLimit:   resource.MustParse(dispatcherMemoryLimit),
+		dispatcherMemoryRequest: resource.MustParse(dispatcherMemoryRequest),
+		dispatcherReplicas:      dispatcherReplicas,
+		receiverCpuLimit:        resource.MustParse(receiverCpuLimit),
+		receiverCpuRequest:      resource.MustParse(receiverCpuRquest),
+		receiverMemoryLimit:     resource.MustParse(receiverMemoryLimit),
+		receiverMemoryRequest:   resource.MustParse(receiverMemoryRequest),
+		receiverReplicas:        receiverReplicas,
+		expectedError:           nil,
 	}
 }
 
@@ -100,21 +90,6 @@ func TestVerifyConfiguration(t *testing.T) {
 	testCase := getValidTestCase("Valid Complete Config")
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Invalid Config - Kafka.Topic.DefaultNumPartitions")
-	testCase.kafkaTopicDefaultNumPartitions = -1
-	testCase.expectedError = ControllerConfigurationError("Kafka.Topic.DefaultNumPartitions must be > 0")
-	testCases = append(testCases, testCase)
-
-	testCase = getValidTestCase("Invalid Config - Kafka.Topic.DefaultReplicationFactor")
-	testCase.kafkaTopicDefaultReplicationFactor = -1
-	testCase.expectedError = ControllerConfigurationError("Kafka.Topic.DefaultReplicationFactor must be > 0")
-	testCases = append(testCases, testCase)
-
-	testCase = getValidTestCase("Invalid Config - Kafka.Topic.DefaultRetentionMillis")
-	testCase.kafkaTopicDefaultRetentionMillis = -1
-	testCase.expectedError = ControllerConfigurationError("Kafka.Topic.DefaultRetentionMillis must be > 0")
-	testCases = append(testCases, testCase)
-
 	testCase = getValidTestCase("Valid Config - Dispatcher.CpuLimit = Zero (unlimited)")
 	testCase.dispatcherCpuLimit = resource.Quantity{}
 	testCases = append(testCases, testCase)
@@ -165,9 +140,6 @@ func TestVerifyConfiguration(t *testing.T) {
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
 			testConfig := &commonconfig.EventingKafkaConfig{}
-			testConfig.Kafka.Topic.DefaultNumPartitions = testCase.kafkaTopicDefaultNumPartitions
-			testConfig.Kafka.Topic.DefaultReplicationFactor = testCase.kafkaTopicDefaultReplicationFactor
-			testConfig.Kafka.Topic.DefaultRetentionMillis = testCase.kafkaTopicDefaultRetentionMillis
 			testConfig.Channel.AdminType = testCase.kafkaAdminType
 			testConfig.Channel.Dispatcher.CpuLimit = testCase.dispatcherCpuLimit
 			testConfig.Channel.Dispatcher.CpuRequest = testCase.dispatcherCpuRequest
@@ -186,9 +158,6 @@ func TestVerifyConfiguration(t *testing.T) {
 			// Verify The Results
 			if testCase.expectedError == nil {
 				assert.Nil(t, err)
-				assert.Equal(t, testCase.kafkaTopicDefaultNumPartitions, testConfig.Kafka.Topic.DefaultNumPartitions)
-				assert.Equal(t, testCase.kafkaTopicDefaultReplicationFactor, testConfig.Kafka.Topic.DefaultReplicationFactor)
-				assert.Equal(t, testCase.kafkaTopicDefaultRetentionMillis, testConfig.Kafka.Topic.DefaultRetentionMillis)
 				assert.Equal(t, testCase.kafkaAdminType, testConfig.Channel.AdminType)
 				assert.Equal(t, testCase.dispatcherCpuLimit, testConfig.Channel.Dispatcher.CpuLimit)
 				assert.Equal(t, testCase.dispatcherCpuRequest, testConfig.Channel.Dispatcher.CpuRequest)
diff --git a/pkg/channel/distributed/controller/kafkachannel/topic_test.go b/pkg/channel/distributed/controller/kafkachannel/topic_test.go
index 3dd9ab592..b5a7a937a 100644
--- a/pkg/channel/distributed/controller/kafkachannel/topic_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/topic_test.go
@@ -47,8 +47,8 @@ type TopicTestCase struct {
 // Test The Kafka Topic Reconciliation
 //
 // Ideally the Knative Eventing test runner implementation would have provided a hook for additional
-// channel-type-specific (ie Kafka, NATS, etc) validation, but unfortunately it is solely focused
-// on the K8S objects existing/not.  Therefore we're left to test the actual Topic handling separately.
+// channel-type-specific (ie Kafka, NATS, etc.) validation, but unfortunately it is solely focused
+// on the K8S objects existing/not.  Therefore, we're left to test the actual Topic handling separately.
 //
 func TestReconcileTopic(t *testing.T) {
 
@@ -85,7 +85,7 @@ func TestReconcileTopic(t *testing.T) {
 			WantTopicDetail: &sarama.TopicDetail{
 				NumPartitions:     controllertesting.NumPartitions,
 				ReplicationFactor: controllertesting.ReplicationFactor,
-				ConfigEntries:     map[string]*string{commonconstants.KafkaTopicConfigRetentionMs: &controllertesting.DefaultRetentionMillisString},
+				ConfigEntries:     map[string]*string{commonconstants.KafkaTopicConfigRetentionMs: &controllertesting.RetentionMillisString},
 			},
 		},
 		{
@@ -104,7 +104,7 @@ func TestReconcileTopic(t *testing.T) {
 			WantTopicDetail: &sarama.TopicDetail{
 				NumPartitions:     controllertesting.NumPartitions,
 				ReplicationFactor: controllertesting.ReplicationFactor,
-				ConfigEntries:     map[string]*string{commonconstants.KafkaTopicConfigRetentionMs: &controllertesting.DefaultRetentionMillisString},
+				ConfigEntries:     map[string]*string{commonconstants.KafkaTopicConfigRetentionMs: &controllertesting.RetentionMillisString},
 			},
 			MockErrorCode: sarama.ErrTopicAlreadyExists,
 		},
@@ -124,7 +124,7 @@ func TestReconcileTopic(t *testing.T) {
 			WantTopicDetail: &sarama.TopicDetail{
 				NumPartitions:     controllertesting.NumPartitions,
 				ReplicationFactor: controllertesting.ReplicationFactor,
-				ConfigEntries:     map[string]*string{commonconstants.KafkaTopicConfigRetentionMs: &controllertesting.DefaultRetentionMillisString},
+				ConfigEntries:     map[string]*string{commonconstants.KafkaTopicConfigRetentionMs: &controllertesting.RetentionMillisString},
 			},
 			MockErrorCode: sarama.ErrBrokerNotAvailable,
 			WantError:     sarama.ErrBrokerNotAvailable.Error() + " - " + controllertesting.ErrorString,
diff --git a/pkg/channel/distributed/controller/testing/data.go b/pkg/channel/distributed/controller/testing/data.go
index c43b14259..16e28a4db 100644
--- a/pkg/channel/distributed/controller/testing/data.go
+++ b/pkg/channel/distributed/controller/testing/data.go
@@ -57,19 +57,16 @@ const (
 	MetricsPortName = "metrics"
 
 	// Environment Test Data
-	ServiceAccount           = "TestServiceAccount"
-	KafkaAdminType           = "kafka"
-	MetricsPort              = 9876
-	MetricsDomain            = "eventing-kafka"
-	HealthPort               = 8082
-	ResyncPeriod             = 3600 * time.Minute
-	ReceiverImage            = "TestReceiverImage"
-	ReceiverReplicas         = 1
-	DispatcherImage          = "TestDispatcherImage"
-	DispatcherReplicas       = 1
-	DefaultNumPartitions     = 4
-	DefaultReplicationFactor = 1
-	DefaultRetentionMillis   = 99999
+	ServiceAccount     = "TestServiceAccount"
+	KafkaAdminType     = "kafka"
+	MetricsPort        = 9876
+	MetricsDomain      = "eventing-kafka"
+	HealthPort         = 8082
+	ResyncPeriod       = 3600 * time.Minute
+	ReceiverImage      = "TestReceiverImage"
+	ReceiverReplicas   = 1
+	DispatcherImage    = "TestDispatcherImage"
+	DispatcherReplicas = 1
 
 	// Channel Test Data
 	KafkaChannelNamespace  = "kafkachannel-namespace"
@@ -89,6 +86,8 @@ const (
 	NumPartitions = 123
 	// ReplicationFactor - ChannelSpec Test Data
 	ReplicationFactor = 456
+	// RetentionDurationISO8601 - ChannelSpec Test Data
+	RetentionDurationISO8601 = "P3D"
 
 	// ErrorString - Mock Test Error MetaData
 	ErrorString = "Expected Mock Test Error"
@@ -161,10 +160,6 @@ kafka:
   authSecretName: ` + KafkaSecretName + `
   authSecretNamespace: ` + KafkaSecretNamespace + `
   enableSaramaLogging: false
-  topic:
-    defaultNumPartitions: 4
-    defaultReplicationFactor: 1
-    defaultRetentionMillis: 604800000
   adminType: kafka
 `
 	SaramaConfigYaml = `
@@ -194,8 +189,9 @@ Producer:
 )
 
 var (
-	DefaultRetentionMillisString = strconv.FormatInt(DefaultRetentionMillis, 10)
-	DeletionTimestamp            = metav1.Now()
+	RetentionDuration     = 3 * 24 * time.Hour // Match RetentionDurationISO8601 Value Above !
+	RetentionMillisString = strconv.FormatInt(RetentionDuration.Milliseconds(), 10)
+	DeletionTimestamp     = metav1.Now()
 
 	DispatcherDeploymentAnnotations = map[string]string{"DDAKey1": "DDAValue1", "DDAKey2": "DDAValue2", "DDAKey3": "DDAValue3"}
 	DispatcherDeploymentLabels      = map[string]string{"DDLKey1": "DDLValue1", "DDLKey2": "DDLValue2", "DDLKey3": "DDLValue3"}
@@ -446,11 +442,6 @@ func NewConfig(options ...KafkaConfigOption) *commonconfig.EventingKafkaConfig {
 		Kafka: commonconfig.EKKafkaConfig{
 			AuthSecretName:      KafkaSecretName,
 			AuthSecretNamespace: KafkaSecretNamespace,
-			Topic: commonconfig.EKKafkaTopicConfig{
-				DefaultNumPartitions:     DefaultNumPartitions,
-				DefaultReplicationFactor: DefaultReplicationFactor,
-				DefaultRetentionMillis:   DefaultRetentionMillis,
-			},
 		},
 		Channel: commonconfig.EKChannelConfig{
 			AdminType: KafkaAdminType,
@@ -576,7 +567,7 @@ func NewKafkaChannel(options ...KafkaChannelOption) *kafkav1beta1.KafkaChannel {
 		Spec: kafkav1beta1.KafkaChannelSpec{
 			NumPartitions:     NumPartitions,
 			ReplicationFactor: ReplicationFactor,
-			// TODO RetentionMillis:   RetentionMillis,
+			RetentionDuration: RetentionDurationISO8601,
 		},
 	}
 
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index 64c37bc65..cb2583e9a 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -88,7 +88,7 @@ func TestNewController(t *testing.T) {
 			stopChan := make(chan struct{})
 
 			// Perform The Test
-			c := NewController(logger, channelKey, mockDispatcher, kafkaChannelInformer, fakeK8sClientSet, fakeKafkaChannelClientSet, stopChan, testCase.managerEvents)
+			c := NewController(context.TODO(), logger, channelKey, mockDispatcher, kafkaChannelInformer, fakeK8sClientSet, fakeKafkaChannelClientSet, stopChan, testCase.managerEvents)
 
 			// Verify Results
 			assert.NotNil(t, c)
@@ -122,7 +122,7 @@ func TestProcessManagerEvents(t *testing.T) {
 		kafkachannelInformer: kafkaChannelInformer.Informer(),
 		kafkachannelLister:   kafkaChannelInformer.Lister(),
 		kafkaClientSet:       fakeKafkaChannelClientSet,
-		impl:                 controller.NewImplFull(mockReconciler, controller.ControllerOptions{Logger: logger.Sugar()}),
+		impl:                 controller.NewContext(context.TODO(), mockReconciler, controller.ControllerOptions{Logger: logger.Sugar()}),
 	}
 	events := make(chan consumer.ManagerEvent)
 
diff --git a/pkg/common/config/util_test.go b/pkg/common/config/util_test.go
index 880022597..9da524b49 100644
--- a/pkg/common/config/util_test.go
+++ b/pkg/common/config/util_test.go
@@ -10,21 +10,12 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
-	logtesting "knative.dev/pkg/logging/testing"
 	"knative.dev/pkg/system"
 
-	kafkav1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	"knative.dev/eventing-kafka/pkg/common/constants"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
 )
 
-const (
-	numPartitions            = int32(123)
-	defaultNumPartitions     = int32(987)
-	replicationFactor        = int16(22)
-	defaultReplicationFactor = int16(33)
-)
-
 func TestConfigmapDataCheckSum(t *testing.T) {
 	cases := []struct {
 		name          string
@@ -217,46 +208,6 @@ func getSaramaTestSecret(t *testing.T, name string,
 	return commontesting.GetTestSaramaSecret(name, username, password, namespace, saslType)
 }
 
-// Test The NumPartitions Accessor
-func TestNumPartitions(t *testing.T) {
-
-	// Test Logger
-	logger := logtesting.TestLogger(t)
-
-	// Test Data
-	configuration := &EventingKafkaConfig{Kafka: EKKafkaConfig{Topic: EKKafkaTopicConfig{DefaultNumPartitions: defaultNumPartitions}}}
-
-	// Test The Default Failover Use Case
-	channel := &kafkav1beta1.KafkaChannel{}
-	actualNumPartitions := NumPartitions(channel, configuration, logger)
-	assert.Equal(t, defaultNumPartitions, actualNumPartitions)
-
-	// Test The Valid NumPartitions Use Case
-	channel = &kafkav1beta1.KafkaChannel{Spec: kafkav1beta1.KafkaChannelSpec{NumPartitions: numPartitions}}
-	actualNumPartitions = NumPartitions(channel, configuration, logger)
-	assert.Equal(t, numPartitions, actualNumPartitions)
-}
-
-// Test The ReplicationFactor Accessor
-func TestReplicationFactor(t *testing.T) {
-
-	// Test Logger
-	logger := logtesting.TestLogger(t)
-
-	// Test Data
-	configuration := &EventingKafkaConfig{Kafka: EKKafkaConfig{Topic: EKKafkaTopicConfig{DefaultReplicationFactor: defaultReplicationFactor}}}
-
-	// Test The Default Failover Use Case
-	channel := &kafkav1beta1.KafkaChannel{}
-	actualReplicationFactor := ReplicationFactor(channel, configuration, logger)
-	assert.Equal(t, defaultReplicationFactor, actualReplicationFactor)
-
-	// Test The Valid ReplicationFactor Use Case
-	channel = &kafkav1beta1.KafkaChannel{Spec: kafkav1beta1.KafkaChannelSpec{ReplicationFactor: replicationFactor}}
-	actualReplicationFactor = ReplicationFactor(channel, configuration, logger)
-	assert.Equal(t, replicationFactor, actualReplicationFactor)
-}
-
 // Test The JoinStringMaps() Functionality
 func TestJoinStringMaps(t *testing.T) {
 	map1 := map[string]string{"key1": "value1a", "key2": "value2", "key3": "value3"}
diff --git a/pkg/common/kafka/sarama/sarama_test.go b/pkg/common/kafka/sarama/sarama_test.go
index 781c9f3d2..1ef258abf 100644
--- a/pkg/common/kafka/sarama/sarama_test.go
+++ b/pkg/common/kafka/sarama/sarama_test.go
@@ -52,10 +52,6 @@ channel:
 kafka:
   brokers: REPLACE_WITH_CLUSTER_URL
   enableSaramaLogging: false
-  topic:
-    defaultNumPartitions: 4
-    defaultReplicationFactor: 1
-    defaultRetentionMillis: 604800000
 `
 	EKDefaultSaramaConfig = `
 enableLogging: false
@@ -91,10 +87,6 @@ dispatcher:
 kafka:
   brokers: REPLACE_WITH_CLUSTER_URL
   enableSaramaLogging: false
-  topic:
-    defaultNumPartitions: 4
-    defaultReplicationFactor: 1
-    defaultRetentionMillis: 604800000
   adminType: azure
 `
 	EKDefaultSaramaConfigOld = `
@@ -226,9 +218,6 @@ func verifyEventingKafkaSettings(t *testing.T, settings *commonconfig.EventingKa
 	assert.Equal(t, resource.Quantity{}, settings.Channel.Receiver.MemoryLimit)
 	assert.Equal(t, resource.MustParse("50Mi"), settings.Channel.Receiver.MemoryRequest)
 	assert.Equal(t, 1, settings.Channel.Receiver.Replicas)
-	assert.Equal(t, int32(4), settings.Kafka.Topic.DefaultNumPartitions)
-	assert.Equal(t, int16(1), settings.Kafka.Topic.DefaultReplicationFactor)
-	assert.Equal(t, int64(604800000), settings.Kafka.Topic.DefaultRetentionMillis)
 	assert.Equal(t, resource.Quantity{}, settings.Channel.Dispatcher.CpuLimit)
 	assert.Equal(t, resource.MustParse("100m"), settings.Channel.Dispatcher.CpuRequest)
 	assert.Equal(t, resource.Quantity{}, settings.Channel.Dispatcher.MemoryLimit)
diff --git a/pkg/common/testing/constants.go b/pkg/common/testing/constants.go
index 1685ce8d1..aedffc46b 100644
--- a/pkg/common/testing/constants.go
+++ b/pkg/common/testing/constants.go
@@ -55,10 +55,6 @@ channel:
     replicas: ` + DispatcherReplicas + `
 kafka:
   brokers: ` + BrokerString + `
-  topic:
-    defaultNumPartitions: 4
-    defaultReplicationFactor: 1 # Cannot exceed the number of Kafka Brokers!
-    defaultRetentionMillis: 604800000  # 1 week
 `
 
 	OldSaramaConfig = `
diff --git a/test/rekt/features/kafkachannel/util.go b/test/rekt/features/kafkachannel/util.go
index 64ffcd18c..c5f5c402d 100644
--- a/test/rekt/features/kafkachannel/util.go
+++ b/test/rekt/features/kafkachannel/util.go
@@ -25,6 +25,7 @@ import (
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cloudeventtest "github.com/cloudevents/sdk-go/v2/test"
+	subscriptionresources "knative.dev/eventing/test/rekt/resources/subscription"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/eventshub/assert"
@@ -33,7 +34,6 @@ import (
 
 	kafkachannelresources "knative.dev/eventing-kafka/test/rekt/resources/kafkachannel"
 	resetoffsetresources "knative.dev/eventing-kafka/test/rekt/resources/resetoffset"
-	subscriptionresources "knative.dev/eventing/test/rekt/resources/subscription"
 )
 
 const (
@@ -76,6 +76,7 @@ func setupKafkaChannel(f *feature.Feature, name string) {
 	f.Setup("Install A KafkaChannel", kafkachannelresources.Install(name,
 		kafkachannelresources.WithNumPartitions("3"),
 		kafkachannelresources.WithReplicationFactor("1"),
+		kafkachannelresources.WithRetentionDuration("P1D"),
 	))
 }
 
diff --git a/test/rekt/resources/kafkachannel/kafkachannel.go b/test/rekt/resources/kafkachannel/kafkachannel.go
index 9408eae1e..5ec67d453 100644
--- a/test/rekt/resources/kafkachannel/kafkachannel.go
+++ b/test/rekt/resources/kafkachannel/kafkachannel.go
@@ -77,3 +77,10 @@ func WithReplicationFactor(replicationFactor string) manifest.CfgFn {
 		cfg["replicationFactor"] = replicationFactor
 	}
 }
+
+// WithRetentionDuration adds the retentionDuration config to a KafkaChannel spec.
+func WithRetentionDuration(retentionDuration string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		cfg["retentionDuration"] = retentionDuration
+	}
+}
diff --git a/test/rekt/resources/kafkachannel/kafkachannel_test.go b/test/rekt/resources/kafkachannel/kafkachannel_test.go
index 7142cf87a..18b619b6e 100644
--- a/test/rekt/resources/kafkachannel/kafkachannel_test.go
+++ b/test/rekt/resources/kafkachannel/kafkachannel_test.go
@@ -33,6 +33,7 @@ func Example() {
 	}
 	WithNumPartitions("10")(cfg)
 	WithReplicationFactor("3")(cfg)
+	WithRetentionDuration("P1D")
 
 	files, err := manifest.ExecuteYAML(yaml, images, cfg)
 	if err != nil {