Flink-cdc 2.1.0 Oracle CDC使用XStream捕获数据初步实践以及后续问题。

[peniridis]

1. 环境

Oracle

SQL*Plus: Release 19.0.0.0.0 - Production on Fri Dec 17 18:49:43 2021
Version 19.3.0.0.0

Flink

Version: 1.13.3

Flink-cdc

git branch master
Version: 2.2-SNAPSHOT

MacOS Oracle Client

Client: instantclient-basic-macos.x64-19.3.0.0.0dbru.zip
JVM Options: -Djava.library.path=instantclient_19_3

2. Oracle 配置

CDB&PDB数据库实例

dbca -silent -createDatabase -templateName General_Purpose.dbc -responseFile NO_VALUE \
-gdbname HCCDB  -sid HCCDB \
-createAsContainerDatabase TRUE \
-numberOfPDBs 1 \
-pdbName HCPDB1 \
-pdbAdminPassword dbz \
-sysPassword dbz -systemPassword dbz \
-datafileDestination '/opt/oracle/oradata/' \
-recoveryAreaDestination '/opt/oracle/oradata/recovery_area' \
-redoLogFileSize 50 \
-storageType FS \
-characterset ZHS16GBK -nationalCharacterSet AL16UTF16 \
-sampleSchema true \
-totalMemory 1024 \
-databaseType OLTP  \
-emConfiguration NONE

参考XStream配置脚本

https://github.com/opengauss-mirror/openGauss-tools-onlineMigration/blob/master/instruction/setup-xtream.sh

Oracle源表初始化

CREATE TABLE customers (
id NUMBER(4) NOT NULL PRIMARY KEY,
first_name VARCHAR2(255) NOT NULL,
last_name VARCHAR2(255) NOT NULL,
email VARCHAR2(255) NOT NULL
);

3. Flink捕获XStream数据

pom.xml

<dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>1.13.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>1.13.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>1.13.3</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java-bridge -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.11</artifactId>
            <version>1.13.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.11</artifactId>
            <version>1.13.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>1.13.3</version>
        </dependency>
        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-connector-oracle-cdc</artifactId>
            <version>2.2-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>com.oracle.ojdbc</groupId>
            <artifactId>ojdbc10</artifactId>
            <version>19.3.0.0</version>
        </dependency>
        <dependency>
            <groupId>oracle</groupId>
            <artifactId>streams</artifactId>
            <scope>system</scope>
            <version>1.0</version>
            <systemPath>${basedir}/lib/xstreams.jar</systemPath>
        </dependency>
    </dependencies>

OracleCDCExample

public static void main(String[] args) throws Exception {
        Properties deProps = new Properties();
        deProps.setProperty("database.connection.adapter", "xstream");
        deProps.setProperty("database.out.server.name", "dbzxout");
        deProps.setProperty("database.pdb.name", "HCPDB1");
        SourceFunction<String> sourceFunction = OracleSource.<String>builder()
                .hostname("host")
                .port(1521)
                .database("HCCDB") // monitor XE database
                .schemaList("DEBEZIUM") // monitor inventory schema
                .tableList("DEBEZIUM.CUSTOMERS") // monitor products table
                .username("c##dbzuser")
                .password("dbz")
                .debeziumProperties(deProps)
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(sourceFunction).print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute();
    }

SOUT

可以正确解析数据库变化JSON数据

3. FlinkSQL捕获XStream数据

OracleCDCSQLExample

public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //2.创建 Flink-MySQL-CDC 的 Source
        tableEnv.executeSql("CREATE TABLE customers (\n" +
                "    ID INT,\n" +
                "    FIRST_NAME STRING,\n" +
                "    LAST_NAME STRING,\n" +
                "    EMAIL STRING,\n" +
                "    PRIMARY KEY (ID) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'oracle-cdc',\n" +
                "    'hostname' = 'hostname',\n" +
                "    'port' = '1521',\n" +
                "    'username' = 'c##dbzuser',\n" +
                "    'password' = 'dbz',\n" +
                "    'database-name' = 'HCCDB',\n" +
                "    'schema-name' = 'DEBEZIUM', \n" +
                "    'table-name' = 'DEBEZIUM.CUSTOMERS',\n" +
                "    'debezium.database.connection.adpter' = 'xstream',\n" +
                "    'debezium.database.out.server.name' = 'dbzxout',\n" +
                "    'debezium.database.pdb.name' = 'HCPDB1'\n" +
                "  )");
        tableEnv.executeSql("select * from customers").print();
        env.execute();
    }

