LogMinerDmlParser可能导致Insert DLL生成的事件中每列的值为Null

[tcodz]

在embedded-debezium中使用默认LogMinerDmlParser会导致部分Int类型变为String类型，最后导致每列的value都变为了Null。请问作者为何默认使用LogMinerDmlParser，其SimplerDmlParser存在问题吗？

[qianmoQ]

SimplerDmlParser 会导致部分 SQL 解析有问题，关于 Int类型变为String类型的问题能否给一个具体的示例呢？

[tcodz]

采用Insert插入数据时，LogMinerDmlParser获取的原始Sql
INSERT INTO "DATABASE"."TABLE"("UNITTYPE", "CONFIGTYPE", "PARENTID", "CONFIGINFO", "CREATETIME", "UPDATETIME", "VERSIONS") VALUES(1, 1, 141213025904376, '{"hiddenFlag":1}', TIMESTAMP'2024-03-14 06:11:56.102528', TIMESTAMP'2024-03-14 06:35:36.34255', '1710398136');
上述sql的values部分，每个值之前会有个空格，所以LogMinerDmlParser解析时，会将数值 1 转换为 ' 1'，继承的jdbcValueConverters转换失败。

[pool-1-thread-1] ERROR org.devlive.connector.DamengDebeziumConnectorTest - Process status [ false ] with error message [ Error while trying to run connector class 'org.devlive.connector.dameng.DamengConnector' ], full stack trace: org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped. at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:50) at org.devlive.connector.dameng.logminer.TransactionalBuffer.commit(TransactionalBuffer.java:250) at org.devlive.connector.dameng.logminer.TransactionalBuffer.commit(TransactionalBuffer.java:186) at org.devlive.connector.dameng.logminer.TransactionalBuffer.checkAndAutoCommitTransactions(TransactionalBuffer.java:291) at org.devlive.connector.dameng.logminer.LogMinerQueryResultProcessor.processResult(LogMinerQueryResultProcessor.java:255) at org.devlive.connector.dameng.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:210) at org.devlive.connector.dameng.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:62) at io.debezium.pipeline.ChangeEventSourceCoordinator.streamEvents(ChangeEventSourceCoordinator.java:174) at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:141) at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266) at java.util.concurrent.FutureTask.run(FutureTask.java) at --- Async.Stack.Trace --- (captured by IntelliJ IDEA debugger) at java.util.concurrent.FutureTask.<init>(FutureTask.java:151) at java.util.concurrent.AbstractExecutorService.newTaskFor(AbstractExecutorService.java:87) at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:111) at java.util.concurrent.Executors$DelegatedExecutorService.submit(Executors.java:678) at io.debezium.pipeline.ChangeEventSourceCoordinator.start(ChangeEventSourceCoordinator.java:98) at org.devlive.connector.dameng.DamengConnectorTask.start(DamengConnectorTask.java:130) at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:133) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:760) at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:192) 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) Caused by: org.apache.kafka.connect.errors.ConnectException: Error while processing event at offset {commit_scn=486389413, transaction_id=null, scn=486389413} at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:246) at org.devlive.connector.dameng.logminer.TransactionalBuffer.commit(TransactionalBuffer.java:225) ... 24 more Caused by: java.lang.NumberFormatException: For input string: " 141213025904376" at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65) at java.lang.Long.parseLong(Long.java:578) at java.lang.Long.valueOf(Long.java:803) at io.debezium.jdbc.JdbcValueConverters.lambda$convertBigInt$42(JdbcValueConverters.java:928) at io.debezium.jdbc.JdbcValueConverters.convertValue(JdbcValueConverters.java:1297) at io.debezium.jdbc.JdbcValueConverters.convertBigInt(JdbcValueConverters.java:916) at io.debezium.jdbc.JdbcValueConverters.lambda$converter$6(JdbcValueConverters.java:297) at io.debezium.relational.TableSchemaBuilder.lambda$createKeyGenerator$3(TableSchemaBuilder.java:224)
PS: update也存在同样问题

[tcodz]

补充一点，issue标题所说的每列值为Null，可能存在描述误导

[qianmoQ]

提供一下完整的SQL，建表语句，插入语句，更新语句等，方便根据提供的内容进行测试

[qianmoQ]

采用Insert插入数据时，LogMinerDmlParser获取的原始Sql INSERT INTO "DATABASE"."TABLE"("UNITTYPE", "CONFIGTYPE", "PARENTID", "CONFIGINFO", "CREATETIME", "UPDATETIME", "VERSIONS") VALUES(1, 1, 141213025904376, '{"hiddenFlag":1}', TIMESTAMP'2024-03-14 06:11:56.102528', TIMESTAMP'2024-03-14 06:35:36.34255', '1710398136'); 上述sql的values部分，每个值之前会有个空格，所以LogMinerDmlParser解析时，会将数值 1 转换为 ' 1'，继承的jdbcValueConverters转换失败。

