AI 助理
备案控制台
开发者社区 > 大数据与机器学习 > 实时计算 Flink > 正文

请教一下 flink cdc oracle 报错

maven

        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-connector-oracle-cdc</artifactId>
            <version>${flinkcdc.vsersion}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.12</artifactId>
            <version>1.13.6</version>
        </dependency>
         <dependency>
             <groupId>com.oracle.database.jdbc</groupId>
             <artifactId>ojdbc10</artifactId>
             <version>19.10.0.0</version>
         </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-cep_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>${fastjson.vsersion}</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.20</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

代码
```public static void main(String[] args) throws Exception {
//反序列化定义
Properties debeziumProperties = new Properties();
debeziumProperties.setProperty("log.mining.strategy","online_catalog");
//Oracle数据源定义（ip、username、password、、、、、）
JdbcIncrementalSource oracleChangeEventSource = new OracleSourceBuilder()
.hostname("192.168.1.253")
.port(1521)
.databaseList("newdb")
.schemaList("payment")
.tableList("payment.order")
.username("payment")
.password("payment")
.deserializer(new JsonDebeziumDeserializationSchema())
.includeSchemaChanges(true)
.startupOptions(StartupOptions.initial())
.debeziumProperties(debeziumProperties)
.splitSize(2)
.build();
StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
environment.enableCheckpointing(3000L);
environment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
environment.fromSource(oracleChangeEventSource,
WatermarkStrategy.noWatermarks(),
"OracleParallelSource")
.print()
.setParallelism(1);
environment.execute("Print Oracle Snapshot + RedoLog");
}



maven打包后，在webUI提交jar，并submit，执行失败，

![image.png](https://ucc.alicdn.com/pic/developer-ecology/m666eo5xcszli_2622060522c744c9b7b655e0ac915da7.png)


失败日志
```2023-10-24 20:07:25
org.apache.flink.runtime.JobException: Recovery is suppressed by FixedDelayRestartBackoffTimeStrategy(maxNumberRestartAttempts=3, backoffTimeMS=2000)
    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:139)
    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getGlobalFailureHandlingResult(ExecutionFailureHandler.java:102)
    at org.apache.flink.runtime.scheduler.DefaultScheduler.handleGlobalFailure(DefaultScheduler.java:301)
    at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.lambda$failJob$0(OperatorCoordinatorHolder.java:618)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRunAsync$4(AkkaRpcActor.java:453)
    at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:453)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:218)
    at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:84)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:168)
    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24)
    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20)
    at scala.PartialFunction.applyOrElse(PartialFunction.scala:127)
    at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126)
    at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
    at akka.actor.Actor.aroundReceive(Actor.scala:537)
    at akka.actor.Actor.aroundReceive$(Actor.scala:535)
    at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:579)
    at akka.actor.ActorCell.invoke(ActorCell.scala:547)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
    at akka.dispatch.Mailbox.run(Mailbox.scala:231)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175)
Caused by: org.apache.flink.util.FlinkException: Global failure triggered by OperatorCoordinator for 'Source: OracleParallelSource -> Sink: Print to Std. Out' (operator cbc357ccb763df2852fee8c4fc7d55f2).
    at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:600)
    at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$QuiesceableContext.failJob(RecreateOnResetOperatorCoordinator.java:237)
    at org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext.failJob(SourceCoordinatorContext.java:374)
    at org.apache.flink.runtime.source.coordinator.SourceCoordinator.start(SourceCoordinator.java:226)
    at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$DeferrableCoordinator.resetAndStart(RecreateOnResetOperatorCoordinator.java:405)
    at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator.lambda$resetToCheckpoint$6(RecreateOnResetOperatorCoordinator.java:150)
    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.complete(CompletableFuture.java:1975)
    at org.apache.flink.runtime.operators.coordination.ComponentClosingUtils.lambda$closeAsyncWithTimeout$0(ComponentClosingUtils.java:77)
    at java.lang.Thread.run(Thread.java:750)
Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to discover captured tables for enumerator
    at com.ververica.cdc.connectors.base.source.IncrementalSource.createEnumerator(IncrementalSource.java:151)
    at org.apache.flink.runtime.source.coordinator.SourceCoordinator.start(SourceCoordinator.java:222)
    ... 8 more
Caused by: java.lang.RuntimeException: Failed to resolve Oracle database version
    at io.debezium.connector.oracle.OracleConnection.resolveOracleDatabaseVersion(OracleConnection.java:177)
    at io.debezium.connector.oracle.OracleConnection.<init>(OracleConnection.java:87)
    at io.debezium.connector.oracle.OracleConnection.<init>(OracleConnection.java:82)
    at com.ververica.cdc.connectors.oracle.source.utils.OracleConnectionUtils.createOracleConnection(OracleConnectionUtils.java:61)
    at com.ververica.cdc.connectors.oracle.source.OracleDialect.openJdbcConnection(OracleDialect.java:90)
    at com.ververica.cdc.connectors.oracle.source.OracleDialect.discoverDataCollections(OracleDialect.java:107)
    at com.ververica.cdc.connectors.oracle.source.OracleDialect.discoverDataCollections(OracleDialect.java:51)
    at com.ververica.cdc.connectors.base.source.IncrementalSource.createEnumerator(IncrementalSource.java:139)
    ... 9 more
Caused by: java.sql.SQLException: No suitable driver found for jdbc:oracle:thin:@192.168.1.253:1521/newecpss
    at java.sql.DriverManager.getConnection(DriverManager.java:689)
    at java.sql.DriverManager.getConnection(DriverManager.java:208)
    at io.debezium.jdbc.JdbcConnection.lambda$patternBasedFactory$0(JdbcConnection.java:192)
    at io.debezium.jdbc.JdbcConnection$ConnectionFactoryDecorator.connect(JdbcConnection.java:129)
    at io.debezium.jdbc.JdbcConnection.connection(JdbcConnection.java:888)
    at io.debezium.jdbc.JdbcConnection.connection(JdbcConnection.java:883)
    at io.debezium.jdbc.JdbcConnection.queryAndMap(JdbcConnection.java:636)
    at io.debezium.jdbc.JdbcConnection.queryAndMap(JdbcConnection.java:510)
    at io.debezium.connector.oracle.OracleConnection.resolveOracleDatabaseVersion(OracleConnection.java:147)
    ... 16 more

展开
收起
songweixing 2023-10-24 20:26:34 220 0
3 条回答
写回答
取消 提交回答
  • 听风de歌

    看起来像是Flink执行失败了，原因是出现了全局失败。具体原因是"Recovery is suppressed by FixedDelayRestartBackoffTimeStrategy(maxNumberRestartAttempts=3, backoffTimeMS=2000)"，表示恢复被固定延迟重试策略抑制，最大重试次数为3次，每次重试之间的间隔为2000毫秒。这可能是由于某些作业在执行过程中出现了严重的问题，导致无法恢复。建议你可以检查一下你的Flink配置文件，看看是否有错误的地方。另外，你也可以检查一下你的作业逻辑，看看是否有潜在的问题。

    2023-10-25 16:08:56
    赞同 展开评论 打赏
  • 小周sir
    面对过去，不要迷离；面对未来，不必彷徨；活在今天，你只要把自己完全展示给别人看。

    从日志信息中可以看出，Job执行失败的原因是由于执行失败后采用了固定延迟重试策略，并且已经达到了最大重试次数，因此恢复被抑制。建议检查您的代码逻辑，确保源数据的可用性，以及设置合理的重试策略。

    2023-10-25 14:02:57
    赞同 展开评论 打赏
  • sunrr

    从你提供的依赖来看，你已经包含了所有必要的依赖。Flink CDC Oracle的错误可能由多种原因引起，包括Oracle版本不匹配、Oracle配置问题、Flink配置问题等。

    以下是一些可能的解决方案：

    1. 检查Oracle的版本是否与Flink CDC Oracle的版本相匹配。例如，如果你的Oracle是12c，那么你需要使用Flink CDC Oracle 12c的版本。

    2. 检查Oracle的配置，确保CDC功能已经启用。

    3. 检查Flink的配置，确保Flink正确连接到Oracle，并且Flink的并行度设置得当。

    4. 检查Flink的任务管理器和工作节点的资源分配，确保它们有足够的资源来处理CDC数据。

    5. 如果问题仍然存在，你可能需要查看Flink的日志，以获取更多的错误信息。

    2023-10-25 10:39:27
    赞同 展开评论 打赏
问答分类：
关系型数据库 Oracle Java Maven 流计算 数据库连接 实时计算 Flink版 日志服务
问答标签：
实时计算 Flink版CDC 实时计算 Flink版报错 flink Oracle数据库一键上云 实时计算 Flink版oracle 实时计算 Flink版cdc报错
问答地址：
开发者社区 > 大数据与机器学习 > 实时计算 Flink > 问答
相关产品：
实时计算 Flink版

版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

