Missing filed in dbz

[yufansong]

Describe the bug

There is one customer from sunbit feedback an issue with the sink Debezium.
There is a missing ts_ms in the source block. (There are only DB and Table in this block)

When i manually added this property it resolved the issue.

Error message/log

This caused the following error while consuming the message:
org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:611)
	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.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.IllegalArgumentException: Unexpected field name: ts_ms
	at io.debezium.transforms.ExtractNewRecordState$FieldReference.getSchema(ExtractNewRecordState.java:382)
	at io.debezium.transforms.ExtractNewRecordState.updateSchema(ExtractNewRecordState.java:273)
	at io.debezium.transforms.ExtractNewRecordState.makeUpdatedSchema(ExtractNewRecordState.java:266)
	at io.debezium.transforms.ExtractNewRecordState.lambda$addFields$2(ExtractNewRecordState.java:232)
	at java.base/java.util.concurrent.ConcurrentMap.computeIfAbsent(ConcurrentMap.java:330)
	at io.debezium.transforms.ExtractNewRecordState.addFields(ExtractNewRecordState.java:231)
	at io.debezium.transforms.ExtractNewRecordState.apply(ExtractNewRecordState.java:180)
	at com.getindata.kafka.connect.iceberg.sink.converter.SinkRecordToIcebergChangeEventConverter.convert(SinkRecordToIcebergChangeEventConverter.java:40)
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
	at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1655)
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
	at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:913)
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:578)
	at com.getindata.kafka.connect.iceberg.sink.IcebergChangeConsumer.accept(IcebergChangeConsumer.java:46)
	at com.getindata.kafka.connect.iceberg.sink.IcebergSinkTask.put(IcebergSinkTask.java:38)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:581)

To Reproduce

No response

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response

[tabVersion]

close as completed