[pool-1-thread-1] ERROR org.devlive.connector.DamengDebeziumConnectorTest - Process status [ false ] with error message [ Error while trying to run connector class 'org.devlive.connector.dameng.DamengConnector' ], full stack trace: org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped. at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:50) at org.devlive.connector.dameng.logminer.TransactionalBuffer.commit(TransactionalBuffer.java:250) at org.devlive.connector.dameng.logminer.TransactionalBuffer.commit(TransactionalBuffer.java:186) at org.devlive.connector.dameng.logminer.TransactionalBuffer.checkAndAutoCommitTransactions(TransactionalBuffer.java:291) at org.devlive.connector.dameng.logminer.LogMinerQueryResultProcessor.processResult(LogMinerQueryResultProcessor.java:255) at org.devlive.connector.dameng.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:210) at org.devlive.connector.dameng.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:62) at io.debezium.pipeline.ChangeEventSourceCoordinator.streamEvents(ChangeEventSourceCoordinator.java:174) at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:141) at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266) at java.util.concurrent.FutureTask.run(FutureTask.java) at --- Async.Stack.Trace --- (captured by IntelliJ IDEA debugger) at java.util.concurrent.FutureTask.<init>(FutureTask.java:151) at java.util.concurrent.AbstractExecutorService.newTaskFor(AbstractExecutorService.java:87) at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:111) at java.util.concurrent.Executors$DelegatedExecutorService.submit(Executors.java:678) at io.debezium.pipeline.ChangeEventSourceCoordinator.start(ChangeEventSourceCoordinator.java:98) at org.devlive.connector.dameng.DamengConnectorTask.start(DamengConnectorTask.java:130) at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:133) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:760) at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:192) 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) Caused by: org.apache.kafka.connect.errors.ConnectException: Error while processing event at offset {commit_scn=486389413, transaction_id=null, scn=486389413} at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:246) at org.devlive.connector.dameng.logminer.TransactionalBuffer.commit(TransactionalBuffer.java:225) ... 24 more Caused by: java.lang.NumberFormatException: For input string: " 141213025904376" at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65) at java.lang.Long.parseLong(Long.java:578) at java.lang.Long.valueOf(Long.java:803) at io.debezium.jdbc.JdbcValueConverters.lambda$convertBigInt$42(JdbcValueConverters.java:928) at io.debezium.jdbc.JdbcValueConverters.convertValue(JdbcValueConverters.java:1297) at io.debezium.jdbc.JdbcValueConverters.convertBigInt(JdbcValueConverters.java:916) at io.debezium.jdbc.JdbcValueConverters.lambda$converter$6(JdbcValueConverters.java:297) at io.debezium.relational.TableSchemaBuilder.lambda$createKeyGenerator$3(TableSchemaBuilder.java:224) PS: update也存在同样问题

INSERT INTO TEST.T00003(NAME, AGE) VALUES ('qianmoQ', 11);

获取到的日志：

[pool-1-thread-1] INFO org.devlive.connector.DamengDebeziumConnectorTest - Record: EmbeddedEngineChangeEvent [key=null, value={"before":null,"after":{"NAME":"'qianmoQ'","AGE":11},"source":{"version":"${project.version}","connector":"dameng","name":"my-dameng-connector-2025-0516","ts_ms":1747376904849,"snapshot":"false","db":"TEST","sequence":null,"schema":"TEST","table":"T00003","txId":"0000000000031A8D","scn":"472018","commit_scn":"472018","lcr_position":null},"op":"c","ts_ms":1747376904853,"transaction":null}, sourceRecord=SourceRecord{sourcePartition={server=my-dameng-connector-2025-0516}, sourceOffset={commit_scn=472018, transaction_id=null, scn=472018}} ConnectRecord{topic='my-dameng-connector-2025-0516.TEST.T00003', kafkaPartition=null, key=null, keySchema=null, value=Struct{after=Struct{NAME='qianmoQ',AGE=11},source=Struct{version=${project.version},connector=dameng,name=my-dameng-connector-2025-0516,ts_ms=1747376904849,db=TEST,schema=TEST,table=T00003,txId=0000000000031A8D,scn=472018,commit_scn=472018},op=c,ts_ms=1747376904853}, valueSchema=Schema{my_dameng_connector_2025_0516.TEST.T00003.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}]

AGE字段是数字类型，你使用的是哪个依赖，我使用的是 DamengDebeziumConnectorTest.java

[tcodz]

我使用的同样是DamengDebeziumConnectorTest.java。我目前使用的SimplerDmlParser.java来再REDO日志的解析，后续有时间我把LogMinerDmlParser的补上。