Repository: knative-sandbox/eventing-kafka
Pull request #883 (2021-09-22 08:56:43)

Move initOffsets functionality to a common package
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Move initOffsets functionality to a common package - so that we can also use it in Channels later
- Do not create the admin client within the function. Get that as parameter. In channel impl (which will use this func soon), we already have the adminClient and we would just pass that as a parameter.
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ 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

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/883#" title="Author self-approved">aliok</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).

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/blob/main/OWNERS)~~ [aliok]

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":[]} -->
aliok: cc @lionelvillard 
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/883?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#883](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/883?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0331712) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/d7dde1ec4628d262a7501f88fc035797d6a118d1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (d7dde1e) will **decrease** coverage by `0.03%`.
> The diff coverage is `0.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/883/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&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/pull/883?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     #883      +/-   ##
==========================================
- Coverage   74.64%   74.60%   -0.04%     
==========================================
  Files         154      154              
  Lines        7173     7174       +1     
==========================================
- Hits         5354     5352       -2     
- Misses       1529     1533       +4     
+ Partials      290      289       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/883?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/common/kafka/offset/offsets.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/883/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9rYWZrYS9vZmZzZXQvb2Zmc2V0cy5nbw==) | `58.82% <ø> (ø)` | |
| [pkg/source/reconciler/source/kafkasource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/883/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9yZWNvbmNpbGVyL3NvdXJjZS9rYWZrYXNvdXJjZS5nbw==) | `0.00% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/883?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/pull/883?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [d7dde1e...0331712](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/883?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-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/kafka/offset/offsets.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/883/pull-knative-sandbox-eventing-kafka-go-coverage/1440601148783333376/artifacts/line-cov.html#file0) | Do not exist | 81.1% | 

aliok: /retest

Infra
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/883#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/883#pullrequestreview-760753472" title="Approved">devguyio</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).

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/blob/main/OWNERS)~~ [aliok,devguyio]

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":[]} -->
devguyio: /cherrypick release-0.25
knative-prow-robot: @devguyio: once the present PR merges, I will cherry-pick it on top of release-0.25 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka/pull/883#issuecomment-924817195):

>/cherrypick release-0.25


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
devguyio: /cherrypick release-0.24
knative-prow-robot: @devguyio: once the present PR merges, I will cherry-pick it on top of release-0.24 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka/pull/883#issuecomment-924817317):

>/cherrypick release-0.24


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

Patch:
diff --git a/pkg/source/client/offsets.go b/pkg/common/kafka/offset/offsets.go
similarity index 87%
rename from pkg/source/client/offsets.go
rename to pkg/common/kafka/offset/offsets.go
index 93fc0972e..345ce2d9b 100644
--- a/pkg/source/client/offsets.go
+++ b/pkg/common/kafka/offset/offsets.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package client
+package offset
 
 import (
 	"context"
@@ -28,23 +28,17 @@ import (
 )
 
 // We want to make sure that ALL consumer group offsets are set before marking
-// the source as ready, to avoid "losing" events in case the consumer group session
+// the resource as ready, to avoid "losing" events in case the consumer group session
 // is closed before at least one message is consumed from ALL partitions.
 // Without InitOffsets, an event sent to a partition with an uninitialized offset
 // will not be forwarded when the session is closed (or a rebalancing is in progress).
-func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string, consumerGroup string) (int32, error) {
+func InitOffsets(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 	totalPartitions := 0
 	offsetManager, err := sarama.NewOffsetManagerFromClient(consumerGroup, kafkaClient)
 	if err != nil {
 		return -1, err
 	}
 
-	kafkaAdminClient, err := sarama.NewClusterAdminFromClient(kafkaClient)
-	if err != nil {
-		return -1, fmt.Errorf("failed to create a Kafka admin client: %w", err)
-	}
-	defer kafkaAdminClient.Close()
-
 	// Retrieve all partitions
 	topicPartitions := make(map[string][]int32)
 	for _, topic := range topics {
@@ -103,7 +97,7 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 		logging.FromContext(ctx).Infow("consumer group offsets committed", zap.String("consumergroup", consumerGroup))
 	}
 
-	// At this stage the KafkaSource instance is considered Ready
+	// At this stage the resource is considered Ready
 	return int32(totalPartitions), nil
 
 }
diff --git a/pkg/source/reconciler/mtsource/kafkasource.go b/pkg/source/reconciler/mtsource/kafkasource.go
index 7708dd42f..629013be6 100644
--- a/pkg/source/reconciler/mtsource/kafkasource.go
+++ b/pkg/source/reconciler/mtsource/kafkasource.go
@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/kubernetes"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 	"knative.dev/eventing/pkg/reconciler/source"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
@@ -128,7 +129,14 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1beta1.KafkaSource
 	defer c.Close()
 	src.Status.MarkConnectionEstablished()
 
-	totalPartitions, err := client.InitOffsets(ctx, c, src.Spec.Topics, src.Spec.ConsumerGroup)
+	kafkaAdminClient, err := sarama.NewClusterAdminFromClient(c)
+	if err != nil {
+		src.Status.MarkInitialOffsetNotCommitted("OffsetsNotCommitted", "Unable to initialize consumergroup offsets: %v", err)
+		return fmt.Errorf("failed to create a Kafka admin client: %w", err)
+	}
+	defer kafkaAdminClient.Close()
+
+	totalPartitions, err := offset.InitOffsets(ctx, c, kafkaAdminClient, src.Spec.Topics, src.Spec.ConsumerGroup)
 	if err != nil {
 		logging.FromContext(ctx).Errorw("unable to initialize consumergroup offsets", zap.Error(err))
 		src.Status.MarkInitialOffsetNotCommitted("OffsetsNotCommitted", "Unable to initialize consumergroup offsets: %v", err)
diff --git a/pkg/source/reconciler/source/kafkasource.go b/pkg/source/reconciler/source/kafkasource.go
index b7f0b25dc..21a42726b 100644
--- a/pkg/source/reconciler/source/kafkasource.go
+++ b/pkg/source/reconciler/source/kafkasource.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "knative.dev/control-protocol/pkg"
+	"knative.dev/eventing-kafka/pkg/common/kafka/offset"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/kmeta"
 
@@ -186,7 +187,14 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, src *v1beta1.KafkaSource
 	defer c.Close()
 	src.Status.MarkConnectionEstablished()
 
-	_, err = client.InitOffsets(ctx, c, src.Spec.Topics, src.Spec.ConsumerGroup)
+	kafkaAdminClient, err := sarama.NewClusterAdminFromClient(c)
+	if err != nil {
+		src.Status.MarkInitialOffsetNotCommitted("OffsetsNotCommitted", "Unable to initialize consumergroup offsets: %v", err)
+		return fmt.Errorf("failed to create a Kafka admin client: %w", err)
+	}
+	defer kafkaAdminClient.Close()
+
+	_, err = offset.InitOffsets(ctx, c, kafkaAdminClient, src.Spec.Topics, src.Spec.ConsumerGroup)
 	if err != nil {
 		logging.FromContext(ctx).Errorw("unable to initialize consumergroup offsets", zap.Error(err))
 		src.Status.MarkInitialOffsetNotCommitted("OffsetsNotCommitted", "Unable to initialize consumergroup offsets: %v", err)


Test patch:
diff --git a/pkg/source/client/offsets_test.go b/pkg/common/kafka/offset/offsets_test.go
similarity index 95%
rename from pkg/source/client/offsets_test.go
rename to pkg/common/kafka/offset/offsets_test.go
index 77d5f4ec0..e5effa504 100644
--- a/pkg/source/client/offsets_test.go
+++ b/pkg/common/kafka/offset/offsets_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package client
+package offset
 
 import (
 	"testing"
@@ -138,8 +138,15 @@ func TestInitOffsets(t *testing.T) {
 				t.Errorf("unexpected error: %v", err)
 			}
 			defer sc.Close()
+
+			kac, err := sarama.NewClusterAdminFromClient(sc)
+			if err != nil {
+				t.Errorf("unexpected error: %v", err)
+			}
+			defer kac.Close()
+
 			ctx := logtesting.TestContextWithLogger(t)
-			partitionCt, err := InitOffsets(ctx, sc, tc.topics, group)
+			partitionCt, err := InitOffsets(ctx, sc, kac, tc.topics, group)
 			total := 0
 			for _, partitions := range tc.topicOffsets {
 				total += len(partitions)