{"id":759,"date":"2020-05-05T18:34:25","date_gmt":"2020-05-05T17:34:25","guid":{"rendered":"http:\/\/www.igfasouza.com\/blog\/?p=759"},"modified":"2021-05-20T14:20:33","modified_gmt":"2021-05-20T13:20:33","slug":"sink-kafka-connect","status":"publish","type":"post","link":"http:\/\/www.igfasouza.com\/blog\/sink-kafka-connect\/","title":{"rendered":"Sink Kafka connect"},"content":{"rendered":"

\"\"<\/p>\n

Well?<\/b><\/p>\n

This blog post is part of my series of posts regarding Kafka Connect.
\nIf you’re not familiar with Kafka, I suggest you have a look at some of my previous post;<\/p>\n

What is Kafka?<\/a>
\nKafka Connect Overview<\/a>
\nKafka Connector Architecture<\/a>
\nKafka Connect<\/a>
\nSource Kafka Connect<\/a><\/p>\n

This post is a collection of links, videos, tutorials, blogs and books that I found mixed with my opinion. <\/p>\n

TL;DR<\/b><\/p>\n

This post is about code, I\u2019ll show all steps to develop your Sink Kafka Connector.<\/p>\n

Table of contents<\/h2>\n

1. Use Case
\n2. Code
\n3. Links<\/p>\n

1. Use Case<\/h2>\n

One scenario that has become popular and I came across with some questions around in the last months is to use Kafka to trigger functions as service.<\/p>\n

In simple words: Send a message in your kafka topic and your function(s) gets invoked with the payload which you sent to Kafka, you can now have a function trigger in response to messages in Kafka Topics.<\/p>\n

My example is using OCI functions.<\/p>\n

2. Code<\/h2>\n

\"\"<\/p>\n

The connector needs extends SinkConnector. Is almost the same from source and you need to implement six methods here.<\/p>\n

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
<\/div><\/td>
@Override
\n    public<\/span> String<\/span><\/a> version(<\/span>)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n        return<\/span> null<\/span>;<\/span>
\n    }<\/span>
\n
\n    @Override
\n    public<\/span> void<\/span> start(<\/span>Map<<\/span>String<\/span><\/a>, String><\/span> props)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n       
\n    }<\/span>
\n
\n    @Override
\n    public<\/span> Class<?<\/span> extends<\/span> Task><\/span> taskClass(<\/span>)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n        return<\/span> null<\/span>;<\/span>
\n    }<\/span>
\n
\n    @Override
\n    public<\/span> List<<\/span>Map<<\/span>String<\/span><\/a>, String>><\/span> taskConfigs(<\/span>int<\/span> maxTasks)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n        return<\/span> null<\/span>;<\/span>
\n    }<\/span>
\n
\n    @Override
\n    public<\/span> void<\/span> stop(<\/span>)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n       
\n    }<\/span>
\n
\n    @Override
\n    public<\/span> ConfigDef config(<\/span>)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n        return<\/span> null<\/span>;<\/span>
\n    }<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

And the Task needs extends SinkTask Is almost the same from source but now you don’t have the poll method and instead, you have put.<\/p>\n

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
<\/div><\/td>
@Override
\n    public<\/span> String<\/span><\/a> version(<\/span>)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n        return<\/span> null<\/span>;<\/span>
\n    }<\/span>
\n
\n    @Override
\n    public<\/span> void<\/span> start(<\/span>Map<<\/span>String<\/span><\/a>, String><\/span> props)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n    }<\/span>
\n
\n    @Override
\n    public<\/span> void<\/span> put(<\/span>Collection<<\/span>SinkRecord><\/span> records)<\/span> {<\/span>
\n        \/\/ here is where the magic happenings.<\/span>
\n        \/\/ just add a boolean triggerFn method that trigger the function.<\/span>
\n    }<\/span>
\n
\n    @Override
\n    public<\/span> void<\/span> stop(<\/span>)<\/span> {<\/span>
\n        \/\/ TODO Auto-generated method stub<\/span>
\n       
\n    }<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

