Repository: knative-sandbox/eventing-kafka-broker
Pull request #1399 (2021-11-02 18:38:44)

Init offsets when initialOffset is latest in KafkaSource
As per title.

Part of #312

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Init offsets when initialOffset is latest in KafkaSource

<!--
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
None
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->
```
None
```

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1399#" 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/main/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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/source/source.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1399/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1455605315780218880/artifacts/line-cov.html#file2) | 73.9% | 74.2% | 0.4

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1399](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a4e4b23) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/c73237e8abc398e21f8256cbe1759bdf8006c16d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c73237e) will **decrease** coverage by `0.20%`.
> The diff coverage is `44.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1399      +/-   ##
============================================
- Coverage     76.57%   76.37%   -0.21%     
  Complexity      584      584              
============================================
  Files            99       99              
  Lines          3607     3631      +24     
  Branches        165      165              
============================================
+ Hits           2762     2773      +11     
- Misses          635      646      +11     
- Partials        210      212       +2     
```

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

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/1399?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | `0.00% <0.00%> (ø)` | |
| [control-plane/pkg/reconciler/source/source.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2Uvc291cmNlLmdv) | `57.69% <56.25%> (+0.30%)` | :arrow_up: |
| [control-plane/pkg/reconciler/source/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvY29udHJvbGxlci5nbw==) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [c73237e...a4e4b23](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1399?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/source/source.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1399/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1455609769917157376/artifacts/line-cov.html#file2) | 73.9% | 74.2% | 0.4

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/source/source.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1399/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1455639791142965248/artifacts/line-cov.html#file2) | 73.9% | 74.2% | 0.4

knative-prow-robot: @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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | b27a0c22e8258237b5f3f8e5d0fdb3c67bc03149 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1399/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1455639816979877888) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1399). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<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 -->
aliok: /lgtm

I only have a comment about a file's name.
knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/source/source.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1399/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1455805023446372352/artifacts/line-cov.html#file2) | 73.9% | 74.2% | 0.4

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1399#pullrequestreview-796208787" title="Approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1399#" 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/main/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":[]} -->

Patch:
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index 7e8c3c78b..a0c796b3e 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -32,11 +32,12 @@ import (
 )
 
 const (
-	ConditionAddressable        apis.ConditionType = "Addressable"
-	ConditionDataPlaneAvailable apis.ConditionType = "DataPlaneAvailable"
-	ConditionTopicReady         apis.ConditionType = "TopicReady"
-	ConditionConfigMapUpdated   apis.ConditionType = "ConfigMapUpdated"
-	ConditionConfigParsed       apis.ConditionType = "ConfigParsed"
+	ConditionAddressable             apis.ConditionType = "Addressable"
+	ConditionDataPlaneAvailable      apis.ConditionType = "DataPlaneAvailable"
+	ConditionTopicReady              apis.ConditionType = "TopicReady"
+	ConditionConfigMapUpdated        apis.ConditionType = "ConfigMapUpdated"
+	ConditionConfigParsed            apis.ConditionType = "ConfigParsed"
+	ConditionInitialOffsetsCommitted apis.ConditionType = "InitialOffsetsCommitted"
 )
 
 var IngressConditionSet = apis.NewLivingConditionSet(
@@ -51,6 +52,7 @@ var EgressConditionSet = apis.NewLivingConditionSet(
 	ConditionDataPlaneAvailable,
 	ConditionTopicReady,
 	ConditionConfigMapUpdated,
+	ConditionInitialOffsetsCommitted,
 )
 
 const (
@@ -250,3 +252,16 @@ func (manager *StatusConditionManager) TopicsNotPresentOrInvalid(topics []string
 	)
 	return fmt.Errorf("topics %v not present or invalid: check topic configuration", topics)
 }
+
+func (manager *StatusConditionManager) InitialOffsetNotCommitted(err error) error {
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
+		ConditionInitialOffsetsCommitted,
+		"InitialOffsetsNotCommitted",
+		err.Error(),
+	)
+	return err
+}
+
+func (manager *StatusConditionManager) InitialOffsetsCommitted() {
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrue(ConditionInitialOffsetsCommitted)
+}
diff --git a/control-plane/pkg/reconciler/kafka/consumer_group.go b/control-plane/pkg/reconciler/kafka/consumer_group.go
new file mode 100644
index 000000000..8b6b43424
--- /dev/null
+++ b/control-plane/pkg/reconciler/kafka/consumer_group.go
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka
+
+import (
+	"context"
+
+	"github.com/Shopify/sarama"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
+)
+
+// InitOffsetsFunc initialize offsets for a provided set of topics and a provided consumer group id.
+type InitOffsetsFunc func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error)
+
+var (
+	_ InitOffsetsFunc = offset.InitOffsets
+)
diff --git a/control-plane/pkg/reconciler/source/controller.go b/control-plane/pkg/reconciler/source/controller.go
index 875e81831..d1539582d 100644
--- a/control-plane/pkg/reconciler/source/controller.go
+++ b/control-plane/pkg/reconciler/source/controller.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
@@ -51,7 +52,9 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			DispatcherLabel:             base.SourceDispatcherLabel,
 		},
 		Env:                        configs,
