[Bug][Sort] Fix Oracle CDC reads timestamp error

[e-mhui]

What happened

Oracle CDC reads timestamp records and reports the following error.

2023-03-10 12:00:01.748 [Source: TableSourceScan(table=[[default_catalog, default_database, node_11191_161075]], fields=[ID, name, createtime, meta.table_name, meta.database_name, meta.op_ts]) -> Calc(select=[CAST(ID) AS id, name, createtime AS create_time]) -> NotNullEnforcer(fields=[id]) (1/1)#20] WARN  org.apache.flink.runtime.taskmanager.Task - Source: TableSourceScan(table=[[default_catalog, default_database, node_11191_161075]], fields=[ID, name, createtime, meta.table_name, meta.database_name, meta.op_ts]) -> Calc(select=[CAST(ID) AS id, name, createtime AS create_time]) -> NotNullEnforcer(fields=[id]) (1/1)#20 (cfac9a033b806d3be7d7dd4d4b6e0361) switched from RUNNING to FAILED with failure cause: java.lang.IllegalArgumentException: Unable to convert to LocalDateTime from unexpected value '1678444760000000' of type java.lang.Long
	at org.apache.inlong.sort.cdc.base.util.TemporalConversions.toLocalDateTime(TemporalConversions.java:212)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema$8.convert(RowDataDebeziumDeserializeSchema.java:348)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema$13.convert(RowDataDebeziumDeserializeSchema.java:502)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema.convertField(RowDataDebeziumDeserializeSchema.java:487)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema.access$100(RowDataDebeziumDeserializeSchema.java:80)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema$17.convert(RowDataDebeziumDeserializeSchema.java:661)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema$17.convert(RowDataDebeziumDeserializeSchema.java:670)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema$13.convert(RowDataDebeziumDeserializeSchema.java:510)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema.extractAfterRow(RowDataDebeziumDeserializeSchema.java:831)
	at org.apache.inlong.sort.cdc.oracle.debezium.table.RowDataDebeziumDeserializeSchema.deserialize(RowDataDebeziumDeserializeSchema.java:803)
	at org.apache.inlong.sort.cdc.oracle.debezium.DebeziumSourceFunction$2.deserialize(DebeziumSourceFunction.java:503)
	at org.apache.inlong.sort.cdc.oracle.debezium.internal.DebeziumChangeFetcher.handleBatch(DebeziumChangeFetcher.java:239)
	at org.apache.inlong.sort.cdc.oracle.debezium.internal.DebeziumChangeFetcher.runFetchLoop(DebeziumChangeFetcher.java:161)
	at org.apache.inlong.sort.cdc.oracle.debezium.DebeziumSourceFunction.run(DebeziumSourceFunction.java:537)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:104)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:60)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:269)

The original records is:

What you expected to happen

Oracle CDC can correctly read timestamp records.

How to reproduce

Create an Oracle table with a timestamp data type.

Environment

No response

InLong version

master

InLong Component

InLong Sort

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct