Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
Consumer group initial offset not matched
**Describe the bug**
There is a mismatch between the initial offset set in a KafkaSource manifest and the one set in the consumer group.

 **Expected behavior**
The two offset should match. 

 **To Reproduce**
Deploy a kafkaSource with initialOffset set to earliest like the below:

`apiVersion: sources.knative.dev/v1beta1
kind: KafkaSource
metadata:
  name: datafetch-property-records-source
spec:
  bootstrapServers:
    - kafka-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092
  topics:
    - uk-property-records
  consumers: 12
  initialOffset: earliest
  sink:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: platform-lambda-address-publisher
      namespace: data-fetch
    uri: /publish`

 This will render as this:

```
apiVersion: sources.knative.dev/v1beta1
kind: KafkaSource
metadata:
  annotations:
    kubectl.kubernetes.io/last-applied-configuration: |
      {"apiVersion":"sources.knative.dev/v1beta1","kind":"KafkaSource","metadata":{"annotations":{},"labels":{"app.kubernetes.io/instance":"property-xyz-knative-eventing"},"name":"datafetch-property-records-source","namespace":"knative-eventing"},"spec":{"bootstrapServers":["kafka-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092"[],"consumers":12,"initialOffset":"earliest","sink":{"ref":{"apiVersion":"serving.knative.dev/v1","kind":"Service","name":"platform-lambda-address-publisher","namespace":"data-fetch"},"uri":"/publish"},"topics":["uk-property-records"[]}}
  creationTimestamp: "2023-02-15T10:04:04Z"
  generation: 1
  labels:
    app.kubernetes.io/instance: property-xyz-knative-eventing
  name: datafetch-property-records-source
  namespace: knative-eventing
  resourceVersion: "140780692"
  uid: 870caa8e-61f7-4b0d-b806-75da91ff7486
spec:
  bootstrapServers:
  - kafka-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092
  consumerGroup: knative-kafka-source-9fe12ce7-d225-420a-8788-8b4a851e87e2
  consumers: 12
  initialOffset: earliest
  net:
    sasl:
      password: {}
      type: {}
      user: {}
    tls:
      caCert: {}
      cert: {}
      key: {}
  sink:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: platform-lambda-address-publisher
      namespace: data-fetch
    uri: /publish
  topics:
  - uk-property-records
status:
  conditions:
  - lastTransitionTime: "2023-02-15T10:04:05Z"
    status: "True"
    type: ConsumerGroup
  - lastTransitionTime: "2023-02-15T10:04:05Z"
    status: "True"
    type: Ready
  - lastTransitionTime: "2023-02-15T10:04:05Z"
    status: "True"
    type: SinkProvided
  consumers: 12
  observedGeneration: 1
  placements:
  - podName: kafka-source-dispatcher-3
    vreplicas: 6
  - podName: kafka-source-dispatcher-4
    vreplicas: 6
  selector: eventing.knative.dev/source=kafka-source-controller,eventing.knative.dev/sourceName=datafetch-property-records-source
  sinkUri: http://platform-lambda-address-publisher.data-fetch.svc.cluster.local/publish
```   

and will create a consumer group like the one below: 

