用doris16.2版本flinkcdc的jar包同步sqlserver数据到doris，doris代码出现死锁问题

Viewed 66
2.1

5662371b48e8b40bf61e9b234904d9e.png
因为默认flinkcdc是2.X 要手动设置scan.incremental.snapshot.enabled
我的代码

bin/flink run -Dyarn.application.name=TrayLotBinding_Log  -Dexecution.checkpointing.interval=3s -Dparallelism.default=3 -c org.apache.doris.flink.tools.cdc.CdcTools lib/flink-doris-connector-1.18-1.6.2.jar sqlserver-sync-database  --database ods   --table-prefix TEST_ --sqlserver-conf hostname=10.50.2.XXX --sqlserver-conf port=1433 --sqlserver-conf username=XXXX --sqlserver-conf password="XXXX" --sqlserver-conf database-name=F1SUB --sqlserver-conf schema-name="dbo" --sqlserver-conf scan.startup.mode='initial' --including-tables "AOI" --sqlserver-conf scan.incremental.snapshot.enabled='true' --debezium.snapshot.isolation.mode='read-uncommitted'  --sink-conf fenodes=10.48.XXX.XXX:8030 --sink-conf username=root --sink-conf password="XXXXX" --sink-conf jdbc-url=jdbc:mysql://10.48.XXX.XXX:9030 --sink-conf sink.label-prefix=label --table-conf replication_num=3

执行后会报错
1732845009023.png

xu chaoqiang1
2 Answers

配置中改为 --sqlserver-conf debezium.snapshot.isolation.mode='read-uncommitted'

vinlee303 

 The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Failed to execute job 'SqlServer-Doris Sync Database: F1SUB'.
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
        at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
        at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:105)
        at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:851)
        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:245)
        at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1095)
        at org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$9(CliFrontend.java:1189)
        at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
        at org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1189)
        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1157)
Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'SqlServer-Doris Sync Database: F1SUB'.
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2253)
        at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:189)
        at org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:118)
        at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:2099)
        at org.apache.doris.flink.tools.cdc.CdcTools.syncDatabase(CdcTools.java:168)
        at org.apache.doris.flink.tools.cdc.CdcTools.createSqlServerSyncDatabase(CdcTools.java:107)
        at org.apache.doris.flink.tools.cdc.CdcTools.main(CdcTools.java:63)
        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.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
        ... 9 more
Caused by: java.lang.RuntimeException: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
        at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:321)
        at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:75)
        at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
        at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
        at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:457)
        at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
        at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1067)
        at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1703)
        at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:172)
Caused by: org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
        at org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97)
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
        at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1609)
        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: java.util.concurrent.CompletionException: java.lang.RuntimeException: org.apache.flink.runtime.JobException: Cannot instantiate the coordinator for operator Source: S                             qlServer IncrSource -> Process
        at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
        at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
        at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1606)
        ... 3 more
Caused by: java.lang.RuntimeException: org.apache.flink.runtime.JobException: Cannot instantiate the coordinator for operator Source: SqlServer IncrSource -> Process
        at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:321)
        at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:114)
        at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
        ... 3 more
Caused by: org.apache.flink.runtime.JobException: Cannot instantiate the coordinator for operator Source: SqlServer IncrSource -> Process
        at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.initialize(ExecutionJobVertex.java:234)
        at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.initializeJobVertex(DefaultExecutionGraph.java:894)
        at org.apache.flink.runtime.executiongraph.ExecutionGraph.initializeJobVertex(ExecutionGraph.java:224)
        at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.initializeJobVertices(DefaultExecutionGraph.java:875)
        at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.attachJobGraph(DefaultExecutionGraph.java:829)
        at org.apache.flink.runtime.executiongraph.DefaultExecutionGraphBuilder.buildGraph(DefaultExecutionGraphBuilder.java:221)
        at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:163)
        at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:371)
        at org.apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:214)
        at org.apache.flink.runtime.scheduler.DefaultScheduler.<init>(DefaultScheduler.java:140)
        at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:156)
        at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:122)
        at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:379)
        at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:356)
        at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:128)
        at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:100)
        at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112)
        ... 4 more
Caused by: java.lang.ClassNotFoundException: com.ververica.cdc.connectors.sqlserver.source.SqlServerSourceBuilder$SqlServerIncrementalSource
        at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67)
        at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
        at org.apache.flink.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:192)
        at java.lang.Class.forName0(Native Method)
        at java.lang.Class.forName(Class.java:348)
        at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:78)
        at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2011)
        at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1875)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2209)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1692)
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2454)
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2378)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2236)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1692)
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:508)
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:466)
        at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:539)
        at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:527)
        at org.apache.flink.util.SerializedValue.deserializeValue(SerializedValue.java:67)
        at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.create(OperatorCoordinatorHolder.java:477)
        at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.createOperatorCoordinatorHolder(ExecutionJobVertex.java:292)
        at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.initialize(ExecutionJobVertex.java:225)
        ... 20 more
xu chaoqiang1