add initial offset to kafka source spec #779

itsmurugappan · 2021-07-26T17:11:49Z

Fixes #289

Proposed Changes

Adds initial offset to the kafkasource spec

Release Note

🎁  - Provide an option for the user to specify the initial offset for a consumer group in Kafka source, this field is honored only if there are no prior offsets committed for the consumer group.

/cc @lionelvillard

codecov · 2021-07-26T17:49:55Z

Codecov Report

Merging #779 (a86f128) into main (c1d2251) will increase coverage by 0.36%.
The diff coverage is 90.90%.

@@            Coverage Diff             @@
##             main     #779      +/-   ##
==========================================
+ Coverage   75.62%   75.98%   +0.36%     
==========================================
  Files         142      144       +2     
  Lines        6445     6518      +73     
==========================================
+ Hits         4874     4953      +79     
+ Misses       1350     1337      -13     
- Partials      221      228       +7

Impacted Files	Coverage Δ
pkg/apis/sources/v1beta1/kafka_types.go	`66.66% <ø> (ø)`
pkg/common/client/config.go	`88.61% <75.00%> (-0.95%)`	⬇️
pkg/apis/sources/v1beta1/kafka_defaults.go	`100.00% <100.00%> (ø)`
pkg/apis/sources/v1beta1/kafka_validation.go	`76.66% <100.00%> (+3.58%)`	⬆️
pkg/source/client/client.go	`57.64% <100.00%> (+41.98%)`	⬆️
...rce/reconciler/source/resources/receive_adapter.go	`100.00% <100.00%> (ø)`
test/rekt/resources/kafkachannel/kafkachannel.go	`26.66% <0.00%> (ø)`
test/rekt/resources/resetoffset/resetoffset.go	`42.30% <0.00%> (ø)`
...el/distributed/controller/kafkachannel/receiver.go	`89.66% <0.00%> (+0.03%)`	⬆️
.../distributed/controller/kafkachannel/dispatcher.go	`90.25% <0.00%> (+0.08%)`	⬆️
... and 2 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update c1d2251...a86f128. Read the comment docs.

matzew · 2021-07-27T09:15:05Z

pkg/source/reconciler/source/resources/receive_adapter_test.go

