flink-cdc icon indicating copy to clipboard operation
flink-cdc copied to clipboard

Published 20 hours ago • verified publisher icon apache

Oracle Logminer 正常输出数据 异常关闭

Open TYangXiaoChen opened this issue 3 years ago • 10 comments

Describe the bug A clear and concise description of what the bug is.

监控Oracle Logminer 正常输出 初始化数据后，程序异常关闭

报错： Caused by: java.lang.IllegalStateException: Retrieve schema history failed, the schema records for engine 90d183a3-f783-4ed3-bd07-e3d807ca8bb5 has been removed, this might because the debezium engine has been shutdown due to other errors.

Environment :

  • Flink version : 1.13.3
  • Flink CDC version: 2.1
  • Database and version: Oracle 11g 只读

To Reproduce Steps to reproduce the behavior:

  1. Thes test data :
  2. The test code :
  3. The error :

Additional Description If applicable, add screenshots to help explain your

源码中 DatabaseHistoryUtil#HISTORY_CLEANUP_STATUS 变量 engineName 未被清掉

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:602) at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) at org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$0(AkkaInvocationHandler.java:237) at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) 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(Promise.scala:60) at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:73) at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:68) at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:284) at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:284) at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:284) at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:573) 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(Future.scala:532) at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:29) at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:29) at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60) at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55) at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:91) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12) at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:81) at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:91) 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.GeneratedMethodAccessor6.invoke(Unknown Source) 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) ... 4 more Caused by: java.lang.IllegalStateException: Retrieve schema history failed, the schema records for engine fec54704-e416-4dad-ab34-28b42fafd2ce has been removed, this might because the debezium engine has been shutdown due to other errors. at com.ververica.cdc.debezium.utils.DatabaseHistoryUtil.retrieveHistory(DatabaseHistoryUtil.java:77) at com.ververica.cdc.debezium.internal.FlinkDatabaseSchemaHistory.configure(FlinkDatabaseSchemaHistory.java:82) at io.debezium.relational.HistorizedRelationalDatabaseConnectorConfig.getDatabaseHistory(HistorizedRelationalDatabaseConnectorConfig.java:105) at io.debezium.relational.HistorizedRelationalDatabaseSchema.(HistorizedRelationalDatabaseSchema.java:39) at io.debezium.connector.oracle.OracleDatabaseSchema.(OracleDatabaseSchema.java:35) at io.debezium.connector.oracle.OracleConnectorTask.start(OracleConnectorTask.java:54) at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:130) at io.debezium.connector.common.BaseSourceTask.startIfNeededAndPossible(BaseSourceTask.java:207) at io.debezium.connector.common.BaseSourceTask.poll(BaseSourceTask.java:148) at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:788) 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:748)

TYangXiaoChen avatar Jan 21 '22 10:01 TYangXiaoChen

i'v the same problem,have you solved it?

lidong2021 avatar Jan 26 '22 10:01 lidong2021

俺也一样，请问你们解决了吗？

mengyejiang avatar Mar 02 '22 07:03 mengyejiang

i'v the same problem,have you solved it?

TigerBossAA avatar Mar 10 '22 07:03 TigerBossAA

Could you use English for better communication? feel free to reopen once updated the description.

leonardBang avatar Mar 14 '22 08:03 leonardBang

I also encounter this problem, how to solve it?

hehetown avatar May 13 '22 03:05 hehetown

I also encounter this problem, how to solve it?

yqzdy1207 avatar Aug 08 '22 06:08 yqzdy1207

I also encounter this problem, how to solve it?

wdzhanghao avatar Aug 09 '22 05:08 wdzhanghao

I tried several versions, flink version is 1.13.1 and cdc version is 2.2.1 oracle database version is 11g and flink version is 1.13.1 and cdc version is 2.1.1 and flink version is 1.13.6 cdc version is 2.1.0 is to read oracle data incrementally. This error will be reported after more than ten hours. Is there any good solution? If so, please leave a message

wdzhanghao avatar Aug 11 '22 01:08 wdzhanghao

I also encounter this problem, how to solve it?

Mindfaker avatar Aug 28 '23 01:08 Mindfaker

Considering collaboration with developers around the world, please re-create your issue in English on Apache Jira under project Flink with component tag Flink CDC. Thank you! cc @GOODBOY008

gong avatar Feb 03 '24 06:02 gong

Closing this issue because it was created before version 2.3.0 (2022-11-10). Please try the latest version of Flink CDC to see if the issue has been resolved. If the issue is still valid, kindly report it on Apache Jira under project Flink with component tag Flink CDC. Thank you!

PatrickRen avatar Feb 28 '24 15:02 PatrickRen