+		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
+		InitOffsetsFunc:            offset.InitOffsets,
 	}
 
 	impl := kafkasource.NewImpl(ctx, r)
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index c714db84f..bf5fb73e5 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -57,9 +57,16 @@ type Reconciler struct {
 
 	Resolver *resolver.URIResolver
 
+	// NewKafkaClient creates new sarama Client. It's convenient to add this as Reconciler field so that we can
+	// mock the function used during the reconciliation loop.
+	NewKafkaClient kafka.NewClientFunc
 	// 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.
 	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
+	// InitOffsetsFunc initialize offsets for a provided set of topics and a provided consumer group id.
+	// It's convenient to add this as Reconciler field so that we can mock the function used during the
+	// reconciliation loop.
+	InitOffsetsFunc kafka.InitOffsetsFunc
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
@@ -104,9 +111,15 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 		return fmt.Errorf("error getting cluster admin sarama config: %w", err)
 	}
 
+	kafkaClient, err := r.NewKafkaClient(ks.Spec.BootstrapServers, saramaConfig)
+	if err != nil {
+		return statusConditionManager.TopicsNotPresentOrInvalidErr(ks.Spec.Topics, fmt.Errorf("error getting sarama config: %w", err))
+	}
+	defer kafkaClient.Close()
+
 	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(ks.Spec.BootstrapServers, saramaConfig)
 	if err != nil {
-		return fmt.Errorf("cannot obtain Kafka cluster admin, %w", err)
+		return statusConditionManager.TopicsNotPresentOrInvalidErr(ks.Spec.Topics, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
 	}
 	defer kafkaClusterAdminClient.Close()
 
@@ -119,6 +132,20 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 	}
 	statusConditionManager.TopicReady(strings.Join(ks.Spec.Topics, ", "))
 
+	if ks.Spec.InitialOffset == sources.OffsetLatest {
+		logger.Debug("Initializing initial offset",
+			zap.String("initialOffset", string(ks.Spec.InitialOffset)),
+			zap.String("consumerGroup", ks.Spec.ConsumerGroup),
+			zap.Strings("topics", ks.Spec.Topics),
+		)
+		if _, err := r.InitOffsetsFunc(ctx, kafkaClient, kafkaClusterAdminClient, ks.Spec.Topics, ks.Spec.ConsumerGroup); err != nil {
+			return statusConditionManager.InitialOffsetNotCommitted(
+				fmt.Errorf("unable to initialize consumer group %s offsets: %w", ks.Spec.ConsumerGroup, err),
+			)
+		}
+	}
+	statusConditionManager.InitialOffsetsCommitted()
+
 	// Get contract config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {


