Kafka sink #1406

aslom · 2020-07-23T21:25:08Z

Fixes #

Proposed Changes

Add Kafka Sink, a simple receive adapter that can be run standalone (described in README)

Ref: #682

Immediately useful for users that need to get events from knative to outside Kafka broker

Current version is limited: no security support, no unit tests, no controller

Docs etc. will be addressed in future when limitations are removed and Kafka Sink was tested for some time.

aslom · 2020-07-23T21:27:39Z

I followed code example provided by @slinkydeveloper and I think that current version can be useful for our users so hope we can get it merged. I listed limitations that will take longer time to implement such as creating CR and controller for Kafka Sink.

@lionelvillard was mentioning Kafka Sink may be useful for async PoC implementation?

/assign @slinkydeveloper @lionelvillard

slinkydeveloper · 2020-07-24T06:50:27Z

kafka/sink/pkg/adapter/adapter.go

+}
+
+// NewAdapter returns the instance of gitHubReceiveAdapter that implements adapter.Adapter interface
+func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, ceClient cloudevents.Client) adapter.Adapter {


Why you create a source like adapter if then you don't use the provided client?

That is to structure it as receive adapter for future PR that add controller support and KafkaSink CR

slinkydeveloper · 2020-07-24T06:51:04Z

kafka/sink/cmd/receive_adapter/main.go

+)
+
+func main() {
+	adapter.Main("kafkasink", kafkasinkadapter.NewEnvConfig, kafkasinkadapter.NewAdapter)


Does it works on kube? adapter.Main requires a service account that can watch config maps and stuff, and from the yamls below it doesn't seem like you're using any

It worked in k8s v1.18 - see example output in README.md

slinkydeveloper · 2020-07-24T06:51:39Z

kafka/sink/README.md

+
+TODO: coming soon
+
+## Deploying Kafka Sink to Kubernetes


I wonder, can we show an example with a sinkbinding?

Do you mean example that shows how to use kafkasink instead of event-display? https://knative.dev/docs/eventing/samples/sinkbinding/

matzew · 2020-07-24T06:58:20Z

kafka/sink/samples/kafka-sink-deployment.yaml

@@ -0,0 +1,30 @@
+apiVersion: apps/v1
+kind: Deployment


would it make sense to have this thing being reflected by a CRD ?

Yes in future

matzew · 2020-07-24T07:08:44Z

@aslom how about adding this to the examples? I do not really see this being "library" code.

And it would be nice if we just go w/ a deployment, to take the binding into account here

slinkydeveloper · 2020-07-24T07:14:51Z

I was wondering, how much it takes to go from this code to a crd like:

type: KafkaSink
...
spec:
  topics:
    - aaa
    - bbbb

and then this KafkaSink implements the addressable duck type?

aslom · 2020-07-24T18:04:13Z

Thank you for your comments @matzew @slinkydeveloper

About Kafka Sink CR - yes that is the goal. However I would like to get there is smaller incremental steps getting to full functionality in small-medium size PRs instead of large XXL (that is hard for me to write and others to test).

So this is first step - MVP - Kafka sink functionality for receive adapter. Do you think this approach could work?

I started looking into Kafka Sink CR but it may require more discussion - for example could Kafka Source and Kafka Sink share the same controller?

knative-prow-robot · 2020-07-24T18:48:02Z

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by: aslom
To complete the pull request process, please assign lionelvillard
You can assign the PR to them by writing /assign @lionelvillard in a comment when ready.

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

Needs approval from an approver in each of these files:

kafka/OWNERS

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

knative-metrics-robot · 2020-07-24T18:53:16Z

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

File	Old Coverage	New Coverage	Delta
kafka/sink/pkg/adapter/adapter.go	Do not exist	0.0%

knative-prow-robot · 2020-07-24T18:53:46Z

@aslom: The following test failed, say /retest to rerun all failed tests:

Test name	Commit	Details	Rerun command
pull-knative-eventing-contrib-go-coverage	`afacb21`	link	`/test pull-knative-eventing-contrib-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.

aslom · 2020-07-27T20:53:21Z

@slinkydeveloper do you have any links (or ideas) how i could do unit test for kafka protocol (and to get code coverage pull-knative-eventing-contrib-go-coverage to pass)? Is there some way I can run unti test without requiring Kafka broker - intercept messages?

slinkydeveloper · 2020-07-28T08:04:37Z

@aslom I think you should mock the sync producer (here you create a sender https://github.com/knative/eventing-contrib/pull/1406/files#diff-985279ee1eab4df1be848de7a44a119aR94 but you can create the same sender directly from the built sync producer). So you start the adapter, send events to it and then assert that on the other side of the producer you received something

pierDipi · 2020-07-28T08:27:47Z

FYI, with some changes (that I will do anyway) to the receiver component from this project https://github.com/knative-sandbox/eventing-kafka-broker we get a Kafka Sink data plane. It already has tests (unit / integration / e2e).
We can use it in both a multi-tenant or single-tenant setup.

aslom · 2020-07-28T14:34:15Z

@pierDipi could you point me to kafka sink code and how you run unit tests (and do kafka sender mock)? I could not find it https://github.com/knative-sandbox/eventing-kafka-broker/search?p=2&q=kafka+sink&unscoped_q=kafka+sink

pierDipi · 2020-07-28T14:58:06Z

It's written in Java: https://github.com/knative-sandbox/eventing-kafka-broker/tree/master/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver
Basically, the idea is to mount a config map with all configurations where each of them binds an HTTP request path (Addressable) to a Kafka topic.
Is this the goal of a KafkaSink?

For testing, we use MockProducer for unit tests, we create an embedded Kafka cluster so that we can debug things in an IDE out of k8s for integration tests, and for e2e tests the same as in this repo.

pierDipi · 2020-07-28T15:02:30Z

All you need to do for implementing a KafkaSink is to write a control plane that updates ConfigMap(s) depending on single-tenant vs multi-tenant setup

pierDipi · 2020-07-28T15:10:41Z

Let me know if you find it interesting so that I can prioritize the changes I mentioned before.

aslom · 2020-07-28T21:32:05Z

@pierDipi the goal of Kafka Sink is to be very simple to use Sink that send events ot Kafka topic: #682

I would like to make it very easy to use from CLI - the same way that event display can be deployed as a sink to show events. That is why it is just one file that listen over HTTP for CloudEvents, send them to Kafka topic.

I would like to keep using golang to have it symmetrical to Kafka Source.

I think mapping config maps looks more complex than the CLI use case I have described in the github issue? However, I do not know the details how alternative kafka broker is implemented so hard to compare for me.

pierDipi · 2020-07-29T07:46:24Z

I would like to make it very easy to use from CLI - the same way that event display can be deployed as a sink to show events. That is why it is just one file that listen over HTTP for CloudEvents, send them to Kafka topic

I think mapping config maps looks more complex than the CLI use case I have described in the github issue?

I don't see many differences from a usability perspective between putting configurations in a ConfigMap vs environment variables.

I would like to keep using golang to have it symmetrical to Kafka Source.

So, in this case, re-using that component is not an option.

slinkydeveloper · 2020-07-29T09:11:08Z

I would like to keep using golang to have it symmetrical to Kafka Source.

Is there any reason for that?

aslom · 2020-07-29T12:53:52Z

@slinkydeveloper I could not figure how to get to HTTP server started by SDK to send events for unit testing - i need to know HTTP server URL to send cloud events to it.

The closes I could find in SDK was https://github.com/cloudevents/sdk-go/blob/master/v2/client/http_receiver_test.go#L15

aslom · 2020-07-30T19:37:42Z

@slinkydeveloper I am not able to access syncProducer in kafka_sarama.Sender as the field is private
in protocol/kafka_sarama/v2/sender.go

type Sender struct {
	topic        string
	syncProducer sarama.SyncProducer
}

so that code will not work even though I have syncProducerMock

	sender := &cloudeventskafka.Sender{topic: topic, syncProducer: syncProducerMock}

I created PR to make possible to run

	sender, err := cloudeventskafka.NewSenderFromSyncProducer(topic, syncProducerMock)

PR: cloudevents/sdk-go#564

Any other/better way to run unit test then?

matzew · 2020-08-31T09:13:18Z

@aslom I'd suggest we close this, in fav of: knative-extensions/eventing-kafka-broker#134

matzew · 2020-09-09T15:49:56Z

@aslom here is a quick sample on the code:
https://gist.github.com/matzew/e2c2fcd2696a346f25b8bc9e64bfd0fa

Let's close this?

aslom · 2020-09-09T16:25:08Z

@matzew if I understand correctly the Kafka Sink form knative-extensions/eventing-kafka-broker#134 requires new broker and can not run independently? Maybe there can be more than one implementation of Kafka Sink depending how users what to use it? In particular Kafka Sink that runs in user namespace as ksvc? @pierDipi

slinkydeveloper · 2020-09-09T16:27:03Z

requires new broker and can not run independently?

Not really, the data plane is abstract from the broker concept and this sink is "multi tenant"

matzew · 2020-09-10T08:13:24Z

Just read this text: https://gist.github.com/matzew/e2c2fcd2696a346f25b8bc9e64bfd0fa

I do not even have the broker deployed on my cluster 😄

/close

Closing this now, since we have a decent type, instead of more example work, and we can improve things over there

knative-prow-robot · 2020-09-10T08:13:31Z

@matzew: Closed this PR.

In response to this:

Just read this text: https://gist.github.com/matzew/e2c2fcd2696a346f25b8bc9e64bfd0fa

I do not even have the broker deployed on my cluster 😄

/close

Closing this now, since we have a decent type, instead of more example work, and we can improve things over there

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.

aslom added 3 commits July 22, 2020 18:23

initial Kafka Sink RA impl

2d13f32

documented and tested running local and in k8s

aafa478

add example output

7bf32ae

googlebot added the cla: yes Indicates the PR's author has signed the CLA. label Jul 23, 2020

knative-prow-robot added the size/L Denotes a PR that changes 100-499 lines, ignoring generated files. label Jul 23, 2020

knative-prow-robot requested review from lberk and matzew July 23, 2020 21:25

knative-prow-robot assigned lionelvillard and slinkydeveloper Jul 23, 2020

slinkydeveloper reviewed Jul 24, 2020

View reviewed changes

matzew reviewed Jul 24, 2020

View reviewed changes

update README, remove not needed comments

afacb21

aslom mentioned this pull request Jul 24, 2020

Add Kafka Sink #682

Closed

knative-prow-robot closed this Sep 10, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka sink #1406

Kafka sink #1406

aslom commented Jul 23, 2020

aslom commented Jul 23, 2020

slinkydeveloper Jul 24, 2020

aslom Jul 24, 2020

slinkydeveloper Jul 24, 2020

aslom Jul 24, 2020

slinkydeveloper Jul 24, 2020

aslom Jul 24, 2020

matzew Jul 24, 2020

aslom Jul 24, 2020

matzew commented Jul 24, 2020

slinkydeveloper commented Jul 24, 2020

aslom commented Jul 24, 2020

knative-prow-robot commented Jul 24, 2020

knative-metrics-robot commented Jul 24, 2020

knative-prow-robot commented Jul 24, 2020

aslom commented Jul 27, 2020

slinkydeveloper commented Jul 28, 2020

pierDipi commented Jul 28, 2020

aslom commented Jul 28, 2020

pierDipi commented Jul 28, 2020

pierDipi commented Jul 28, 2020

pierDipi commented Jul 28, 2020

aslom commented Jul 28, 2020

pierDipi commented Jul 29, 2020

slinkydeveloper commented Jul 29, 2020

aslom commented Jul 29, 2020

aslom commented Jul 30, 2020

matzew commented Aug 31, 2020

matzew commented Sep 9, 2020

aslom commented Sep 9, 2020

slinkydeveloper commented Sep 9, 2020

matzew commented Sep 10, 2020

knative-prow-robot commented Sep 10, 2020

Kafka sink #1406

Kafka sink #1406

Conversation

aslom commented Jul 23, 2020

Proposed Changes

aslom commented Jul 23, 2020

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

matzew commented Jul 24, 2020

slinkydeveloper commented Jul 24, 2020

aslom commented Jul 24, 2020

knative-prow-robot commented Jul 24, 2020

knative-metrics-robot commented Jul 24, 2020

knative-prow-robot commented Jul 24, 2020

aslom commented Jul 27, 2020

slinkydeveloper commented Jul 28, 2020

pierDipi commented Jul 28, 2020

aslom commented Jul 28, 2020

pierDipi commented Jul 28, 2020

pierDipi commented Jul 28, 2020

pierDipi commented Jul 28, 2020

aslom commented Jul 28, 2020

pierDipi commented Jul 29, 2020

slinkydeveloper commented Jul 29, 2020

aslom commented Jul 29, 2020

aslom commented Jul 30, 2020

matzew commented Aug 31, 2020

matzew commented Sep 9, 2020

aslom commented Sep 9, 2020

slinkydeveloper commented Sep 9, 2020

matzew commented Sep 10, 2020

knative-prow-robot commented Sep 10, 2020