相关问答
实时计算 Flink版
Flink CDC 在采集Oracle数据的时候，是不是这个原因导致呢？
63
0
0
在Flink CDC中如下报错，如何解决？
41
0
0
flink CDC里oracle怎么删除归档日志？
42
1
0
在Flink CDC中，有遇到此报错吗？将检查点写入到oss，启动job报错。
32
0
0
在Flink CDC中这个语句报错如下如何解决？
59
0
0
在Flink CDC中如图所示这个报错是因为什么原因？
44
1
0
在Flink CDC中明明有创建如下，但是使用同步消费kafka中数据的时候，会显示报错？
43
1
0
在Flink CDC中oraclecdc抓取数据的时候报错，是需要其他配置吗？
43
0
0
在Flink CDC中如图所示报错如何解决？
40
1
0
在Flink CDC中oracle这个数据只能读到一行数据，需要怎么处理呢?
50
7
0
大数据与机器学习

实时计算 Flink

实时计算Flink版是阿里云提供的全托管Serverless Flink云服务，基于 Apache Flink 构建的企业级、高性能实时大数据处理系统。提供全托管版 Flink 集群和引擎，提高作业开发运维效率。

我要提问

相关产品

  • 实时计算 Flink版
    文档详情 产品详情

    • 热门讨论

    热门文章

  • FFA 2024 大会门票免费送！AI时代下大数据技术未来路在何方？
    1066
  • Slot request bulk is not fulfillable，20以上job同时部署失败
    699
  • Caused by: org.apache.kafka.common.errors.TimeoutE
    3230
  • FlinkCDC MySQL 中 scan.startup.mode 用的是什么模式啊？
    3510
  • 大佬，Caused by: java.lang.ClassNotFoundException: or
    440
  • Flink如何配置Task Manager？
    604
  • 请问flink-connector-jdbc在sink时，如何设置批量写入？
    3085
  • 不同版本的 flink 如何适配不同版本 kafka，flink-connect-kafka 版本
    939
  • Flink sql将数组炸开，实现hive的explode函数的效果，还有什么其他好的方式？
    1487
  • mysql是cst时区，不修改msyq时区的情况下， flink改怎么配置时区？
    1260
    • 展开全部
  • 权威详解 | 阿里新一代实时计算引擎 Blink，每秒支持数十亿次计算
    24446
  • OPPO数据中台之基石：基于Flink SQL构建实数据仓库
    21908
  • 独家专访阿里集团副总裁贾扬清：我为什么选择加入阿里巴巴？
    17610
  • Flink SQL 功能解密系列 —— 维表 JOIN 与异步优化
    27921
  • Flume+Kafka+Flink+Redis构建大数据实时处理系统：实时统计网站PV、UV展示
    23341
  • 阿里云实时计算产品案例&解决方案汇总
    27832
  • 基于实时计算（Flink）打造一个简单的实时推荐系统
    14382
  • 运维场景下的实时计算应用
    6958
  • 【阿里内部应用】基于Blink为新商业调控打造实时大数据交互查询服务
    6549
  • 基于 Flink 构建 CEP 引擎的挑战和实践
    6655
    • 展开全部

    相关课程

    更多
  • 实时计算 Flink 版产品入门与实操
    4020
    10
    去学习
  • 开源 Flink 极速上手教程
    1861
    7
    去学习
  • 大数据实时计算框架Spark快速入门
    1021
    93
    去学习
  • Apache Flink 入门
    5119
    9
    去学习

    • 相关文章

  • PHP内置函数
  • Eclipse 教程
  • 深入探索Android系统架构：从Linux内核到应用层
  • 探索Java中的Lambda表达式
  • 【Azure App Service】在App Service中调用Stroage SDK上传文件时遇见 System.OutOfMemoryException

    • 相关电子书

    更多
    • PostgresChina2018_樊文凯_ORACLE数据库和应用异构迁移最佳实践 立即下载
    PostgresChina2018_王帅_从Oracle到PostgreSQL的数据迁移 立即下载
    Oracle云上最佳实践 立即下载

    相关实验场景

    更多
  • 使用Flink实时发现最热Github项目
    1124
    1.0小时
    去实验
  • 5分钟上手Flink MySQL连接器
    427
    1.0小时
    去实验
  • 基于Flink+Tair搭建实时监控大屏
    147
    1.0小时
    去实验
  • 基于Hologres+Flink搭建GitHub实时数据大屏
    139
    1.0小时
    去实验

    • 相关镜像

  • oracle
  • mysql
  • postgresql