Prerequisites

• An active SingleStore deployment running SingleStore version 8.7.16 or later. Run the following command to check the SingleStore version:

  SQL

  Copy

  SELECT @@memsql_version;

• Enable Change Data Capture (OBSERVE queries) on the cluster. Run the following command to enable CDC:

  SQL

  Copy

  SET GLOBAL enable_observe_queries = 1;

• Install the following:

  • ZooKeeper

  • Kafka

  • Kafka Connect

  • Java 11+

Install the SingleStore Debezium Connector

To install the connector,

1. Download the SingleStore Debezium connector (singlestore-debezium-connector-<version>-plugin.tar.gz) plugin from its GitHub repository.

2. Extract the archive contents to a directory.

3. Add the path of the directory where the plugin is extracted to the Kafka Connect's plugin path. Set the plugin.path property in the connect-distributed.properties file. For example, if the archive is extracted to the /home/user/kafka/plugins/singlestore-debezium-connector directory, set plugin.path to /home/user/kafka/plugins/ directory.

4. Restart the Kafka Connect workers (process) to identify and add the plugin JARs.

Configure the Connection

After installing the connector, run the following curl command to register the connector and configure the connection to your SingleStore deployment.

Update the following properties before running the curl command:

• "name": A unique name for the connector.

• "database.hostname": IP address or hostname of the SingleStore deployment.

• "database.port": Port of the SingleStore deployment.

• "database.user": Name of the SingleStore database user with which to access the database.

• "database.password": Password for the SingleStore database user.

• "topic.prefix": Prefix for the Kafka topic name.

• "database.dbname": Name of the SingleStore database to connect to.

• "database.table": Name of the SingleStore table in the specified database.

curl -i -X POST \
 -H "Accept:application/json" \
 -H "Content-Type:application/json" \
 <Kafka_Connect_URL>/connectors/ \
 -d '{ 
    "name": "<connector_name>", 
    "config": 
    { 
        "connector.class": "com.singlestore.debezium.SingleStoreConnector", 
        "tasks.max": "1", 
        "database.hostname": "<SingleStore_endpoint>", 
        "database.port": "<port>", 
        "database.user": "<user>", 
        "database.password": "<password>", 
        "topic.prefix": "<topic_prefix>", 
        "database.dbname": "<database>",
        "database.table": "<table_name>",
        "delete.handling.mode": "none"
    } 
}'