[0.24] KafkaChannel to init offsets before dispatcher #886

aliok · 2021-09-22T12:25:12Z

Fixes #549

Proposed Changes

Init offsets, similar to KafkaSource, to make sure dispatcher doesn't fetch events from latest when it starts (more details on the issue)
Consumer groups are now started after making sure the initial offsets are committed by the controller(s)
NOTE: Check for the initial offsets are done in the ConsumerFactory. This means, KafkaSource adapters and distributed channel can get this functionality for free. However, I didn't enable it for them.
Get rid of status probing mechanism for subscription readiness in Kafka Channel
Needed to touch some distributed channel code as I changed some interfaces. But, no business logic changes

Release Note

- 🐛  Fixed a bug that was causing Kafka Channel dispatcher to miss some events in case the offset for the consumer group was not initialized

Docs

knative-prow-robot · 2021-09-22T12:25:16Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: aliok

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~~ [aliok]

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

aliok · 2021-09-22T12:36:12Z

/retest

codecov · 2021-09-22T13:31:21Z

Codecov Report

❗ No coverage uploaded for pull request base (release-0.24@bd92902). Click here to learn what that means.
The diff coverage is n/a.

❗ Current head 21ee494 differs from pull request most recent head 62cf6d0. Consider uploading reports for the commit 62cf6d0 to get more accurate results

@@               Coverage Diff               @@
##             release-0.24     #886   +/-   ##
===============================================
  Coverage                ?   74.06%           
===============================================
  Files                   ?      134           
  Lines                   ?     5954           
  Branches                ?        0           
===============================================
  Hits                    ?     4410           
  Misses                  ?     1319           
  Partials                ?      225

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 bd92902...62cf6d0. Read the comment docs.

aliok · 2021-09-23T13:58:09Z

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

aliok · 2021-09-27T11:22:43Z

/retest

Let's see what happens

aliok · 2021-09-28T06:56:19Z

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

devguyio · 2021-09-28T09:34:47Z

/assign

devguyio

I did a short review, will continue later today.

devguyio · 2021-09-28T10:00:12Z

pkg/channel/consolidated/reconciler/controller/kafkachannel.go

@@ -477,7 +492,20 @@ func (r *Reconciler) reconcileChannelService(ctx context.Context, dispatcherName
 	return svc, nil
 }

