mysql->oceanbase pipeline同步的时候报错

[liht-github]

我在实现mysql->oceanbase同步的时候报错，无法同步启动pipeline
有没有人遇到相同的情况，怎么解决呢

版本信息：
flink 1.19.1
flink cdc ：3.4.0

pipeline 如下：
source:
type: mysql
hostname: {hostname}
port: 3306
username: root
password: {mysql passwd}
tables: test..*
server-id: 5400-5404
server-time-zone: Asia/Shanghai

sink:
type: oceanbase
url: jdbc:mysql://{hostname}:2881/oceanbase
username: {oceanbase username}
password: {oceanbase passwd}

pipeline:
name: MySQL to OceanBase Pipeline
parallelism: 1

运行时报错如下：
Exception in thread "main" org.apache.flink.util.FlinkException: Failed to execute job 'MySQL to OceanBase Pipeline'.
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2455)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2421)
at org.apache.flink.cdc.composer.flink.FlinkPipelineExecution.execute(FlinkPipelineExecution.java:43)
at org.apache.flink.cdc.cli.CliExecutor.run(CliExecutor.java:91)
at org.apache.flink.cdc.cli.CliFrontend.main(CliFrontend.java:74)
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:748)
Caused by: java.util.concurrent.CompletionException: java.lang.RuntimeException: org.apache.flink.runtime.JobException: Cannot instantiate the coordinator for operator Source: Flink CDC Event Source: mysql -> SchemaOperator -> PrePartition
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: Flink CDC Event Source: mysql -> SchemaOperator -> PrePartition
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: Flink CDC Event Source: mysql -> SchemaOperator -> PrePartition
at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.(ExecutionJobVertex.java:182)
at org.apache.flink.runtime.executiongraph.ExecutionJobVertex$Factory.createExecutionJobVertex(ExecutionJobVertex.java:671)
at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.attachJobVertices(DefaultExecutionGraph.java:860)
at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.attachJobGraph(DefaultExecutionGraph.java:833)
at org.apache.flink.runtime.executiongraph.DefaultExecutionGraphBuilder.buildGraph(DefaultExecutionGraphBuilder.java:221)
at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:173)
at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:381)
at org.apache.flink.runtime.scheduler.SchedulerBase.(SchedulerBase.java:224)
at org.apache.flink.runtime.scheduler.DefaultScheduler.(DefaultScheduler.java:140)
at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:162)
at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:121)
at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:379)
at org.apache.flink.runtime.jobmaster.JobMaster.(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.ClassCastException: cannot assign instance of com.oceanbase.connector.flink.connection.OceanBaseConnectionProvider to field org.apache.flink.cdc.connectors.oceanbase.catalog.OceanBaseCatalog.connectionProvider of type com.oceanbase.connector.flink.connection.OceanBaseConnectionProvider in instance of org.apache.flink.cdc.connectors.oceanbase.catalog.OceanBaseMySQLCatalog
at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2302)
at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1432)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2409)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2327)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2185)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1665)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2403)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2327)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2185)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1665)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2403)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2327)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2185)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1665)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:501)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:459)
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:496)
at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.createOperatorCoordinatorHolder(ExecutionJobVertex.java:294)
at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.(ExecutionJobVertex.java:173)

[whhe]

3.4.0 是指的 master 分支吗？从报错来看可能是 lib 下边放的 jar 包冲突了，因为 pipeline 包里包含 flink-connector-oceanbase，所以如果当前 lib 下边有 flink-connector-oceanbase-xx.jar，需要把它先去掉。

[liht-github]

lib 下没有 flink-connector-oceanbase-xx.jar, 只在flink/lib下加了mysql-connector-java-8.0.27.jar，flink-cdc-pipeline-connector-mysql-3.4-SNAPSHOT.jar，flink-cdc-pipeline-connector-oceanbase-3.4-SNAPSHOT.jar，所以感觉更大的可能性是flink-cdc-pipeline-connector-oceanbase-3.4-SNAPSHOT.jar打包的时候就出错了

[liht-github]

是master分支的代码

[yuanoOo]

@liht-github hi，我从零尝试复现了一下，没有发现问题，下面是我的部分测试步骤, 你看看有什么问题吗？

1. git clone https://github.com/apache/flink-cdc.git 拉取master分支
2. 执行mvn clean package -DskipTests，编译打包所需的jar包和flink-cdc分发tar包
   
3. 部署准备flink cdc
   
4. chmod +x bin/flink-cdc.sh
5. ./bin/flink-cdc.sh cdc.yaml --flink-home ~/opt/flink-1.19.1/

################################################################################
# Description: Sync MySQL all tables to OceanBase
################################################################################
source:
  type: mysql
  hostname: localhost
  port: 13306
  username: root
  password: 123456
  tables: app_db.\.*
  server-id: 5400-5404
  server-time-zone: UTC

sink:
  type: oceanbase
  url: jdbc:mysql://localhost:2881/test
  username: root@test
  password: 123456

pipeline:
  name: Sync MySQL Database to OceanBase

[liht-github]

特别感谢您的详细解答，
我对照了一下我的步骤，重新做了一遍，发现部署方式和您的不一样，

我是把部署在了 $FLINK_HOME/lib

flink-cdc-pipeline-connector-mysql-3.4-SNAPSHOT.jar
flink-cdc-pipeline-connector-oceanbase-3.4-SNAPSHOT.jar 

当我把这两个jar 部署在/root/flink-cdc-3.4-SNAPSHOT/lib/下后，运行没有报错了
应该是FLINK的加载顺序不一样导致的