Repository: knative-sandbox/eventing-kafka-broker
Pull request #2966 (2023-02-16 11:53:51)

Bump Quarkus BOM to 2.13.7.Final
## Proposed Changes

- bump to latest version from `2.13.x` release

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
Bump Quarkus BOM to 2.13.7.Final
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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/2966#" title="Author self-approved">matzew</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/main/OWNERS)~~ [matzew]

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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2966](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (e31dd58) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/79ee8403fac4d9e9b338dad4cc6c21395ef80ee5?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (79ee840) will **decrease** coverage by `5.49%`.
> The diff coverage is `n/a`.

```diff
@@             Coverage Diff              @@
##               main    #2966      +/-   ##
============================================
- Coverage     64.05%   58.57%   -5.49%     
============================================
  Files           155       78      -77     
  Lines         10873     8254    -2619     
  Branches        231        0     -231     
============================================
- Hits           6965     4835    -2130     
+ Misses         3410     3054     -356     
+ Partials        498      365     -133     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

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/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...roker/dispatcher/main/ConsumerVerticleBuilder.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvbWFpbi9Db25zdW1lclZlcnRpY2xlQnVpbGRlci5qYXZh) | | |
| [...ive/eventing/kafka/broker/core/utils/Shutdown.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvU2h1dGRvd24uamF2YQ==) | | |
| [...spatcher/impl/consumer/CloudEventDeserializer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9DbG91ZEV2ZW50RGVzZXJpYWxpemVyLmphdmE=) | | |
| [...her/impl/filter/subscriptionsapi/PrefixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9QcmVmaXhGaWx0ZXIuamF2YQ==) | | |
| [...enting/kafka/broker/core/utils/Configurations.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29uZmlndXJhdGlvbnMuamF2YQ==) | | |
| [...ker/core/reconciler/IngressReconcilerListener.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9JbmdyZXNzUmVjb25jaWxlckxpc3RlbmVyLmphdmE=) | | |
| [...g/kafka/broker/receiver/impl/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1JlY2VpdmVyVmVydGljbGUuamF2YQ==) | | |
| [...tcher/impl/consumer/UnorderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9Vbm9yZGVyZWRDb25zdW1lclZlcnRpY2xlLmphdmE=) | | |
| [...ting/kafka/broker/core/utils/CollectionsUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29sbGVjdGlvbnNVdGlscy5qYXZh) | | |
| [.../kafka/broker/core/eventbus/ContractPublisher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RQdWJsaXNoZXIuamF2YQ==) | | |
| ... and [67 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2966?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)

knative-prow[bot]: @matzew: 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
--- | --- | --- | --- | ---
reconciler-tests-namespaced-broker_eventing-kafka-broker_main | e31dd5899bf10f2ff8a283879b40fc3e38afcd24 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2966/reconciler-tests-namespaced-broker_eventing-kafka-broker_main/1626188110754746368) | true | `/test reconciler-tests-namespaced-broker`

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

<details>

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

Patch:
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/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>


Test patch:
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/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/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
index 2b2739fa2..0bbabc426 100644
--- a/test/rekt/features/kafka_sink.go
+++ b/test/rekt/features/kafka_sink.go
@@ -18,17 +18,12 @@ package features
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/cloudevents/sdk-go/v2/test"
 	"github.com/google/uuid"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
-	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
-	testpkg "knative.dev/eventing-kafka-broker/test/pkg"
-	brokerconfigmap "knative.dev/eventing-kafka-broker/test/rekt/resources/configmap/broker"
-	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkasink"
-	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/trigger"
@@ -36,6 +31,15 @@ import (
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/resources/job"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	brokerreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	testpkg "knative.dev/eventing-kafka-broker/test/pkg"
+	brokerconfigmap "knative.dev/eventing-kafka-broker/test/rekt/resources/configmap/broker"
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkasink"
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
 )
 
 const (
@@ -138,3 +142,26 @@ func BrokerWithTriggersAndKafkaSink(env environment.Environment) *feature.Featur
 
 	return f
 }
+
+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 e89b9c049..a58a6b52d 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/pkg/resources/service"
 
+	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"
 
@@ -177,3 +182,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(service.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 c75439683..6a01b5d04 100644
--- a/test/rekt/resources/kafkasink/kafkasink.go
+++ b/test/rekt/resources/kafkasink/kafkasink.go
@@ -32,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"}
 }
 
@@ -83,13 +83,13 @@ func Install(name, topic string, bootstrapServers []string, opts ...manifest.Cfg
 
 // 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),
+		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 }}