@rmoff#kafkasummit

From Zero to Herowith Kafka Connect
a.k.a.

A practical guide to becoming l33t with Kafka Connect

@rmoff #kafkasummit

What isKafka
Connect?
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Streaming Integration with Kafka Connect
syslogSourcesKafka Connect
Kafka Brokers
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Streaming Integration with Kafka Connect
Amazon S3

Sinks

Google BigQuery

Kafka Connect
Kafka Brokers
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Streaming Integration with Kafka Connect
Amazon S3

syslog

Google BigQuery

Kafka Connect
Kafka Brokers
From Zero to Hero with Kafka Connect

Look Ma, No Code!

@rmoff #kafkasummit{"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector","jdbc:mysql://asgard:3306/demo","table.whitelist":"sales,orders,customers"}

https://docs.confluent.io/current/connect/"connection.url":

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Streaming Pipelines

Amazon S3

RDBMS

Kafka

Connect

Kafka

Connect

HDFS

From Zero to Hero with Kafka Connect

Writing to data stores from Kafka@rmoff #kafkasummit

App

Kaf

ka

Con
nec

t

Data

Store

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Evolve processing from old systems to new
ExistingApp

New App
<x>

a
k
f
Ka
t
c
e
n
n
o
C

RDBMS

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Demo #1
http://rmoff.dev/ksldn19_demo-code

From Zero to Hero with Kafka Connect

REST API - tricks@rmoff #kafkasummit

http://go.rmoff.net/connector-status
From Zero to Hero with Kafka Connect

REST API - tricks@rmoff #kafkasummit(h/t to @madewithtea)

http://go.rmoff.net/selectively-delete-connectors

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Configuring
Kafka
Connect

Inside the API - connectors, transforms, converters
From Zero to Hero with Kafka Connect

Kafka Connect basics

SourceKafka Connect@rmoff #kafkasummit

Kafka
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Connectors
Connector

SourceKafka Connect

Kafka
From Zero to Hero with Kafka Connect

Connectors@rmoff #kafkasummit"config": {[...]"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector","connection.url": "jdbc:postgresql://postgres:5432/","topics": "asgard.demo.orders",}

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Connectors
Connector
Native data

Connect
RecordSourceKafka Connect

Kafka
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Converters
Converter

Connector
Native data

Connect

bytes[]

RecordSourceKafka Connect

Kafka
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Serialisation & Schemas

Avro-> Confluent
Schema Registry

Protobuf

JSON

CSV

https://qconnewyork.com/system/files/presentation-slides/qcon_17_-_schemas_and_apis.pdf
From Zero to Hero with Kafka Connect

The Confluent Schema Registry
Avro
Schema@rmoff #kafkasummit

SchemaRegistry
TargetSourceKafka
Connect

Avro
Message

Avro
Message

Kafka
Connect

From Zero to Hero with Kafka Connect

Converters@rmoff #kafkasummit

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081

Set as a global default per-worker; optionally can be overriden per-connector
From Zero to Hero with Kafka Connect

What about internal converters?

@rmoff #kafkasummit

value.converter=org.apache.kafka.connect.json.JsonConverterinternal.value.converter=org.apache.kafka.connect.json.JsonConverter
key.internal.value.converter=org.apache.kafka.connect.json.JsonConverter
value.internal.value.converter=org.apache.kafka.connect.json.JsonConverter
key.internal.value.converter.bork.bork.bork=org.apache.kafka.connect.json.JsonConverter
key.internal.value.please.just.work.converter=org.apache.kafka.connect.json.JsonConverter

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Single Message Transforms
Connector

SourceTransform(s)Converter

Kafka Connect

Kafka
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Single Message Transforms"config": {Do these transforms[...]"transforms": "addDateToTopic,labelFooBar","transforms.addDateToTopic.type": "org.apache.kafka.connect.transforms.TimestampRouter","transforms.addDateToTopic.topic.format": "${topic}-${timestamp}","transforms.addDateToTopic.timestamp.format": "YYYYMM","transforms.labelFooBar.type": "org.apache.kafka.connect.transforms.ReplaceField$Value","transforms.labelFooBar.renames": "delivery_address:shipping_address",}

Transforms config

Config per transform
From Zero to Hero with Kafka Connect

Extensible
Connector@rmoff #kafkasummit

Transform(s)Converter

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Confluent Hub

hub.confluent.io

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Deploying
Kafka
Connect
Connectors, Tasks, and Workers

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Connectors and Tasks
JDBC Source

S3 Sink

S3 Task #1

JDBC Task #1

JDBC Task #2

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Connectors and Tasks
JDBC Source

S3 Sink