-func (r *Reconciler) createClient(ctx context.Context) (sarama.ClusterAdmin, error) {
+func (r *Reconciler) createClients(ctx context.Context) (sarama.Client, sarama.ClusterAdmin, error) {
+	kafkaClient := r.kafkaClient


you're not setting
r.kafkaClient anywhere, is that intentional? wouldn't that mean you're creating a new client with every reconcile loop?

I refactored the code that creates the admin client and created the normal client there, the same way.

For clusterAdmin, there's this ticket: IBM/sarama#1162. I see that ticket didn't really get any resolution.

I haven't tried reusing the regular kafkaClient. I can try that, once I made the failing tests working.

pretty bad sarama bug 😞

Can we add a comment that this thing returns both clients? Normal and ClusterAdmin?

func (r *Reconciler) createSaramaClients instead? not just client?

Do we really always need both? See Ahmed's comment with stashing and defering the "regular" client.

Just wondering

one alternative can be that we have two functions, one for each.

We now have 2 functions. Code looks much better, thanks for the suggestions.

About stashing into r.kafkaClient -> I will try this, one I see tests are passing again.

Ok, when I stash the client to r.kafkaClient and reuse it, things break. I have no desire to debug why.

See my comment here: #886 (comment)

You can have a look at the commit list here, https://github.com/knative-sandbox/eventing-kafka/pull/886/commits, and see the job history for each commit there (not always consistent, especially when there's new code pushed while the jobs were running).

. I have no desire to debug why.

that is OK w/ me. That said - mind creating an issue we are having to track this?

pkg/channel/consolidated/reconciler/controller/kafkachannel.go

aliok · 2021-09-28T11:57:23Z

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

Retesting to see if there's a pattern

pkg/channel/consolidated/reconciler/controller/controller.go

pkg/common/consumer/consumer_factory.go

…be used later

aliok · 2021-10-01T10:01:15Z

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

Stashing the kafka client broke things. I am gonna try here again, before reverting commit 4d57547.

This reverts commit 4d57547

aliok · 2021-10-01T10:23:35Z

Stashing the kafka client broke things. I am gonna try here again, before reverting commit 4d57547.

I reverted it now. See my comment here: #886 (comment)

…ed ready

knative-metrics-robot · 2021-10-01T11:00:07Z

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/channel/consolidated/dispatcher/dispatcher.go	64.5%	66.1%	1.6
pkg/channel/consolidated/reconciler/controller/kafkachannel.go	54.8%	54.2%	-0.6
pkg/channel/distributed/dispatcher/controller/kafkachannel.go	87.5%	71.8%	-15.7
pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go	97.7%	97.5%	-0.1
pkg/common/consumer/consumer_factory.go	87.2%	84.8%	-2.4
pkg/common/consumer/consumer_manager.go	96.9%	98.7%	1.8
pkg/common/consumer/consumergroup_offsets_checker.go	Do not exist	4.0%
pkg/common/kafka/offset/offsets.go	81.1%	81.5%	0.4
pkg/source/adapter/adapter.go	52.1%	55.4%	3.3

knative-prow-robot · 2021-10-01T11:00:45Z

@aliok: 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-go-coverage	`62cf6d0`	link	false	`/test pull-knative-sandbox-eventing-kafka-go-coverage`

Full PR test history. Your PR dashboard.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. I understand the commands that are listed here.

aliok · 2021-10-01T11:17:02Z

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

TestEnableSaramaLogging

devguyio · 2021-10-01T12:16:13Z

/lgtm

[0.24] KafkaChannel to init offsets before dispatcher (knative-extensions#886)

* Cherry pick 3f2a9d7 [0.24] KafkaChannel to init offsets before dispatcher (#886) * ./hack/update-codegen.sh

…ions#886) * Init offsets in Kafka channel controller - first iteration * Do not check prober for subscription readiness * Better code style * Get rid of probing for checking subscription readiness * Get rid of unused deps * Ooops, fixed the tests * Pass context along the functions, which will be necessary later * Fix unit test * Move partition retrieval into a separate function, which is going to be used later * Check if offsets are initialized in the consumerFactory * IDE "extracts" method LOL * Make unit tests working * Move MockClusterAdmin into the common package and reuse it * Copy paste tests for CheckIfAllOffsetsInitialized * Unify tests for CheckIfAllOffsetsInitialized and InitOffsets * Separate tests for CheckIfAllOffsetsInitialized and InitOffsets * Do not block main reconciliation thread for offset checking * Remove last crumbs of probing * Change log level for offset init message * Move some consts to right place * Rename checkOffsets.. func to WaitForOffsets... * Rename consumerOffsetInitializer to ConsumerGroupOffsetsChecker * Do not handle deleted topics or partitions when checking the offsets * Copy the partitions array when retrieving partitions * Address comments * Separate client and clusteradminclient creation * Stash kafka client * Revert "Stash kafka client" This reverts commit 4d57547 * Do not do any offset initialization when subscription is already marked ready

…ions#886) (#383) * [0.24] KafkaChannel to init offsets before dispatcher (knative-extensions#886) * Init offsets in Kafka channel controller - first iteration * Do not check prober for subscription readiness * Better code style * Get rid of probing for checking subscription readiness * Get rid of unused deps * Ooops, fixed the tests * Pass context along the functions, which will be necessary later * Fix unit test * Move partition retrieval into a separate function, which is going to be used later * Check if offsets are initialized in the consumerFactory * IDE "extracts" method LOL * Make unit tests working * Move MockClusterAdmin into the common package and reuse it * Copy paste tests for CheckIfAllOffsetsInitialized * Unify tests for CheckIfAllOffsetsInitialized and InitOffsets * Separate tests for CheckIfAllOffsetsInitialized and InitOffsets * Do not block main reconciliation thread for offset checking * Remove last crumbs of probing * Change log level for offset init message * Move some consts to right place * Rename checkOffsets.. func to WaitForOffsets... * Rename consumerOffsetInitializer to ConsumerGroupOffsetsChecker * Do not handle deleted topics or partitions when checking the offsets * Copy the partitions array when retrieving partitions * Address comments * Separate client and clusteradminclient creation * Stash kafka client * Revert "Stash kafka client" This reverts commit 4d57547 * Do not do any offset initialization when subscription is already marked ready * [0.24] KafkaChannel dispatcher offset checking improvements (knative-extensions#924) * Change poll loop Signed-off-by: Pierangelo Di Pilato <[email protected]> * Change poll loop and requeue channel on failure * Get rid of unused func * Fix unit tests Co-authored-by: Pierangelo Di Pilato <[email protected]> Co-authored-by: Pierangelo Di Pilato <[email protected]>

…ions#913) * Cherry pick 3f2a9d7 [0.24] KafkaChannel to init offsets before dispatcher (knative-extensions#886) * ./hack/update-codegen.sh

* [0.25] KafkaChannel to init offsets before dispatcher (knative-extensions#913) * Cherry pick 3f2a9d7 [0.24] KafkaChannel to init offsets before dispatcher (knative-extensions#886) * ./hack/update-codegen.sh * [0.25] KafkaChannel dispatcher offset checking improvements (knative-extensions#929) * Fix Kafka channel event loss during subscription becoming ready * Make it look like knative-extensions#926

knative-prow-robot added the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Sep 22, 2021

knative-prow-robot added approved Indicates a PR has been approved by an approver from all required OWNERS files. size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. labels Sep 22, 2021

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

aliok force-pushed the kafka-ch-event-loss-024 branch from 05189a6 to 1058793 Compare September 22, 2021 12:30

knative-prow-robot added the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Sep 23, 2021

aliok force-pushed the kafka-ch-event-loss-024 branch from ab79292 to 0903a03 Compare September 23, 2021 14:07

knative-prow-robot removed the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Sep 23, 2021

knative-prow-robot assigned devguyio Sep 28, 2021

devguyio reviewed Sep 28, 2021

View reviewed changes

matzew reviewed Sep 28, 2021

View reviewed changes

pkg/channel/consolidated/reconciler/controller/controller.go Outdated Show resolved Hide resolved

matzew reviewed Sep 28, 2021

View reviewed changes

pkg/common/consumer/consumer_factory.go Outdated Show resolved Hide resolved

aliok added 10 commits September 28, 2021 18:20

Init offsets in Kafka channel controller - first iteration

8a994b4

Do not check prober for subscription readiness

6de8a7d

Better code style

a63d7a1

Get rid of probing for checking subscription readiness

ab11d6d

Get rid of unused deps

94f8608

Ooops, fixed the tests

1a61524

Pass context along the functions, which will be necessary later

4bbfe95

Fix unit test

f0249db

Move partition retrieval into a separate function, which is going to …

d56515c

…be used later

Check if offsets are initialized in the consumerFactory

4d1a6b1

Revert "Stash kafka client"

21ee494

This reverts commit 4d57547

Do not do any offset initialization when subscription is already mark…

62cf6d0

…ed ready

This was referenced Oct 1, 2021

[0.26] KafkaChannel to init offsets before dispatcher #910

Closed

[main] KafkaChannel to init offsets before dispatcher #911

Merged

aliok requested review from devguyio and matzew October 1, 2021 11:39

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

knative-prow-robot merged commit 3f2a9d7 into knative-extensions:release-0.24 Oct 1, 2021

aliok mentioned this pull request Oct 1, 2021

KafkaChannel reconciler creates Kafka client too often #912

Open

aliok deleted the kafka-ch-event-loss-024 branch October 1, 2021 12:23

aliok added a commit to aliok/eventing-kafka that referenced this pull request Oct 1, 2021

Cherry pick 3f2a9d7

7caf7cf

[0.24] KafkaChannel to init offsets before dispatcher (knative-extensions#886)

aliok mentioned this pull request Oct 1, 2021

[0.25] KafkaChannel to init offsets before dispatcher #913

Merged

knative-prow-robot pushed a commit that referenced this pull request Oct 1, 2021

[0.25] KafkaChannel to init offsets before dispatcher (#913)

813d807

* Cherry pick 3f2a9d7 [0.24] KafkaChannel to init offsets before dispatcher (#886) * ./hack/update-codegen.sh

aliok mentioned this pull request Oct 1, 2021

[0.26] KafkaChannel to init offsets before dispatcher #915

Closed

pierDipi mentioned this pull request Oct 4, 2021

Avoid waiting for offsets initialization in dispatcher openshift-knative/eventing-kafka#382

Closed

This was referenced Oct 4, 2021

[0.24] KafkaChannel to init offsets before dispatcher (#886) openshift-knative/eventing-kafka#383

Merged

[0.24] KafkaChannel dispatcher offset checking improvements #924

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[0.24] KafkaChannel to init offsets before dispatcher #886

[0.24] KafkaChannel to init offsets before dispatcher #886

aliok commented Sep 22, 2021 •

edited

Loading

knative-prow-robot commented Sep 22, 2021

aliok commented Sep 22, 2021

codecov bot commented Sep 22, 2021 •

edited

Loading

aliok commented Sep 23, 2021

aliok commented Sep 27, 2021

aliok commented Sep 28, 2021

devguyio commented Sep 28, 2021

devguyio left a comment

devguyio Sep 28, 2021

aliok Sep 28, 2021 •

edited

Loading

matzew Sep 28, 2021

matzew Oct 1, 2021

matzew Oct 1, 2021

devguyio Oct 1, 2021

aliok Oct 1, 2021

aliok Oct 1, 2021

matzew Oct 1, 2021

aliok Oct 1, 2021

aliok commented Sep 28, 2021

aliok commented Oct 1, 2021

aliok commented Oct 1, 2021

knative-metrics-robot commented Oct 1, 2021

knative-prow-robot commented Oct 1, 2021 •

edited

Loading

aliok commented Oct 1, 2021

devguyio commented Oct 1, 2021

[0.24] KafkaChannel to init offsets before dispatcher #886

[0.24] KafkaChannel to init offsets before dispatcher #886

Conversation

aliok commented Sep 22, 2021 • edited Loading

Proposed Changes

knative-prow-robot commented Sep 22, 2021

aliok commented Sep 22, 2021

codecov bot commented Sep 22, 2021 • edited Loading

Codecov Report

aliok commented Sep 23, 2021

aliok commented Sep 27, 2021

aliok commented Sep 28, 2021

devguyio commented Sep 28, 2021

devguyio left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aliok Sep 28, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aliok commented Sep 28, 2021

aliok commented Oct 1, 2021

aliok commented Oct 1, 2021

knative-metrics-robot commented Oct 1, 2021

knative-prow-robot commented Oct 1, 2021 • edited Loading

aliok commented Oct 1, 2021

devguyio commented Oct 1, 2021

aliok commented Sep 22, 2021 •

edited

Loading

codecov bot commented Sep 22, 2021 •

edited

Loading

aliok Sep 28, 2021 •

edited

Loading

knative-prow-robot commented Oct 1, 2021 •

edited

Loading