[SUPPORT] Handling of DELETE operation using Debezium Kafka connector

[seethb]

Hi I am doing CRUD operation in YugabyteDB and my changes are tracked like below thru my Debezium Kafka connector config.

INSERT operation:

"before_id_value":null
"after_id_value":1000
"op": "c"

UPDATE operation:

"before_id_value":null
"after_id_value":1000
"op": "u"

DELETE operation
"before_id_value":1000
"after_id_value":null
"op": "d"

The HoodieSink connector perfectly handles Insert and update operations with after_id_value. it fails when i do DELETE operation as It will carry NULL value in the after_id_value. In Delete operation we will get before_id only. How do we dynamically assign this or is there any other way that we have to handle NULL values in Hoodie Kafka sink.

            "hoodie.datasource.write.recordkey.field": "after_id_value",
            "hoodie.datasource.write.partitionpath.field": "after_product_id_value",

My Kafka connector config below.

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8085/connectors/ -d '{
"name": "hudi-mors-reviews",
"config": {
"bootstrap.servers": "localhost:9092",
"connector.class": "org.apache.hudi.connect.HoodieSinkConnector",
"tasks.max": "1",
"topics": "review1.public.reviews",
"hoodie.table.name": "public-mors-review-table",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter.schemas.enable": "false",
"hoodie.table.type": "MERGE_ON_READ",
"hoodie.base.path": "file:///tmp/hoodie/public-mors-review-table",
"hoodie.datasource.write.recordkey.field": "after_id_value",
"hoodie.datasource.write.partitionpath.field": "after_product_id_value",
"hoodie.schemaprovider.class": "org.apache.hudi.schema.SchemaRegistryProvider",
"hoodie.deltastreamer.schemaprovider.registry.url": "http://localhost:8081/subjects/review1.public.reviews-value/versions/latest",
"hoodie.kafka.commit.interval.secs": 60
}
}'

getting below error

[2023-11-27 08:31:47,194] WARN [hudi-mors-baltest|task-0] Error received while writing records for transaction 20231127083047010 in partition 0 (org.apache.hudi.connect.transaction.ConnectTransactionParticipant:238)
org.apache.hudi.exception.HoodieKeyException: recordKey value: "null" for field: "after_sno_value" cannot be null or empty.
at org.apache.hudi.keygen.KeyGenUtils.getRecordKey(KeyGenUtils.java:146)
at org.apache.hudi.keygen.SimpleAvroKeyGenerator.getRecordKey(SimpleAvroKeyGenerator.java:50)
at org.apache.hudi.keygen.BaseKeyGenerator.getKey(BaseKeyGenerator.java:69)
at org.apache.hudi.connect.writers.AbstractConnectWriter.writeRecord(AbstractConnectWriter.java:85)
at org.apache.hudi.connect.transaction.ConnectTransactionParticipant.writeRecords(ConnectTransactionParticipant.java:219)
at org.apache.hudi.connect.transaction.ConnectTransactionParticipant.processRecords(ConnectTransactionParticipant.java:137)
at org.apache.hudi.connect.HoodieSinkTask.put(HoodieSinkTask.java:114)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:581)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:333)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
[2023-11-27 08:31:47,194] WARN [hudi-mors-baltest|task-0] Error received while writing records for transaction 20231127083047010 in partition 0 (org.apache.hudi.connect.transaction.ConnectTransactionParticipant:238)

Environment Description

• Hudi version : 0.13.1

• Spark version : 3.2

• Hive version : N/A

• Hadoop version : N/A

• Storage (HDFS/S3/GCS..) : Local Storage /tmp

• Running on Docker? (yes/no) : No

[danny0405]

cc @ad1happy2go for the clarification ~

[ad1happy2go]

@seethb Full Details on this similar issue - #9143

Go over it and let us know in case you have any doubts. Thanks.

[seethb]

Thanks @ad1happy2go I couldn't find any flag that we need to set for ignore.partition.for.delete please could you help me here or we can have quick call tomorrow first half or second half please. Thanks

[seethb]

Also @ad1happy2go we are not using PG (Postgres as the source) and the DB is YuagabyteDB however the CDC mechanism between YugabyteDB and PG is slightly different. here we use neither REPLICA IDENTITY FULL nor REPLICA IDENTITY as our storage engine is different (DocDB). In such cases, we have op flag "d" which I shown above comes up with before and after image values. The challenge here is how dynamically we swap the before_field to after_field or after_field to before_field for both key and partition for diff. operations like "C", "U" and "D".

[ad1happy2go]

@seethb Can you ping me on slack. We can connect and call and understand more. Thanks

[seethb]

Hi @ad1happy2go Sure, I am available now and just pinged in slack please could you join. Thanks

[seethb]

Hi @ad1happy2go as discussed we need to either match the YugabyteDB debezium payload structure to PG Standard Debezium format or implement our custom YugabyteDB payload which can parse the similar structure. Thanks again.