Repository: knative-sandbox/eventing-kafka-broker
Pull request #2700 (2022-10-04 12:50:30)

Use full TopicDetail when creating new topics
Fixes propagation of KafkaChannel's `retentionDuration` into KafkaTopic resource. Before this PR, `retentionDuration` was ignored.

## Proposed Changes

- :gift: Consume entire sarama.TopicDetail when creating new KafkaTopic's
- :bug: And so propagate KafkaChannel's retentionDuration to KafkaTopic's config (default's to 7 days)

**Release Note**


```release-note
New Kafka topics will be created with the default 7 day retention duration. Existing topics will _**not**_ be updated.
```




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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2700#" title="Author self-approved">EndPositive</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign dprotaso for approval by writing `/assign @dprotaso` in a comment. For more information see:[The Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["dprotaso"]} -->
linux-foundation-easycla[bot]:  <a href="https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/25619798/273315692/2700/#/?version=2"><img src="https://s3.amazonaws.com/cla-project-logo-prod/cla-not-signed.svg" alt="CLA Not Signed" align="left" height="28" width="328"></a><br/><br /><ul><li><a href='https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/25619798/273315692/2700/#/?version=2' target='_blank'>:x:</a> - login: @EndPositive / name: Jop Zitman . The commit (4969479a8a30ce55e86edf6e81819f6e734df0a3) is not authorized under a signed CLA. <a href='https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/25619798/273315692/2700/#/?version=2' target='_blank'>Please click here to be authorized</a>. For further assistance with EasyCLA, <a href='https://jira.linuxfoundation.org/servicedesk/customer/portal/4' target='_blank'>please submit a support request ticket</a>.</li></ul>
knative-prow[bot]: Welcome @EndPositive! It looks like this is your first PR to knative-sandbox/eventing-kafka-broker 🎉
knative-prow[bot]: Hi @EndPositive. Thanks for your PR.

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

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

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

<details>

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

pierDipi: Thanks for the PR @EndPositive! Let me know if you have trouble with the Linux Foundation CLA as described by the bot.

/ok-to-test
knative-prow[bot]: @EndPositive: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 4969479a8a30ce55e86edf6e81819f6e734df0a3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2700/unit-tests_eventing-kafka-broker_main/1577426884034039808) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @EndPositive: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 4969479a8a30ce55e86edf6e81819f6e734df0a3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2700/unit-tests_eventing-kafka-broker_main/1577426884034039808) | true | `/test unit-tests`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 4969479a8a30ce55e86edf6e81819f6e734df0a3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2700/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1577426884549939200) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @EndPositive: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 4969479a8a30ce55e86edf6e81819f6e734df0a3 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2700/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1577426884549939200) | true | `/test channel-integration-tests-sasl-plain`
unit-tests_eventing-kafka-broker_main | ba361b13e75cf7a2d735214b42370e48e24066e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2700/unit-tests_eventing-kafka-broker_main/1577751353738924032) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @EndPositive: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | ba361b13e75cf7a2d735214b42370e48e24066e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2700/unit-tests_eventing-kafka-broker_main/1577751353738924032) | true | `/test unit-tests`
reconciler-tests-namespaced-broker_eventing-kafka-broker_main | ba361b13e75cf7a2d735214b42370e48e24066e4 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2700/reconciler-tests-namespaced-broker_eventing-kafka-broker_main/1577751354095439872) | true | `/test reconciler-tests-namespaced-broker`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
EndPositive: @pierDipi ready for review :+1: . Looks like the pipeline fails, but unrelated to my pr?

Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 3a991563e..e6831b278 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -114,16 +114,13 @@ aliases:
   - nainaz
   - salaboy
   - zroubalik
-  func-writers:
+  functastic-writers:
   - lance
   - lkingland
   - matejvasek
   - nainaz
   - salaboy
   - zroubalik
-  functions-wg-leads:
-  - lance
-  - salaboy
   homebrew-kn-plugins-approvers:
   - dsimansk
   - maximilien
@@ -140,14 +137,6 @@ aliases:
   kn-plugin-event-approvers:
   - cardil
   - rhuss