Test patch:
diff --git a/control-plane/pkg/reconciler/kafka/testing/client_mock.go b/control-plane/pkg/reconciler/kafka/testing/client_mock.go
new file mode 100644
index 000000000..09a1d33be
--- /dev/null
+++ b/control-plane/pkg/reconciler/kafka/testing/client_mock.go
@@ -0,0 +1,111 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package testing
+
+import (
+	"github.com/Shopify/sarama"
+)
+
+type MockKafkaClient struct {
+	CloseError error
+	IsClosed   bool
+}
+
+var (
+	_ sarama.Client = &MockKafkaClient{}
+)
+
+func (m MockKafkaClient) Config() *sarama.Config {
+	return sarama.NewConfig()
+}
+
+func (m MockKafkaClient) Controller() (*sarama.Broker, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) RefreshController() (*sarama.Broker, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) Brokers() []*sarama.Broker {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) Broker(brokerID int32) (*sarama.Broker, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) Topics() ([]string, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) Partitions(topic string) ([]int32, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) WritablePartitions(topic string) ([]int32, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) Leader(topic string, partitionID int32) (*sarama.Broker, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) Replicas(topic string, partitionID int32) ([]int32, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) InSyncReplicas(topic string, partitionID int32) ([]int32, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) OfflineReplicas(topic string, partitionID int32) ([]int32, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) RefreshBrokers(addrs []string) error {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) RefreshMetadata(topics ...string) error {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) GetOffset(topic string, partitionID int32, time int64) (int64, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) Coordinator(consumerGroup string) (*sarama.Broker, error) {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) RefreshCoordinator(consumerGroup string) error {
+	panic("implement me")
+}
+
+func (m MockKafkaClient) InitProducerID() (*sarama.InitProducerIDResponse, error) {
+	panic("implement me")
+}
+
+func (m *MockKafkaClient) Close() error {
+	m.IsClosed = true
+	return m.CloseError
+}
+
+func (m MockKafkaClient) Closed() bool {
+	return m.IsClosed
+}
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 2aa8128a6..93c19d2cf 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -116,6 +116,7 @@ func TestReconcileKind(t *testing.T) {
 						SourceConfigMapUpdatedReady(&configs),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
+						InitialOffsetsCommitted,
 					),
 				},
 			},
@@ -182,6 +183,7 @@ func TestReconcileKind(t *testing.T) {
 						SourceConfigMapUpdatedReady(&configs),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
+						InitialOffsetsCommitted,
 					),
 				},
 			},
@@ -242,6 +244,7 @@ func TestReconcileKind(t *testing.T) {
 						SourceConfigMapUpdatedReady(&configs),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
+						InitialOffsetsCommitted,
 					),
 				},
 			},
@@ -302,6 +305,7 @@ func TestReconcileKind(t *testing.T) {
 						SourceConfigMapUpdatedReady(&configs),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
+						InitialOffsetsCommitted,
 					),
 				},
 			},
@@ -362,6 +366,7 @@ func TestReconcileKind(t *testing.T) {
 						SourceConfigMapUpdatedReady(&configs),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
+						InitialOffsetsCommitted,
 					),
 				},
 			},
@@ -422,6 +427,7 @@ func TestReconcileKind(t *testing.T) {
 						SourceConfigMapUpdatedReady(&configs),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
+						InitialOffsetsCommitted,
 					),
 				},
 			},
@@ -457,6 +463,12 @@ func useTable(t *testing.T, table TableTest, configs broker.Configs) {
 				DispatcherLabel:             base.SourceDispatcherLabel,
 			},
 			Env: &configs.Env,
+			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
+				return 1, nil
+			},
+			NewKafkaClient: func(addrs []string, config *sarama.Config) (sarama.Client, error) {
+				return &kafkatesting.MockKafkaClient{}, nil
+			},
 			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName:                      "",
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index 3991b30ae..41a3e0b33 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -131,6 +131,10 @@ func SourceTopicsReady(source *sources.KafkaSource) {
 	)
 }
 
+func InitialOffsetsCommitted(source *sources.KafkaSource) {
+	source.GetConditionSet().Manage(source.GetStatus()).MarkTrue(base.ConditionInitialOffsetsCommitted)
+}
+
 func SourceDataPlaneAvailable(source *sources.KafkaSource) {
 	source.GetConditionSet().Manage(source.GetStatus()).MarkTrue(base.ConditionDataPlaneAvailable)
 }