ReThink Sink

Download connector ReThinkDB Connector 1.2 for Kafka ReThinkDB Connector 1.1 for Kafka

This ReThink sink Connector allows you to write events from Kafka to RethinkDb. The connector takes the value from the Kafka Connect SinkRecords and inserts a new entry to RethinkDb.

Prerequisites

  • Apache Kafka 0.11.x of above
  • Kafka Connect 0.11.x or above
  • RethinkDb 2.3.3 or above
  • Java 1.8

Features

  1. The KCQL routing querying - Kafka topic payload field selection is supported, allowing you to select fields written to RethinkDb
  2. Error policies for handling failures
  3. SSL/TLS support
  4. Payload support for Schema.Struct and payload Struct, Schema.String and JSON payload and JSON payload with no schema.

KCQL Support

{ INSERT | UPSERT } INTO table_name SELECT { FIELD, ... } FROM kafka_topic_name [AUTOCREATE] [PK FIELD, ...]

Tip

You can specify multiple KCQL statements separated by ; to have a the connector sink multiple topics.

The ReThinkDB sink supports KCQL, Kafka Connect Query Language. The following support KCQL is available:

  1. Field selection
  2. Target rethinkDB table selection
  3. RethinkDB write modes - Two write modes are supported insert and upsert which map to RethinkDb’s conflict policies, ERROR and REPLACE respectively
  4. Auto created tables
  5. Target table primary key field selection - Which fields will compose the primary key in the rethinkDB table, only one key is supported. If none specified a - concatentation of the topic name, partition and offset are used and a primary key called id is created.

Examples:

-- AutoCreate the target table
INSERT INTO table1 SELECT * FROM topic AUTOCREATE PK field1

-- AutoCreate the target table with default primary key called id with
-- RethinkDB conflict policy REPLACE
UPSERT INTO table1 SELECT field1, field2 FROM topic AUTOCREATE

Payload Support

Schema.Struct and a Struct Payload

If you follow the best practice while producing the events, each message should carry its schema information. The best option is to send AVRO. Your Connector configurations options include:

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

This requires the SchemaRegistry.

Note

This needs to be done in the connect worker properties if using Kafka versions prior to 0.11

Schema.String and a JSON Payload

Sometimes the producer would find it easier to just send a message with Schema.String and a JSON string. In this case your connector configuration should be set to value.converter=org.apache.kafka.connect.json.JsonConverter. This doesn’t require the SchemaRegistry.

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

Note

This needs to be done in the connect worker properties if using Kafka versions prior to 0.11

No schema and a JSON Payload

There are many existing systems which are publishing Json over Kafka and bringing them in line with best practices is quite a challenge, hence we added the support. To enable this support you must change the converters in the connector configuration.

key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

Note

This needs to be done in the connect worker properties if using Kafka versions prior to 0.11

Error Polices

Landoop sink connectors support error polices. These error polices allow you to control the behaviour of the sink if it encounters an error when writing records to the target system. Since Kafka retains the records, subject to the configured retention policy of the topic, the sink can ignore the error, fail the connector or attempt redelivery.

Throw

Any error on write to the target system will be propagated up and processing is stopped. This is the default behavior.

Noop

Any error on write to the target database is ignored and processing continues.

Warning

This can lead to missed errors if you don’t have adequate monitoring. Data is not lost as it’s still in Kafka subject to Kafka’s retention policy. The sink currently does not distinguish between integrity constraint violations and or other exceptions thrown by any drivers or target system.

Retry

Any error on write to the target system causes the RetryIterable exception to be thrown. This causes the Kafka Connect framework to pause and replay the message. Offsets are not committed. For example, if the table is offline it will cause a write failure, the message can be replayed. With the Retry policy, the issue can be fixed without stopping the sink.

Lenses QuickStart

The easiest way to try out this is using Lenses Box the pre-configured docker, that comes with this connector pre-installed. You would need to Connectors –> New Connector –> Sink –> ReThink and paste your configuration