Run Docker<\/b><\/p>\n

1
2
3
4
5
6
7
<\/div><\/td>
docker run -<\/span>it --<\/span>rm --<\/span>name fn-<\/span>oci-<\/span>connect-<\/span>demo -<\/span>p 8083<\/span>:<\/span>8083<\/span> -<\/span>e GROUP_ID=<\/span>1<\/span> \\
\n    -<\/span>e BOOTSTRAP_SERVERS=<\/span>"bootstrap_URL"<\/span> \\
\n    -<\/span>e CONFIG_STORAGE_TOPIC=<\/span>\u201dID-<\/span>config\u201d \\
\n    -<\/span>e OFFSET_STORAGE_TOPIC=<\/span>\u201dID-<\/span>offset\u201d \\
\n    -<\/span>e STATUS_STORAGE_TOPIC=<\/span>\u201dID-<\/span>status\u201d \\
\n    -<\/span>v fn-<\/span>sink-<\/span>kafka-<\/span>connector\/<\/span>target\/<\/span>fn-<\/span>sink-<\/span>kafka-<\/span>connector-<\/span>0.0.1-<\/span>SNAPSHOT-<\/span>jar-<\/span>with-<\/span>dependencies.jar<\/span>\/:\/<\/span>kafka\/<\/span>connect\/<\/span>fn-<\/span>connector \\
\n    debezium\/<\/span>connect:<\/span>latest<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

Configure<\/b><\/p>\n

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
<\/div><\/td>
curl -<\/span>X POST \\
\n    http:<\/span>\/\/localhost:8082\/connectors \\<\/span>
\n    -<\/span>H 'content-type: application\/json'<\/span> \\
\n    -<\/span>d '{
\n    "name": "FnSinkConnector",
\n    "config": {
\n      "connector.class": "com.fn.sink.kafka.connect.FnSinkConnector",
\n      "tasks.max": "1",
\n      "topics": "test-sink-topic",
\n      "tenant_ocid": "<tenant_ocid>",
\n      "user_ocid": "<user_ocid>",
\n      "public_fingerprint": "<public_fingerprint>",
\n      "private_key_location": "\/path\/to\/kafka-connect\/secrets\/<private_key_name>",
\n      "function_url": "<FUNCTION_URL>"
\n    }
\n  }'<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

I created a package “http” where I add the OCI FN part, the code is based on this one here<\/a>. <\/p>\n

GitHub<\/b>
\nYou can check the full code in my GitHub<\/a>.<\/p>\n

3. Links<\/h2>\n

https:\/\/docs.confluent.io\/current\/connect\/devguide.html<\/a><\/p>\n

https:\/\/github.com\/oracle\/oci-java-sdk\/blob\/master\/bmc-examples\/src\/main\/java\/InvokeFunctionExample.java<\/a><\/p>\n

https:\/\/docs.cloud.oracle.com\/en-us\/iaas\/Content\/Functions\/Concepts\/functionsoverview.htm<\/a><\/p>\n

Happy Coding.<\/b><\/p>\n","protected":false},"excerpt":{"rendered":"

Well? This blog post is part of my series of posts regarding Kafka Connect. If you’re not familiar with Kafka, I suggest you have a look at some of my previous post; What is Kafka? Kafka Connect Overview Kafka Connector… Continue Reading →<\/span><\/a><\/p>\n","protected":false},"author":1,"featured_media":760,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[25],"tags":[7,11],"class_list":["post-759","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-kafka","tag-kafka","tag-kafka-connect"],"_links":{"self":[{"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/posts\/759","targetHints":{"allow":["GET"]}}],"collection":[{"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/comments?post=759"}],"version-history":[{"count":6,"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/posts\/759\/revisions"}],"predecessor-version":[{"id":1247,"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/posts\/759\/revisions\/1247"}],"wp:featuredmedia":[{"embeddable":true,"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/media\/760"}],"wp:attachment":[{"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/media?parent=759"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/categories?post=759"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/www.igfasouza.com\/blog\/wp-json\/wp\/v2\/tags?post=759"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}