Repository: knative-sandbox/eventing-kafka
Pull request #885 (2021-09-22 11:51:00)

[0.24] Move initOffsets functionality to a common package
Cherry pick https://github.com/knative-sandbox/eventing-kafka/pull/883

- Return partition count in `initOffsets` function, but do not use it as it is not used in 0.24 codebase

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/885#" 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/release-0.24/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 @devguyio @lionelvillard 

Cherry pick https://github.com/knative-sandbox/eventing-kafka/pull/883
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/885?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> :exclamation: No coverage uploaded for pull request base (`release-0.24@778680a`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#section-missing-base-commit).
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/885/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/885?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@               Coverage Diff               @@
##             release-0.24     #885   +/-   ##
===============================================
  Coverage                ?   74.62%           
===============================================
  Files                   ?      136           
  Lines                   ?     6184           
  Branches                ?        0           
===============================================
  Hits                    ?     4615           
  Misses                  ?     1346           
  Partials                ?      223           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/885?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/885?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [778680a...aec60ed](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/885?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).

lionelvillard: /lgtm

Patch:
diff --git a/pkg/source/client/offsets.go b/pkg/common/kafka/offset/offsets.go
similarity index 78%
rename from pkg/source/client/offsets.go
rename to pkg/common/kafka/offset/offsets.go
index dac1574a0..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,29 +28,24 @@ 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) 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 err
+		return -1, err
 	}
 
-	kafkaAdminClient, err := sarama.NewClusterAdminFromClient(kafkaClient)
-	if err != nil {
-		return 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 {
 		partitions, err := kafkaClient.Partitions(topic)
-
+		totalPartitions += len(partitions)
 		if err != nil {
-			return fmt.Errorf("failed to get partitions for topic %s: %w", topic, err)
+			return -1, fmt.Errorf("failed to get partitions for topic %s: %w", topic, err)
 		}
 
 		topicPartitions[topic] = partitions
@@ -59,13 +54,13 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 	// Fetch topic offsets
 	topicOffsets, err := knsarama.GetOffsets(kafkaClient, topicPartitions, sarama.OffsetNewest)
 	if err != nil {
-		return fmt.Errorf("failed to get the topic offsets: %w", err)
+		return -1, fmt.Errorf("failed to get the topic offsets: %w", err)
 	}
 
 	// Look for uninitialized offset (-1)
 	offsets, err := kafkaAdminClient.ListConsumerGroupOffsets(consumerGroup, topicPartitions)
 	if err != nil {
-		return err
+		return -1, err
 	}
 
 	dirty := false
@@ -88,7 +83,7 @@ func InitOffsets(ctx context.Context, kafkaClient sarama.Client, topics []string
 
 				pm, err := offsetManager.ManagePartition(topic, partitionID)
 				if err != nil {
-					return fmt.Errorf("failed to create the partition manager for topic %s and partition %d: %w", topic, partitionID, err)
+					return -1, fmt.Errorf("failed to create the partition manager for topic %s and partition %d: %w", topic, partitionID, err)
 				}
 
 				pm.MarkOffset(offset, "")
@@ -102,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
-	return nil
+	// 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 f80ff386c..720e3c3e2 100644
--- a/pkg/source/reconciler/mtsource/kafkasource.go
+++ b/pkg/source/reconciler/mtsource/kafkasource.go
@@ -26,6 +26,7 @@ import (
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/labels"
 	"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"
 	"knative.dev/pkg/logging"
@@ -120,7 +121,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)
diff --git a/pkg/source/reconciler/source/kafkasource.go b/pkg/source/reconciler/source/kafkasource.go
index b31f2eb67..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 91%
rename from pkg/source/client/offsets_test.go
rename to pkg/common/kafka/offset/offsets_test.go
index 269d8f9dc..e5effa504 100644
--- a/pkg/source/client/offsets_test.go
+++ b/pkg/common/kafka/offset/offsets_test.go
@@ -14,12 +14,13 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package client
+package offset
 
 import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
 
 	logtesting "knative.dev/pkg/logging/testing"
 )
@@ -137,9 +138,20 @@ func TestInitOffsets(t *testing.T) {
 				t.Errorf("unexpected error: %v", err)
 			}
 			defer sc.Close()
-			ctx := logtesting.TestContextWithLogger(t)
-			err = InitOffsets(ctx, sc, tc.topics, group)
 
+			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, kac, tc.topics, group)
+			total := 0
+			for _, partitions := range tc.topicOffsets {
+				total += len(partitions)
+			}
+			assert.Equal(t, int(partitionCt), total)
 			if err != nil {
 				t.Errorf("unexpected error: %v", err)
 			}