../../_images/lenses-create-rethink-sink-connector.png

RethinkDB Setup

Download and install RethinkDb. Follow the instruction here dependent on your operating system.

Installing the Connector

Connect, in production should be run in distributed mode

  1. Install and configure a Kafka Connect cluster
  2. Create a folder on each server called plugins/lib
  3. Copy into the above folder the required connector jars from the stream reactor download
  4. Edit connect-avro-distributed.properties in the etc/schema-registry folder and uncomment the plugin.path option. Set it to the root directory i.e. plugins you deployed the stream reactor connector jars in step 2.
  5. Start Connect, bin/connect-distributed etc/schema-registry/connect-avro-distributed.properties

Connect Workers are long running processes so set an init.d or systemctl service accordingly.

Sink Connector QuickStart

Start Kafka Connect in distributed mode (see install). In this mode a Rest Endpoint on port 8083 is exposed to accept connector configurations. We developed Command Line Interface to make interacting with the Connect Rest API easier. The CLI can be found in the Stream Reactor download under the bin folder. Alternatively the Jar can be pulled from our GitHub releases page.

Starting the Connector (Distributed)

Download, and install Stream Reactor. Follow the instructions here if you haven’t already done so. All paths in the quickstart are based on the location you installed the Stream Reactor.

Once the Connect has started we can now use the kafka-connect-tools cli to post in our distributed properties file for ReThinkDB. For the CLI to work including when using the dockers you will have to set the following environment variable to point the Kafka Connect Rest API.

export KAFKA_CONNECT_REST="http://myserver:myport"

➜  bin/connect-cli create rethink-sink < rethink-sink.properties

name=rethink-sink
connector.class=com.datamountaineer.streamreactor.connect.rethink.sink.ReThinkSinkConnector
tasks.max=1
topics=rethink-topic
connect.rethink.db=dbname
connect.rethink.host=localhost
connect.rethink.port=28015
connect.rethink.kcql=INSERT INTO TABLE1 SELECT * FROM rethink_topic

If you switch back to the terminal you started Kafka Connect in, you should see the Redis Sink being accepted and the task starting.

We can use the CLI to check if the connector is up but you should be able to see this in logs as well.

