Scylla Capture CDC events

[Vikas Gupta]

Hi Team

Recently we are using confluent kafka connect to capture CDC from Scylla. We are facing the issue where heartbeat events and cdc events are appearing in same topic. I need help on how to separate out these events. Here are the details

Create Table

CREATE TABLE attribute_store_controlplane.orders(order_id int, product text, PRIMARY KEY(order_id)) WITH cdc = {'enabled': true};

Schema

{ "type": "record", "name": "TestScyllaAvroEvent", "namespace": "com.expediagroup.dataplatform", "fields": [ { "name": "order_id", "type": "int" }, { "name": "product", "type": { "type": "string", "avro.java.string": "String" } } ] }

Connector Config

{ "name": "scylla-source-connector-create-test", "config": { "connector.class": "com.scylladb.cdc.debezium.connector.ScyllaConnector", "key.converter": "io.confluent.connect.avro.AvroConverter", "value.converter": "io.confluent.connect.avro.AvroConverter", "scylla.cluster.ip.addresses": "<database-host>:9042", "scylla.name": "controlplane_scylla_source_test_stream_v1", "scylla.table.names": "attribute_store_********", "scylla.user": "<username>", "scylla.password": "<password>", "tasks.max": "1", "key.converter.schema.registry.url": "**********************", "value.converter.schema.registry.url": "***************", "producer.override.bootstrap.servers": "kafka-1j-us-east-1.egdp-test.aws.away.black:9093", "producer.override.security.protocol": "SSL", "producer.override.ssl.keystore.location": "${env::CONNECT_SSL_KEYSTORE_LOCATION}", "producer.override.ssl.keystore.password": "${env::CONNECT_SSL_KEYSTORE_PASSWORD}", "producer.override.ssl.key.password": "${env::CONNECT_SSL_KEY_PASSWORD}", "producer.override.ssl.truststore.password": "${env::CONNECT_SSL_TRUSTSTORE_PASSWORD}", "producer.override.ssl.truststore.location": "${env::CONNECT_SSL_TRUSTSTORE_LOCATION}", "value.converter.value.subject.name.strategy": "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy", "key.converter.key.subject.name.strategy": "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy", "transforms": "route, t2", "transforms.route.type":"org.apache.kafka.connect.transforms.RegexRouter", "transforms.route.replacement": "controlplane_scylla_source_test_stream_v1", "transforms.route.regex": ".*", "transforms.t2.predicate": "has-my-prefix", "transforms.t2.negate": "true", "transforms.t2.type": "org.apache.kafka.connect.transforms.ExtractField$Key", "transforms.t2.field": "serverName", "predicates": "has-my-prefix", "predicates.has-my-prefix.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches", "predicates.has-my-prefix.pattern": "xyz_*" } }

Output in Kafka (edited) 

cdc_events.png

[Vikas Gupta]

Confluent kafka connect used for CDC is https://www.confluent.io/hub/scylladb/scylla-cdc-source-connector/