Pro-actively detect issues with LogMiner records

[2022-05-19 10:09:14,701] ERROR [DBZ_INCREMENTAL|task-0] WorkerSourceTask{id=DBZ_INCREMENTAL-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:195)
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 io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:199)
        at io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:59)
        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.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: io.debezium.connector.oracle.logminer.parser.DmlParserException: DML statement couldn't be parsed. Please open a Jira issue with the statement 'insert into "TEST"."DOC1"("DCID","COL2","COL3","COL4","COL5","COL6","COL7","COL8","COL9","COL10","COL11","COL12","COL13","COL14","COL15","COL16","COL17","COL18","COL19","COL20","COL21","COL22","COL23","COLCREATED","COLUPDATED","SRID","COLTYPE1","COL24","COL25","COL26","COL27","COL28","COL29","COL30") values ('6209',NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,TO_TIMESTAMP('2022-05-18 09:26:03.934000'),TO_TIMESTAMP('2022-05-18 09:26:03.934000'),'5011','0','0','2',NULL,'asdf',NULL,NULL,NULL);'.
        at io.debezium.connector.oracle.logminer.processor.AbstractLogMinerEventProcessor.parseDmlStatement(AbstractLogMinerEventProcessor.java:907)
        at io.debezium.connector.oracle.logminer.processor.AbstractLogMinerEventProcessor.lambda$handleDataEvent$5(AbstractLogMinerEventProcessor.java:716)
        at io.debezium.connector.oracle.logminer.processor.memory.MemoryLogMinerEventProcessor.addToTransaction(MemoryLogMinerEventProcessor.java:213)
        at io.debezium.connector.oracle.logminer.processor.AbstractLogMinerEventProcessor.handleDataEvent(AbstractLogMinerEventProcessor.java:715)
        at io.debezium.connector.oracle.logminer.processor.AbstractLogMinerEventProcessor.processRow(AbstractLogMinerEventProcessor.java:292)
        at io.debezium.connector.oracle.logminer.processor.AbstractLogMinerEventProcessor.processResults(AbstractLogMinerEventProcessor.java:242)
        at io.debezium.connector.oracle.logminer.processor.AbstractLogMinerEventProcessor.process(AbstractLogMinerEventProcessor.java:188)
        at io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:186)
        ... 9 more
Caused by: io.debezium.connector.oracle.logminer.parser.DmlParserException: Failed to parse insert DML: 'insert into "TEST"."DOC1"("DCID","COL2","COL3","COL4","COL5","COL6","COL7","COL8","COL9","COL10","COL11","COL12","COL13","COL14","COL15","COL16","COL17","COL18","COL19","COL20","COL21","COL22","COL23","COLCREATED","COLUPDATED","SRID","COLTYPE1","COL24","COL25","COL26","COL27","COL28","COL29","COL30") values ('6209',NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,TO_TIMESTAMP('2022-05-18 09:26:03.934000'),TO_TIMESTAMP('2022-05-18 09:26:03.934000'),'5011','0','0','2',NULL,'asdf',NULL,NULL,NULL);'
        at io.debezium.connector.oracle.logminer.parser.LogMinerDmlParser.parseInsert(LogMinerDmlParser.java:113)
        at io.debezium.connector.oracle.logminer.parser.LogMinerDmlParser.parse(LogMinerDmlParser.java:77)
        at io.debezium.connector.oracle.logminer.processor.AbstractLogMinerEventProcessor.parseDmlStatement(AbstractLogMinerEventProcessor.java:901)
        ... 16 more
Caused by: java.lang.ArrayIndexOutOfBoundsException: Index 32 out of bounds for length 32
        at io.debezium.connector.oracle.logminer.parser.LogMinerDmlParser.parseColumnListClause(LogMinerDmlParser.java:246)
        at io.debezium.connector.oracle.logminer.parser.LogMinerDmlParser.parseInsert(LogMinerDmlParser.java:104)
        ... 18 more
[2022-05-19 10:09:14,701] INFO [DBZ_INCREMENTAL|task-0] Stopping down connector (io.debezium.connector.common.BaseSourceTask:238)
[2022-05-19 10:09:14,716] INFO [DBZ_INCREMENTAL|task-0] Connection gracefully closed (io.debezium.jdbc.JdbcConnection:956)

 

----connector config
listeners=HTTP://:8083
bootstrap.servers=localhost:9092
group.id=connect-cluster
key.converter.schemas.enable=true
value.converter.schemas.enable=true
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage.topic=debezium-offset
offset.storage.replication.factor=1
config.storage.topic=debezium-configs
config.storage.replication.factor=1
status.storage.topic=debezium-status
status.storage.replication.factor=1
rest.advertised.host.name=localhostip
plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/kafka/connectors/debezium-connector-oracle/

---worker config
curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
    "name": "DBZ_INCREMENTAL",
    "config":

{     "connector.class" : "io.debezium.connector.oracle.OracleConnector",     "db_type" : "oracle",     "tasks.max" : "1",     "database.server.name" : "DEVP",     "database.hostname" : "dummyhostname",     "rac.nodes" : "RAC1IP,RAC2IP",     "database.port" : "1521",     "database.user" : "SCOTT",     "database.password" : "TIGER",     "database.dbname" : "DEVDB",     "database.oracle.version": "12+",     "database.tablename.case.insensitive": "true",     "database.history.kafka.bootstrap.servers" : "localhost:9092",     "database.history.kafka.topic": "dbz-schema-history-devdb",     "max.request.size": "20971520",     "include.schema.changes": "true",     "schema.include.list": "TEST",     "errors.log.enable": "true",     "time.precision.mode": "adaptive_time_microseconds",     "decimal.handling.mode": "double",     "snapshot.mode": "schema_only",     "internal.log.mining.dml.parser": "legacy"         }

}'