#check for running connectors with the CLI
➜ bin/connect-cli ps
rethink-sink

 [2016-05-08 22:37:05,616] INFO
    __                    __
   / /   ____ _____  ____/ /___  ____  ____
  / /   / __ `/ __ \/ __  / __ \/ __ \/ __ \
 / /___/ /_/ / / / / /_/ / /_/ / /_/ / /_/ /
/_____/\__,_/_/ /_/\__,_/\____/\____/ .___/
                                   /_/
     ____     ________    _       __   ____  ____
    / __ \___/_  __/ /_  (_)___  / /__/ __ \/ __ )
   / /_/ / _ \/ / / __ \/ / __ \/ //_/ / / / __  |
  / _, _/  __/ / / / / / / / / / ,< / /_/ / /_/ /
 /_/ |_|\___/_/ /_/ /_/_/_/ /_/_/|_/_____/_____/

  (com.datamountaineer.streamreactor.connect.rethink.sink.config.RethinkSinkConfig)

Test Records

Tip

If your input topic doesn’t match the target use Lenses SQL to transform in real-time the input, no Java or Scala required!

Now we need to put some records it to the rethink_topic topics. We can use the kafka-avro-console-producer to do this. Start the producer and pass in a schema to register in the Schema Registry. The schema has a firstname field of type string a lastname field of type string, an age field of type int and a salary field of type double.

bin/kafka-avro-console-producer \
  --broker-list localhost:9092 --topic rethink_topic \
  --property value.schema='{"type":"record","name":"User","namespace":"com.datamountaineer.streamreactor.connect.rethink"
  ,"fields":[{"name":"firstName","type":"string"},{"name":"lastName","type":"string"},{"name":"age","type":"int"},{"name":"salary","type":"double"}]}'

Now the producer is waiting for input. Paste in the following:

{"firstName": "John", "lastName": "Smith", "age":30, "salary": 4830}

Check for records in Rethink

Now check the logs of the connector you should see this:

INFO Received record from topic:person_rethink partition:0 and offset:0 (com.datamountaineer.streamreactor.connect.rethink.sink.writer.rethinkDbWriter:48)
INFO Empty list of records received. (com.datamountaineer.streamreactor.connect.rethink.sink.RethinkSinkTask:75)

Check for records in Rethink

Configurations

The Kafka Connect framework requires the following in addition to any connectors specific configurations:

Config Description Type Value
name Name of the connector string This must be unique across the Connect cluster
topics
The topics to sink.
The connector will check this matchs the KCQL statement
 string  
tasks.max The number of tasks to scale output int 1
connector.class Name of the connector class string com.datamountaineer.streamreactor.connect.rethink.sink.ReThinkSinkConnector

Connector Configurations

Config Description Type
connect.rethink.kcql Kafka connect query language expression string
connect.rethink.host Specifies the rethink server string
connect.rethink.port Specifies the rethink server port number int

Optional Configurations

Config Description Type Default
connect.rethink.db Specifies the rethink database to connect to string connect_rethink_sink
connect.rethink.cert.file
Certificate file to connect to a TLS
enabled ReThink cluster.
Cannot be used in conjunction with username/password.
connect.rethink.auth.key must be set
 string  
connect.rethink.auth.key
Authentication key to connect to a
TLS enabled ReThink cluster.
Cannot be used in conjunction
with username/password.
connect.rethink.cert.file must be set
 string  
connect.rethink.username
Username to connect to ReThink with
 string  
connect.rethink.password Password to connect to ReThink with string  
connect.rethink.ssl.enabled
Enables SSL communication against
an SSL enabled Rethink cluster
 boolean false
connect.rethink.trust.store.password Password for truststore string  
connect.rethink.key.store.path Path to truststore string  
connect.rethink.key.store.password Password for key store string  
connect.rethink.ssl.client.cert.auth Path to keystore string  
connect.rethink.error.policy
Specifies the action to be
taken if an error occurs while inserting the data.
There are three available options, NOOP, the error
is swallowed, THROW, the error is allowed
to propagate and retry.
For RETRY the Kafka message is redelivered up
to a maximum number of times specified by the
connect.rethink.max.retries option
 string THROW
connect.rethink.max.retries
The maximum number of times a message
is retried. Only valid when the
connect.rethink.error.policy is set to RETRY
 string 10
connect.rethink.retry.interval
The interval, in milliseconds between retries,
if the sink is using
connect.rethink.error.policy set to RETRY
 string 60000
connect.progress.enabled
Enables the output for how many
records have been processed
 boolean false

Example

name=rethink-sink
connector.class=com.datamountaineer.streamreactor.connect.rethink.sink.ReThinkSinkConnector
tasks.max=1
topics=person_rethink
connect.rethink.db=dbname
connect.rethink.host=localhost
connect.rethink.port=28015
connect.rethink.kcql=INSERT INTO TABLE1 SELECT * FROM person_rethink

Schema Evolution

Upstream changes to schemas are handled by Schema registry which will validate the addition and removal or fields, data type changes and if defaults are set. The Schema Registry enforces AVRO schema evolution rules. More information can be found here.

The Rethink Sink will automatically write and update the Rethink table if new fields are added to the Source topic, if fields are removed the Kafka Connect framework will return the default value for this field, dependent of the compatibility settings of the Schema registry.

Kubernetes

Helm Charts are provided at our repo, add the repo to your Helm instance and install. We recommend using the Landscaper to manage Helm Values since typically each Connector instance has its own deployment.

Add the Helm charts to your Helm instance:

helm repo add landoop https://landoop.github.io/kafka-helm-charts/

TroubleShooting

Please review the FAQs and join our slack channel