S3 Task #1

JDBC Task #1

JDBC Task #2

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Connectors and Tasks
JDBC Source

S3 Sink

S3 Task #1

JDBC Task #1

JDBC Task #2

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Tasks and Workers
JDBC Source

S3 Sink

S3 Task #1

JDBC Task #1

JDBC Task #2

WorkerFrom Zero to Hero with Kafka Connect

Kafka Connect Standalone Worker@rmoff #kafkasummit

S3 Task #1

JDBC Task #1

JDBC Task #2

WorkerOffsets

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit"Scaling" the Standalone Worker
JDBC Task #1

S3 Task #1

JDBC Task #2

WorkerOffsets

WorkerOffsets

Fault-tolerant? Nope.

From Zero to Hero with Kafka Connect

Kafka Connect Distributed Worker@rmoff #kafkasummit

S3 Task #1

JDBC Task #1

JDBC Task #2

Kafka Connect cluster

WorkerOffsets
ConfigStatusFault-tolerant? Yeah!

From Zero to Hero with Kafka Connect

Scaling the Distributed Worker@rmoff #kafkasummit

S3 Task #1

JDBC Task #1

Kafka Connect cluster

JDBC Task #2

WorkerWorkerOffsets
ConfigStatusFault-tolerant? Yeah!

From Zero to Hero with Kafka Connect

Distributed Worker - fault tolerance@rmoff #kafkasummit

S3 Task #1

JDBC Task #1

Kafka Connect cluster

WorkerWorkerOffsets
ConfigStatusFrom Zero to Hero with Kafka Connect

Distributed Worker - fault tolerance@rmoff #kafkasummit

S3 Task #1

JDBC Task #1

JDBC Task #2

Kafka Connect cluster

WorkerOffsets
ConfigStatusFrom Zero to Hero with Kafka Connect

Multiple Distributed Clusters@rmoff #kafkasummit

S3 Task #1
JDBC Task #1

Kafka Connect cluster #1

JDBC Task #2

Kafka Connect cluster #2

Offsets

Offsets

ConfigConfigStatusStatusFrom Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Troubleshooting
Kafka Connect
From Zero to Hero with Kafka ConnectTroubleshooting Kafka Connect@rmoff #kafkasummit

Task

FAILED

Connector

RUNNING$ curl -s "http://localhost:8083/connectors/source-debezium-orders/status" | \jq '.connector.state'"RUNNING"$ curl -s "http://localhost:8083/connectors/source-debezium-orders/status" | \jq '.tasks[0].state'"FAILED"
http://go.rmoff.net/connector-status

From Zero to Hero with Kafka ConnectTroubleshooting Kafka Connect@rmoff #kafkasummit

curl -s "http:"//localhost:8083/connectors/source-debezium-orders-00/status"| jq '.tasks[0].trace'"org.apache.kafka.connect.errors.ConnectException\n\tatio.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:230)\n\tatio.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:197)\n\tatio.debezium.connector.mysql.BinlogReader$ReaderThreadLifecycleListener.onCommunicationFailure(BinlogReader.java:1018)\n\tat com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:950)\n\tatcom.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:580)\n\tatcom.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:825)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: java.io.EOFException\n\tatcom.github.shyiko.mysql.binlog.io.ByteArrayInputStream.read(ByteArrayInputStream.java:190)\n\tatcom.github.shyiko.mysql.binlog.io.ByteArrayInputStream.readInteger(ByteArrayInputStream.java:46)\n\tatcom.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer.deserialize(EventHeaderV4Deserializer.java:35)\n\tatcom.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer.deserialize(EventHeaderV4Deserializer.java:27)\n\tat com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.nextEvent(EventDeserializer.java:212)\n\tat io.debezium.connector.mysql.BinlogReader$1.nextEvent(BinlogReader.java:224)\n\tatcom.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:922)\n\t""... 3 more\n"From Zero to Hero with Kafka Connect

The log is the source of truth@rmoff #kafkasummit$ confluent log connect$ docker-compose logs kafka-connect$ cat /var/log/kafka/connect.log

From Zero to Hero with Kafka Connect

Kafka Connect@rmoff #kafkasummit

From Zero to Hero with Kafka Connect

Kafka Connect@rmoff #kafkasummit"Task is being killed and will

not recover until manually restarted"

Symptom not Cause
From Zero to Hero with Kafka Connect

Kafka Connect@rmoff #kafkasummit

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

CommonerrorsFrom Zero to Hero with Kafka Connect

@rmoff #kafkasummit

org.apache.kafka.common.errors.SerializationException:

Unknown magic byte!

From Zero to Hero with Kafka Connect