-  kn-plugin-func-approvers:
-  - jrangelramos
-  - lance
-  - lkingland
-  - matejvasek
-  - nainaz
-  - salaboy
-  - zroubalik
   kn-plugin-migration-approvers:
   - maximilien
   - zhangtbj
diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go
index fe73316c1..4c2931a24 100644
--- a/control-plane/cmd/webhook-kafka/main.go
+++ b/control-plane/cmd/webhook-kafka/main.go
@@ -24,6 +24,7 @@ import (
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection/sharedmain"
+	"knative.dev/pkg/logging"
 	"knative.dev/pkg/signals"
 	"knative.dev/pkg/webhook"
 	"knative.dev/pkg/webhook/certificates"
@@ -33,12 +34,14 @@ import (
 
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	eventingcorev1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/pkg/apis/feature"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/core"
 	eventingv1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1"
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
 
 const (
@@ -50,6 +53,8 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 	sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"):    &sourcesv1beta1.KafkaSource{},
 	messagingv1beta1.SchemeGroupVersion.WithKind("KafkaChannel"): &messagingv1beta1.KafkaChannel{},
 	eventingcorev1.SchemeGroupVersion.WithKind("Broker"):         &eventingv1.BrokerStub{},
+	kafkainternals.SchemeGroupVersion.WithKind("ConsumerGroup"):  &kafkainternals.ConsumerGroup{},
+	kafkainternals.SchemeGroupVersion.WithKind("Consumer"):       &kafkainternals.Consumer{},
 }
 
 var defaultingCallbacks = map[schema.GroupVersionKind]defaulting.Callback{
@@ -110,7 +115,15 @@ func NewPodDefaultingAdmissionController(ctx context.Context, _ configmap.Watche
 	)
 }
 
-func NewValidationAdmissionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
+func NewValidationAdmissionController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+	featureStore := feature.NewStore(logging.FromContext(ctx).Named("feature-config-store"))
+	featureStore.WatchConfigs(cmw)
+
+	// Decorate contexts with the current state of the config.
+	ctxFunc := func(ctx context.Context) context.Context {
+		return featureStore.ToContext(ctx)
+	}
+
 	return validation.NewAdmissionController(ctx,
 		// Name of the resource webhook.
 		"validation.webhook.kafka.eventing.knative.dev",
@@ -122,9 +135,7 @@ func NewValidationAdmissionController(ctx context.Context, _ configmap.Watcher)
 		types,
 
 		// A function that infuses the context passed to Validate/SetDefaults with custom metadata.
-		func(ctx context.Context) context.Context {
-			return ctx
-		},
+		ctxFunc,
 
 		// Whether to disallow unknown fields.
 		true,
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
index c95b782b5..68fedcbb4 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
@@ -49,10 +49,15 @@ func (cgs *ConsumerGroupSpec) Validate(ctx context.Context) *apis.FieldError {
 
 func (cts *ConsumerTemplateSpec) Validate(ctx context.Context) *apis.FieldError {
 	specCtx := ctx
+	var err *apis.FieldError
 	if apis.IsInUpdate(ctx) {
 		specCtx = apis.WithinUpdate(ctx, apis.GetBaseline(ctx).(*ConsumerGroup).Spec.Template.Spec)
 	}
-	return cts.Spec.Validate(specCtx).ViaField("spec")
+	err = err.Also( // fields with defaults
+		cts.Spec.Delivery.Validate(specCtx).ViaField("delivery"),
+		cts.Spec.Subscriber.Validate(specCtx).ViaField("subscriber"),
+	)
+	return err
 }
 
 func (c *ConsumerGroup) CheckImmutableFields(ctx context.Context, original map[string]string) *apis.FieldError {
diff --git a/control-plane/pkg/kafka/topic.go b/control-plane/pkg/kafka/topic.go
index ba20afed9..44d4cd9bd 100644
--- a/control-plane/pkg/kafka/topic.go
+++ b/control-plane/pkg/kafka/topic.go
@@ -159,19 +159,13 @@ func ChannelTopic(prefix string, obj metav1.Object) string {
 // If the topic already exists, it will return no errors.
 // TODO: what happens if the topic exists but it has a different config?
 func CreateTopicIfDoesntExist(admin sarama.ClusterAdmin, logger *zap.Logger, topic string, config *TopicConfig) (string, error) {
-
-	topicDetail := &sarama.TopicDetail{
-		NumPartitions:     config.TopicDetail.NumPartitions,
-		ReplicationFactor: config.TopicDetail.ReplicationFactor,
-	}
-
 	logger.Debug("create topic",
 		zap.String("topic", topic),
-		zap.Int16("replicationFactor", topicDetail.ReplicationFactor),
-		zap.Int32("numPartitions", topicDetail.NumPartitions),
+		zap.Int16("replicationFactor", config.TopicDetail.ReplicationFactor),
+		zap.Int32("numPartitions", config.TopicDetail.NumPartitions),
 	)
 
-	createTopicError := admin.CreateTopic(topic, topicDetail, false)
+	createTopicError := admin.CreateTopic(topic, &config.TopicDetail, false)
 	if err, ok := createTopicError.(*sarama.TopicError); ok && err.Err == sarama.ErrTopicAlreadyExists {
 		return topic, nil
 	}
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 1929e286b..593114c8e 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -69,8 +69,8 @@ Lists of 163 third-party dependencies.
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: JUnit :: Server Mock (io.fabric8:kubernetes-server-mock:6.0.0 - http://fabric8.io/junit/kubernetes-server-mock/)
      (Apache License, Version 2.0) Fabric8 :: Mock Web Server (io.fabric8:mockwebserver:0.2.2 - http://fabric8.io/)
      (The Apache Software License, Version 2.0) zjsonpatch (io.fabric8:zjsonpatch:0.3.0 - https://github.com/fabric8io/zjsonpatch/)
-     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.9.3 - https://github.com/micrometer-metrics/micrometer)
-     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.9.3 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.9.4 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.9.4 - https://github.com/micrometer-metrics/micrometer)
      (Apache License, Version 2.0) Netty/Buffer (io.netty:netty-buffer:4.1.74.Final - https://netty.io/netty-buffer/)
      (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.74.Final - https://netty.io/netty-codec/)
      (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.74.Final - https://netty.io/netty-codec-dns/)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index a2548f216..0bb808522 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -44,7 +44,7 @@
     <vertx.version>4.2.5</vertx.version>
     <vertx.kafka.clients.version>4.3.2</vertx.kafka.clients.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
-    <micrometer.version>1.9.3</micrometer.version>
+    <micrometer.version>1.9.4</micrometer.version>
     <opentelemetry.version>1.17.0</opentelemetry.version>
     <jackson.version>2.13.3</jackson.version>
     <protobuf.version>3.20.0</protobuf.version>


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
index 42a62f714..04418ff67 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
@@ -22,12 +22,20 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
+const (
+	DefaultDeliveryOrder = internals.Ordered
+)
+
 func TestConsumerGroup_Validate(t *testing.T) {
+	backoffPolicy := eventingduck.BackoffPolicyExponential
+
 	tests := []struct {
 		name    string
 		ctx     context.Context
@@ -51,19 +59,18 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "no topics",
+			name: "no consumer template",
 			ctx:  context.Background(),
 			given: &ConsumerGroup{
 				Spec: ConsumerGroupSpec{
 					Replicas: pointer.Int32Ptr(1),
 					Selector: map[string]string{"app": "app"},
-					Template: ConsumerTemplateSpec{},
 				},
 			},
 			wantErr: true,
 		},
 		{
-			name: "missing group.id",
+			name: "no subscriber",
 			ctx:  context.Background(),
 			given: &ConsumerGroup{
 				Spec: ConsumerGroupSpec{
@@ -71,12 +78,14 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Subscriber: duckv1.Destination{
-								URI: &apis.URL{
-									Scheme: "http",
-									Host:   "127.0.0.1",
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{
+									Retry:         pointer.Int32(10),
+									BackoffPolicy: &backoffPolicy,
+									BackoffDelay:  pointer.String("PT0.3S"),
+									Timeout:       pointer.String("PT600S"),
 								},
+								Ordering: DefaultDeliveryOrder,
 							},
 						},
 					},
@@ -85,7 +94,7 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "valid",
+			name: "no delivery",
 			ctx:  context.Background(),
 			given: &ConsumerGroup{
 				Spec: ConsumerGroupSpec{
@@ -93,26 +102,12 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
-							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
-							},
 							Subscriber: duckv1.Destination{
 								URI: &apis.URL{
 									Scheme: "http",
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
@@ -120,7 +115,7 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			wantErr: false,
 		},
 		{
-			name: "invalid multiple reply strategies (topic, destination)",
+			name: "invalid delivery - timeout feature disabled",
 			ctx:  context.Background(),
 			given: &ConsumerGroup{
 				Spec: ConsumerGroupSpec{
@@ -128,54 +123,12 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
-							Reply: &ReplyStrategy{
-								TopicReply: &TopicReply{Enabled: true},
-								URLReply:   &DestinationReply{Enabled: true},
-							},
 							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
-							},
-							Subscriber: duckv1.Destination{
-								URI: &apis.URL{
-									Scheme: "http",
-									Host:   "127.0.0.1",
+								DeliverySpec: &eventingduck.DeliverySpec{
+									Retry:   pointer.Int32(-10),
+									Timeout: pointer.String("PT600S"),
 								},
-							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
-						},
-					},
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name: "invalid no pod name",
-			ctx:  context.Background(),
-			given: &ConsumerGroup{
-				Spec: ConsumerGroupSpec{
-					Replicas: pointer.Int32Ptr(1),
-					Selector: map[string]string{"app": "app"},
-					Template: ConsumerTemplateSpec{
-						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
-							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
+								Ordering: DefaultDeliveryOrder,
 							},
 							Subscriber: duckv1.Destination{
 								URI: &apis.URL{
@@ -183,9 +136,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
@@ -193,86 +143,19 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "invalid no pod namespace",
-			ctx:  context.Background(),
+			name: "valid delivery - timeout feature enabled",
+			ctx:  feature.ToContext(context.Background(), feature.Flags{feature.DeliveryTimeout: feature.Enabled}),
 			given: &ConsumerGroup{
 				Spec: ConsumerGroupSpec{
 					Replicas: pointer.Int32Ptr(1),
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
 							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
-							},
-							Subscriber: duckv1.Destination{
-								URI: &apis.URL{
-									Scheme: "http",
-									Host:   "127.0.0.1",
-								},
-							},
-							PodBind: &PodBind{
-								PodName: "p-0",
-							},
-						},
-					},
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name: "valid - no updates",
-			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
-				Spec: ConsumerGroupSpec{
-					Replicas: pointer.Int32Ptr(1),
-					Selector: map[string]string{"app": "app"},
-					Template: ConsumerTemplateSpec{
-						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
+								DeliverySpec: &eventingduck.DeliverySpec{
+									Timeout: pointer.String("PT600S"),
 								},
-							},
-							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
-							},
-							Subscriber: duckv1.Destination{
-								URI: &apis.URL{
-									Scheme: "http",
-									Host:   "127.0.0.1",
-								},
-							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
-						},
-					},
-				},
-			}),
-			given: &ConsumerGroup{
-				Spec: ConsumerGroupSpec{
-					Replicas: pointer.Int32Ptr(1),
-					Selector: map[string]string{"app": "app"},
-					Template: ConsumerTemplateSpec{
-						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
-							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
+								Ordering: DefaultDeliveryOrder,
 							},
 							Subscriber: duckv1.Destination{
 								URI: &apis.URL{
@@ -280,10 +163,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
@@ -291,114 +170,14 @@ func TestConsumerGroup_Validate(t *testing.T) {
 			wantErr: false,
 		},
 		{
-			name: "invalid pod name update",
-			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
-				Spec: ConsumerGroupSpec{
-					Replicas: pointer.Int32Ptr(1),
-					Selector: map[string]string{"app": "app"},
-					Template: ConsumerTemplateSpec{
-						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
-							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
-							},
-							Subscriber: duckv1.Destination{
-								URI: &apis.URL{
-									Scheme: "http",
-									Host:   "127.0.0.1",
-								},
-							},
-							PodBind: &PodBind{
-								PodName:      "p-1",
-								PodNamespace: "ns",
-							},
-						},
-					},
-				},
-			}),
-			given: &ConsumerGroup{
-				Spec: ConsumerGroupSpec{
-					Replicas: pointer.Int32Ptr(1),
-					Selector: map[string]string{"app": "app"},
-					Template: ConsumerTemplateSpec{
-						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
-							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
-							},
-							Subscriber: duckv1.Destination{
-								URI: &apis.URL{
-									Scheme: "http",
-									Host:   "127.0.0.1",
-								},
-							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
-						},
-					},
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name: "invalid pod name update",
-			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
-				Spec: ConsumerGroupSpec{
-					Replicas: pointer.Int32Ptr(1),
-					Selector: map[string]string{"app": "app"},
-					Template: ConsumerTemplateSpec{
-						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
-							Delivery: &DeliverySpec{
-								DeliverySpec: &eventingduck.DeliverySpec{},
-							},
-							Subscriber: duckv1.Destination{
-								URI: &apis.URL{
-									Scheme: "http",
-									Host:   "127.0.0.1",
-								},
-							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns-1",
-							},
-						},
-					},
-				},
-			}),
+			name: "valid",
+			ctx:  context.Background(),
 			given: &ConsumerGroup{
 				Spec: ConsumerGroupSpec{
 					Replicas: pointer.Int32Ptr(1),
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
 							Delivery: &DeliverySpec{
 								DeliverySpec: &eventingduck.DeliverySpec{},
 							},
@@ -408,15 +187,11 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns-2",
-							},
 						},
 					},
 				},
 			},
-			wantErr: true,
+			wantErr: false,
 		},
 		{
 			name: "valid with channel label",
@@ -431,13 +206,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
 							Delivery: &DeliverySpec{
 								DeliverySpec: &eventingduck.DeliverySpec{},
 							},
@@ -447,10 +215,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
@@ -466,13 +230,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
 							Delivery: &DeliverySpec{
 								DeliverySpec: &eventingduck.DeliverySpec{},
 							},
@@ -482,10 +239,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
@@ -505,13 +258,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
 							Delivery: &DeliverySpec{
 								DeliverySpec: &eventingduck.DeliverySpec{},
 							},
@@ -521,10 +267,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
@@ -535,13 +277,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
 							Delivery: &DeliverySpec{
 								DeliverySpec: &eventingduck.DeliverySpec{},
 							},
@@ -551,10 +286,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
@@ -574,13 +305,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
 							Delivery: &DeliverySpec{
 								DeliverySpec: &eventingduck.DeliverySpec{},
 							},
@@ -590,10 +314,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
@@ -609,13 +329,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 					Selector: map[string]string{"app": "app"},
 					Template: ConsumerTemplateSpec{
 						Spec: ConsumerSpec{
-							Topics: []string{"t1"},
-							Configs: ConsumerConfigs{
-								Configs: map[string]string{
-									"group.id":          "g1",
-									"bootstrap.servers": "kafka:9092",
-								},
-							},
 							Delivery: &DeliverySpec{
 								DeliverySpec: &eventingduck.DeliverySpec{},
 							},
@@ -625,10 +338,6 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
-							PodBind: &PodBind{
-								PodName:      "p-0",
-								PodNamespace: "ns",
-							},
 						},
 					},
 				},
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation_test.go
new file mode 100644
index 000000000..87f10a9b0
--- /dev/null
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation_test.go
@@ -0,0 +1,342 @@
+/*
+ * 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 v1alpha1
+
+import (
+	"context"
+	"testing"
+
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+func TestConsumer_Validate(t *testing.T) {
+	tests := []struct {
+		name    string
+		ctx     context.Context
+		given   *Consumer
+		wantErr bool
+	}{
+		{
+			name:    "no topics",
+			ctx:     context.Background(),
+			given:   &Consumer{},
+			wantErr: true,
+		},
+		{
+			name: "missing group.id",
+			ctx:  context.Background(),
+			given: &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "valid",
+			ctx:  context.Background(),
+			given: &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName:      "p-0",
+						PodNamespace: "ns",
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "invalid multiple reply strategies (topic, destination)",
+			ctx:  context.Background(),
+			given: &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Reply: &ReplyStrategy{
+						TopicReply: &TopicReply{Enabled: true},
+						URLReply:   &DestinationReply{Enabled: true},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName:      "p-0",
+						PodNamespace: "ns",
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "invalid no pod name",
+			ctx:  context.Background(),
+			given: &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodNamespace: "ns",
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "invalid no pod namespace",
+			ctx:  context.Background(),
+			given: &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName: "p-0",
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "valid - no updates",
+			ctx: apis.WithinUpdate(context.Background(), &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName:      "p-0",
+						PodNamespace: "ns",
+					},
+				},
+			},
+			),
+			given: &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName:      "p-0",
+						PodNamespace: "ns",
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "invalid pod name update",
+			ctx: apis.WithinUpdate(context.Background(), &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName:      "p-1",
+						PodNamespace: "ns",
+					},
+				},
+			},
+			),
+			given: &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName:      "p-0",
+						PodNamespace: "ns",
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "invalid pod name update",
+			ctx: apis.WithinUpdate(context.Background(), &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName:      "p-0",
+						PodNamespace: "ns-1",
+					},
+				},
+			},
+			),
+			given: &Consumer{
+				Spec: ConsumerSpec{
+					Topics: []string{"t1"},
+					Configs: ConsumerConfigs{
+						Configs: map[string]string{
+							"group.id":          "g1",
+							"bootstrap.servers": "kafka:9092",
+						},
+					},
+					Delivery: &DeliverySpec{
+						DeliverySpec: &eventingduck.DeliverySpec{},
+					},
+					Subscriber: duckv1.Destination{
+						URI: &apis.URL{
+							Scheme: "http",
+							Host:   "127.0.0.1",
+						},
+					},
+					PodBind: &PodBind{
+						PodName:      "p-0",
+						PodNamespace: "ns-2",
+					},
+				},
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := tt.given.Validate(tt.ctx); (err != nil) != tt.wantErr {
+				t.Errorf("want err = %v, got err %v", tt.wantErr, err)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 4f92bd5c1..a45dc9433 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -19,6 +19,7 @@ package channel_test
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"testing"
 
 	"github.com/Shopify/sarama"
@@ -52,9 +53,11 @@ import (
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 
+	"github.com/rickb777/date/period"
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	messagingv1beta1kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
+	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 )
 
 const (
@@ -63,6 +66,7 @@ const (
 	finalizerName                 = "kafkachannels.messaging.knative.dev"
 	TestExpectedDataNumPartitions = "TestExpectedDataNumPartitions"
 	TestExpectedReplicationFactor = "TestExpectedReplicationFactor"
+	TestExpectedRetentionDuration = "TestExpectedRetentionDuration"
 )
 
 var finalizerUpdatedEvent = Eventf(
@@ -1108,7 +1112,7 @@ func TestReconcileKind(t *testing.T) {
 				NewChannel(
 					WithNumPartitions(3),
 					WithReplicationFactor(4),
-					WithRetentionDuration("1000"),
+					WithRetentionDuration("PT10M"),
 				),
 				NewService(),
 				NewPerChannelService(DefaultEnv),
@@ -1127,6 +1131,7 @@ func TestReconcileKind(t *testing.T) {
 			OtherTestData: map[string]interface{}{
 				TestExpectedDataNumPartitions: int32(3),
 				TestExpectedReplicationFactor: int16(4),
+				TestExpectedRetentionDuration: "PT10M",
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1162,7 +1167,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewChannel(
 						WithNumPartitions(3),
 						WithReplicationFactor(4),
-						WithRetentionDuration("1000"),
+						WithRetentionDuration("PT10M"),
 						WithInitKafkaChannelConditions,
 						StatusConfigParsed,
 						StatusConfigMapUpdatedReady(&env),
@@ -1741,7 +1746,6 @@ func TestFinalizeKind(t *testing.T) {
 
 func useTable(t *testing.T, table TableTest, env config.Env) {
 	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
-
 		proberMock := probertesting.MockProber(prober.StatusReady)
 		if p, ok := row.OtherTestData[testProber]; ok {
 			proberMock = p.(prober.Prober)
@@ -1757,6 +1761,17 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 			replicationFactor = v.(int16)
 		}
 
+		retentionDuration := commonconstants.DefaultRetentionDuration
+		if v, ok := row.OtherTestData[TestExpectedRetentionDuration]; ok {
+			retentionPeriod, err := period.Parse(v.(string))
+			if err != nil {
+				t.Errorf("couldn't parse retention duration: %s", err.Error())
+			}
+			retentionDuration, _ = retentionPeriod.Duration()
+		}
+
+		retentionMillisString := strconv.FormatInt(retentionDuration.Milliseconds(), 10)
+
 		reconciler := &Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
@@ -1785,6 +1800,9 @@ func useTable(t *testing.T, table TableTest, env config.Env) {
 					ExpectedTopicDetail: sarama.TopicDetail{
 						NumPartitions:     numPartitions,
 						ReplicationFactor: replicationFactor,
+						ConfigEntries: map[string]*string{
+							commonconstants.KafkaTopicConfigRetentionMs: &retentionMillisString,
+						},
 					},
 					T: t,
 				}, nil
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
index ab267d9f8..1d15cb5ca 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
@@ -19,6 +19,7 @@ package v2
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"testing"
 
 	"github.com/Shopify/sarama"
@@ -52,8 +53,10 @@ import (
 	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	messagingv1beta1kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
 
+	"github.com/rickb777/date/period"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	fakeconsumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
+	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 )
 
 const (
@@ -62,6 +65,7 @@ const (
 	finalizerName                 = "kafkachannels.messaging.knative.dev"
 	TestExpectedDataNumPartitions = "TestExpectedDataNumPartitions"
 	TestExpectedReplicationFactor = "TestExpectedReplicationFactor"
+	TestExpectedRetentionDuration = "TestExpectedRetentionDuration"
 )
 
 var finalizerUpdatedEvent = Eventf(
@@ -1041,7 +1045,7 @@ func TestReconcileKind(t *testing.T) {
 				NewChannel(
 					WithNumPartitions(3),
 					WithReplicationFactor(4),
-					WithRetentionDuration("1000"),
+					WithRetentionDuration("PT10M"),
 				),
 				NewConfigMapWithTextData(env.SystemNamespace, DefaultEnv.GeneralConfigMapName, map[string]string{
 					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
@@ -1050,6 +1054,7 @@ func TestReconcileKind(t *testing.T) {
 			OtherTestData: map[string]interface{}{
 				TestExpectedDataNumPartitions: int32(3),
 				TestExpectedReplicationFactor: int16(4),
+				TestExpectedRetentionDuration: "PT10M",
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1078,7 +1083,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewChannel(
 						WithNumPartitions(3),
 						WithReplicationFactor(4),
-						WithRetentionDuration("1000"),
+						WithRetentionDuration("PT10M"),
 						WithInitKafkaChannelConditions,
 						StatusConfigParsed,
 						StatusConfigMapUpdatedReady(&env),
@@ -1330,6 +1335,17 @@ func TestReconcileKind(t *testing.T) {
 			replicationFactor = v.(int16)
 		}
 
+		retentionDuration := commonconstants.DefaultRetentionDuration
+		if v, ok := row.OtherTestData[TestExpectedRetentionDuration]; ok {
+			retentionPeriod, err := period.Parse(v.(string))
+			if err != nil {
+				t.Errorf("couldn't parse retention duration: %s", err.Error())
+			}
+			retentionDuration, _ = retentionPeriod.Duration()
+		}
+
+		retentionMillisString := strconv.FormatInt(retentionDuration.Milliseconds(), 10)
+
 		reconciler := &Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
@@ -1352,6 +1368,9 @@ func TestReconcileKind(t *testing.T) {
 					ExpectedTopicDetail: sarama.TopicDetail{
 						NumPartitions:     numPartitions,
 						ReplicationFactor: replicationFactor,
+						ConfigEntries: map[string]*string{
+							commonconstants.KafkaTopicConfigRetentionMs: &retentionMillisString,
+						},
 					},
 					T: t,
 				}, nil
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
index 54441899e..c4abe0c9f 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
@@ -612,9 +612,8 @@ public class RecordDispatcherTest {
           .summaries()
           .stream()
           .reduce((a, b) -> b)
-          .get()
-          .max()
-      ).isGreaterThan(0)
+          .isPresent()
+      ).isTrue()
     );
   }