[ https://issues.apache.org/jira/browse/FLINK-24961?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17447177#comment-17447177 ]
Fangliang Liu commented on FLINK-24961: --------------------------------------- Hi [~jark] [~lzljs3620320], Take a look, thanks. > When the DDL statement is different from the actual schema in the database, > ArrayIndexOutOfBoundsException will be reported > ---------------------------------------------------------------------------------------------------------------------------- > > Key: FLINK-24961 > URL: https://issues.apache.org/jira/browse/FLINK-24961 > Project: Flink > Issue Type: Bug > Components: Connectors / JDBC > Affects Versions: 1.13.2 > Reporter: Fangliang Liu > Priority: Major > > DDL > {code:java} > CREATE TABLE if not exists table_a ( > `user_id` BIGINT NULL COMMENT '', > `id` BIGINT NULL COMMENT '', > `position_id` BIGINT NULL COMMENT '', > `status` STRING NULL COMMENT '', > `transaction_id` BIGINT NULL COMMENT '', > PRIMARY KEY (`user_id`, `id`) NOT ENFORCED > ) WITH( > 'connector'='kafka', > 'topic'='xxxx', > 'properties.bootstrap.servers'='xxx', > 'properties.group.id'='xxx', > 'properties.auto.offset.reset'='earliest', > 'scan.startup.mode'='earliest-offset', > 'format'='debezium-avro-confluent', > 'debezium-avro-confluent.schema-registry.url'='xxxx' > ); > CREATE TABLE if not exists table_b ( > `user_id` BIGINT NULL COMMENT '', > `id` BIGINT NULL COMMENT '', > `position_id` BIGINT NULL COMMENT '', > `status` STRING NULL COMMENT '', > `transaction_id` BIGINT NULL COMMENT '', > ) WITH ( > 'connector' = 'tidb', > 'tidb.database.url' = 'jdbc:mysql://xxxx', > 'tidb.username' = 'xxxx', > 'tidb.password' = 'xxxxx', > 'tidb.database.name' = 'xxxxx', > 'tidb.maximum.pool.size' = '1', > 'tidb.minimum.idle.size' = '1', > 'tidb.table.name' = 'withdraws', > 'tidb.write_mode' = 'upsert', > 'sink.buffer-flush.max-rows' = '0' > ); > insert into table_b select * from table_a; > {code} > The actual schema in tidb has one more auto-increment column than table_b, > and the following error is reported when the task is started > {code:java} > java.lang.ArrayIndexOutOfBoundsException: -1 > at > org.apache.flink.connector.jdbc.table.JdbcDynamicOutputFormatBuilder.lambda$createBufferReduceExecutor$1(JdbcDynamicOutputFormatBuilder.java:145) > ~[flink-connector-jdbc_2.12-1.13.2.jar:1.13.2] > at java.util.stream.IntPipeline$4$1.accept(IntPipeline.java:250) > ~[?:1.8.0_291] > at > java.util.Spliterators$IntArraySpliterator.forEachRemaining(Spliterators.java:1032) > ~[?:1.8.0_291] > at java.util.Spliterator$OfInt.forEachRemaining(Spliterator.java:693) > ~[?:1.8.0_291] > at > java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) > ~[?:1.8.0_291] > at > java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472) > ~[?:1.8.0_291] > at > java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:546) > ~[?:1.8.0_291] > at > java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260) > ~[?:1.8.0_291] > at > java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438) > ~[?:1.8.0_291] > at > org.apache.flink.connector.jdbc.table.JdbcDynamicOutputFormatBuilder.createBufferReduceExecutor(JdbcDynamicOutputFormatBuilder.java:145) > ~[flink-connector-jdbc_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.connector.jdbc.table.JdbcDynamicOutputFormatBuilder.lambda$build$edc08011$1(JdbcDynamicOutputFormatBuilder.java:106) > ~[flink-connector-jdbc_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat.createAndOpenStatementExecutor(JdbcBatchingOutputFormat.java:142) > ~[flink-connector-jdbc_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.connector.jdbc.internal.JdbcBatchingOutputFormat.open(JdbcBatchingOutputFormat.java:116) > ~[flink-connector-jdbc_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.connector.jdbc.internal.GenericJdbcSinkFunction.open(GenericJdbcSinkFunction.java:49) > ~[flink-connector-jdbc_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34) > ~[flink-tidb-connector-1.13-0.0.4.jar:?] > at > org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.table.runtime.operators.sink.SinkOperator.open(SinkOperator.java:58) > ~[flink-table-blink_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:442) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:582) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.call(StreamTaskActionExecutor.java:100) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:562) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:647) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at > org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:537) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) > ~[flink-dist_2.12-1.13.2.jar:1.13.2] > at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_291] > 2021-11-19 07:55:36,996 INFO > org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager > [] - Clearing resource requirements of job 25e6800fc67392651c32db54b2fcc483 > {code} > > -- This message was sent by Atlassian Jira (v8.20.1#820001)