Repository: knative-sandbox/eventing-kafka-broker
Pull request #2701 (2022-10-04 23:28:54)

Bump protobuf-java from 3.20.0 to 3.20.3 in /data-plane
Bumps [protobuf-java](https://github.com/protocolbuffers/protobuf) from 3.20.0 to 3.20.3.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/protocolbuffers/protobuf/releases">protobuf-java's releases</a>.</em></p>
<blockquote>
<h2>Protocol Buffers v3.20.3</h2>
<h1>Java</h1>
<ul>
<li>Refactoring java full runtime to reuse sub-message builders and prepare to
migrate parsing logic from parse constructor to builder.</li>
<li>Move proto wireformat parsing functionality from the private &quot;parsing
constructor&quot; to the Builder class.</li>
<li>Change the Lite runtime to prefer merging from the wireformat into mutable
messages rather than building up a new immutable object before merging. This
way results in fewer allocations and copy operations.</li>
<li>Make message-type extensions merge from wire-format instead of building up
instances and merging afterwards. This has much better performance.</li>
<li>Fix TextFormat parser to build up recurring (but supposedly not repeated)
sub-messages directly from text rather than building a new sub-message and
merging the fully formed message into the existing field.</li>
<li>This release addresses a <a href="https://github.com/protocolbuffers/protobuf/security/advisories/GHSA-h4h5-3hr4-j3g2">Security Advisory for Java users</a></li>
</ul>
<h2>Protocol Buffers v3.20.2</h2>
<h1>C++</h1>
<ul>
<li>Reduce memory consumption of MessageSet parsing</li>
<li>This release addresses a <a href="https://github.com/protocolbuffers/protobuf/security/advisories/GHSA-8gq9-2x98-w8hf">Security Advisory for C++ and Python users</a></li>
</ul>
<h2>Protocol Buffers v3.20.1</h2>
<h1>PHP</h1>
<ul>
<li>Fix building packaged PHP extension (<a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/9727">#9727</a>)</li>
<li>Fixed composer.json to only advertise compatibility with PHP 7.0+.  (<a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/9819">#9819</a>)</li>
</ul>
<h1>Ruby</h1>
<ul>
<li>Disable the aarch64 build on macOS until it can be fixed. (<a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/9816">#9816</a>)</li>
</ul>
<h1>Other</h1>
<ul>
<li>Fix versioning issues in 3.20.0</li>
</ul>
<h2>Protocol Buffers v3.20.1-rc1</h2>
<p>#PHP</p>
<ul>
<li>Fix building packaged PHP extension (<a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/9727">#9727</a>)</li>
</ul>
<p>#Other</p>
<ul>
<li>Fix versioning issues in 3.20.0</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/fe271ab76f2ad2b2b28c10443865d2af21e27e0e"><code>fe271ab</code></a> Updating changelog</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/4d837c0845fc4ea00763be86de1a0e094e2be336"><code>4d837c0</code></a> Updating version.json and repo version numbers to: 20.3</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/24b65038cfe36191268fda355aca0e43d3ecbcbe"><code>24b6503</code></a> Fix TextFormat parser (<a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/10674">#10674</a>)</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/ae6d69dba64334d77dfb633a146c479982620c1c"><code>ae6d69d</code></a> Refactoring Java parsing (3.20.x) (<a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/10666">#10666</a>)</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/139068ba0965766d64a4ad6a7b29abaa659ae694"><code>139068b</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/10607">#10607</a> from deannagarcia/buildZip20</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/ca680efb928ef16996a6df2f2fedd17c7980b2ff"><code>ca680ef</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/10602">#10602</a> from deannagarcia/codespell</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/274188f866f59742f739b2b9b0740a468d29fe7b"><code>274188f</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/protocolbuffers/protobuf/issues/10601">#10601</a> from deannagarcia/3.20.x</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/661b490de7d3fc892d36fa89e9b2efe7281ebfba"><code>661b490</code></a> Use local artifacts</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/0973b7c26ea665194c8a6f641dbdec7b3e8f85c9"><code>0973b7c</code></a> Fix spelling</li>
<li><a href="https://github.com/protocolbuffers/protobuf/commit/d5d2b815acf46e8713e186522a287a9d0479d876"><code>d5d2b81</code></a> Remove osx-aarch_64</li>
<li>Additional commits viewable in <a href="https://github.com/protocolbuffers/protobuf/compare/v3.20.0...v3.20.3">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=com.google.protobuf:protobuf-java&package-manager=maven&previous-version=3.20.0&new-version=3.20.3)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)
You can disable automated security fix PRs for this repo from the [Security Alerts page](https://github.com/knative-sandbox/eventing-kafka-broker/network/alerts).

</details>

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/2701#" title="Author self-approved">dependabot[bot]</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign evankanderson for approval by writing `/assign @evankanderson` 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":["evankanderson"]} -->
knative-prow[bot]: Hi @dependabot[bot]. 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>

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2701?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2701](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2701?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (140ab17) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/d0d8aeb62f813ba69018656d2ab92475ac864cb8?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (d0d8aeb) will **not change** coverage.
> The diff coverage is `n/a`.

> :exclamation: Current head 140ab17 differs from pull request most recent head 6525871. Consider uploading reports for the commit 6525871 to get more accurate results

```diff
@@           Coverage Diff           @@
##             main    #2701   +/-   ##
=======================================
  Coverage   55.98%   55.98%           
=======================================
  Files          75       75           
  Lines        7888     7888           
=======================================
  Hits         4416     4416           
  Misses       3154     3154           
  Partials      318      318           
```



Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

pierDipi: /ok-to-test
pierDipi: @dependabot rebase
knative-prow[bot]: @dependabot[bot]: 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
--- | --- | --- | --- | ---
build-tests_eventing-kafka-broker_main | f6bc2a97a61d95d8f280db700763baeb30d4db6a | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2701/build-tests_eventing-kafka-broker_main/1578071473967337472) | true | `/test build-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 -->

Patch:
diff --git a/.github/workflows/knative-releasability.yaml b/.github/workflows/knative-releasability.yaml
index 930ad8e62..9c20538e7 100644
--- a/.github/workflows/knative-releasability.yaml
+++ b/.github/workflows/knative-releasability.yaml
@@ -25,8 +25,8 @@ jobs:
   releasability:
     uses: knative/actions/.github/workflows/releasability.yaml@main
     with:
-      releaseFamily: ${{ github.event.inputs.releaseFamily || 'v1.7' }}
-      moduleReleaseFamily: ${{ github.event.inputs.moduleReleaseFamily || 'v0.34' }}
+      releaseFamily: ${{ github.event.inputs.releaseFamily || 'v1.8' }}
+      moduleReleaseFamily: ${{ github.event.inputs.moduleReleaseFamily || 'v0.35' }}
       slackChannel: ${{ github.event.inputs.slackChannel || 'release' }}
     secrets:
       SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
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/config/eventing-kafka-broker/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
index 144909b40..261c0bb0c 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
@@ -75,8 +75,6 @@ spec:
               value: kafka-channel-channels-subscriptions
             - name: SINK_CONTRACT_CONFIG_MAP_NAME
               value: kafka-sink-sinks
-            - name: SOURCE_CONTRACT_CONFIG_MAP_NAME
-              value: kafka-source-sources
 
             - name: BROKER_DATA_PLANE_CONFIG_CONFIG_MAP_NAME
               value: config-kafka-broker-data-plane
diff --git a/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-pod-defaulting.yaml b/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-pod-defaulting.yaml
index c1357bda6..6fa59dd05 100644
--- a/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-pod-defaulting.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-webhook/400-webhook-pod-defaulting.yaml
@@ -32,7 +32,6 @@ webhooks:
     reinvocationPolicy: IfNeeded
     matchPolicy: Equivalent
     namespaceSelector:
-      matchExpressions: [ ]
       matchLabels:
         app.kubernetes.io/name: knative-eventing
     objectSelector:
diff --git a/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
index 5480985f6..bcc2bff7d 100644
--- a/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-source/200-controller/500-controller.yaml
@@ -62,8 +62,6 @@ spec:
           env:
             - name: SOURCE_DATA_PLANE_CONFIG_MAP_NAMESPACE
               value: knative-eventing
-            - name: SOURCE_CONTRACT_CONFIG_MAP_NAME
-              value: kafka-source-sources
             - name: SOURCE_CONTRACT_CONFIG_MAP_FORMAT
               value: json
             - name: SOURCE_INGRESS_NAME
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/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 551baec02..96ab3746b 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -22,6 +22,7 @@ import (
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
@@ -398,15 +399,22 @@ func cmNameFromPod(p *corev1.Pod, c *kafkainternals.Consumer) (string, error) {
 }
 
 func podOwnerReference(p *corev1.Pod) base.ConfigMapOption {
+
+	expected := metav1.OwnerReference{
+		APIVersion:         corev1.SchemeGroupVersion.String(),
+		Kind:               "Pod",
+		Name:               p.Name,
+		UID:                p.UID,
+		Controller:         pointer.Bool(true),
+		BlockOwnerDeletion: pointer.Bool(true),
+	}
 	return func(cm *corev1.ConfigMap) {
-		cm.OwnerReferences = append(cm.OwnerReferences, metav1.OwnerReference{
-			APIVersion:         corev1.SchemeGroupVersion.String(),
-			Kind:               "Pod",
-			Name:               p.Name,
-			UID:                p.UID,
-			Controller:         pointer.Bool(true),
-			BlockOwnerDeletion: pointer.Bool(true),
-		})
+		for _, or := range cm.OwnerReferences {
+			if equality.Semantic.DeepDerivative(expected, or) {
+				return
+			}
+		}
+		cm.OwnerReferences = append(cm.OwnerReferences, expected)
 	}
 }
 
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 593114c8e..2449fbf06 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -20,8 +20,8 @@ Lists of 163 third-party dependencies.
      (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.1.1-jre - https://github.com/google/guava/guava)
      (The Apache Software License, Version 2.0) Guava ListenableFuture only (com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava - https://github.com/google/guava/listenablefuture)
      (The Apache Software License, Version 2.0) J2ObjC Annotations (com.google.j2objc:j2objc-annotations:1.3 - https://github.com/google/j2objc/)
-     (BSD-3-Clause) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.20.0 - https://developers.google.com/protocol-buffers/protobuf-java/)
-     (BSD-3-Clause) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.20.0 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
+     (BSD-3-Clause) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.20.3 - https://developers.google.com/protocol-buffers/protobuf-java/)
+     (BSD-3-Clause) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.20.3 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
      (Apache 2.0) OkHttp Logging Interceptor (com.squareup.okhttp3:logging-interceptor:3.12.12 - https://github.com/square/okhttp/logging-interceptor)
      (Apache 2.0) MockWebServer (com.squareup.okhttp3:mockwebserver:3.12.12 - https://github.com/square/okhttp/mockwebserver)
      (Apache 2.0) OkHttp (com.squareup.okhttp3:okhttp:3.14.9 - https://github.com/square/okhttp/okhttp)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 0bb808522..fb0b4b8fd 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -47,7 +47,7 @@
     <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>
+    <protobuf.version>3.20.3</protobuf.version>
     <bucket4j.version>7.4.0</bucket4j.version>
     <slf4j.version>1.7.36</slf4j.version>
     <ch.qos.logback.version>1.4.1</ch.qos.logback.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