```
Name:         9dc7c195-5512-437e-b114-8e5299cdf28e
Namespace:    knative-eventing
Labels:       kafka.eventing.knative.dev/metadata.kind=kafkasource
Annotations:  internal.kafka.eventing.knative.dev/creator: system:serviceaccount:knative-eventing:kafka-controller
              internal.kafka.eventing.knative.dev/lastModifier: system:serviceaccount:knative-eventing:kafka-controller
API Version:  internal.kafka.eventing.knative.dev/v1alpha1
Kind:         ConsumerGroup
Metadata:
  Creation Timestamp:  2023-02-15T09:05:21Z
  Finalizers:
    consumergroups.internal.kafka.eventing.knative.dev
  Generation:  1
  Managed Fields:
    API Version:  internal.kafka.eventing.knative.dev/v1alpha1
    Fields Type:  FieldsV1
    fieldsV1:
      f:status:
        .:
        f:conditions:
        f:observedGeneration:
        f:placements:
        f:replicas:
        f:subscriberUri:
    Manager:      kafka-controller
    Operation:    Update
    Subresource:  status
    Time:         2023-02-15T09:29:56Z
    API Version:  internal.kafka.eventing.knative.dev/v1alpha1
    Fields Type:  FieldsV1
    fieldsV1:
      f:metadata:
        f:finalizers:
          .:
          v:"consumergroups.internal.kafka.eventing.knative.dev":
        f:labels:
          .:
          f:kafka.eventing.knative.dev/metadata.kind:
        f:ownerReferences:
          .:
          k:{"uid":"9dc7c195-5512-437e-b114-8e5299cdf28e"}:
      f:spec:
        .:
        f:replicas:
        f:template:
          .:
          f:metadata:
            .:
            f:creationTimestamp:
            f:labels:
              .:
              f:kafka.eventing.knative.dev/metadata.uid:
          f:spec:
            .:
            f:auth:
              .:
              f:NetSpec:
                .:
                f:sasl:
                  .:
                  f:password:
                  f:type:
                  f:user:
                f:tls:
                  .:
                  f:caCert:
                  f:cert:
                  f:key:
              f:SecretSpec:
            f:configs:
              .:
              f:Configs:
                .:
                f:bootstrap.servers:
                f:group.id:
            f:delivery:
              .:
              f:backoffDelay:
              f:backoffPolicy:
              f:initialOffset:
              f:ordering:
              f:retry:
              f:timeout:
            f:podBind:
            f:reply:
              .:
              f:noReply:
                .:
                f:enabled:
            f:subscriber:
              .:
              f:ref:
                .:
                f:apiVersion:
                f:kind:
                f:name:
                f:namespace:
              f:uri:
            f:topics:
            f:vReplicas:
    Manager:    kafka-controller
    Operation:  Update
    Time:       2023-02-15T10:03:13Z
  Owner References:
    API Version:           sources.knative.dev/v1beta1
    Block Owner Deletion:  true
    Controller:            true
    Kind:                  KafkaSource
    Name:                  datafeed-feed-processor-source
    UID:                   9dc7c195-5512-437e-b114-8e5299cdf28e
  Resource Version:        140779233
  UID:                     8be4b8ba-3a41-44e1-814a-a2dc30772054
Spec:
  Replicas:  12
  Selector:
    kafka.eventing.knative.dev/metadata.uid:  9dc7c195-5512-437e-b114-8e5299cdf28e
  Template:
    Metadata:
      Creation Timestamp:  <nil>
      Labels:
        kafka.eventing.knative.dev/metadata.uid:  9dc7c195-5512-437e-b114-8e5299cdf28e
      Namespace:                                  knative-eventing
    Spec:
      Auth:
        Net Spec:
          Sasl:
            Password:
            Type:
            User:
          Tls:
            Ca Cert:
            Cert:
            Key:
        Secret Spec:  <nil>
      Configs:
        Configs:
          bootstrap.servers:  kafka-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092
          group.id:           knative-kafka-source-034552df-0867-4a27-9eeb-428e723f76cf
      Delivery:
        Backoff Delay:   PT0.3S
        Backoff Policy:  exponential
        Initial Offset:  latest
        Ordering:        ordered
        Retry:           10
        Timeout:         PT600S
      Pod Bind:          <nil>
      Reply:
        No Reply:
          Enabled:  true
      Subscriber:
        Ref:
          API Version:  serving.knative.dev/v1
          Kind:         Service
          Name:         platform-datafeed-feed-processor
          Namespace:    data-feed
        Uri:            /processProperty
      Topics:
        uk-properties
      V Replicas:  <nil>
Status:
  Conditions:
    Last Transition Time:  2023-02-15T09:29:55Z
    Reason:                Autoscaler is disabled
    Status:                True
    Type:                  Autoscaler
    Last Transition Time:  2023-02-15T09:29:56Z
    Status:                True
    Type:                  Consumers
    Last Transition Time:  2023-02-15T09:29:55Z
    Status:                True
    Type:                  ConsumersScheduled
    Last Transition Time:  2023-02-15T09:29:56Z
    Status:                True
    Type:                  Ready
  Observed Generation:     1
  Placements:
    Pod Name:      kafka-source-dispatcher-2
    Vreplicas:     12
  Replicas:        12
  Subscriber Uri:  http://platform-datafeed-feed-processor.data-feed.svc.cluster.local/processProperty
Events:
  Type    Reason           Age   From                      Message
  ----    ------           ----  ----                      -------
  Normal  FinalizerUpdate  51m   consumergroup-controller  Updated "9dc7c195-5512-437e-b114-8e5299cdf28e" finalizers
  Normal  FinalizerUpdate  98s   consumergroup-controller  Updated "9dc7c195-5512-437e-b114-8e5299cdf28e" finalizers
```
 
you can see the initial offset in the consumer group is Initial Offset:  latest

 **Knative release version**
1.9.0



Pull request:
Properly handle KafkaSource earliest offset configuration
Fixes #2961

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Properly handle KafkaSource initial offset configuration

<!--
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
Fixed a bug where KafkaSource initial offset configuration wasn't applied when using the earliest offset configuration.
```


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,


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 }}