Mismatched converters@rmoff #kafkasummit

org.apache.kafka.common.errors.SerializationException:

Unknown magic byte!
Messages are not Avro

"value.converter":"AvroConverter"ⓘ

Use the correct Converter for thesource data
From Zero to Hero with Kafka Connect

Mixed serialisation methods@rmoff #kafkasummit

org.apache.kafka.common.errors.SerializationException:

Unknown magic byte!
Some messages are not Avro

"value.converter":"AvroConverter"ⓘ

Use error handling to deal
with bad messages
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Error Handling and DLQ
HandledNot Handled

Convert (read/write from Kafka,

Start (Connections to a data[de]-serialisation)

store)

Transform

Poll / Put (Read/Write from/to*data store)* can be retried by Connect

https://cnfl.io/connect-dlq
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Fail Fast
Source topic messages

Kafka Connect

https://cnfl.io/connect-dlq

Sink messages

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

YOLO ¯\_(ツ)_/¯
Source topic messages

errors.tolerance=all

Kafka Connect

https://cnfl.io/connect-dlq

Sink messages

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Dead Letter Queue
Dead
letterqueue

Source topic messages

Kafka Connect
errors.tolerance=all
errors.deadletterqueue.topic.name=my_dlq

https://cnfl.io/connect-dlq

Sink messages

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Re-processing the Dead Letter Queue
Source topic messages

Deadletterqueue

Kafka Connect (Avro sink)Kafka Connect (JSON sink)
https://cnfl.io/connect-dlq

Sink messages
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

No fields found using key and value schemas fortable: foo-bar

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

No fields found using key and value schemas fortable: foo-bar

JsonDeserializer with schemas.enable requires"schema" and "payload" fields and may not containadditional fields
From Zero to Hero with Kafka Connect

Schema, where art thou?

@rmoff #kafkasummit

No fields found using key and value schemas for table: foo-bar

