CREATE EXTERNAL SOURCE

Setup Kafka Broker. Download Kafka 3.3.2 (or above). Start Zookeeper and a Kafka broker.

Setup Confluent Schema Registry. For example:

docker run -it --rm --name csr -p 8085:8085 
  -e SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=<kafka_broker_ip>:9092 
  -e SCHEMA_REGISTRY_HOST_NAME=localhost 
  -e SCHEMA_REGISTRY_LISTENERS=http://0.0.0.0:8085 
  -e SCHEMA_REGISTRY_KAFKASTORE_TIMEOUT_MS=5000 confluentinc/cp-schema-registry:latest

Setup Kafka Connect service with the connector plugin. For example:

docker run -it --rm --name connect -p 8083:8083 
  -e BOOTSTRAP_SERVERS=<kafka_broker_ip>:9092 
  -e GROUP_ID=connect_cluster 
  -e CONFIG_STORAGE_TOPIC=connect_configs 
  -e OFFSET_STORAGE_TOPIC=connect_offsets 
  -e STATUS_STORAGE_TOPIC=connect_status quay.io/yugabyte/debezium-connector:latest

Set up the database instance. For example:

docker run -d -p 7000:7000 -p 7100:7100 -p 9000:9000 -p 9100:9100 -p 15433:15433 
  -p 5433:5433 -p 9042:9042 --name yugabyte yugabytedb/yugabyte:latest bin/yugabyted
  start --daemon=false

In docker perform the following steps:

docker exec -it  yugabyte /bin/bash
(a) yugabyted status
(b) bin/ysqlsh -h <docker_host_ip>  -U yugabyte -d yugabyte
(c) bin/yb-admin --master_addresses <docker_host_ip>:7100 
  create_change_data_stream ysql.yugabyte IMPLICIT ALL

Note the CDC Stream ID, it will be required in the connector configuration.

Setup PostgreSQL as the target database. Create the target tables in the database.

docker run -d -p 5432:5432 --name postgres12 
  123456789012.dkr.ecr.us-west-1.amazonaws.com/postgres-12.2:latest

Start the Striim server. The following sample TQL creates an external stream and write to the PostgreSQL target:

create application demo recovery 30 second interval;
create PropertySet KConnPropSet (
    zk.address: '<ip-address>:<port>',
    bootstrap.brokers: '<ip-address>:<port>',
    kafkaversion: '2.1',
    partitions: '1',
    replication.factor: '1',
    dataformat: 'avro-debezium',
    schemaregistry: 'http://<ip-address>:<port>'
);
create stream yb_data of Global.WAEvent persist using KConnPropSet;
create external source yb_source (
    connector: 'yugabyteDB'
) output to yb_data;
create target pg_replica using Global.DatabaseWriter (
    ConnectionURL:'jdbc:postgresql://<postgresql_ip>:<port/example>',
    ConnectionRetryPolicy: 'retryInterval=5, maxRetries=3',
    CDDLAction: 'Process',
    Username:'<example>',
    Password:'<example>',
    BatchPolicy:'Eventcount:10,Interval:10',
    CommitPolicy:'Interval:10,Eventcount:10',
    Tables: '<list_of_table_mappings>'
) input from yb_data;
create target con_disp using SysOut(name:Disp) input from yb_data;
end application demo;
deploy application demo;
start demo;

Start the source connector.

curl -H "Content-Type: application/json" 
  -X POST http://<kafka_connect_ip>:8083/connectors 
  -d @"debezium-yugabyte-source-connector.json"

Configure the source connector as follows:

{
    "name" : "debezium-yugabyte-source-connector",
    "config": {
		"connector.class" : "io.debezium.connector.yugabytedb.YugabyteDBConnector",
		"tasks.max" : 1,
		"snapshot.mode" : "initial",
		"exactly.once.support" : "requested",
		"database.hostname" : "<yugabytedb_ip>",
		"database.port" : 5433,
		"database.master.addresses" : "<yugabytedb_ip>:7100",
		"database.user" : "yugabyte",
		"database.password" : "yugabyte",
		"database.dbname" : "yugabyte",
		"database.server.name" : "dbserver1",
		"database.streamid" : "<cdc_stream_id>",
		"table.include.list" : "<list_of_tables>",
		"tombstones.on.delete" : false,
		"topic.creation.enable" : true,
		"topic.prefix" : "debezium",
		"topic.creation.default.partitions" : 1,
		"topic.creation.default.replication.factor" : 1,
		"include.schema.changes" : true,
		"schema.history.internal.kafka.bootstrap.servers" : "<kafka_broker_ip>:9092", 
		"schema.history.internal.kafka.topic" : "debezium_schemahistory",
		"key.converter" : "io.confluent.connect.avro.AvroConverter",
		"value.converter" : "io.confluent.connect.avro.AvroConverter",
		"key.converter.schema.registry.url" : "http://<schema_registry_ip>:8085",
		"value.converter.schema.registry.url" : "http://<schema_registry_ip>:8085",
		"key.converter.key.subject.name.strategy" :
                  "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy",
		"value.converter.value.subject.name.strategy":
                  "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy",
		"transforms" : "Reroute",
		"transforms.Reroute.type" : "org.apache.kafka.connect.transforms.RegexRouter",
		"transforms.Reroute.regex" : ".*",
		"transforms.Reroute.replacement" : "admin_yb_data"
    }
}

The existing records from the source database tables will appear in the PostgreSQL tables. Do some DMLs in the source database tables, and verify that they appear in the PostgreSQL tables.