问题

1. 使用SQL语法未能正常捕获数据库变化
   

2. 使用Flink sql-client.sh 执行SQL提示缺少DebeziumSourceFunction

[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: com.ververica.cdc.debezium.DebeziumSourceFunction

当前 Flink1.13.3 lib目录

-rwxrwxrwx 1 root root 102K Dec 15 14:48     connect-api-2.7.0.jar
-rwxrwxrwx 1 root root  86K Dec 15 14:33      flink-connector-debezium-2.2-SNAPSHOT.jar
-rwxrwxrwx 1 root root  91K Oct 13 00:15       flink-csv-1.13.3.jar
-rwxrwxrwx 1 root root 111M Oct 13 00:29     flink-dist_2.11-1.13.3.jar
-rwxrwxrwx 1 root root 145K Oct 13 00:13     flink-json-1.13.3.jar
-rwxrwxrwx 1 root root 7.4M Jun  8  2021       flink-shaded-zookeeper-3.4.14.jar
-rwxrwxrwx 1 root root  22M Dec 15 14:00     flink-sql-connector-oracle-cdc-2.2-SNAPSHOT.jar
-rwxrwxrwx 1 root root  35M Oct 13 00:25     flink-table_2.11-1.13.3.jar
-rwxrwxrwx 1 root root  40M Oct 13 00:26     flink-table-blink_2.11-1.13.3.jar
-rwxrwxrwx 1 root root  66K Mar 31  2021      log4j-1.2-api-2.12.1.jar
-rwxrwxrwx 1 root root 271K Mar 31  2021      log4j-api-2.12.1.jar
-rwxrwxrwx 1 root root 1.6M Mar 31  2021      log4j-core-2.12.1.jar
-rwxrwxrwx 1 root root  23K Mar 31  2021      log4j-slf4j-impl-2.12.1.jar
-rw-r--r--    1 root root  73K Dec 17 17:12       xstreams.jar

[LiaoMinYu]

为什么我的一直报这个 java.lang.UnsatisfiedLinkError: no ocijdbc19 in java.library.path，不过我用的是flink sql client flink lib中都已经放了xstream.jar、ojdbc8.jar，以及/etc/profile配置了环境变量，大佬有遇到过这问题吗

[DKodak]

这是pom.xml
`


org.springframework.boot
spring-boot-starter

	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-test</artifactId>
		<scope>test</scope>
	</dependency>

	<dependency>
		<groupId>org.apache.flink</groupId>
		<artifactId>flink-core</artifactId>
		<version>1.13.3</version>
	</dependency>
	<dependency>
		<groupId>org.apache.flink</groupId>
		<artifactId>flink-clients_2.11</artifactId>
		<version>1.13.3</version>
	</dependency>
	<dependency>
		<groupId>org.apache.flink</groupId>
		<artifactId>flink-table-common</artifactId>
		<version>1.13.3</version>
	</dependency>
	<dependency>
		<groupId>org.apache.flink</groupId>
		<artifactId>flink-table-api-java-bridge_2.11</artifactId>
		<version>1.13.3</version>
	</dependency>
	<dependency>
		<groupId>org.apache.flink</groupId>
		<artifactId>flink-table-planner-blink_2.11</artifactId>
		<version>1.13.3</version>
	</dependency>
	<dependency>
		<groupId>org.apache.flink</groupId>
		<artifactId>flink-streaming-java_2.11</artifactId>
		<version>1.13.3</version>
	</dependency>
	<dependency>
		<groupId>com.ververica</groupId>
		<artifactId>flink-connector-oracle-cdc</artifactId>
		<version>2.2.0</version>
	</dependency>
	<dependency>
		<groupId>com.oracle.ojdbc</groupId>
		<artifactId>ojdbc10</artifactId>
		<version>19.3.0.0</version>
	</dependency>
	<dependency>
		<groupId>org.apache.flink</groupId>
		<artifactId>flink-connector-kafka_2.11</artifactId>
		<version>1.13.5</version>
		<exclusions>
			<exclusion>
				<artifactId>kafka-clients</artifactId>
				<groupId>org.apache.kafka</groupId>
			</exclusion>
		</exclusions>
	</dependency>

	<dependency>
		<groupId>xstream</groupId>
		<artifactId>xstream</artifactId>
		<version>1.1</version>
	</dependency>
</dependencies>`

这是demo

` public static void main(String[] args) throws Exception {

    Properties deProps = new Properties();
    deProps.setProperty("database.connection.adapter", "xstream");
    //deProps.setProperty("database.out.server.name", "dbzxout");
    //deProps.setProperty("database.pdb.name", "HCPDB1");
    SourceFunction<String> sourceFunction = OracleSource.<String>builder()
      
            .hostname("localhost")
            .port(1521)
            .database("helowin")
            // monitor inventory schema//表头
            .schemaList("a")
            //.tableList("inventory.products") 
            .tableList("a.ITEMS")
            .username("root")
            .password("123456")
            .debeziumProperties(deProps)
            .deserializer(new JsonDebeziumDeserializationSchema())
            .build();

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.addSource(sourceFunction).print().setParallelism(1);
    env.execute();
}`

执行后
15:44:40.362 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job Flink Streaming Job (2f9dae8548bb1729c45c98f1dc60d58c) switched from state RUNNING to FAILING. org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138) at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82) at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:216) at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:206) at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:197) at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:682) at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79) at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:435) 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$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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: org.apache.kafka.connect.errors.ConnectException: Error configuring an instance of OracleConnectorTask; check the logs for details at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:120) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759) at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:188) 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) 15:44:40.367 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job Flink Streaming Job (2f9dae8548bb1729c45c98f1dc60d58c) switched from state FAILING to FAILED. org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138) at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82) at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:216) at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:206) at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:197) at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:682) at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79) at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:435) 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$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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: org.apache.kafka.connect.errors.ConnectException: Error configuring an instance of OracleConnectorTask; check the logs for details at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:120) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759) at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:188) 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) 15:44:40.370 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.executiongraph.ExecutionGraph - ExecutionGraph 2f9dae8548bb1729c45c98f1dc60d58c reached terminal state FAILED. 15:44:40.370 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Stopping checkpoint coordinator for job 2f9dae8548bb1729c45c98f1dc60d58c. 15:44:40.376 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore - Shutting down 15:44:40.382 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.jobmaster.JobMaster - Archive global failure. org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138) at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82) at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:216) at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:206) at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:197) at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:682) at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79) at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:435) 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$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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: org.apache.kafka.connect.errors.ConnectException: Error configuring an instance of OracleConnectorTask; check the logs for details at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:120) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759) at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:188) 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) 15:44:40.386 [jobmanager-future-thread-1] DEBUG org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess - Job 2f9dae8548bb1729c45c98f1dc60d58c under leader id 43ca8930-bcbd-4bc5-afcc-e4f3c33a663a reached a globally terminal state FAILED. 15:44:40.387 [jobmanager-future-thread-1] DEBUG org.apache.flink.runtime.jobmaster.JobMasterServiceLeadershipRunner - Completing the result for job 2f9dae8548bb1729c45c98f1dc60d58c. 15:44:40.389 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.minicluster.MiniCluster - Shutting down Flink Mini Cluster 15:44:40.390 [flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Job 2f9dae8548bb1729c45c98f1dc60d58c reached terminal state FAILED. org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138) at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82) at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:216) at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:206) at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:197) at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:682) at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79) at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:435) 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$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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: org.apache.kafka.connect.errors.ConnectException: Error configuring an instance of OracleConnectorTask; check the logs for details at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:120) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759) at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:188) 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) 15:44:40.390 [flink-akka.actor.default-dispatcher-9] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Stopping TaskExecutor akka://flink/user/rpc/taskmanager_0. 15:44:40.390 [flink-akka.actor.default-dispatcher-6] INFO org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint - Shutting down rest endpoint. 15:44:40.390 [flink-akka.actor.default-dispatcher-9] DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor - Close ResourceManager connection 7578d69d2c92a59db21bb8674e2e6ab8. org.apache.flink.util.FlinkException: The TaskExecutor is shutting down. at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:438) at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:214) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StartedState.terminate(AkkaRpcActor.java:563) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleControlMessage(AkkaRpcActor.java:186) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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) 15:44:40.390 [flink-akka.actor.default-dispatcher-3] DEBUG org.apache.flink.runtime.jobmaster.JobMasterServiceLeadershipRunner - Terminating the leadership runner for job 2f9dae8548bb1729c45c98f1dc60d58c. 15:44:40.390 [flink-akka.actor.default-dispatcher-3] DEBUG org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess - Terminating the JobMasterService process for job 2f9dae8548bb1729c45c98f1dc60d58c under leader id 43ca8930-bcbd-4bc5-afcc-e4f3c33a663a. 15:44:40.391 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager - Closing TaskExecutor connection 0b6ae8cc-8c45-4961-b33b-595a4b6053c2 because: The TaskExecutor is shutting down. 15:44:40.391 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager - Unregistering task executor 62ccaf28db91fb30955f491e7557285e from the slot manager. 15:44:40.391 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.resourcemanager.slotmanager.DefaultResourceTracker - Stopping tracking of resources for job 2f9dae8548bb1729c45c98f1dc60d58c. 15:44:40.391 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.io.network.partition.ResourceManagerPartitionTrackerImpl - Processing shutdown of task executor 0b6ae8cc-8c45-4961-b33b-595a4b6053c2. 15:44:40.392 [flink-akka.actor.default-dispatcher-9] DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor - Close JobManager connection for job 2f9dae8548bb1729c45c98f1dc60d58c. org.apache.flink.util.FlinkException: The TaskExecutor is shutting down. at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:438) at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:214) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StartedState.terminate(AkkaRpcActor.java:563) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleControlMessage(AkkaRpcActor.java:186) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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) 15:44:40.393 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.jobmaster.JobMaster - Stopping the JobMaster for job Flink Streaming Job(2f9dae8548bb1729c45c98f1dc60d58c). 15:44:40.394 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.scheduler.SlotSharingExecutionSlotAllocator - There is no SharedSlot for ExecutionSlotSharingGroup of ExecutionVertexID cbc357ccb763df2852fee8c4fc7d55f2_0 15:44:40.395 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.jobmaster.JobMaster - Disconnect TaskExecutor 0b6ae8cc-8c45-4961-b33b-595a4b6053c2 because: Stopping JobMaster for job Flink Streaming Job(2f9dae8548bb1729c45c98f1dc60d58c). 15:44:40.396 [flink-akka.actor.default-dispatcher-9] DEBUG org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl - Free slot TaskSlot(index:0, state:ALLOCATED, resource profile: ResourceProfile{taskHeapMemory=256.000gb (274877906944 bytes), taskOffHeapMemory=256.000gb (274877906944 bytes), managedMemory=32.000mb (33554432 bytes), networkMemory=16.000mb (16777216 bytes)}, allocationId: 319aacb2ce0cfca86953d2199aeab0cb, jobId: 2f9dae8548bb1729c45c98f1dc60d58c). org.apache.flink.util.FlinkException: Closing task slot table at org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl.closeAsync(TaskSlotTableImpl.java:173) at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:455) at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:214) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StartedState.terminate(AkkaRpcActor.java:563) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleControlMessage(AkkaRpcActor.java:186) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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) 15:44:40.396 [flink-akka.actor.default-dispatcher-10] INFO org.apache.flink.runtime.jobmaster.slotpool.DefaultDeclarativeSlotPool - Releasing slot [319aacb2ce0cfca86953d2199aeab0cb]. org.apache.flink.util.FlinkException: Stopping JobMaster for job Flink Streaming Job(2f9dae8548bb1729c45c98f1dc60d58c). at org.apache.flink.runtime.jobmaster.JobMaster.onStop(JobMaster.java:400) at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:214) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StartedState.terminate(AkkaRpcActor.java:563) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleControlMessage(AkkaRpcActor.java:186) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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) 15:44:40.398 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.jobmaster.JobMaster - Close ResourceManager connection 7578d69d2c92a59db21bb8674e2e6ab8. org.apache.flink.util.FlinkException: Stopping JobMaster for job Flink Streaming Job(2f9dae8548bb1729c45c98f1dc60d58c). at org.apache.flink.runtime.jobmaster.JobMaster.onStop(JobMaster.java:400) at org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:214) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StartedState.terminate(AkkaRpcActor.java:563) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleControlMessage(AkkaRpcActor.java:186) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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) 15:44:40.399 [flink-akka.actor.default-dispatcher-3] DEBUG org.apache.flink.runtime.resourcemanager.DefaultJobLeaderIdService - Remove job 2f9dae8548bb1729c45c98f1dc60d58c from job leader id monitoring. 15:44:40.400 [flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.resourcemanager.StandaloneResourceManager - Disconnect job manager afcce4f3c33a663a43ca8930bcbd4bc5@akka://flink/user/rpc/jobmanager_3 for job 2f9dae8548bb1729c45c98f1dc60d58c from the resource manager. 15:44:40.401 [flink-akka.actor.default-dispatcher-3] DEBUG org.apache.flink.runtime.resourcemanager.slotmanager.DefaultResourceTracker - Initiating tracking of resources for job 2f9dae8548bb1729c45c98f1dc60d58c. 15:44:40.401 [flink-akka.actor.default-dispatcher-3] DEBUG org.apache.flink.runtime.resourcemanager.slotmanager.DefaultResourceTracker - Stopping tracking of resources for job 2f9dae8548bb1729c45c98f1dc60d58c. 15:44:40.402 [flink-akka.actor.default-dispatcher-9] DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor - Free slot with allocation id 319aacb2ce0cfca86953d2199aeab0cb because: Stopping JobMaster for job Flink Streaming Job(2f9dae8548bb1729c45c98f1dc60d58c). 15:44:40.402 [flink-akka.actor.default-dispatcher-9] DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor - Could not free slot for allocation id 319aacb2ce0cfca86953d2199aeab0cb. org.apache.flink.runtime.taskexecutor.slot.SlotNotFoundException: Could not find slot for 319aacb2ce0cfca86953d2199aeab0cb. at org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl.freeSlot(TaskSlotTableImpl.java:429) at org.apache.flink.runtime.taskexecutor.TaskExecutor.freeSlotInternal(TaskExecutor.java:1906) at org.apache.flink.runtime.taskexecutor.TaskExecutor.freeSlot(TaskExecutor.java:1125) 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.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$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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) 15:44:40.402 [flink-akka.actor.default-dispatcher-9] DEBUG org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager - Releasing local state under allocation id 319aacb2ce0cfca86953d2199aeab0cb. 15:44:40.403 [flink-akka.actor.default-dispatcher-10] DEBUG org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor - The RpcEndpoint jobmanager_3 terminated successfully. 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.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:137) at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616) at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975) at org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$0(AkkaInvocationHandler.java:237) 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.concurrent.FutureUtils$1.onComplete(FutureUtils.java:1081) at akka.dispatch.OnComplete.internal(Future.scala:264) at akka.dispatch.OnComplete.internal(Future.scala:261) at akka.dispatch.japi$CallbackBridge.apply(Future.scala:191) at akka.dispatch.japi$CallbackBridge.apply(Future.scala:188) at scala.concurrent.impl.CallbackRunnable.run$$$capture(Promise.scala:36) at scala.concurrent.impl.CallbackRunnable.run(Promise.scala) at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:73) at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44) at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252) at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:572) at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:22) at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:21) at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:436) at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:435) at scala.concurrent.impl.CallbackRunnable.run$$$capture(Promise.scala:36) at scala.concurrent.impl.CallbackRunnable.run(Promise.scala) at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55) at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91) at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90) at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40) at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:44) 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: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138) at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82) at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:216) at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:206) at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:197) at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:682) at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79) at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:435) 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$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:517) at akka.actor.Actor$class.aroundReceive(Actor.scala) 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) ... 4 more Caused by: org.apache.kafka.connect.errors.ConnectException: Error configuring an instance of OracleConnectorTask; check the logs for details at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:120) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759) at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:188) 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)

[gujincheng]

我这边也遇到这个情况， oracle-cdc使用 xstream 可以采集存量的数据，但是，后面 数据的更新无法捕捉到，请问你们解决了吗？是什么原因？