[Bug] [sync] mysql数据同步到达梦8 当wirteMode为update的时候执行job报错

### Search before asking

- [X] I had searched in the [issues](https://github.com/DTStack/chunjun/issues) and found no similar issues.

### What happened

mysql数据同步到达梦8 当wirteMode为update的时候执行job报错，这边需要覆盖写， 目的表如果没有数据使用insert模式没有问题

### What you expected to happen

java.sql.BatchUpdateException: Communication error
at dm.jdbc.driver.DBError.throwBatchUpdateException(DBError.java:713) ~[chunjun-connector-dm.jar:?]
at dm.jdbc.b.a.a(DBAccess.java:874) ~[chunjun-connector-dm.jar:?]
at dm.jdbc.driver.DmdbPreparedStatement.do_executeLargeBatch(DmdbPreparedStatement.java:567) ~[chunjun-connector-dm.jar:?]
at dm.jdbc.driver.DmdbPreparedStatement.do_executeBatch(DmdbPreparedStatement.java:514) ~[chunjun-connector-dm.jar:?]
at dm.jdbc.driver.DmdbPreparedStatement.executeBatch(DmdbPreparedStatement.java:1518) ~[chunjun-connector-dm.jar:?]
at com.dtstack.chunjun.connector.jdbc.statement.FieldNamedPreparedStatementImpl.executeBatch(FieldNamedPreparedStatementImpl.java:131) ~[chunjun-connector-dm.jar:?]
at com.dtstack.chunjun.connector.jdbc.sink.PreparedStmtProxy.executeBatch(PreparedStmtProxy.java:251) ~[chunjun-connector-dm.jar:?]
at com.dtstack.chunjun.connector.jdbc.sink.JdbcOutputFormat.writeMultipleRecordsInternal(JdbcOutputFormat.java:164) ~[chunjun-connector-dm.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeRecordInternal(BaseRichOutputFormat.java:496) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$initTimingSubmitTask$0(BaseRichOutputFormat.java:456) ~[chunjun-clients.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_121]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) [?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) [?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_121]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
2023-05-10 10:01:20.943 [timer-data-write-thread-pool-1-thread-1] ERROR com.dtstack.chunjun.connector.jdbc.sink.JdbcOutputFormat - Writing records failed. com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.addConsumed(DirtyDataCollector.java:105)
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.offer(DirtyDataCollector.java:79)
at com.dtstack.chunjun.dirty.manager.DirtyManager.collect(DirtyManager.java:140)
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeSingleRecord(BaseRichOutputFormat.java:482)
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$writeRecordInternal$1(BaseRichOutputFormat.java:500)
at java.util.ArrayList.forEach(ArrayList.java:1249)
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeRecordInternal(BaseRichOutputFormat.java:500)
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$initTimingSubmitTask$0(BaseRichOutputFormat.java:456)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

2023-05-10 10:01:20.946 [dirty-consumer-pool-2-thread-2] INFO com.dtstack.chunjun.dirty.utils.TablePrintUtil -
| 1 | 6c0ad7815de19efca51841df5e9f4332 | Flink_Job | Sink: dmsinkfactory | {"extHeader":[],"byteSize":187,"string":"(1,Kong Xiuying,kongxiuying@icloud.com,2009-02-21 02:12:04)","headers":null,"arity":4,"rowKind":"INSERT","headerInfo":null} | com.dtstack.chunjun.throwable.WriteRecordException:
JdbcOutputFormat [Flink_Job] writeRecord error: when converting field[0] in Row(+I(1,Kong Xiuying,kongxiuying@icloud.com,2009-02-21 02:12:04))
dm.jdbc.driver.DMException: Statement handle is closed
at com.dtstack.chunjun.connector.jdbc.sink.JdbcOutputFormat.processWriteException(JdbcOutputFormat.java:305)
at com.dtstack.chunjun.connector.jdbc.sink.JdbcOutputFormat.writeSingleRecordInternal(JdbcOutputFormat.java:141)
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeSingleRecord(BaseRichOutputFormat.java:479)
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$writeRecordInternal$1(BaseRichOutputFormat.java:500)
at java.util.ArrayList.forEach(ArrayList.java:1249)
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeRecordInternal(BaseRichOutputFormat.java:500)
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$initTimingSubmitTask$0(BaseRichOutputFormat.java:456)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: dm.jdbc.driver.DMException: Statement handle is closed
at dm.jdbc.driver.DBError.throwz(DBError.java:739)
at dm.jdbc.driver.DmdbStatement.checkClosed(DmdbStatement.java:752)
at dm.jdbc.driver.DmdbStatement.beforeExecute(DmdbStatement.java:634)
at dm.jdbc.driver.DmdbPreparedStatement.do_execute(DmdbPreparedStatement.java:464)
at dm.jdbc.driver.DmdbPreparedStatement.execute(DmdbPreparedStatement.java:1836)
at com.dtstack.chunjun.connector.jdbc.statement.FieldNamedPreparedStatementImpl.execute(FieldNamedPreparedStatementImpl.java:141)
at com.dtstack.chunjun.connector.jdbc.sink.PreparedStmtProxy.writeSingleRecordInternal(PreparedStmtProxy.java:205)
at com.dtstack.chunjun.connector.jdbc.sink.JdbcOutputFormat.writeSingleRecordInternal(JdbcOutputFormat.java:133)
... 12 more
| null | 2023-05-10 10:01:20.895 |
+------------------+----------------------------------+-----------+---------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+-----------+-------------------------+

2023-05-10 10:01:30.994 [Legacy Source Thread - Source: mysqlsourcefactory -> Sink: dmsinkfactory (1/1)#0] INFO com.dtstack.chunjun.connector.jdbc.source.JdbcInputFormat - subtask input close finished
2023-05-10 10:01:30.998 [Source: mysqlsourcefactory -> Sink: dmsinkfactory (1/1)#0] INFO com.dtstack.chunjun.connector.jdbc.sink.JdbcOutputFormat - taskNumber[0] close()
2023-05-10 10:01:51.006 [Source: mysqlsourcefactory -> Sink: dmsinkfactory (1/1)#0] WARN org.apache.flink.runtime.taskmanager.Task - Source: mysqlsourcefactory -> Sink: dmsinkfactory (1/1)#0 (b441b8c5c64e7055cd8da3f2d2f804ae) switched from RUNNING to FAILED.
java.util.concurrent.ExecutionException: java.lang.RuntimeException: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) ~[?:1.8.0_121]
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915) ~[?:1.8.0_121]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.quiesceTimeServiceAndCloseOperator(StreamOperatorWrapper.java:168) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:131) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:135) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.OperatorChain.closeOperators(OperatorChain.java:444) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:629) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:591) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:758) [chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:573) [chunjun-clients.jar:?]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]
Caused by: java.lang.RuntimeException: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.close(BaseRichOutputFormat.java:366) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.DtOutputFormatSinkFunction.close(DtOutputFormatSinkFunction.java:127) ~[chunjun-clients.jar:?]
at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:41) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.close(AbstractUdfStreamOperator.java:109) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.lambda$closeOperator$5(StreamOperatorWrapper.java:213) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.closeOperator(StreamOperatorWrapper.java:210) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.lambda$deferCloseOperatorToMailbox$3(StreamOperatorWrapper.java:185) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.tryYield(MailboxExecutorImpl.java:97) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.quiesceTimeServiceAndCloseOperator(StreamOperatorWrapper.java:162) ~[chunjun-clients.jar:?]
... 8 more
Caused by: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.addConsumed(DirtyDataCollector.java:105) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.offer(DirtyDataCollector.java:79) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.dirty.manager.DirtyManager.collect(DirtyManager.java:140) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeSingleRecord(BaseRichOutputFormat.java:482) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$writeRecordInternal$1(BaseRichOutputFormat.java:500) ~[chunjun-clients.jar:?]
at java.util.ArrayList.forEach(ArrayList.java:1249) ~[?:1.8.0_121]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeRecordInternal(BaseRichOutputFormat.java:500) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$initTimingSubmitTask$0(BaseRichOutputFormat.java:456) ~[chunjun-clients.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_121]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) ~[?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[?:1.8.0_121]
... 1 more
2023-05-10 10:01:51.007 [Source: mysqlsourcefactory -> Sink: dmsinkfactory (1/1)#0] INFO org.apache.flink.runtime.taskmanager.Task - Freeing task resources for Source: mysqlsourcefactory -> Sink: dmsinkfactory (1/1)#0 (b441b8c5c64e7055cd8da3f2d2f804ae).
2023-05-10 10:01:51.015 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Un-registering task and sending final execution state FAILED to JobManager for task Source: mysqlsourcefactory -> Sink: dmsinkfactory (1/1)#0 b441b8c5c64e7055cd8da3f2d2f804ae.
2023-05-10 10:01:51.023 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: mysqlsourcefactory -> Sink: dmsinkfactory (1/1) (b441b8c5c64e7055cd8da3f2d2f804ae) switched from RUNNING to FAILED on dc44d445-ab76-49a8-b95a-09a5cc47f5ce @ localhost (dataPort=-1).
java.util.concurrent.ExecutionException: java.lang.RuntimeException: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) ~[?:1.8.0_121]
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915) ~[?:1.8.0_121]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.quiesceTimeServiceAndCloseOperator(StreamOperatorWrapper.java:168) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:131) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:135) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.OperatorChain.closeOperators(OperatorChain.java:444) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:629) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:591) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:758) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:573) ~[chunjun-clients.jar:?]
at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_121]
Caused by: java.lang.RuntimeException: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.close(BaseRichOutputFormat.java:366) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.DtOutputFormatSinkFunction.close(DtOutputFormatSinkFunction.java:127) ~[chunjun-clients.jar:?]
at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:41) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.close(AbstractUdfStreamOperator.java:109) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.lambda$closeOperator$5(StreamOperatorWrapper.java:213) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.closeOperator(StreamOperatorWrapper.java:210) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.lambda$deferCloseOperatorToMailbox$3(StreamOperatorWrapper.java:185) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.tryYield(MailboxExecutorImpl.java:97) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.quiesceTimeServiceAndCloseOperator(StreamOperatorWrapper.java:162) ~[chunjun-clients.jar:?]
... 8 more
Caused by: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.addConsumed(DirtyDataCollector.java:105) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.offer(DirtyDataCollector.java:79) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.dirty.manager.DirtyManager.collect(DirtyManager.java:140) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeSingleRecord(BaseRichOutputFormat.java:482) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$writeRecordInternal$1(BaseRichOutputFormat.java:500) ~[chunjun-clients.jar:?]
at java.util.ArrayList.forEach(ArrayList.java:1249) ~[?:1.8.0_121]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeRecordInternal(BaseRichOutputFormat.java:500) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$initTimingSubmitTask$0(BaseRichOutputFormat.java:456) ~[chunjun-clients.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_121]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) ~[?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[?:1.8.0_121]
... 1 more
2023-05-10 10:01:51.034 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy - Calculating tasks to restart to recover the failed task cbc357ccb763df2852fee8c4fc7d55f2_0.
2023-05-10 10:01:51.035 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.failover.flip1.RestartPipelinedRegionFailoverStrategy - 1 tasks should be restarted to recover the failed task cbc357ccb763df2852fee8c4fc7d55f2_0.
2023-05-10 10:01:51.040 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job Flink_Job (6c0ad7815de19efca51841df5e9f4332) switched from state RUNNING to FAILING.
org.apache.flink.runtime.JobException: The failure is not recoverable
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:107) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:666) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:446) ~[chunjun-clients.jar:?]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_121]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_121]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_121]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_121]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158) ~[chunjun-clients.jar:?]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) [chunjun-clients.jar:?]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) [chunjun-clients.jar:?]
at scala.PartialFunction.applyOrElse(PartialFunction.scala:123) [chunjun-clients.jar:?]
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122) [chunjun-clients.jar:?]
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) [chunjun-clients.jar:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) [chunjun-clients.jar:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172) [chunjun-clients.jar:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172) [chunjun-clients.jar:?]
at akka.actor.Actor.aroundReceive(Actor.scala:517) [chunjun-clients.jar:?]
at akka.actor.Actor.aroundReceive$(Actor.scala:515) [chunjun-clients.jar:?]
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225) [chunjun-clients.jar:?]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592) [chunjun-clients.jar:?]
at akka.actor.ActorCell.invoke(ActorCell.scala:561) [chunjun-clients.jar:?]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) [chunjun-clients.jar:?]
at akka.dispatch.Mailbox.run(Mailbox.scala:225) [chunjun-clients.jar:?]
at akka.dispatch.Mailbox.exec(Mailbox.scala:235) [chunjun-clients.jar:?]
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [chunjun-clients.jar:?]
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [chunjun-clients.jar:?]
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [chunjun-clients.jar:?]
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [chunjun-clients.jar:?]
Caused by: java.util.concurrent.ExecutionException: java.lang.RuntimeException: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) ~[?:1.8.0_121]
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915) ~[?:1.8.0_121]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.quiesceTimeServiceAndCloseOperator(StreamOperatorWrapper.java:168) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:131) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:135) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.OperatorChain.closeOperators(OperatorChain.java:444) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:629) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:591) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:758) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:573) ~[chunjun-clients.jar:?]
at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_121]
Caused by: java.lang.RuntimeException: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.close(BaseRichOutputFormat.java:366) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.DtOutputFormatSinkFunction.close(DtOutputFormatSinkFunction.java:127) ~[chunjun-clients.jar:?]
at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:41) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.close(AbstractUdfStreamOperator.java:109) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.lambda$closeOperator$5(StreamOperatorWrapper.java:213) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.closeOperator(StreamOperatorWrapper.java:210) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.lambda$deferCloseOperatorToMailbox$3(StreamOperatorWrapper.java:185) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.tryYield(MailboxExecutorImpl.java:97) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.quiesceTimeServiceAndCloseOperator(StreamOperatorWrapper.java:162) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:131) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:135) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.OperatorChain.closeOperators(OperatorChain.java:444) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:629) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:591) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:758) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:573) ~[chunjun-clients.jar:?]
at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_121]
Caused by: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.addConsumed(DirtyDataCollector.java:105) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.offer(DirtyDataCollector.java:79) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.dirty.manager.DirtyManager.collect(DirtyManager.java:140) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeSingleRecord(BaseRichOutputFormat.java:482) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$writeRecordInternal$1(BaseRichOutputFormat.java:500) ~[chunjun-clients.jar:?]
at java.util.ArrayList.forEach(ArrayList.java:1249) ~[?:1.8.0_121]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeRecordInternal(BaseRichOutputFormat.java:500) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$initTimingSubmitTask$0(BaseRichOutputFormat.java:456) ~[chunjun-clients.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_121]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) ~[?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[?:1.8.0_121]
at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_121]
2023-05-10 10:01:51.045 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job Flink_Job (6c0ad7815de19efca51841df5e9f4332) switched from state FAILING to FAILED.
org.apache.flink.runtime.JobException: The failure is not recoverable
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:107) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:666) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:446) ~[chunjun-clients.jar:?]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_121]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_121]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_121]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_121]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158) ~[chunjun-clients.jar:?]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) [chunjun-clients.jar:?]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) [chunjun-clients.jar:?]
at scala.PartialFunction.applyOrElse(PartialFunction.scala:123) [chunjun-clients.jar:?]
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122) [chunjun-clients.jar:?]
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) [chunjun-clients.jar:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) [chunjun-clients.jar:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172) [chunjun-clients.jar:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172) [chunjun-clients.jar:?]
at akka.actor.Actor.aroundReceive(Actor.scala:517) [chunjun-clients.jar:?]
at akka.actor.Actor.aroundReceive$(Actor.scala:515) [chunjun-clients.jar:?]
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225) [chunjun-clients.jar:?]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592) [chunjun-clients.jar:?]
at akka.actor.ActorCell.invoke(ActorCell.scala:561) [chunjun-clients.jar:?]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) [chunjun-clients.jar:?]
at akka.dispatch.Mailbox.run(Mailbox.scala:225) [chunjun-clients.jar:?]
at akka.dispatch.Mailbox.exec(Mailbox.scala:235) [chunjun-clients.jar:?]
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [chunjun-clients.jar:?]
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [chunjun-clients.jar:?]
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [chunjun-clients.jar:?]
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [chunjun-clients.jar:?]
Caused by: java.util.concurrent.ExecutionException: java.lang.RuntimeException: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) ~[?:1.8.0_121]
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915) ~[?:1.8.0_121]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.quiesceTimeServiceAndCloseOperator(StreamOperatorWrapper.java:168) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:131) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:135) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.OperatorChain.closeOperators(OperatorChain.java:444) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:629) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:591) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:758) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:573) ~[chunjun-clients.jar:?]
at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_121]
Caused by: java.lang.RuntimeException: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.close(BaseRichOutputFormat.java:366) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.DtOutputFormatSinkFunction.close(DtOutputFormatSinkFunction.java:127) ~[chunjun-clients.jar:?]
at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:41) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.close(AbstractUdfStreamOperator.java:109) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.lambda$closeOperator$5(StreamOperatorWrapper.java:213) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.closeOperator(StreamOperatorWrapper.java:210) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.lambda$deferCloseOperatorToMailbox$3(StreamOperatorWrapper.java:185) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.tryYield(MailboxExecutorImpl.java:97) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.quiesceTimeServiceAndCloseOperator(StreamOperatorWrapper.java:162) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:131) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:135) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.OperatorChain.closeOperators(OperatorChain.java:444) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:629) ~[chunjun-clients.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:591) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:758) ~[chunjun-clients.jar:?]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:573) ~[chunjun-clients.jar:?]
at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_121]
Caused by: com.dtstack.chunjun.throwable.NoRestartException: The dirty consumer shutdown, due to the consumed count exceed the max-consumed [0]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.addConsumed(DirtyDataCollector.java:105) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.dirty.consumer.DirtyDataCollector.offer(DirtyDataCollector.java:79) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.dirty.manager.DirtyManager.collect(DirtyManager.java:140) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeSingleRecord(BaseRichOutputFormat.java:482) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$writeRecordInternal$1(BaseRichOutputFormat.java:500) ~[chunjun-clients.jar:?]
at java.util.ArrayList.forEach(ArrayList.java:1249) ~[?:1.8.0_121]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.writeRecordInternal(BaseRichOutputFormat.java:500) ~[chunjun-clients.jar:?]
at com.dtstack.chunjun.sink.format.BaseRichOutputFormat.lambda$initTimingSubmitTask$0(BaseRichOutputFormat.java:456) ~[chunjun-clients.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_121]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) ~[?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_121]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[?:1.8.0_121]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[?:1.8.0_121]
at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_121]
2023-05-10 10:01:51.049 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Stopping checkpoint coordinator for job 6c0ad7815de19efca51841df5e9f4332.
2023-05-10 10:01:51.056 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore - Shutting down
2023-05-10 10:01:51.060 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job 6c0ad7815de19efca51841df5e9f4332 reached terminal state FAILED.
2023-05-10 10:01:51.060 [main] INFO org.apache.flink.runtime.minicluster.MiniCluster - Shutting down Flink Mini Cluster
2023-05-10 10:01:51.061 [main] INFO org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint - Shutting down rest endpoint.
2023-05-10 10:01:51.062 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Stopping TaskExecutor akka://flink/user/rpc/taskmanager_0.
2023-05-10 10:01:51.062 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Close ResourceManager connection 6ebe3dddeb8ea7ecc69a383ff1f5ed0f.
2023-05-10 10:01:51.063 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Close JobManager connection for job 6c0ad7815de19efca51841df5e9f4332.
2023-05-10 10:01:51.065 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl - Free slot TaskSlot(index:0, state:ALLOCATED, resource profile: ResourceProfile{taskHeapMemory=1024.000gb (1099511627776 bytes), taskOffHeapMemory=1024.000gb (1099511627776 bytes), managedMemory=512.000mb (536870912 bytes), networkMemory=64.000mb (67108864 bytes)}, allocationId: 3a74ad9e36804c57759133e8af4a5367, jobId: 6c0ad7815de19efca51841df5e9f4332).
2023-05-10 10:01:51.071 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.jobmaster.JobMaster - Stopping the JobMaster for job Flink_Job(6c0ad7815de19efca51841df5e9f4332).
2023-05-10 10:01:51.076 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService - Stop job leader service.
2023-05-10 10:01:51.076 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager - Shutting down TaskExecutorLocalStateStoresManager.
2023-05-10 10:01:51.081 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl - Suspending SlotPool.
2023-05-10 10:01:51.085 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.jobmaster.JobMaster - Close ResourceManager connection 6ebe3dddeb8ea7ecc69a383ff1f5ed0f: Stopping JobMaster for job Flink_Job(6c0ad7815de19efca51841df5e9f4332)..
2023-05-10 10:01:51.085 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl - Stopping SlotPool.
2023-05-10 10:01:51.113 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.io.disk.FileChannelManagerImpl - FileChannelManager removed spill file directory /tmp/flink-io-b08dc492-b2f2-4d9f-a337-2dfd9d12cef6
2023-05-10 10:01:51.118 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.io.network.NettyShuffleEnvironment - Shutting down the network environment and its components.
2023-05-10 10:01:51.121 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.io.disk.FileChannelManagerImpl - FileChannelManager removed spill file directory /tmp/flink-netty-shuffle-d60bfe18-59e0-49b6-8378-f3a02791fdba
2023-05-10 10:01:51.121 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.KvStateService - Shutting down the kvState service and its components.
2023-05-10 10:01:51.121 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService - Stop job leader service.
2023-05-10 10:01:51.124 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager - Closing TaskExecutor connection dc44d445-ab76-49a8-b95a-09a5cc47f5ce because: The TaskExecutor is shutting down.
2023-05-10 10:01:51.124 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager - Disconnect job manager 8b482b4b5a7ebf5f1706f6348522459e@akka://flink/user/rpc/jobmanager_3 for job 6c0ad7815de19efca51841df5e9f4332 from the resource manager.
2023-05-10 10:01:51.125 [Thread-29] INFO org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint - Removing cache directory /tmp/flink-web-ui
2023-05-10 10:01:51.126 [Thread-29] INFO org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint - Shut down complete.
2023-05-10 10:01:51.128 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager - Shut down cluster because application is in CANCELED, diagnostics DispatcherResourceManagerComponent has been closed..
2023-05-10 10:01:51.128 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent - Closing components.
2023-05-10 10:01:51.129 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.dispatcher.runner.SessionDispatcherLeaderProcess - Stopping SessionDispatcherLeaderProcess.
2023-05-10 10:01:51.130 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Stopping dispatcher akka://flink/user/rpc/dispatcher_2.
2023-05-10 10:01:51.131 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Stopping all currently running jobs of dispatcher akka://flink/user/rpc/dispatcher_2.
2023-05-10 10:01:51.131 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerImpl - Closing the SlotManager.
2023-05-10 10:01:51.131 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerImpl - Suspending the SlotManager.
2023-05-10 10:01:51.141 [mini-cluster-io-thread-8] INFO org.apache.flink.runtime.rest.handler.legacy.backpressure.BackPressureRequestCoordinator - Shutting down back pressure request coordinator.
2023-05-10 10:01:51.142 [mini-cluster-io-thread-8] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Stopped dispatcher akka://flink/user/rpc/dispatcher_2.
2023-05-10 10:01:56.124 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.filecache.FileCache - removed file cache directory /tmp/flink-dist-cache-a4decc26-2c7c-4db8-8ef6-9546eedd1c9e
2023-05-10 10:01:56.124 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Stopped TaskExecutor akka://flink/user/rpc/taskmanager_0.
2023-05-10 10:01:56.126 [AkkaRpcService-Supervisor-Termination-Future-Executor-thread-1] INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService - Stopping Akka RPC service.
2023-05-10 10:01:56.170 [flink-metrics-2] INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService - Stopping Akka RPC service.
2023-05-10 10:01:56.171 [flink-metrics-2] INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService - Stopped Akka RPC service.
2023-05-10 10:01:56.189 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.blob.PermanentBlobCache - Shutting down BLOB cache
2023-05-10 10:01:56.195 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.blob.TransientBlobCache - Shutting down BLOB cache
2023-05-10 10:01:56.197 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.blob.BlobServer - Stopped BLOB server at 0.0.0.0:19468
2023-05-10 10:01:56.198 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.rpc.akka.AkkaRpcService - Stopped Akka RPC service.
Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
at org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:811)
at com.dtstack.chunjun.environment.MyLocalStreamEnvironment.execute(MyLocalStreamEnvironment.java:194)
at com.dtstack.chunjun.Main.exeSyncJob(Main.java:227)
at com.dtstack.chunjun.Main.main(Main.java:118)
at com.dtstack.chunjun.client.local.LocalClusterClientHelper.submit(LocalClusterClientHelper.java:35)
at com.dtstack.chunjun.client.Launcher.main(Launcher.java:119)
Caused by: org.apache.flink.runtime.JobException: The failure is not recoverable
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:107)
at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80)
at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233)
at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224)
at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215)
at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:666)
at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89)
at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:446)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305)
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212)
at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77)
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
at scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
at akka.actor.Actor.aroundReceive(Actor.scala:517)
at akka.actor.Actor.aroundReceive$(Actor.scala:515)
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
at akka.actor.ActorCell.invoke(ActorCell.scala:561)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
at akka.dispatch.Mailbox.run(Mailbox.scala:225)
at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: java.util.concurrent.ExecutionException: java.lang.RuntimeException:

### How to reproduce

{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"column": ["*"],
"username": "root",
"password": "passwd",
"connection": [
{
"jdbcUrl": [
"jdbc:mysql://192.168.31.22:3307/test"
],
"table": [
"users"
]
}
]
}
},
"writer": {
"name": "dmwriter",
"parameter": {
"username": "USERTEST",
"password": "cQWX5^PQl!GT",
"connection": [
{
"jdbcUrl": "jdbc:dm://192.168.31.22:55566",
"schema": "USERTEST",
"table": [
"USERS"
]
}
],
"writeMode": "insert",
"column": ["*"]
}
}
}
],
"setting": {
"speed": {
"channel": 1,
"bytes": 0
}
}
}
}

### Anything else

_No response_

### Version

master

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)