Kafka Channel Dispatcher by neosab · Pull Request #589 · knative/eventing

neosab · 2018-11-07T22:44:13Z

Fixes #441

Proposed Changes

StatefulSet with the Kafka dispatcher
Kafka Channel controller creates a config map with subscribers (uses the same multichannelfanout config structure for uniformity)

TODO (in a subsequent PR)

Re-uses a lot of code from the in-memory channel provisioner. This needs to be moved to a new util package (Tracked by issue Move in-memory channel dispatcher config map related code to utils #607)

Release Note

Adds a new Kafka Channel Provisioner

googlebot · 2018-11-07T22:44:23Z

So there's good news and bad news.

👍 The good news is that everyone that needs to sign a CLA (the pull request submitter and all commit authors) have done so. Everything is all good there.

😕 The bad news is that it appears that one or more commits were authored or co-authored by someone other than the pull request submitter. We need to confirm that all authors are ok with their commits being contributed to this project. Please have them confirm that here in the pull request.

Note to project maintainer: This is a terminal state, meaning the cla/google commit status will not change from this state. It's up to you to confirm consent of all the commit author(s), set the cla label to yes (if enabled on your project), and then merge this pull request when appropriate.

neosab · 2018-11-07T22:46:47Z

cc @matzew

This is still missing unit tests which I am happy to add once the dispatcher design is validated by others.

neosab · 2018-11-08T02:29:57Z

/test pull-knative-eventing-build-tests

matzew · 2018-11-08T03:39:06Z

@neosab Is that based on latest of master ?

matzew · 2018-11-08T03:39:22Z

/assign matzew

will take a look at this PR

matzew · 2018-11-08T15:16:13Z

/lgtm
/hold

as said, I got this working end-to-end, also nice to see messages in the kafka topics, using kafkacat

NOTE: other mentioned PRs need to go in first

neosab · 2018-11-09T21:52:13Z

/assign @evankanderson

Tested that this works for me. I will work on moving the common code to helpers in a subsequent PR.

matzew · 2018-11-09T22:06:37Z

I will work on moving the common code to helpers in a subsequent PR.

👍 @neosab let's create an issue in Github for that

googlebot · 2018-11-12T19:41:19Z

CLAs look good, thanks!

matzew · 2018-11-12T19:45:50Z

/hold cancel

neosab · 2018-11-13T00:28:33Z

/cc @adamharwayne

knative-prow-robot · 2018-11-13T00:28:34Z

@neosab: GitHub didn't allow me to request PR reviews from the following users: adamharwayne.

Note that only knative members and repo collaborators can review this PR, and authors cannot review their own PRs.

Details

In response to this:

/cc @adamharwayne

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.

neosab · 2018-11-13T00:30:19Z

/cc @Harwayne

Didn't realize your github username changed :)

evankanderson

This looks fairly good, but I had a few comments on kubernetes interactions, including with the configmaps.

evankanderson · 2018-11-13T00:17:45Z

+  - apiGroups:
+      - "" # Core API group.
+    resources:
+      - configmaps


Should this be all configmaps, or just the one?

The dispatcher uses github.com/knative/pkg/configmap watcher that needs these cluster-scoped privileges.

evankanderson · 2018-11-13T00:32:08Z

+      volumes:
+        - name: kafka-channel-controller-config
+          configMap:
+            name: kafka-channel-controller-config


If you don't mark this optional and the ConfigMap isn't defined above, I think the pod will fail until the configmap is defined.

kafka-channel-controller-config is required as it contains the broker info and we check if it is defined while starting up https://github.com/knative/eventing/pull/589/files#diff-8634243f0697f44c13a150873e6cc457R72. Do you want to change this behavior?

I was wondering about crash-looping vs sitting and waiting for configuration, vs shipping a default configuration that the user might update later. I guess the current ClusterChannelProvisioner only supports a single set of bootstrap servers, so it may be that crash looping is the right choice for now. At some point in the future, it seems like we might want to support multiple Kafka clusters, at which point 0 clusters would be a valid configuration.

For now, this may be fine, but I have a sort of knee-jerk reaction to configs which fail unless the user has followed a precondition. Maybe a comment here or near the top of the file that creating this ConfigMap should be a precondition to applying this yaml?

The yaml actualy creates the kafka-channel-controller-config configmap in L78. It defaults to a broker URL that one would get if they created the kafka cluster using the sample in ./broker/kafka-broker.yaml. I will add a comment for the end-user to update this config map in cases where they use a different kafka cluster.

evankanderson · 2018-11-13T00:59:10Z

+
+	if value, ok := configMap[BrokerConfigMapKey]; ok {
+		bootstrapServers := strings.Split(value, ",")
+		if len(bootstrapServers) == 0 {


strings.Split will never return a zero-length array (it always returns at least one entry). I think you want to validate that all entries are non-empty (and add a test)?

A test would have caught it :). I added it now.

evankanderson · 2018-11-13T01:01:49Z

+
+	config := &KafkaProvisionerConfig{}
+
+	if value, ok := configMap[BrokerConfigMapKey]; ok {


Access to a non-existent value in a map will return a default value, so you could write this as simply:

config.Brokers := strings.Split(configMap[BrokerConfigMapKey], ",") for _, s := range(config.Brokers) { if len(s) == 0 { return nil, fmt.Errorf("Empty bootstrap_servers in configuration %s: %q", path, configMap[BrokerConfigMapKey]) } } return config, nil

Ah, I misread it. Thanks! I am fixing it.

evankanderson · 2018-11-13T01:05:02Z

+				if _, ok := d.kafkaConsumers[channelRef][sub]; ok {
+					// subscribe can be called multiple times for the same subscription,
+					//unsubscribe before we resubscribe
+					err := d.unsubscribe(channelRef, sub)


If we unsubscribe and resubscribe, will we potentially drop messages?

Again, this may be another TODO at the moment.

Actually, I am not sure why we need to unsubscribe and re-subscribe in case of existing subscription. I retained this from the original kafka bus implementation. cc @markfisher @scothis @matzew if they know.

Feel free to TODO and file an issue for this. At some point, we'll want to actually write some tests to stress this, I expect.

Done, filed #620

evankanderson · 2018-11-13T01:06:06Z

+	d.updateLock.Lock()
+	defer d.updateLock.Unlock()
+
+	if diff := d.ConfigDiff(config); diff != "" {


Presumably if we end up running more than one replica, we'll need to change this to filter the subscriptions by some sort of hash of the replica ID? (Fine to leave as a TODO.)

Similar to the comment on duplicate subscribers, if we run more replicas, the kafka consumer/subscription/replica will still be added to the same consumer group that will ensure exactly one of them consumes a partition. Wouldn't this eliminate the need to filter subscriptions for each replica?

Hmm, if you have multiple replicas, won't they each attempt to consume from the same consumer group? Will all but one block (I'm not familiar enough with Kafka and this library to be sure, it's possible that this is simply a competing consumer model)?

Obviously, with only 1 partition, the benefit of replicas will be limited, but I'm guessing that we'll support wider fanout soon.

The sarama-cluster library provides a balanced consumer group implementation on top of sarama. Even with multiple replicas, since consumer instances are part of the same consumer group, they coordinate so that at most one consumer will have access to a partition in a topic subscribed by the group.

Obviously, with only 1 partition, the benefit of replicas will be limited, but I'm guessing that we'll support wider fanout soon.

We already support NumPartitions as a channel argument.

neosab · 2018-11-14T05:55:02Z

@evankanderson I pushed an update and responded to your comments. Please take a look when it's possible.

evankanderson · 2018-11-15T01:19:49Z

/approve

I'm going to leave to @matzew or @scothis to LGTM this, since I'm not a Kafka expert, but I'm okay with this going in with a few TODOs for later.

knative-prow-robot · 2018-11-15T01:19:57Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: evankanderson, neosab

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

The pull request process is described here

Details

Needs approval from an approver in each of these files:

~~OWNERS~~ [evankanderson]

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

…nal_new

knative-metrics-robot · 2018-11-15T06:56:16Z

The following is the coverage report on pkg/.
Say /test pull-knative-eventing-go-coverage to re-run this coverage report

File	Old Coverage	New Coverage	Delta
pkg/provisioners/kafka/cmd/controller/main.go	Do not exist	0.0%
pkg/provisioners/kafka/cmd/dispatcher/main.go	Do not exist	0.0%
pkg/provisioners/kafka/controller/channel/reconcile.go	79.4%	76.1%	-3.3
pkg/provisioners/kafka/controller/util.go	Do not exist	100.0%
pkg/provisioners/kafka/dispatcher/dispatcher.go	Do not exist	69.4%

matzew · 2018-11-15T11:42:14Z

/lgtm

there are a few minor things, there are some issues filed for that...

I tested (on Minishift) the k8sevent source demo, and used this channel:

apiVersion: eventing.knative.dev/v1alpha1
kind: Channel
metadata:
  name: testchannel
spec:
  provisioner:
    apiVersion: eventing.knative.dev/v1alpha1
    kind: ClusterChannelProvisioner
    name: kafka-channel

REceiving messages on the dumper, like:

2018/11/15 11:41:47 Message Dumper received a message: POST / HTTP/1.1
Host: message-dumper.myproject.svc.cluster.local
Accept-Encoding: gzip
Ce-Cloudeventsversion: 0.1
Ce-Eventid: 4a60a086-e8b2-11e8-8a83-52540070a16c
Ce-Eventtime: 2018-11-15T08:41:48Z
Ce-Eventtype: dev.knative.k8s.event
Ce-Source: /apis/v1/namespaces/myproject/services/my-cluster-kafka-0
Content-Length: 725
Content-Type: application/json
User-Agent: Go-http-client/1.1
X-B3-Parentspanid: a23806bdb563af6a
X-B3-Sampled: 1
X-B3-Spanid: eb1c6949dcf5174c
X-B3-Traceid: a23806bdb563af6a
X-Forwarded-For: 127.0.0.1
X-Forwarded-Proto: http
X-Request-Id: 4cc00841-3752-91ef-9d8a-a86552d465a7

{"metadata":{"name":"my-cluster-kafka-0.15673f956e57f476","namespace":"myproject","selfLink":"/api/v1/namespaces/myproject/events/my-cluster-kafka-0.15673f956e57f476","uid":"4a60a086-e8b2-11e8-8a83-52540070a16c","resourceVersion":"55770","creationTimestamp":"2018-11-15T08:41:48Z"},"involvedObject":{"kind":"Service","namespace":"myproject","name":"my-cluster-kafka-0","uid":"17a1fb19-e8b2-11e8-8a83-52540070a16c","apiVersion":"v1","resourceVersion":"7942"},"reason":"ExternalIP","message":"Count: 1 -\u003e 0","source":{"component":"service-controller"},"firstTimestamp":"2018-11-15T08:41:48Z","lastTimestamp":"2018-11-15T11:41:47Z","count":92,"type":"Normal","eventTime":null,"reportingComponent":"","reportingInstance":""}

``

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 Nov 7, 2018

knative-prow-robot requested review from grantr and scothis November 7, 2018 22:44

knative-prow-robot added the size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. label Nov 7, 2018

googlebot added the cla: no Indicates the PR's author has not signed the CLA. label Nov 7, 2018

neosab changed the title ~~WIP: Kafka delivery~~ WIP: Kafka Channel Dispatcher Nov 7, 2018

matzew reviewed Nov 8, 2018

View reviewed changes

Comment thread pkg/provisioners/kafka/dispatcher/dispatcher.go

knative-prow-robot assigned matzew Nov 8, 2018

knative-prow-robot added lgtm Indicates that a PR is ready to be merged. do-not-merge/hold Indicates that a PR should not merge because someone has issued a /hold command. labels Nov 8, 2018

neosab mentioned this pull request Nov 8, 2018

Remove (seemingly??) unused pkg/buses #594

Closed

neosab force-pushed the kafka_delivery branch from eba5235 to b60006b Compare November 9, 2018 09:50

knative-prow-robot removed the lgtm Indicates that a PR is ready to be merged. label Nov 9, 2018

matzew reviewed Nov 9, 2018

View reviewed changes

Comment thread config/provisioners/kafka/kafka-provisioner.yaml Outdated

neosab changed the title ~~WIP: Kafka Channel Dispatcher~~ Kafka Channel Dispatcher Nov 9, 2018

knative-prow-robot removed the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Nov 9, 2018

knative-prow-robot assigned evankanderson Nov 9, 2018

Kafka dispatcher

8ee2bf1

neosab force-pushed the kafka_delivery branch from 3302cf1 to 8ee2bf1 Compare November 12, 2018 19:41

googlebot added cla: yes Indicates the PR's author has signed the CLA. and removed cla: no Indicates the PR's author has not signed the CLA. labels Nov 12, 2018

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

neosab mentioned this pull request Nov 12, 2018

Move in-memory channel dispatcher config map related code to utils #607

Closed

knative-prow-robot requested a review from Harwayne November 13, 2018 00:30

evankanderson reviewed Nov 13, 2018

View reviewed changes

Address PR Comments

14bb019

neosab force-pushed the kafka_delivery branch from 6ba0991 to 14bb019 Compare November 14, 2018 05:53

knative-prow-robot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Nov 15, 2018

neosab mentioned this pull request Nov 15, 2018

Kafka dispatcher unsubscribe and re-subscribe behavior #620

Closed

neosab added 2 commits November 14, 2018 22:51

Add some more TODO's

2fda1ba

Merge remote-tracking branch 'upstream/master' into kafka_delivery_fi…

e7e4fed

…nal_new

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Nov 15, 2018

knative-prow-robot merged commit ac7ecfe into knative:master Nov 15, 2018

neosab mentioned this pull request Nov 16, 2018

Move delivery utlitities out of pkg/busses #625

Closed

grantr added this to the v0.2.1 milestone Nov 28, 2018

matzew mentioned this pull request Oct 10, 2019

Adding matzew to approvers #2033

Merged


		config := &KafkaProvisionerConfig{}

		if value, ok := configMap[BrokerConfigMapKey]; ok {

Conversation

neosab commented Nov 7, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Proposed Changes

Uh oh!

googlebot commented Nov 7, 2018

Uh oh!

neosab commented Nov 7, 2018

Uh oh!

neosab commented Nov 8, 2018

Uh oh!

Uh oh!

matzew commented Nov 8, 2018

Uh oh!

matzew commented Nov 8, 2018

Uh oh!

matzew commented Nov 8, 2018

Uh oh!

Uh oh!

neosab commented Nov 9, 2018

Uh oh!

matzew commented Nov 9, 2018

Uh oh!

googlebot commented Nov 12, 2018

Uh oh!

matzew commented Nov 12, 2018

Uh oh!

neosab commented Nov 13, 2018

Uh oh!

knative-prow-robot commented Nov 13, 2018

Uh oh!

neosab commented Nov 13, 2018

Uh oh!

evankanderson left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

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!

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!

neosab commented Nov 14, 2018

Uh oh!

evankanderson commented Nov 15, 2018

Uh oh!

neosab commented Nov 7, 2018 •

edited

Loading