Integrate Apache Kafka with Timescale Cloud

Extract the Kafka binaries to a local folder

curl https://dlcdn.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz | tar -xzf - 
cd kafka_2.13-3.9.0

Copy

From now on, the folder where you extracted the Kafka binaries is called <KAFKA_HOME>.

Configure and run Apache Kafka

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
./bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/kraft/reconfig-server.properties
./bin/kafka-server-start.sh config/kraft/reconfig-server.properties

Copy

Use the -daemon flag to run this process in the background.

Create Kafka topics

In another Terminal window, navigate to <KAFKA_HOME>, then call kafka-topics.sh and create the following topics:

• accounts: publishes JSON messages that are consumed by the timescale-sink connector and inserted into your Timescale Cloud service.
• deadletter: stores messages that cause errors and that Kafka Connect workers cannot process.

./bin/kafka-topics.sh \
     --create \
     --topic accounts \
     --bootstrap-server localhost:9092 \
     --partitions 10
     
./bin/kafka-topics.sh \
     --create \
     --topic deadletter \
     --bootstrap-server localhost:9092 \
     --partitions 10

Copy

Test that your topics are working correctly

1. Run kafka-console-producer to send messages to the accounts topic:

   bin/kafka-console-producer.sh --topic accounts --bootstrap-server localhost:9092

   Copy
2. Send some events. For example, type the following:

   >Timescale Cloud
   >How Cool

   Copy
3. In another Terminal window, navigate to <KAFKA_HOME>, then run kafka-console-consumer to consume the events you just sent:

   bin/kafka-console-consumer.sh --topic accounts --from-beginning --bootstrap-server localhost:9092

   Copy
   You see

   Timescale Cloud
   How Cool

   Copy

Install the PostgreSQL connector

In another Terminal window, navigate to <KAFKA_HOME>, then download and configure the PostgreSQL sink and driver.

mkdir -p "plugins/camel-postgresql-sink-kafka-connector"
curl https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-postgresql-sink-kafka-connector/3.21.0/camel-postgresql-sink-kafka-connector-3.21.0-package.tar.gz \
| tar -xzf - -C "plugins/camel-postgresql-sink-kafka-connector" --strip-components=1
curl  -H "Accept: application/zip" https://jdbc.postgresql.org/download/postgresql-42.7.5.jar -o  "plugins/camel-postgresql-sink-kafka-connector/postgresql-42.7.5.jar"
echo "plugin.path=`pwd`/plugins/camel-postgresql-sink-kafka-connector" >> "config/connect-distributed.properties"
echo "plugin.path=`pwd`/plugins/camel-postgresql-sink-kafka-connector" >> "config/connect-standalone.properties"

Copy

Start Kafka Connect

export CLASSPATH=`pwd`/plugins/camel-postgresql-sink-kafka-connector/*
./bin/connect-standalone.sh config/connect-standalone.properties

Copy

Use the -daemon flag to run this process in the background.

Verify Kafka Connect is running

In yet another another Terminal window, run the following command:

curl http://localhost:8083

Copy

You see something like:

{"version":"3.9.0","commit":"a60e31147e6b01ee","kafka_cluster_id":"J-iy4IGXTbmiALHwPZEZ-A"}

Copy

Connect to your Timescale Cloud service

Create a table to ingest Kafka events

CREATE TABLE accounts (created_at TIMESTAMPTZ DEFAULT NOW(),
 name TEXT,
 city TEXT);

Copy

Turn the table into a hypertable

SELECT create_hypertable('accounts', 'created_at');

Copy

Create the connection configuration

1. In the terminal running Kafka Connect, stop the process by pressing Ctrl+C.

2. Write the following configuration to <KAFKA_HOME>/config/timescale-standalone-sink.properties, then update the <properties> with your connection details.

   name=timescale-standalone-sink
   connector.class=org.apache.camel.kafkaconnector.postgresqlsink.CamelPostgresqlsinkSinkConnector
   errors.tolerance=all
   errors.deadletterqueue.topic.name=deadletter
   tasks.max=10
   value.converter=org.apache.kafka.connect.storage.StringConverter
   key.converter=org.apache.kafka.connect.storage.StringConverter
   topics=accounts
   camel.kamelet.postgresql-sink.databaseName=<dbname>
   camel.kamelet.postgresql-sink.username=<user>
   camel.kamelet.postgresql-sink.password=<password>
   camel.kamelet.postgresql-sink.serverName=<host>
   camel.kamelet.postgresql-sink.serverPort=<port>
   camel.kamelet.postgresql-sink.query=INSERT INTO accounts (name,city) VALUES (:#name,:#city)

   Copy

3. Restart Kafka Connect with the new configuration:

   export CLASSPATH=`pwd`/plugins/camel-postgresql-sink-kafka-connector/*
   ./bin/connect-standalone.sh config/connect-standalone.properties config/timescale-standalone-sink.properties

   Copy

Test the connection

To see your sink, query the /connectors route in a GET request:

curl -X GET http://localhost:8083/connectors

Copy

You see:

#["timescale-standalone-sink"]

Copy

In the terminal running kafka-console-producer.sh enter the following json strings

{"name":"Lola","city":"Copacabana"} 
{"name":"Holly","city":"Miami"}
{"name":"Jolene","city":"Tennessee"}
{"name":"Barbara Ann ","city":"California"}

Copy

Look in your terminal running kafka-console-consumer to see the messages being processed.

Query your Timescale Cloud service for all rows in the accounts table

SELECT * FROM accounts;

Copy

You see something like: