In this brief write-up, I demonstrate how to build a working pipeline to ingest Kafka records into CockroachDB via Kafka Connect's JDBC Sink Connector, locally, using Docker.
This serves as functional testing, that is, for understanding the concepts and all moving parts before moving on to performance testing, on real Production grade clusters.
Stay tuned for a follow-up blog where we will use this content to build and run a performance test.
UPDATE: performance test is here!
Setup
Save below as file kafka2crdb.yaml.
# kafka2crdb.yaml
---
version: '2'
services:
zookeeper:
image: confluentinc/cp-zookeeper:7.3.0
hostname: zookeeper
container_name: zookeeper
ports:
- 2181:2181
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
broker:
image: confluentinc/cp-server:7.3.0
hostname: broker
container_name: broker
depends_on:
- zookeeper
ports:
- 9092:9092
- 9101:9101
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_JMX_PORT: 9101
KAFKA_JMX_HOSTNAME: localhost
KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL: http://schema-registry:8081
CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
CONFLUENT_METRICS_ENABLE: 'true'
CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
schema-registry:
image: confluentinc/cp-schema-registry:7.3.0
hostname: schema-registry
container_name: schema-registry
depends_on:
- broker
ports:
- 8081:8081
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
kafka-connect:
image: cnfldemos/cp-server-connect-datagen:0.6.0-7.3.0
hostname: connect
container_name: connect
depends_on:
- broker
- schema-registry
ports:
- 8083:8083
environment:
CONNECT_BOOTSTRAP_SERVERS: 'broker:29092'
CONNECT_REST_ADVERTISED_HOST_NAME: connect
CONNECT_GROUP_ID: compose-connect-group
CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-7.3.0.jar
CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR
command:
- bash
- -c
- |
# Installing connector plugin
confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:latest
# Downloading newer Postgres JDBC driver
cd /usr/share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib
rm -rf postgresql*
wget https://jdbc.postgresql.org/download/postgresql-42.5.0.jar
# Launching Kafka Connect worker
/etc/confluent/docker/run &
sleep infinity
control-center:
image: confluentinc/cp-enterprise-control-center:7.3.0
hostname: control-center
container_name: control-center
depends_on:
- broker
- schema-registry
- kafka-connect
ports:
- 9021:9021
environment:
CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
CONTROL_CENTER_CONNECT_CONNECT-DEFAULT_CLUSTER: 'connect:8083'
CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://ksqldb-server:8088"
CONTROL_CENTER_KSQL_KSQLDB1_ADVERTISED_URL: "http://localhost:8088"
CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
CONTROL_CENTER_REPLICATION_FACTOR: 1
CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
CONFLUENT_METRICS_TOPIC_REPLICATION: 1
PORT: 9021
cockroach:
image: cockroachdb/cockroach:latest
container_name: cockroach-1
command: start-single-node --insecure
ports:
- 26257:26257
- 8080:8080
Notice in the kafka-connect definition how we're installing the latest version of the JDBC Sink Connector and update the JDBC Postgresql driver.
Bring the Docker Compose up.
Note:
You might have to tweak your Docker env to allow for more resources.
I use Docker Desktop for macOS configured with 12 CPUs and 22GB Memory.
docker-compose -f kafka2crdb.yaml up -d
Make sure everything is up and running
$ docker-compose -f kafka2crdb.yaml ps
NAME COMMAND SERVICE STATUS PORTS
broker "/etc/confluent/dock…" broker running 0.0.0.0:9092->9092/tcp, 0.0.0.0:9101->9101/tcp
cockroach-1 "/cockroach/cockroac…" cockroach running 0.0.0.0:8080->8080/tcp, 0.0.0.0:26257->26257/tcp
connect "bash -c '# Installi…" kafka-connect running 0.0.0.0:8083->8083/tcp, 9092/tcp
control-center "/etc/confluent/dock…" control-center running 0.0.0.0:9021->9021/tcp
schema-registry "/etc/confluent/dock…" schema-registry running 0.0.0.0:8081->8081/tcp
zookeeper "/etc/confluent/dock…" zookeeper running 2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp
Open the Control Center at http://localhost:9021 and wait until the broker shows up as healthy.
Open another tab for the CockroachDB Console at http://localhost:8080.
Demo
We create the below pipeline:
datagen(source) --> Kafka Topic --> Kafka JDBC Sink Connector --> CockroachDB(target)
You can do most of below tasks from within the Control Center.
Create Kafka topic
Connect to the broker container, and create a topic transactions with 4 partitions
docker exec -it broker /bin/bash
Once in the broker shell
# $ create topic with 4 partitions
kafka-topics --bootstrap-server broker:9092 --create --topic transactions --partitions 4
# describe it
$ kafka-topics --bootstrap-server broker:9092 --topic transactions --describe
Topic: transactions TopicId: F6NF00Z8Ry2xIWnbYVF9vg PartitionCount: 4 ReplicationFactor: 1 Configs:
Topic: transactions Partition: 0 Leader: 1 Replicas: 1 Isr: 1 Offline:
Topic: transactions Partition: 1 Leader: 1 Replicas: 1 Isr: 1 Offline:
Topic: transactions Partition: 2 Leader: 1 Replicas: 1 Isr: 1 Offline:
Topic: transactions Partition: 3 Leader: 1 Replicas: 1 Isr: 1 Offline:
With the topic partitioned into 4, we can have an equal amount of consumer processes, or tasks, consuming from the topic and ingesting into CockroachDB, in parallel.
All set, now we are ready to ingest data into this topic.
Configure the Source Connector
Our source will be a built-in data generator.
Open a new terminal, and create the Datagen Connector.
# the datagen-connect container listens on port 8083
curl -s -X POST http://localhost:8083/connectors/ \
-H "Content-Type: application/json" -d '{
"name": "datagen-transactions",
"config": {
"connector.class": "io.confluent.kafka.connect.datagen.DatagenConnector",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"kafka.topic": "transactions",
"tasks.max": "64",
"max.interval": "100",
"quickstart": "transactions"
}
}' | jq '.'
Confirm in the Control Center that messages are getting generated by visiting the Topics section.
Check also on the Messages and Schema tabs to see what the records look like
Configure the Sink Connector
We're ready to setup the JDBC Sink Connector to ingest data into CockroachDB.
Note how we set batch.size the same as max.poll.records to make sure 1 transaction includes only 128 records.
Because we've set reWriteBatchedInserts=true, the JDBC Postgres driver will conflate the 128 individual INSERT statements into a single, multi-record INSERT statement.
Note however that this single statement transaction is still an explicit transaction.
Note:
I've re-built the connector set with autocommit=true to leverage implicit transactions.
I have therefore replaced the original filekafka-connect-jdbc-10.6.1.jarin/usr/share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib/with my custom build JAR file.
Below screenshots show the results from using this custom build.
We also set tasks.max=4 to have 4 consumer processes reading from the 4 topic partitions and ingesting data into CockroachDB in parallel. Each task creates its own database connection and reads from a specific topic partition.
# register the connector
curl -s -X POST http://localhost:8083/connectors/ \
-H "Content-Type: application/json" -d '{
"name": "sink-crdb",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"connection.url": "jdbc:postgresql://cockroach-1:26257/defaultdb?reWriteBatchedInserts=true&ApplicationName=txns",
"topics": "transactions",
"tasks.max": "4",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://schema-registry:8081",
"connection.user": "root",
"connection.password": "",
"auto.create": true,
"auto.evolve": true,
"insert.mode": "insert",
"pk.mode": "none",
"pk.fields": "none",
"batch.size": 128,
"consumer.override.max.poll.records": 128
}
}' | jq '.'
You should see now 4 active SQL connections established to CockroachDB, and activity on the SQL Statement chart
Here, in CockroachDB Console in the SQL Activity > Transaction page, we confirm each transaction writes 128 rows.
Drilling down into the transaction, we can see of how many statements it is composed. In this case, as expected, it is just 1 INSERT statement
Further drilling down into the Statement itself, we can see this is a multi-record INSERT statements with 128 records, executed as an implicit transaction.
Query data in CockroachDB
Open a SQL prompt
docker exec -it cockroach-1 cockroach sql --insecure
-- the table was automatically created by the Kafka JDBC Sink Connector
-- this behavior is of course configurable
> SHOW TABLES;
schema_name | table_name | type | owner | estimated_row_count | locality
--------------+--------------+-------+-------+---------------------+-----------
public | transactions | table | root | 0 | NULL
-- check the schema of the created table
-- note how CockroachDB created its own Primary Key
-- this is also configurable
> SHOW CREATE transactions;
table_name | create_statement
---------------+--------------------------------------------------------------
transactions | CREATE TABLE public.transactions (
| transaction_id INT8 NOT NULL,
| card_id INT8 NOT NULL,
| user_id STRING NOT NULL,
| purchase_id INT8 NOT NULL,
| store_id INT8 NOT NULL,
| rowid INT8 NOT VISIBLE NOT NULL DEFAULT unique_rowid(),
| CONSTRAINT transactions_pkey PRIMARY KEY (rowid ASC)
| )
-- inspect few rows
> SELECT * FROM transactions LIMIT 5;
transaction_id | card_id | user_id | purchase_id | store_id
-----------------+---------+---------+-------------+-----------
1 | 7 | User_6 | 0 | 1
1 | 19 | User_7 | 0 | 6
1 | 18 | User_5 | 0 | 6
1 | 6 | User_ | 0 | 3
1 | 13 | User_ | 0 | 3
-- avoid contention by using AS OF SYSTEM TIME when running large aggregate queries
> SELECT count(*) FROM transactions AS OF SYSTEM TIME '-5s';
count
----------
594633
-- wait few seconds, then query again to see count increasing...
> SELECT count(*) FROM transactions AS OF SYSTEM TIME '-5s';
count
----------
594992
You can view more metrics and statement statistics on the DB Console, at http://localhost:8080.
Clean up
Bring down and remove all containers with
docker-compose -f kafka2crdb.yaml down
Top comments (0)