JsonDeserializer with schemas.enable requires "schema" and "payload" fields and may not contain additional fields

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Schema, where art thou?
$ http localhost:8081/subjects[...]
jq '.schema|fromjson'{"type": "record","name": "Value","namespace": "asgard.demo.ORDERS","fields": [{"name": "id","type": "int"},{"name": "order_id","type": ["null","int"],"default": null},{"name": "customer_id","type": ["null","int"],"default": null},

Schema(From Schema Registry)

Messages(Avro)$ kafkacat -b localhost:9092 -C -t mysqlQF@Land RoverDefender 90SheffieldSwift LL
Parkway(2019-05-09T13:42:28Z(2019-05-09T1

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Schema, where art thou?

No fields found using key and value schemas for table: foo-bar{

No schema!

Messages(JSON)"id": 7,"order_id": 7,"customer_id": 849,
"order_total_usd": 98062.21,"make": "Pontiac","model": "Aztek","delivery_city": "Leeds","delivery_company": "Price-Zieme""delivery_address": "754 Becker W"CREATE_TS": "2019-05-09T13:42:28"UPDATE_TS": "2019-05-09T13:42:28}

From Zero to Hero with Kafka Connect

Schema, where art thou?

@rmoff #kafkasummit

No fields found using key and value schemas for table: foo-bar-> You need a schema!
-> Use Avro, or use JSON with schemas.enable=true
Either way, you need to re-configure the producer of the dataOr, use KSQL to impose a schema on the JSON/CSV data andreserialise it to Avro !
https://www.confluent.io/blog/data-wrangling-apache-kafka-ksql

From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Schema, where art thou?

JsonDeserializer with schemas.enable requires "schema" and "payload" fields and may not contain additional fields{

Schema(Embedded per
JSON message)

Messages(JSON)"schema": {"type": "struct","fields": [{"type": "int32","optional": false,"field": "id"},[!!...]],"optional": true,"name": "asgard.demo.OR},"payload": {"id": 611,
"order_id": 111,
"customer_id": 851,
"order_total_usd": 1821
"make": "Kia","model": "Sorento",
From Zero to Hero with Kafka Connect"delivery_city": "Swind@rmoff #kafkasummit

Schema, where art thou?

JsonDeserializer with schemas.enable requires "schema" and"payload" fields and may not contain additional fields-> Your JSON must be structured as
expected.

{"schema": {"type": "struct","fields": [{"type": "int32","optional": false,"field": "id"},[!!...]],"optional": true,"name": "asgard.demo.ORDERS},"payload": {"id": 611,
"order_id": 111,
"customer_id": 851,
From Zero to Hero with Kafka Connect"order_total_usd": 182190.9

@rmoff #kafkasummit

Using KSQL to apply schema to your data
JSON

Avro

CREATE STREAM ORDERS_JSON(id INT,order_total_usd DOUBLE,
delivery_city VARCHAR)
WITH (KAFKA_TOPIC='orders-json',
VALUE_FORMAT='JSON');

Kafka
Connect

CREATE STREAM ORDERS_AVRO WITH (VALUE_FORMAT='AVRO',
KAFKA_TOPIC='orders-avro')
AS SELECT * FROM ORDERS_JSON;

KSQL
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Using KSQL to apply schema to your data
JSON

Avro

KSQL

CREATE STREAM ORDERS_JSON(id INT,order_total_usd DOUBLE,
delivery_city VARCHAR)
WITH (KAFKA_TOPIC='orders-json',
VALUE_FORMAT='JSON');

Kafka
Connect

CREATE STREAM ORDERS_AVRO WITH (VALUE_FORMAT='AVRO',
KAFKA_TOPIC='orders-avro')
AS SELECT * FROM ORDERS_JSON;
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Containers
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

Kafka Connect images on Docker Hub

kafka-connect-elasticsearchkafka-connect-jdbc
kafka-connect-hdfs[…]

confluentinc/cp-kafka-connect-baseconfluentinc/cp-kafka-connect

From Zero to Hero with Kafka Connect

Adding connectors to a container@rmoff #kafkasummit

Confluent Hub

JAR

confluentinc/cp-kafka-connect-baseFrom Zero to Hero with Kafka Connect

@rmoff #kafkasummit

At runtime

kafka-connect:
image: confluentinc/cp-kafka-connect:5.2.1
environment:
CONNECT_PLUGIN_PATH: '/usr/share/java,/usr/share/confluent-hub-components'
command:- bash- -c- |confluent-hub install --no-prompt neo4j/kafka-connect-neo4j:1.0.0/etc/confluent/docker/run

JAR

confluentinc/cp-kafka-connect-base

http://rmoff.dev/ksln19-connect-docker
From Zero to Hero with Kafka Connect

Build a new image@rmoff #kafkasummit

FROM confluentinc/cp-kafka-connect:5.2.1
ENV CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components"
RUN confluent-hub install --no-prompt neo4j/kafka-connect-neo4j:1.0.0

JAR

confluentinc/cp-kafka-connect-baseFrom Zero to Hero with Kafka Connect

Automating connector creation@rmoff #kafkasummit# # Download JDBC drivers
cd /usr/share/java/kafka-connect-jdbc/
curl https:!//cdn.mysql.com/Downloads/Connector-J/mysql-connector-java-8.0.13.tar.gz | tar xz## Now launch Kafka Connect/etc/confluent/docker/run &## Wait for Kafka Connect listener
while [ $$(curl -s -o /dev/null -w %{http_code} http:!//$$CONNECT_REST_ADVERTISED_HOST_NAME:$…
echo -e $$(date) " Kafka Connect listener HTTP state: " $$(curl -s -o /dev/null -w %{http_…
sleep 5done#
# Create JDBC Source connector
curl -X POST http:!//localhost:8083/connectors -H "Content-Type: application/json" -d '{"name": "jdbc_source_mysql_00","config": {"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector","connection.url": "jdbc:mysql:!//mysql:3306/demo","connection.user": "connect_user","connection.password": "asgard","topic.prefix": "mysql-00-","table.whitelist" : "demo.customers",}}'# Don't let the container die
http://rmoff.dev/ksln19-connect-docker
sleep infinity
From Zero to Hero with Kafka Connect

#EOF

Confluent Community Components@rmoff #kafkasummit

Apache Kafka with a bunch of cool stuff! For free!
Log Events

Database Changes

loT Data

Web Events…

Confluent Platform

Data
Integration

Real-time Applications
Monitoring & Administration
Confluent Control
Center | Security
Confluent
Platform
Transformations

HadoopOperations
Replicator | Auto Data Balancing
Custom Apps

Database

Data Compatibility
Schema Registry

SQL Stream Processing
KSQL

Data Warehouse

Development and Connectivity
Clients | Connectors | REST Proxy | CLI

CRM

Monitoring

Apache Kafka®
Core | Connect API | Streams API…

CUSTOMER SELF-MANAGED

Datacenter

Public Cloud

Analytics…

CONFLUENT FULLY-MANAGED

Confluent Cloud
From Zero to Hero with Kafka Connect

@rmoff #kafkasummit

http://cnfl.io/book-bundle

From Zero to Hero with Kafka Connect

@rmoff

Please rate my session in
the Kafka Summit app :-)#kafkasummit@rmoff #kafkasummit

http://cnfl.io/slack

From Zero to Hero with Kafka Connect