Repository: knative-sandbox/eventing-kafka-broker
Pull request #2967 (2023-02-16 15:05:13)

[release-1.9] Properly handle KafkaSource initial offset (#2962)
Cherry-pick of https://github.com/knative-sandbox/eventing-kafka-broker/pull/2962

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

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

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

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

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.9/OWNERS)~~ [pierDipi]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow[bot]: @pierDipi: 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_release-1.9 | 45737830ba1efc66819c7a210df35d0f899e089f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/unit-tests_eventing-kafka-broker_release-1.9/1626236270545997824) | 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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2967](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4573783) into [release-1.9](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/97f9ceeeac4fd59bc93b34ccd5bca475165ce5f1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (97f9cee) will **increase** coverage by `22.67%`.
> The diff coverage is `n/a`.

> :exclamation: Current head 4573783 differs from pull request most recent head 46b003d. Consider uploading reports for the commit 46b003d to get more accurate results

```diff
@@                Coverage Diff                 @@
##             release-1.9    #2967       +/-   ##
==================================================
+ Coverage          58.57%   81.25%   +22.67%     
- Complexity             0      754      +754     
==================================================
  Files                 78       77        -1     
  Lines               8254     2619     -5635     
  Branches               0      231      +231     
==================================================
- Hits                4835     2128     -2707     
+ Misses              3054      357     -2697     
+ Partials             365      134      -231     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.25% <ø> (?)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/source/source.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2Uvc291cmNlLmdv) | | |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | | |
| [control-plane/pkg/kafka/label.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvbGFiZWwuZ28=) | | |
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | | |
| [...ol-plane/pkg/apis/eventing/v1/broker\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MS9icm9rZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [control-plane/pkg/core/config/egress.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvZWdyZXNzLmdv) | | |
| [control-plane/pkg/counter/counter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY291bnRlci9jb3VudGVyLmdv) | | |
| [...ls/kafka/eventing/v1alpha1/consumer\_group\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfZ3JvdXBfdHlwZXMuZ28=) | | |
| [...plane/pkg/reconciler/trigger/namespaced\_trigger.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL25hbWVzcGFjZWRfdHJpZ2dlci5nbw==) | | |
| [control-plane/pkg/contract/serde.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3Qvc2VyZGUuZ28=) | | |
| ... and [145 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2967?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

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: /test unit-tests
knative-prow[bot]: @pierDipi: 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_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/unit-tests_eventing-kafka-broker_release-1.9/1626238089200406528) | true | `/test unit-tests`
integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9/1626238092534878208) | true | `/test integration-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 -->
knative-prow[bot]: @pierDipi: 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
--- | --- | --- | --- | ---
integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9/1626238092534878208) | true | `/test integration-tests-namespaced-broker`
upgrade-tests_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/upgrade-tests_eventing-kafka-broker_release-1.9/1626238094216794112) | true | `/test upgrade-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 -->
pierDipi: /retest-required
knative-prow[bot]: @pierDipi: 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
--- | --- | --- | --- | ---
integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9/1626238092534878208) | true | `/test integration-tests-namespaced-broker`
upgrade-tests_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/upgrade-tests_eventing-kafka-broker_release-1.9/1626244407269265408) | true | `/test upgrade-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 -->
matzew: /retest
knative-prow[bot]: @pierDipi: 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
--- | --- | --- | --- | ---
integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9/1626238092534878208) | true | `/test integration-tests-namespaced-broker`
upgrade-tests_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/upgrade-tests_eventing-kafka-broker_release-1.9/1626244407269265408) | true | `/test upgrade-tests`
channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.9/1626238103058386944) | 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]: @pierDipi: 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
--- | --- | --- | --- | ---
integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/integration-tests-namespaced-broker_eventing-kafka-broker_release-1.9/1626238092534878208) | true | `/test integration-tests-namespaced-broker`
channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.9/1626238103058386944) | true | `/test channel-integration-tests-sasl-plain`
upgrade-tests_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/upgrade-tests_eventing-kafka-broker_release-1.9/1626245971421696000) | true | `/test upgrade-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 -->
pierDipi: /retest
knative-prow[bot]: @pierDipi: 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_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.9/1626238103058386944) | true | `/test channel-integration-tests-sasl-plain`
upgrade-tests_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/upgrade-tests_eventing-kafka-broker_release-1.9/1626254731972186112) | true | `/test upgrade-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]: @pierDipi: 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
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/upgrade-tests_eventing-kafka-broker_release-1.9/1626254731972186112) | true | `/test upgrade-tests`
channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.9/1626254732030906368) | 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 -->
matzew: /retest
knative-prow[bot]: @pierDipi: 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_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.9/1626254732030906368) | true | `/test channel-integration-tests-sasl-plain`
upgrade-tests_eventing-kafka-broker_release-1.9 | 46b003d66eda34f77881ba9bb919118b612d0eef | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2967/upgrade-tests_eventing-kafka-broker_release-1.9/1626480792123215872) | true | `/test upgrade-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 -->
pierDipi: /test upgrade-tests

Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
index 47324b677..8b6590584 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
@@ -21,6 +21,14 @@ metadata:
   labels:
     kafka.eventing.knative.dev/release: devel
 rules:
+  - apiGroups:
+      - ""
+    resources:
+      - namespaces
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - ""
     resources:
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker.go b/control-plane/pkg/reconciler/broker/namespaced_broker.go
index 58ac11d09..befaf78a4 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_broker.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker.go
@@ -30,6 +30,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	appslisters "k8s.io/client-go/listers/apps/v1"
@@ -42,6 +43,8 @@ import (
 	"knative.dev/pkg/resolver"
 
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
+	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/counter"
@@ -49,6 +52,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/propagator"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/util"
 )
 
 // TODO: 5.
@@ -61,11 +65,13 @@ type NamespacedReconciler struct {
 
 	Resolver *resolver.URIResolver
 
+	NamespaceLister          corelisters.NamespaceLister
 	ConfigMapLister          corelisters.ConfigMapLister
 	ServiceAccountLister     corelisters.ServiceAccountLister
 	ServiceLister            corelisters.ServiceLister
 	ClusterRoleBindingLister rbaclisters.ClusterRoleBindingLister
 	DeploymentLister         appslisters.DeploymentLister
+	BrokerLister             eventinglisters.BrokerLister
 
 	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
@@ -78,14 +84,11 @@ type NamespacedReconciler struct {
 
 	IPsLister          prober.IPListerWithMapping
 	ManifestivalClient mf.Client
+
+	DataplaneLifecycleLocksByNamespace util.LockMap[string]
 }
 
 func (r *NamespacedReconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
-	r.IPsLister.Register(
-		types.NamespacedName{Namespace: broker.Namespace, Name: broker.Name},
-		prober.GetIPForService(types.NamespacedName{Namespace: broker.Namespace, Name: r.Env.IngressName}),
-	)
-
 	if broker.Spec.Config != nil && broker.Spec.Config.Namespace != "" && broker.Spec.Config.Namespace != broker.Namespace {
 		return propagateErrorCondition(broker,
 			fmt.Errorf("broker with %q class need the config in the same namespace with the broker. broker.spec.config.namespace=%q, broker.namespace=%q",
@@ -94,6 +97,27 @@ func (r *NamespacedReconciler) ReconcileKind(ctx context.Context, broker *eventi
 
 	br := r.createReconcilerForBrokerInstance(broker)
 
+	event := r.reconcileDataPlane(ctx, broker)
+	if event != nil {
+		return event
+	}
+
+	return br.ReconcileKind(ctx, broker)
+}
+
+// reconcileDataPlane reconciles the data plane in the namespace of the given broker.
+// A lock is acquired for the namespace of the broker to ensure that the creation of the data plane is not done
+// at the same time with the deletion of the data plane.
+func (r *NamespacedReconciler) reconcileDataPlane(ctx context.Context, broker *eventing.Broker) reconciler.Event {
+	namespaceLock := r.DataplaneLifecycleLocksByNamespace.GetLock(broker.Namespace)
+	namespaceLock.Lock()
+	defer namespaceLock.Unlock()
+
+	r.IPsLister.Register(
+		types.NamespacedName{Namespace: broker.Namespace, Name: broker.Name},
+		prober.GetIPForService(types.NamespacedName{Namespace: broker.Namespace, Name: r.Env.IngressName}),
+	)
+
 	manifest, err := r.getManifest(ctx, broker)
 	if err != nil {
 		return propagateErrorCondition(broker, fmt.Errorf("unable to transform dataplane manifest: %w", err))
@@ -103,7 +127,7 @@ func (r *NamespacedReconciler) ReconcileKind(ctx context.Context, broker *eventi
 		return propagateErrorCondition(broker, fmt.Errorf("unable to apply dataplane manifest: %w", err))
 	}
 
-	return br.ReconcileKind(ctx, broker)
+	return nil
 }
 
 func (r *NamespacedReconciler) FinalizeKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
@@ -112,9 +136,79 @@ func (r *NamespacedReconciler) FinalizeKind(ctx context.Context, broker *eventin
 
 	r.IPsLister.Unregister(types.NamespacedName{Namespace: broker.Namespace, Name: broker.Name})
 
+	if err := r.finalizeDataPlane(ctx, broker); err != nil {
+		return err
+	}
+
 	return result
 }
 
+// finalizeDataPlane finalizes the data plane in the namespace of the given broker.
+// A lock is acquired for the namespace of the broker to ensure that the deletion of the data plane is not done
+// at the same time with the creation of the data plane.
+func (r *NamespacedReconciler) finalizeDataPlane(ctx context.Context, broker *eventing.Broker) reconciler.Event {
+	namespaceLock := r.DataplaneLifecycleLocksByNamespace.GetLock(broker.Namespace)
+	namespaceLock.Lock()
+	defer namespaceLock.Unlock()
+
+	count, err := r.namespacedBrokerCountInNamespace(broker.Namespace)
+	if err != nil {
+		return fmt.Errorf("failed to count namespaced brokers in the namespace for clean up: %w", err)
+	}
+
+	if count == 0 {
+		// there's no more namespaced broker in the namespace, we can clean up the cluster scoped resources
+		err := r.cleanUpClusterScopedResources(ctx, broker)
+		if err != nil {
+			return fmt.Errorf("failed to clean up cluster scoped resources for KafkaNamespaced brokers in the namespace: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// cleanUpClusterScopedResources deletes the cluster scoped resources created for the data plane in the namespace, if
+// this is the last namespaced broker in the namespace
+func (r *NamespacedReconciler) cleanUpClusterScopedResources(ctx context.Context, broker *eventing.Broker) error {
+	manifest, err := r.getManifest(ctx, broker)
+	if err != nil {
+		return fmt.Errorf("unable to transform dataplane manifest for deletion: %w", err)
+	}
+
+	// Delete the cluster scoped resources created for the data plane in the namespace.
+	manifest = manifest.Filter(FilterClusterScoped)
+
+	// Only delete cluster scoped resources, others are garbage collected by Kubernetes already
+	err = manifest.Delete()
+	if err != nil {
+		return fmt.Errorf("failed to delete additional resources: %w", err)
+	}
+
+	return nil
+}
+
+// namespacedBrokerCountInNamespace returns the number of non-deleted namespaced brokers in the namespace
+func (r *NamespacedReconciler) namespacedBrokerCountInNamespace(namespace string) (int, error) {
+	var list []*eventing.Broker
+	var err error
+
+	if list, err = r.BrokerLister.Brokers(namespace).List(labels.Everything()); err != nil {
+		return 0, fmt.Errorf("failed to list brokers in the namespace: %w", err)
+	}
+
+	namespacedBrokerCount := 0
+	for _, b := range list {
+		if !b.DeletionTimestamp.IsZero() {
+			continue
+		}
+		if b.Annotations[brokerreconciler.ClassAnnotationKey] == kafka.NamespacedBrokerClass {
+			namespacedBrokerCount++
+		}
+	}
+
+	return namespacedBrokerCount, err
+}
+
 func (r *NamespacedReconciler) createReconcilerForBrokerInstance(broker *eventing.Broker) *Reconciler {
 
 	return &Reconciler{
@@ -150,6 +244,11 @@ func (r *NamespacedReconciler) createReconcilerForBrokerInstance(broker *eventin
 // getManifest returns the manifest that is transformed from the BaseDataPlaneManifest with the changes
 // for the given broker
 func (r *NamespacedReconciler) getManifest(ctx context.Context, broker *eventing.Broker) (mf.Manifest, error) {
+	namespace, err := r.NamespaceLister.Get(broker.Namespace)
+	if err != nil {
+		return mf.Manifest{}, fmt.Errorf("failed to get namespace of the broker: %w", err)
+	}
+
 	var resources []unstructured.Unstructured
 
 	additionalConfigMaps, err := r.configMapsFromSystemNamespace(broker)
@@ -203,6 +302,14 @@ func (r *NamespacedReconciler) getManifest(ctx context.Context, broker *eventing
 		return mf.Manifest{}, fmt.Errorf("unable to append owner ref: %w", err)
 	}
 
+	// Set the namespace as owner ref to cluster-scoped resources. We delete them manually, but as a safety net we set
+	// the namespace as owner ref so that even when cluster scoped resources cannot be deleted, they will be garbage
+	// collected when the namespace is deleted.
+	manifest, err = manifest.Transform(appendNamespaceAsOwnerRefsToClusterScoped(namespace))
+	if err != nil {
+		return mf.Manifest{}, fmt.Errorf("unable to set namespace as owner ref to cluster-scoped resources: %w", err)
+	}
+
 	manifest, err = manifest.Transform(filterMetadata)
 	if err != nil {
 		return mf.Manifest{}, fmt.Errorf("unable to filter metadata: %w", err)
@@ -433,6 +540,22 @@ func unstructuredFromObject(obj runtime.Object) (unstructured.Unstructured, erro
 	return unstructured.Unstructured{Object: unstr}, nil
 }
 
+func appendNamespaceAsOwnerRefsToClusterScoped(namespace *corev1.Namespace) mf.Transformer {
+	return func(resource *unstructured.Unstructured) error {
+		if !isClusterScoped(resource.GetKind()) || strings.ToLower(resource.GetKind()) == "namespace" {
+			return nil
+		}
+
+		ref := *metav1.NewControllerRef(namespace, corev1.SchemeGroupVersion.WithKind("Namespace"))
+		ref.Controller = pointer.Bool(true)
+		ref.BlockOwnerDeletion = pointer.Bool(false)
+
+		resource.SetOwnerReferences([]metav1.OwnerReference{ref})
+
+		return nil
+	}
+}
+
 func appendNewOwnerRefsToPersisted(client mf.Client, broker *eventing.Broker) mf.Transformer {
 	return func(resource *unstructured.Unstructured) error {
 		if isClusterScoped(resource.GetKind()) {
@@ -549,3 +672,7 @@ func isClusterScoped(kind string) bool {
 	}
 	return false
 }
+
+func FilterClusterScoped(u *unstructured.Unstructured) bool {
+	return isClusterScoped(u.GetKind())
+}
diff --git a/control-plane/pkg/reconciler/broker/namespaced_controller.go b/control-plane/pkg/reconciler/broker/namespaced_controller.go
index 846aadcac..22b2ec85b 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_controller.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_controller.go
@@ -19,6 +19,9 @@ package broker
 import (
 	"context"
 	"net/http"
+	"time"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/util"
 
 	"github.com/Shopify/sarama"
 	mfclient "github.com/manifestival/client-go-client"
@@ -46,6 +49,7 @@ import (
 	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
 	deploymentinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
+	namespaceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/namespace"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 	serviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
@@ -89,15 +93,18 @@ func NewNamespacedController(ctx context.Context, watcher configmap.Watcher, env
 			DispatcherLabel:              base.BrokerDispatcherLabel,
 			ReceiverLabel:                base.BrokerReceiverLabel,
 		},
-		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
-		ConfigMapLister:            configmapInformer.Lister(),
-		ServiceAccountLister:       serviceaccountinformer.Get(ctx).Lister(),
-		ServiceLister:              serviceinformer.Get(ctx).Lister(),
-		ClusterRoleBindingLister:   clusterrolebindinginformer.Get(ctx).Lister(),
-		DeploymentLister:           deploymentinformer.Get(ctx).Lister(),
-		Env:                        env,
-		Counter:                    counter.NewExpiringCounter(ctx),
-		ManifestivalClient:         mfc,
+		NewKafkaClusterAdminClient:         sarama.NewClusterAdmin,
+		NamespaceLister:                    namespaceinformer.Get(ctx).Lister(),
+		ConfigMapLister:                    configmapInformer.Lister(),
+		ServiceAccountLister:               serviceaccountinformer.Get(ctx).Lister(),
+		ServiceLister:                      serviceinformer.Get(ctx).Lister(),
+		ClusterRoleBindingLister:           clusterrolebindinginformer.Get(ctx).Lister(),
+		DeploymentLister:                   deploymentinformer.Get(ctx).Lister(),
+		BrokerLister:                       brokerinformer.Get(ctx).Lister(),
+		Env:                                env,
+		Counter:                            counter.NewExpiringCounter(ctx),
+		ManifestivalClient:                 mfc,
+		DataplaneLifecycleLocksByNamespace: util.NewExpiringLockMap[string](ctx, time.Minute*30),
 	}
 
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.NamespacedBrokerClass, func(impl *controller.Impl) controller.Options {
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index f93e20ee9..9f9aac0d3 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -131,6 +131,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 						NetSpec: &ks.Spec.KafkaAuthSpec.Net,
 					},
 					Delivery: &internalscg.DeliverySpec{
+						InitialOffset: ks.Spec.InitialOffset,
 						DeliverySpec: &eventingduck.DeliverySpec{
 							Retry:         pointer.Int32(10),
 							BackoffPolicy: &backoffPolicy,
diff --git a/control-plane/pkg/util/lock_map.go b/control-plane/pkg/util/lock_map.go
new file mode 100644
index 000000000..a1c9b6e8d
--- /dev/null
+++ b/control-plane/pkg/util/lock_map.go
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2023 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 util
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+// LockMap provides a mechanism to get a lock for a given key.
+type LockMap[K comparable] interface {
+	GetLock(K) *sync.Mutex
+}
+
+type expiringLockMap[K comparable] struct {
+	mu     sync.Mutex
+	locks  map[K]*sync.Mutex
+	access map[K]time.Time
+
+	ttl time.Duration
+}
+
+// NewExpiringLockMap returns a new LockMap that removes entries after the given TTL. Timing of the removal is done
+// in a best-effort fashion, and is not guaranteed to be exact.
+func NewExpiringLockMap[K comparable](ctx context.Context, ttl time.Duration) LockMap[K] {
+	lm := &expiringLockMap[K]{
+		locks:  make(map[K]*sync.Mutex),
+		access: make(map[K]time.Time),
+		ttl:    ttl,
+	}
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(ttl):
+				lm.removeExpiredEntries(time.Now())
+			}
+		}
+	}()
+
+	return lm
+}
+
+func (lm *expiringLockMap[K]) GetLock(key K) *sync.Mutex {
+	lm.mu.Lock()
+	defer lm.mu.Unlock()
+
+	var lock *sync.Mutex
+	var ok bool
+	if lock, ok = lm.locks[key]; !ok {
+		lock = &sync.Mutex{}
+		lm.locks[key] = lock
+	}
+
+	lm.access[key] = time.Now()
+
+	return lock
+}
+
+func (lm *expiringLockMap[K]) removeExpiredEntries(now time.Time) {
+	lm.mu.Lock()
+	defer lm.mu.Unlock()
+
+	for key, lastAccess := range lm.access {
+		if now.Sub(lastAccess) > lm.ttl {
+			delete(lm.locks, key)
+			delete(lm.access, key)
+		}
+	}
+}
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 690f4f708..402cef5cb 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -72,22 +72,22 @@ Lists of 216 third-party dependencies.
      (BSD-2-Clause) org-crac (io.github.crac:org-crac:0.1.1 - https://github.com/crac/org.crac)
      (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.82.Final - https://netty.io/netty-buffer/)
-     (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.82.Final - https://netty.io/netty-codec/)
-     (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.82.Final - https://netty.io/netty-codec-dns/)
-     (Apache License, Version 2.0) Netty/Codec/HAProxy (io.netty:netty-codec-haproxy:4.1.82.Final - https://netty.io/netty-codec-haproxy/)
-     (Apache License, Version 2.0) Netty/Codec/HTTP (io.netty:netty-codec-http:4.1.82.Final - https://netty.io/netty-codec-http/)
-     (Apache License, Version 2.0) Netty/Codec/HTTP2 (io.netty:netty-codec-http2:4.1.82.Final - https://netty.io/netty-codec-http2/)
-     (Apache License, Version 2.0) Netty/Codec/Socks (io.netty:netty-codec-socks:4.1.82.Final - https://netty.io/netty-codec-socks/)
-     (Apache License, Version 2.0) Netty/Common (io.netty:netty-common:4.1.82.Final - https://netty.io/netty-common/)
-     (Apache License, Version 2.0) Netty/Handler (io.netty:netty-handler:4.1.82.Final - https://netty.io/netty-handler/)
-     (Apache License, Version 2.0) Netty/Handler/Proxy (io.netty:netty-handler-proxy:4.1.82.Final - https://netty.io/netty-handler-proxy/)
-     (Apache License, Version 2.0) Netty/Resolver (io.netty:netty-resolver:4.1.82.Final - https://netty.io/netty-resolver/)
-     (Apache License, Version 2.0) Netty/Resolver/DNS (io.netty:netty-resolver-dns:4.1.82.Final - https://netty.io/netty-resolver-dns/)
-     (Apache License, Version 2.0) Netty/Transport (io.netty:netty-transport:4.1.82.Final - https://netty.io/netty-transport/)
-     (Apache License, Version 2.0) Netty/Transport/Classes/Epoll (io.netty:netty-transport-classes-epoll:4.1.82.Final - https://netty.io/netty-transport-classes-epoll/)
-     (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.82.Final - https://netty.io/netty-transport-native-epoll/)
-     (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.82.Final - https://netty.io/netty-transport-native-unix-common/)
+     (Apache License, Version 2.0) Netty/Buffer (io.netty:netty-buffer:4.1.86.Final - https://netty.io/netty-buffer/)
+     (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.86.Final - https://netty.io/netty-codec/)
+     (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.86.Final - https://netty.io/netty-codec-dns/)
+     (Apache License, Version 2.0) Netty/Codec/HAProxy (io.netty:netty-codec-haproxy:4.1.86.Final - https://netty.io/netty-codec-haproxy/)
+     (Apache License, Version 2.0) Netty/Codec/HTTP (io.netty:netty-codec-http:4.1.86.Final - https://netty.io/netty-codec-http/)
+     (Apache License, Version 2.0) Netty/Codec/HTTP2 (io.netty:netty-codec-http2:4.1.86.Final - https://netty.io/netty-codec-http2/)
+     (Apache License, Version 2.0) Netty/Codec/Socks (io.netty:netty-codec-socks:4.1.86.Final - https://netty.io/netty-codec-socks/)
+     (Apache License, Version 2.0) Netty/Common (io.netty:netty-common:4.1.86.Final - https://netty.io/netty-common/)
+     (Apache License, Version 2.0) Netty/Handler (io.netty:netty-handler:4.1.86.Final - https://netty.io/netty-handler/)
+     (Apache License, Version 2.0) Netty/Handler/Proxy (io.netty:netty-handler-proxy:4.1.86.Final - https://netty.io/netty-handler-proxy/)
+     (Apache License, Version 2.0) Netty/Resolver (io.netty:netty-resolver:4.1.86.Final - https://netty.io/netty-resolver/)
+     (Apache License, Version 2.0) Netty/Resolver/DNS (io.netty:netty-resolver-dns:4.1.86.Final - https://netty.io/netty-resolver-dns/)
+     (Apache License, Version 2.0) Netty/Transport (io.netty:netty-transport:4.1.86.Final - https://netty.io/netty-transport/)
+     (Apache License, Version 2.0) Netty/Transport/Classes/Epoll (io.netty:netty-transport-classes-epoll:4.1.86.Final - https://netty.io/netty-transport-classes-epoll/)
+     (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.86.Final - https://netty.io/netty-transport-native-epoll/)
+     (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.86.Final - https://netty.io/netty-transport-native-unix-common/)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.17.0 - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.17.0 - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-common:1.17.0 - https://github.com/open-telemetry/opentelemetry-java)
@@ -106,18 +106,18 @@ Lists of 216 third-party dependencies.
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - Common (io.prometheus:simpleclient_tracer_common:0.15.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_common)
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - OpenTelemetry (io.prometheus:simpleclient_tracer_otel:0.15.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_otel)
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - OpenTelemetry Agent (io.prometheus:simpleclient_tracer_otel_agent:0.15.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_otel_agent)
-     (The Apache Software License, Version 2.0) Quarkus - ArC - Runtime (io.quarkus:quarkus-arc:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (Apache License, Version 2.0) Quarkus - Bootstrap - Runner (io.quarkus:quarkus-bootstrap-runner:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (The Apache Software License, Version 2.0) Quarkus - Core - Runtime (io.quarkus:quarkus-core:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (The Apache Software License, Version 2.0) Quarkus - Development mode - SPI (io.quarkus:quarkus-development-mode-spi:2.13.5.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - ArC - Runtime (io.quarkus:quarkus-arc:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (Apache License, Version 2.0) Quarkus - Bootstrap - Runner (io.quarkus:quarkus-bootstrap-runner:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - Core - Runtime (io.quarkus:quarkus-core:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - Development mode - SPI (io.quarkus:quarkus-development-mode-spi:2.13.7.Final - https://github.com/quarkusio/quarkus)
      (Apache License, Version 2.0) Quarkus - FS Util (io.quarkus:quarkus-fs-util:0.0.9 - https://quarkus.io/)
-     (The Apache Software License, Version 2.0) Quarkus - IDE Launcher (io.quarkus:quarkus-ide-launcher:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (The Apache Software License, Version 2.0) Quarkus - Mutiny - Runtime (io.quarkus:quarkus-mutiny:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (The Apache Software License, Version 2.0) Quarkus - Netty - Runtime (io.quarkus:quarkus-netty:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (The Apache Software License, Version 2.0) Quarkus - SmallRye Context Propagation - Runtime (io.quarkus:quarkus-smallrye-context-propagation:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (The Apache Software License, Version 2.0) Quarkus - Vert.x - Runtime (io.quarkus:quarkus-vertx:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (The Apache Software License, Version 2.0) Quarkus - Quarkus - Vert.x Late Bound MDC Provider (io.quarkus:quarkus-vertx-latebound-mdc-provider:2.13.5.Final - https://github.com/quarkusio/quarkus)
-     (Apache License, Version 2.0) ArC - Runtime (io.quarkus.arc:arc:2.13.5.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - IDE Launcher (io.quarkus:quarkus-ide-launcher:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - Mutiny - Runtime (io.quarkus:quarkus-mutiny:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - Netty - Runtime (io.quarkus:quarkus-netty:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - SmallRye Context Propagation - Runtime (io.quarkus:quarkus-smallrye-context-propagation:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - Vert.x - Runtime (io.quarkus:quarkus-vertx:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (The Apache Software License, Version 2.0) Quarkus - Quarkus - Vert.x Late Bound MDC Provider (io.quarkus:quarkus-vertx-latebound-mdc-provider:2.13.7.Final - https://github.com/quarkusio/quarkus)
+     (Apache License, Version 2.0) ArC - Runtime (io.quarkus.arc:arc:2.13.7.Final - https://github.com/quarkusio/quarkus)
      (The Apache Software License, Version 2.0) smallrye-context-propagation (io.smallrye:smallrye-context-propagation:1.2.2 - https://github.com/smallrye/smallrye-context-propagation/smallrye-context-propagation)
      (The Apache Software License, Version 2.0) smallrye-context-propagation-api (io.smallrye:smallrye-context-propagation-api:1.2.2 - https://github.com/smallrye/smallrye-context-propagation/smallrye-context-propagation-api)
      (The Apache Software License, Version 2.0) smallrye-context-propagation-storage (io.smallrye:smallrye-context-propagation-storage:1.2.2 - https://github.com/smallrye/smallrye-context-propagation/smallrye-context-propagation-storage)
@@ -129,9 +129,9 @@ Lists of 216 third-party dependencies.
      (Apache License, Version 2.0) SmallRye Common: Functions (io.smallrye.common:smallrye-common-function:1.13.1 - http://smallrye.io/smallrye-common-function)
      (Apache License, Version 2.0) SmallRye Common: IO (io.smallrye.common:smallrye-common-io:1.13.1 - http://smallrye.io/smallrye-common-io)
      (Apache License, Version 2.0) SmallRye Common: Vert.x Context Utilities (io.smallrye.common:smallrye-common-vertx-context:1.13.1 - http://smallrye.io/smallrye-common-vertx-context)
-     (Apache License, Version 2.0) SmallRye: MicroProfile Config CDI Implementation (io.smallrye.config:smallrye-config:2.12.0 - http://smallrye.io/smallrye-config)
-     (Apache License, Version 2.0) SmallRye: Common classes (io.smallrye.config:smallrye-config-common:2.12.0 - http://smallrye.io/smallrye-config-common)
-     (Apache License, Version 2.0) SmallRye: MicroProfile Config Core Implementation (io.smallrye.config:smallrye-config-core:2.12.0 - http://smallrye.io/smallrye-config-core)
+     (Apache License, Version 2.0) SmallRye: MicroProfile Config CDI Implementation (io.smallrye.config:smallrye-config:2.12.3 - http://smallrye.io/smallrye-config)
+     (Apache License, Version 2.0) SmallRye: Common classes (io.smallrye.config:smallrye-config-common:2.12.3 - http://smallrye.io/smallrye-config-common)
+     (Apache License, Version 2.0) SmallRye: MicroProfile Config Core Implementation (io.smallrye.config:smallrye-config-core:2.12.3 - http://smallrye.io/smallrye-config-core)
      (Apache License, Version 2.0) SmallRye Mutiny - Core library (io.smallrye.reactive:mutiny:1.7.0 - https://smallrye.io/smallrye-mutiny/mutiny)
      (Apache License, Version 2.0) SmallRye Mutiny - Integration with SmallRye Context Propagation (io.smallrye.reactive:mutiny-smallrye-context-propagation:1.7.0 - https://smallrye.io/smallrye-mutiny/mutiny-smallrye-context-propagation)
      (The Apache Software License, Version 2.0) SmallRye Mutiny - Vert.x Auth Common (io.smallrye.reactive:smallrye-mutiny-vertx-auth-common:2.27.0 - https://smallrye.io/smallrye-mutiny-vertx-bindings/vertx-mutiny-clients/smallrye-mutiny-vertx-auth-common)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index f289a9b1e..6b1b9df92 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -60,7 +60,7 @@
     <kafka.version>3.2.3</kafka.version>
     <debezium.version>1.9.6.Final</debezium.version>
     <jib.version>3.2.1</jib.version>
-    <quarkus.version>2.13.5.Final</quarkus.version>
+    <quarkus.version>2.13.7.Final</quarkus.version>
   </properties>
 
   <modules>
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/namespace/namespace.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/namespace/namespace.go
new file mode 100644
index 000000000..aa57e789f
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/namespace/namespace.go
@@ -0,0 +1,110 @@
+/*
+Copyright 2022 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package namespace
+
+import (
+	context "context"
+
+	apicorev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	v1 "k8s.io/client-go/informers/core/v1"
+	kubernetes "k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/listers/core/v1"
+	cache "k8s.io/client-go/tools/cache"
+	client "knative.dev/pkg/client/injection/kube/client"
+	factory "knative.dev/pkg/client/injection/kube/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Core().V1().Namespaces()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.NamespaceInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch k8s.io/client-go/informers/core/v1.NamespaceInformer from context.")
+	}
+	return untyped.(v1.NamespaceInformer)
+}
+
+type wrapper struct {
+	client kubernetes.Interface
+
+	resourceVersion string
+}
+
+var _ v1.NamespaceInformer = (*wrapper)(nil)
+var _ corev1.NamespaceLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apicorev1.Namespace{}, 0, nil)
+}
+
+func (w *wrapper) Lister() corev1.NamespaceLister {
+	return w
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apicorev1.Namespace, err error) {
+	lo, err := w.client.CoreV1().Namespaces().List(context.TODO(), metav1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apicorev1.Namespace, error) {
+	return w.client.CoreV1().Namespaces().Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 178cf5a95..84d32b553 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1401,6 +1401,7 @@ knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset
 knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/configmap
 knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake
+knative.dev/pkg/client/injection/kube/informers/core/v1/namespace
 knative.dev/pkg/client/injection/kube/informers/core/v1/node
 knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
index 848f90259..7da4b1e5d 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
@@ -20,8 +20,10 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/utils/pointer"
@@ -34,10 +36,12 @@ import (
 	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/util"
 
 	"github.com/Shopify/sarama"
 	"github.com/manifestival/client-go-client"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -79,6 +83,9 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 		{
 			Name: "Reconciled normal",
 			Objects: []runtime.Object{
+				reconcilertesting.NewNamespace(BrokerNamespace, func(ns *corev1.Namespace) {
+					ns.UID = BrokerNamespaceUUID
+				}),
 				NewNamespacedBroker(
 					WithBrokerConfig(
 						KReference(BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace))),
@@ -108,6 +115,24 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					WithClusterRoleBindingSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
 					WithClusterRoleBindingRoleRef("knative-kafka-broker-data-plane"),
 				),
+				NewConfigMapWithTextData(SystemNamespace, NamespacedBrokerAdditionalResourcesConfigMapName, map[string]string{
+					"resources": `
+- apiVersion: rbac.authorization.k8s.io/v1
+  kind: ClusterRole
+  metadata:
+    name: test-role
+    labels:
+      "knative.foo": "foo-{{.Namespace}}"
+    creationTimestamp: null
+  rules:
+    - apiGroups:
+        - "v1"
+      resources:
+        - pods
+      verbs:
+        - get
+`,
+				}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -164,6 +189,16 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					WithNamespacedBrokerOwnerRef,
 					WithNamespacedLabel,
 				),
+				ToManifestivalResource(t, NewClusterRole("test-role",
+					WithClusterRoleLabel("knative.foo", "foo-"+BrokerNamespace),
+					WithClusterRoleRules(rbacv1.PolicyRule{
+						APIGroups: []string{"v1"},
+						Resources: []string{"pods"},
+						Verbs:     []string{"get"},
+					})),
+					WithNamespacedBrokerNamespaceAsOwnerRef,
+					WithNamespacedLabel,
+				),
 				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil,
 					reconcilertesting.WithConfigMapLabels(metav1.LabelSelector{MatchLabels: map[string]string{"eventing.knative.dev/namespaced": "true"}}),
 					WithConfigmapOwnerRef(&metav1.OwnerReference{
@@ -230,6 +265,9 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					),
 				},
 			},
+			// true since we're creating cluster scoped resources and namespace validation is checking if the namespace
+			// of the broker is the same with the namespace of the cluster scoped resource (which is nil)
+			SkipNamespaceValidation: true,
 		},
 	}
 
@@ -258,6 +296,9 @@ func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
 		{
 			Name: "Reconciled normal",
 			Objects: []runtime.Object{
+				reconcilertesting.NewNamespace(BrokerNamespace, func(ns *corev1.Namespace) {
+					ns.UID = BrokerNamespaceUUID
+				}),
 				NewDeletedBroker(reconcilertesting.WithBrokerClass(kafka.NamespacedBrokerClass)),
 				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
@@ -270,6 +311,41 @@ func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
 					},
 					Generation: 1,
 				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
+				reconcilertesting.NewConfigMap(env.DataPlaneConfigConfigMapName, SystemNamespace),
+				reconcilertesting.NewConfigMap("config-tracing", SystemNamespace),
+				reconcilertesting.NewConfigMap("kafka-config-logging", SystemNamespace),
+				reconcilertesting.NewDeployment("kafka-broker-receiver", SystemNamespace),
+				reconcilertesting.NewDeployment("kafka-broker-dispatcher", SystemNamespace),
+				NewServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
+				reconcilertesting.NewService("kafka-broker-ingress", SystemNamespace),
+				NewClusterRoleBinding("knative-kafka-broker-data-plane",
+					WithClusterRoleBindingSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
+				),
+				NewConfigMapWithTextData(SystemNamespace, NamespacedBrokerAdditionalResourcesConfigMapName, map[string]string{
+					"resources": `
+- apiVersion: rbac.authorization.k8s.io/v1
+  kind: ClusterRole
+  metadata:
+    name: test-role
+    labels:
+      "knative.foo": "foo-{{.Namespace}}"
+    creationTimestamp: null
+  rules:
+    - apiGroups:
+        - "v1"
+      resources:
+        - pods
+      verbs:
+        - get
+`,
+				}),
+				NewClusterRole("test-role",
+					WithClusterRoleLabel("knative.foo", "foo-"+BrokerNamespace),
+					WithClusterRoleRules(rbacv1.PolicyRule{
+						APIGroups: []string{"v1"},
+						Resources: []string{"pods"},
+						Verbs:     []string{"get"},
+					})),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
@@ -285,6 +361,18 @@ func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
 					}),
 				),
 			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Resource: schema.GroupVersionResource{
+							Group:    rbacv1.SchemeGroupVersion.Group,
+							Version:  rbacv1.SchemeGroupVersion.Version,
+							Resource: "clusterroles",
+						},
+					},
+					Name: "test-role",
+				},
+			},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
 			},
@@ -354,8 +442,10 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 				ReceiverLabel:               base.BrokerReceiverLabel,
 				Tracker:                     &FakeTracker{},
 			},
+			NamespaceLister:          listers.GetNamespaceLister(),
 			ConfigMapLister:          listers.GetConfigMapLister(),
 			DeploymentLister:         listers.GetDeploymentLister(),
+			BrokerLister:             listers.GetBrokerLister(),
 			ServiceAccountLister:     listers.GetServiceAccountLister(),
 			ServiceLister:            listers.GetServiceLister(),
 			ClusterRoleBindingLister: listers.GetClusterRoleBindingLister(),
@@ -370,9 +460,10 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 					T:                                      t,
 				}, nil
 			},
-			Env:                env,
-			Prober:             proberMock,
-			ManifestivalClient: mfcMockClient,
+			Env:                                env,
+			Prober:                             proberMock,
+			ManifestivalClient:                 mfcMockClient,
+			DataplaneLifecycleLocksByNamespace: util.NewExpiringLockMap[string](ctx, time.Minute*30),
 		}
 
 		r := brokerreconciler.NewReconciler(
@@ -408,6 +499,22 @@ func WithNamespacedBrokerOwnerRef(u *unstructured.Unstructured) {
 	u.SetOwnerReferences(refs)
 }
 
+func WithNamespacedBrokerNamespaceAsOwnerRef(u *unstructured.Unstructured) {
+	refs := u.GetOwnerReferences()
+	if refs == nil {
+		refs = []metav1.OwnerReference{}
+	}
+	refs = append(refs, metav1.OwnerReference{
+		APIVersion:         corev1.SchemeGroupVersion.String(),
+		Kind:               "Namespace",
+		Name:               BrokerNamespace,
+		UID:                BrokerNamespaceUUID,
+		Controller:         pointer.Bool(true),
+		BlockOwnerDeletion: pointer.Bool(false),
+	})
+	u.SetOwnerReferences(refs)
+}
+
 func WithNamespacedLabel(u *unstructured.Unstructured) {
 	labels := u.GetLabels()
 	if labels == nil {
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 066e2b1ce..b7ce36582 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -119,6 +119,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -137,6 +138,100 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal, offset earliest",
+			Objects: []runtime.Object{
+				NewSource(WithInitialOffset(sources.OffsetEarliest)),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+					WithConsumerGroupLabels(ConsumerSourceLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT0.3S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetEarliest),
+							),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						WithInitialOffset(sources.OffsetEarliest),
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal, offset latest",
+			Objects: []runtime.Object{
+				NewSource(WithInitialOffset(sources.OffsetLatest)),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+					WithConsumerGroupLabels(ConsumerSourceLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(NewConsumerSpecAuth()),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT0.3S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
+							),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						WithInitialOffset(sources.OffsetLatest),
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+						StatusSourceSelector(),
+					),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal, key type label",
 			Objects: []runtime.Object{
@@ -165,6 +260,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -293,6 +389,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -405,6 +502,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -454,6 +552,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -506,6 +605,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -552,6 +652,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -584,6 +685,7 @@ func TestReconcileKind(t *testing.T) {
 									NewConsumerRetry(10),
 									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+									ConsumerInitialOffset(sources.OffsetLatest),
 								),
 							),
 							ConsumerSubscriber(NewSourceSink2Reference()),
@@ -643,6 +745,7 @@ func TestReconcileKind(t *testing.T) {
 									NewConsumerRetry(10),
 									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+									ConsumerInitialOffset(sources.OffsetLatest),
 								),
 							),
 							ConsumerSubscriber(NewSourceSinkReference()),
@@ -702,6 +805,7 @@ func TestReconcileKind(t *testing.T) {
 									NewConsumerRetry(10),
 									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+									ConsumerInitialOffset(sources.OffsetLatest),
 								),
 							),
 							ConsumerSubscriber(NewSourceSinkReference()),
@@ -760,6 +864,7 @@ func TestReconcileKind(t *testing.T) {
 									NewConsumerRetry(10),
 									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+									ConsumerInitialOffset(sources.OffsetLatest),
 								),
 							),
 							ConsumerSubscriber(NewSourceSinkReference()),
@@ -805,6 +910,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -850,6 +956,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -894,6 +1001,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -939,6 +1047,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -986,6 +1095,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
@@ -1019,6 +1129,7 @@ func TestReconcileKind(t *testing.T) {
 									NewConsumerRetry(10),
 									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+									ConsumerInitialOffset(sources.OffsetLatest),
 								),
 							),
 							ConsumerSubscriber(NewSourceSinkReference()),
@@ -1066,6 +1177,7 @@ func TestReconcileKind(t *testing.T) {
 								NewConsumerRetry(10),
 								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+								ConsumerInitialOffset(sources.OffsetLatest),
 							),
 						),
 						ConsumerSubscriber(NewSourceSinkReference()),
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 52ce9cb08..b6d57ea04 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -164,6 +164,10 @@ func (l *Listers) GetClusterRoleBindingLister() rbaclisters.ClusterRoleBindingLi
 	return rbaclisters.NewClusterRoleBindingLister(l.indexerFor(&rbacv1.ClusterRoleBinding{}))
 }
 
+func (l *Listers) GetNamespaceLister() corelisters.NamespaceLister {
+	return corelisters.NewNamespaceLister(l.indexerFor(&corev1.Namespace{}))
+}
+
 func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
 	return l.sorter.IndexerForObjectType(obj)
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index c94d3ea45..8b1896289 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -43,10 +43,11 @@ import (
 )
 
 const (
-	BrokerUUID        = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
-	BrokerNamespace   = "test-namespace"
-	BrokerName        = "test-broker"
-	ExternalTopicName = "test-topic"
+	BrokerUUID          = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
+	BrokerNamespaceUUID = "d1234567-8910-1234-5678-901234567890"
+	BrokerNamespace     = "test-namespace"
+	BrokerName          = "test-broker"
+	ExternalTopicName   = "test-topic"
 
 	SecretFinalizerName = "kafka.eventing/" + BrokerUUID
 
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 88605d227..30bcf8633 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -393,6 +393,40 @@ func NewKedaSecret(ns, name string) *corev1.Secret {
 	}
 }
 
+// ClusterRoleOption enables further configuration of a ClusterRole.
+type ClusterRoleOption func(role *rbacv1.ClusterRole)
+
+func NewClusterRole(name string, cro ...ClusterRoleOption) *rbacv1.ClusterRole {
+	cr := &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+
+	for _, opt := range cro {
+		opt(cr)
+	}
+
+	return cr
+}
+
+// WithClusterRoleRules is a ClusterRoleOption that adds the given rules to the ClusterRole.
+func WithClusterRoleRules(rules ...rbacv1.PolicyRule) ClusterRoleOption {
+	return func(role *rbacv1.ClusterRole) {
+		role.Rules = append(role.Rules, rules...)
+	}
+}
+
+// WithClusterRoleLabel is a ClusterRoleOption that adds the given labels to the ClusterRole.
+func WithClusterRoleLabel(key, value string) ClusterRoleOption {
+	return func(role *rbacv1.ClusterRole) {
+		if role.Labels == nil {
+			role.Labels = make(map[string]string)
+		}
+		role.Labels[key] = value
+	}
+}
+
 func loadCerts() (ca, userKey, userCert []byte) {
 	ca, err := os.ReadFile("testdata/ca.crt")
 	if err != nil {
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index bcfa5e072..3a3ead8a6 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -24,13 +24,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/utils/pointer"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
-	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	v1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
+	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
+
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
@@ -114,7 +115,9 @@ func ConsumerUID(uid string) ConsumerOption {
 type DeliverySpecOption func(spec *kafkainternals.DeliverySpec)
 
 func NewConsumerSpecDelivery(order internals.DeliveryOrdering, options ...DeliverySpecOption) *kafkainternals.DeliverySpec {
-	d := &kafkainternals.DeliverySpec{Ordering: order}
+	d := &kafkainternals.DeliverySpec{
+		Ordering: order,
+	}
 
 	for _, opt := range options {
 		opt(d)
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index f49805689..4a338693d 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -21,11 +21,12 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/utils/pointer"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
-	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/bindings/v1beta1"
+	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
@@ -61,6 +62,7 @@ func NewSource(options ...KRShapedOption) *sources.KafkaSource {
 			},
 		},
 	}
+
 	for _, opt := range options {
 		opt(s)
 	}
@@ -151,6 +153,13 @@ func WithSourceConsumers(replicas int32) KRShapedOption {
 	}
 }
 
+func WithInitialOffset(offset sources.Offset) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		s := obj.(*sources.KafkaSource)
+		s.Spec.InitialOffset = offset
+	}
+}
+
 func SourceDispatcherPod(namespace string, annotations map[string]string) runtime.Object {
 	return &corev1.Pod{
 		TypeMeta: metav1.TypeMeta{
diff --git a/control-plane/pkg/util/lock_map_test.go b/control-plane/pkg/util/lock_map_test.go
new file mode 100644
index 000000000..6675279f1
--- /dev/null
+++ b/control-plane/pkg/util/lock_map_test.go
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2023 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 util
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestExpiringLockMap_GetLock(t *testing.T) {
+	chanCount := 1000 // controls concurrency
+	size := 50        // controls how many keys are used and thus the collision rate
+
+	r := rand.New(rand.NewSource(42))
+	randomKey := func() string {
+		return fmt.Sprintf("key-%d", r.Intn(size))
+	}
+
+	wg := sync.WaitGroup{}
+	chans := make([]chan string, chanCount)
+
+	// fill channels with data
+	for i := 0; i < chanCount; i++ {
+		chans[i] = make(chan string, size)
+		for j := 0; j < size; j++ {
+			chans[i] <- randomKey()
+		}
+		close(chans[i])
+	}
+	wg.Add(chanCount)
+
+	// consume data from channels concurrently. Lock and unlock the lock for each key.
+	// TTL is 10 hours, so the locks should never expire.
+	lm := NewExpiringLockMap[string](context.Background(), 10*time.Hour)
+
+	for i := 0; i < chanCount; i++ {
+		go func(i int) {
+			defer wg.Done()
+			for key := range chans[i] {
+				func() {
+					lock := lm.GetLock(key)
+					lock.Lock()
+					defer lock.Unlock()
+				}()
+			}
+		}(i)
+	}
+	wg.Wait()
+}
+
+func TestExpiringLockMapExpiration(t *testing.T) {
+	lm := NewExpiringLockMap[string](context.Background(), 100*time.Millisecond)
+
+	l1_a := lm.GetLock("l1")
+
+	time.Sleep(50 * time.Millisecond)
+	l1_b := lm.GetLock("l1") // should be the same lock as l1_a, as TTL is 100ms and 50 < 100
+
+	time.Sleep(300 * time.Millisecond)
+	l1_c := lm.GetLock("l1") // should be a different lock as TTL is 100ms and 300 > 100
+
+	if l1_a != l1_b {
+		t.Errorf("l1_a != l1_b")
+	}
+	if l1_a == l1_c {
+		t.Errorf("l1_a == l1_c")
+	}
+}
diff --git a/test/e2e_new/kafka_source_test.go b/test/e2e_new/kafka_source_test.go
index f502bc7df..5cec1070a 100644
--- a/test/e2e_new/kafka_source_test.go
+++ b/test/e2e_new/kafka_source_test.go
@@ -24,6 +24,7 @@ import (
 
 	"knative.dev/pkg/system"
 	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 
@@ -78,3 +79,20 @@ func TestKafkaSourceScale(t *testing.T) {
 
 	env.Test(ctx, t, features.ScaleKafkaSource())
 }
+
+func TestKafkaSourceInitialOffsetEarliest(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	topic := feature.MakeRandomK8sName("kafka-topic-earliest")
+
+	env.Test(ctx, t, features.SetupKafkaTopicWithEvents(2, topic))
+	env.Test(ctx, t, features.KafkaSourceInitialOffsetEarliest(2, topic))
+}
diff --git a/test/rekt/features/kafka_sink.go b/test/rekt/features/kafka_sink.go
new file mode 100644
index 000000000..57bd6d948
--- /dev/null
+++ b/test/rekt/features/kafka_sink.go
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2023 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 features
+
+import (
+	"fmt"
+	"time"
+
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+
+	testpkg "knative.dev/eventing-kafka-broker/test/pkg"
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkasink"
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
+)
+
+func SetupKafkaTopicWithEvents(count int, topic string) *feature.Feature {
+
+	f := feature.NewFeatureNamed(fmt.Sprintf("setup Kafka topic with %d events", count))
+
+	ksink := feature.MakeRandomK8sName("ksink")
+	source := feature.MakeRandomK8sName("source-to-ksink")
+
+	f.Setup("install kafka topic", kafkatopic.Install(topic))
+	f.Setup("topic is ready", kafkatopic.IsReady(topic))
+	f.Setup("install kafkasink", kafkasink.Install(ksink, topic, testpkg.BootstrapServersPlaintextArr))
+	f.Setup("KafkaSink is ready", kafkasink.IsReady(ksink))
+
+	f.Requirement("install source for ksink", eventshub.Install(
+		source,
+		eventshub.StartSenderToResource(kafkasink.GVR(), ksink),
+		eventshub.InputEvent(cetest.FullEvent()),
+		eventshub.AddSequence,
+		eventshub.SendMultipleEvents(2, time.Millisecond),
+	))
+
+	return f
+}
diff --git a/test/rekt/features/kafka_source.go b/test/rekt/features/kafka_source.go
index c3a8cade2..01a82e7d2 100644
--- a/test/rekt/features/kafka_source.go
+++ b/test/rekt/features/kafka_source.go
@@ -22,18 +22,23 @@ import (
 	"fmt"
 	"strings"
 
+	cetest "github.com/cloudevents/sdk-go/v2/test"
 	"go.uber.org/zap"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/knative"
 	"knative.dev/reconciler-test/resources/svc"
 
+	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 	kafkaclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client"
-
+	sourcesclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client"
+	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
 	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkasource"
 	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
 
@@ -175,3 +180,69 @@ func ScaleKafkaSource() *feature.Feature {
 
 	return f
 }
+
+func KafkaSourceInitialOffsetEarliest(count int, topic string) *feature.Feature {
+
+	f := feature.NewFeatureNamed("KafkaSource initial offset earliest")
+
+	sink := feature.MakeRandomK8sName("sink")
+	source := feature.MakeRandomK8sName("source")
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
+
+	f.Setup("install kafkasource", kafkasource.Install(
+		source,
+		kafkasource.WithBootstrapServers(testingpkg.BootstrapServersPlaintextArr),
+		kafkasource.WithTopics([]string{topic}),
+		kafkasource.WithInitialOffset(sources.OffsetEarliest),
+		kafkasource.WithSink(svc.AsKReference(sink), ""),
+	))
+	f.Setup("KafkaSource is ready", kafkasource.IsReady(source))
+
+	f.Requirement("consumergroup has earliest offset set", compareConsumerGroup(
+		source,
+		func(group *internalscg.ConsumerGroup) error {
+			got := group.Spec.Template.Spec.Delivery.InitialOffset
+			if got != sources.OffsetEarliest {
+				return fmt.Errorf("expected consumergroup to have offset %s set, got %s", sources.OffsetEarliest, got)
+			}
+			return nil
+		},
+	))
+
+	for i := 1; i <= count; i++ {
+		f.Assert(fmt.Sprintf("received event with sequence %d", i),
+			assert.OnStore(sink).
+				MatchEvent(cetest.HasExtension("sequence", fmt.Sprintf("%d", i))).
+				AtLeast(1),
+		)
+	}
+
+	return f
+}
+
+func compareConsumerGroup(source string, cmp func(*internalscg.ConsumerGroup) error) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+		ns := environment.FromContext(ctx).Namespace()
+
+		ks, err := sourcesclient.Get(ctx).
+			SourcesV1beta1().
+			KafkaSources(ns).
+			Get(ctx, source, metav1.GetOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		InternalsClient := consumergroupclient.Get(ctx)
+		cg, err := InternalsClient.InternalV1alpha1().
+			ConsumerGroups(ns).
+			Get(ctx, string(ks.UID), metav1.GetOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if err := cmp(cg); err != nil {
+			t.Error(err)
+		}
+	}
+}
diff --git a/test/rekt/resources/kafkasink/kafkasink.go b/test/rekt/resources/kafkasink/kafkasink.go
index 0fa409894..6a01b5d04 100644
--- a/test/rekt/resources/kafkasink/kafkasink.go
+++ b/test/rekt/resources/kafkasink/kafkasink.go
@@ -19,9 +19,11 @@ package kafkasink
 import (
 	"context"
 	"embed"
+	"fmt"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/manifest"
@@ -30,7 +32,7 @@ import (
 //go:embed *.yaml
 var yaml embed.FS
 
-func gvr() schema.GroupVersionResource {
+func GVR() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "eventing.knative.dev", Version: "v1alpha1", Resource: "kafkasinks"}
 }
 
@@ -63,7 +65,7 @@ func WithAuthSecretName(name string) manifest.CfgFn {
 }
 
 // Install will create a Trigger resource, augmented with the config fn options.
-func Install(name, topic, bootstrapServers []string, opts ...manifest.CfgFn) feature.StepFn {
+func Install(name, topic string, bootstrapServers []string, opts ...manifest.CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
 		"name":             name,
 		"topic":            topic,
@@ -81,5 +83,14 @@ func Install(name, topic, bootstrapServers []string, opts ...manifest.CfgFn) fea
 
 // IsReady tests to see if a Trigger becomes ready within the time given.
 func IsReady(name string, timing ...time.Duration) feature.StepFn {
-	return k8s.IsReady(gvr(), name, timing...)
+	return k8s.IsReady(GVR(), name, timing...)
+}
+
+func AsKReference(name string, namespace string) *duckv1.KReference {
+	return &duckv1.KReference{
+		Kind:       "KafkaSink",
+		APIVersion: fmt.Sprintf("%s/%s", GVR().Group, GVR().Version),
+		Namespace:  namespace,
+		Name:       name,
+	}
 }
diff --git a/test/rekt/resources/kafkasource/kafkasource.go b/test/rekt/resources/kafkasource/kafkasource.go
index 01ad8a920..34317a027 100644
--- a/test/rekt/resources/kafkasource/kafkasource.go
+++ b/test/rekt/resources/kafkasource/kafkasource.go
@@ -147,6 +147,15 @@ func WithTopics(topics []string) manifest.CfgFn {
 	}
 }
 
+// WithInitialOffset adds the initial offset config to a KafkaSource spec.
+func WithInitialOffset(offset sources.Offset) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if offset != "" {
+			cfg["initialOffset"] = string(offset)
+		}
+	}
+}
+
 // WithTLSEnabled enables TLS to a KafkaSource spec.
 func WithTLSEnabled() manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
diff --git a/test/rekt/resources/kafkasource/kafkasource.yaml b/test/rekt/resources/kafkasource/kafkasource.yaml
index 9813d2d1c..9504682c4 100644
--- a/test/rekt/resources/kafkasource/kafkasource.yaml
+++ b/test/rekt/resources/kafkasource/kafkasource.yaml
@@ -33,6 +33,9 @@ spec:
   {{ if .consumerGroup }}
   consumerGroup: {{ .consumerGroup }}
   {{ end }}
+  {{ if .initialOffset }}
+  initialOffset: {{ .initialOffset }}
+  {{ end }}
   {{ if .consumers }}
   consumers: {{ .consumers }}
   {{ end }}