@@ -38,6 +39,7 @@ func TestMakeReceiveAdapter(t *testing.T) {
 		Spec: v1beta1.KafkaSourceSpec{
 			Topics:        []string{"topic1,topic2"},
 			ConsumerGroup: "group",
+			InitialOffset: ptr.Int64(-2),


what's -2 ?

earliest, i will change to use the constants from sarama library

matzew · 2021-07-27T09:15:28Z

Isn't the official kafka (jvm) client just doing earliest and latest ?

pierDipi · 2021-07-27T15:50:26Z

pkg/apis/sources/v1beta1/kafka_types.go

@@ -72,6 +72,10 @@ type KafkaSourceSpec struct {
 	// +optional
 	ConsumerGroup string `json:"consumerGroup,omitempty"`

+	// InitialOffset is the Initial Offset for the consumer group.
+	// +optional
+	InitialOffset *int64 `json:"initialOffset,omitempty"`


We should not use sarama specific things in the API.

can we use earliest or latest as string?

/hold

how to give a specific offset in that case ?

Do you plan to set a single offset for all partitions?

Or, let me ask another question, as of now, if I set 50 here, what should happen?

sorry, I had a wrong assumption that we could also give a specific offset.
But looks like it needs to be earliest or latest. I will change the API to make the initial offset a string that will accept "earliest" or "latest"

i have made the changes

Thanks!

/unhold

pierDipi · 2021-07-27T19:25:52Z

pkg/apis/sources/v1beta1/kafka_types.go

+	// +optional
+	InitialOffset string `json:"initialOffset,omitempty"`


Since this is optional should it be a pointer *string?

pierDipi · 2021-07-27T21:31:54Z

pkg/apis/sources/v1beta1/kafka_types.go

@@ -72,6 +72,10 @@ type KafkaSourceSpec struct {
 	// +optional
 	ConsumerGroup string `json:"consumerGroup,omitempty"`

+	// InitialOffset is the Initial Offset for the consumer group.
+	// +optional
+	InitialOffset *int64 `json:"initialOffset,omitempty"`


Thanks!

/unhold

pkg/source/reconciler/source/resources/receive_adapter.go

lionelvillard · 2021-07-29T17:08:31Z

thanks @itsmurugappan!

Can you also update the documentation (if not done already), or a least open an issue in knative/docs?

/approve

@pierDipi @matzew for LGTM...

knative-prow-robot · 2021-07-29T17:08:35Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: itsmurugappan, lionelvillard

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~OWNERS~~ [lionelvillard]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

itsmurugappan · 2021-07-29T18:27:09Z

docs pr -> knative/docs#4080

itsmurugappan · 2021-07-29T18:27:25Z

/test pull-knative-sandbox-eventing-kafka-integration-test-channel-consolidated

matzew · 2021-07-30T12:24:43Z

/test all

lionelvillard · 2021-07-30T12:59:50Z

firs time I see this flake in the scheduler. I'll take a look when I get a chance

/test pull-knative-sandbox-eventing-kafka-unit-tests

lionelvillard · 2021-08-04T15:26:53Z

@matzew are you ok with this?

pierDipi · 2021-08-04T16:14:31Z

There is only one comment left #779 (comment) that still applies but at a different line https://github.com/knative-sandbox/eventing-kafka/pull/779/files#diff-aa0fe2380b2926d6809e7204b18c7c78f0d0158748106f0dca91a667ab700d21R88

knative-metrics-robot · 2021-08-05T23:41:43Z

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/apis/sources/v1beta1/kafka_validation.go	90.0%	90.9%	0.9
pkg/common/client/config.go	91.7%	91.3%	-0.4
pkg/source/client/client.go	19.7%	59.0%	39.3

lionelvillard · 2021-08-12T13:05:54Z

/lgtm

add initial offset to spec

501cc01

google-cla bot added the cla: yes Indicates the PR's author has signed the CLA. label Jul 26, 2021

knative-prow-robot added the size/M Denotes a PR that changes 30-99 lines, ignoring generated files. label Jul 26, 2021

add initial offset to kafka source spec

8b91375

itsmurugappan added 2 commits July 26, 2021 11:05

add initial offset for kafka source

fb79264

increase coverage for client

dcd6c53

knative-prow-robot added size/L Denotes a PR that changes 100-499 lines, ignoring generated files. and removed size/M Denotes a PR that changes 30-99 lines, ignoring generated files. labels Jul 26, 2021

matzew reviewed Jul 27, 2021

View reviewed changes

pierDipi suggested changes Jul 27, 2021

View reviewed changes

knative-prow-robot added the do-not-merge/hold Indicates that a PR should not merge because someone has issued a /hold command. label Jul 27, 2021

knative-prow-robot assigned pierDipi Jul 27, 2021

itsmurugappan added 2 commits July 27, 2021 11:13

change offset to string

3b313fd

add offset type

eb6b512

pierDipi reviewed Jul 27, 2021

View reviewed changes

knative-prow-robot removed the do-not-merge/hold Indicates that a PR should not merge because someone has issued a /hold command. label Jul 27, 2021

lionelvillard reviewed Jul 29, 2021

View reviewed changes

pkg/source/reconciler/source/resources/receive_adapter.go Outdated Show resolved Hide resolved

knative-prow-robot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Jul 29, 2021

change env key

45d5afe

itsmurugappan mentioned this pull request Jul 29, 2021

add initial offset to kafka source knative/docs#4080

Merged

add defaulting

a86f128

knative-prow-robot assigned lionelvillard Aug 12, 2021

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Aug 12, 2021

knative-prow-robot merged commit fdf11c5 into knative-extensions:main Aug 12, 2021

		// +optional
		InitialOffset string `json:"initialOffset,omitempty"`

add initial offset to kafka source spec #779

add initial offset to kafka source spec #779

Uh oh!

Conversation

itsmurugappan commented Jul 26, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Proposed Changes

Uh oh!

codecov bot commented Jul 26, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

matzew commented Jul 27, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

itsmurugappan Jul 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

lionelvillard commented Jul 29, 2021

Uh oh!

knative-prow-robot commented Jul 29, 2021

Uh oh!

itsmurugappan commented Jul 29, 2021

Uh oh!

itsmurugappan commented Jul 29, 2021

Uh oh!

matzew commented Jul 30, 2021

Uh oh!

lionelvillard commented Jul 30, 2021

Uh oh!

lionelvillard commented Aug 4, 2021

Uh oh!

pierDipi commented Aug 4, 2021

Uh oh!

knative-metrics-robot commented Aug 5, 2021

Uh oh!

lionelvillard commented Aug 12, 2021

Uh oh!

Uh oh!

itsmurugappan commented Jul 26, 2021 •

edited

Loading

codecov bot commented Jul 26, 2021 •

edited

Loading

itsmurugappan Jul 27, 2021 •

edited

Loading