CODEWITHSUNDEEP
sql-serverdebeziumcdc
Murat Dinc
Murat Dinc

Reputation: 56

In Debezium, all records are coming as op: "r". How can I overcome this situation?

I want to capture only the inserted records using Debezium by enabling the CDC feature on SQL Server, but all the records are coming with the status "r". When I use a condition like transform op: "c", no records are retrieved. Checking the CDC records on SQL Server, I can see that insert statuses are distinguished, but since they all come with the status "r", I am unable to differentiate the inserted records. How can I overcome this situation?

Connector:

{
    "name": "account-activity-connector",
    "config": {
        "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
        "tasks.max": "1",
        "database.hostname": "127.0.0.1",
        "database.port": "50516",
        "database.user": "xx",
        "database.password": "xxxx",
        "database.names": "xxxdb",
        "database.trustServerCertificate": true,
        "database.history.kafka.bootstrap.servers": "kafka:9092",
        "database.history.kafka.topic": "xxx",
        "decimal.handling.mode": "double",
        "schema.history.internal.kafka.topic": "schemahistory",
        "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
        "snapshot.mode": "initial",
        "skipped.operations": "u,d,r",
        "topic.prefix": "data-transfer",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": "false",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "false",
        "include.schema.changes": "false",
        "transforms":"unwrap", 
        "transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.add.fields":"op",
        "transforms.unwrap.delete.handling.mode":"rewrite",
        "transforms.unwrap.drop.tombstones":"false"
    }
}

Debezium Records:

{
    "Id": "5E4AB241-8091-4FA7-AEF3-08DB6CC9CC08",
    "TenantId": "212B4D87-C910-427B-A2EB-0E1CBBA5B43B",
    .....
    "__op": "r",
    "__deleted": "false"
}

I need to receive only the records that are inserts for it to work, but all records are coming with the status "r" regardless of their actual status. This makes it difficult for me to distinguish the records that are inserts.

Upvotes: 0

Views: 1962

Answers (1)

Jogger
Jogger

Reputation: 447

As connector emits snapshot events as READ operations i.e op = "r", you will need to add logic in your code on which events to consider and which not to. Like you can only consider events with "op" = "c" or "u" or "d"

Upvotes: 0

Related Questions