[jira] [Created] (FLINK-20542) flink-1.11.2 - stream join stream savepoint fail

[Shang Yuanchun (Jira)]

谢波 created FLINK-20542:
--------------------------

             Summary: flink-1.11.2 - stream join stream savepoint fail
                 Key: FLINK-20542
                 URL: https://issues.apache.org/jira/browse/FLINK-20542
             Project: Flink
          Issue Type: Bug
         Environment: <flink.version>1.11.2</flink.version>
<scala.binary.version>2.11</scala.binary.version>

{color:#FF0000}tEnv.getConfig.setIdleStateRetentionTime(Time.hours(12), Time.hours(24)){color}

env.enableCheckpointing(2 * 60 * 1000L)
val checkpointConfig = env.getCheckpointConfig
checkpointConfig.setCheckpointTimeout(20 * 60 * 1000)
checkpointConfig.setTolerableCheckpointFailureNumber(2)

 

{color:#FF0000}env.setStateBackend({color}
{color:#FF0000} new RocksDBStateBackend(checkpointPath, true){color}
{color:#FF0000} .asInstanceOf[StateBackend]{color}
{color:#FF0000}){color}
            Reporter: 谢波


I have a SQL job that uses stream join stream, but at savepoint, the job fails.

I can't manually trigger savepoint right now because the task will die once it's triggered.

Can anyone help?Thanks

 

!image-2020-12-09-10-59-58-757.png!

 

2020-12-09 10:46:21,801 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Found Yarn properties file under /tmp/.yarn-properties-xiebo.2020-12-09 10:46:21,801 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Found Yarn properties file under /tmp/.yarn-properties-xiebo.2020-12-09 10:46:21,801 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Found Yarn properties file under /tmp/.yarn-properties-xiebo.2020-12-09 10:46:22,024 WARN  org.apache.flink.yarn.configuration.YarnLogConfigUtil        [] - The configuration directory ('/home/xiebo/module/flink/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.2020-12-09 10:46:22,414 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar2020-12-09 10:46:22,575 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface shd183.yonghui.cn:45855 of application 'application_1594105654926_6844100'.Triggering savepoint for job 749a7a45d6324f57ac859d3eab55a56c.Waiting for response...
------------------------------------------------------------ The program finished with the following exception:
org.apache.flink.util.FlinkException: Triggering a savepoint for the job 749a7a45d6324f57ac859d3eab55a56c failed. at org.apache.flink.client.cli.CliFrontend.triggerSavepoint(CliFrontend.java:668) at org.apache.flink.client.cli.CliFrontend.lambda$savepoint$9(CliFrontend.java:646) at org.apache.flink.client.cli.CliFrontend.runClusterAction(CliFrontend.java:864) at org.apache.flink.client.cli.CliFrontend.savepoint(CliFrontend.java:643) at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:934) at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:992) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:992)Caused by: java.util.concurrent.CompletionException: java.util.concurrent.CompletionException: org.apache.flink.runtime.checkpoint.CheckpointException: The job has failed. at org.apache.flink.runtime.scheduler.SchedulerBase.lambda$triggerSavepoint$3(SchedulerBase.java:764) at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:836) at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:811) at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:456) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:402) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:195) at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152) 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(Actor.scala:517) 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: java.util.concurrent.CompletionException: org.apache.flink.runtime.checkpoint.CheckpointException: The job has failed. at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292) at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308) at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:607) at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1990) at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.lambda$null$0(CheckpointCoordinator.java:467) 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.completeExceptionally(CompletableFuture.java:1990) at org.apache.flink.runtime.checkpoint.PendingCheckpoint.abort(PendingCheckpoint.java:494) at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.abortPendingCheckpoint(CheckpointCoordinator.java:1663) at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.receiveDeclineMessage(CheckpointCoordinator.java:932) at org.apache.flink.runtime.scheduler.SchedulerBase.lambda$declineCheckpoint$5(SchedulerBase.java:827) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) 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: org.apache.flink.runtime.checkpoint.CheckpointException: The job has failed. at org.apache.flink.runtime.checkpoint.PendingCheckpoint.abort(PendingCheckpoint.java:493) ... 10 moreCaused by: java.lang.Exception: Could not materialize checkpoint 439 for operator Join(joinType=[LeftOuterJoin], where=[(transNumber = transNumber0)], select=[transNumber, retailNumber, werks, businessDayDate, goodsId, retailQuantity, salesAmount, rtPlu, updateTime, keyBy, busi_grp_id, busi_grp_name, sales_dist, sales_dist_name, region_id_new, region_name_new, prov_code, prov_name, city_code, city_name, shop_name, vendor_id, vendor_name, firm_id, firm_name, div_id, dept_id, dept_name, catg_l_id, catg_l_name, catg_m_id, catg_m_name, catg_s_id, catg_s_name, goodsname, brand, brand_name, operation_type_id, operation_type_name, efct_sign_id, efct_sign_name, shop_goods_sts_id, shop_goods_sts_name, bravo_region_id, bravo_region_name, shop_belong, shop_belong_desc, zone_id, zone_name, bd_id, bd_name, firm_g1_id, firm_g1_name, firm_g2_id, firm_g2_name, pur_tax_rate, cost_price, vendor_name_new, vendor_id_new, transNumber0, returnFlag, channelFlag, commodityType, endTimestamp], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey]) -> Calc(select=[businessDayDate AS sdt, werks AS shopid, CAST(((CAST((endTimestamp IS NULL IF updateTime IF endTimestamp)) UNIX_TIMESTAMP _UTF-16LE'yyyy-MM-dd HH:mm:ss.SSS') * 1000)) AS updatetime, retailNumber AS serialid, transNumber AS sheetid, goodsId AS goodsid, keyBy AS key_by, CAST((endTimestamp IS NULL IF (FLAG(HOUR) EXTRACT updateTime) IF (FLAG(HOUR) EXTRACT endTimestamp))) AS timeframe, region_id_new AS regionid, region_name_new AS regionname, shop_name AS shopname, bravo_region_id AS serviceregionid, bravo_region_name AS serviceregionname, shop_belong AS shopbelongid, shop_belong_desc AS shopbelongname, sales_dist AS distid, sales_dist_name AS distname, prov_code AS provinceid, prov_name AS provincename, city_code AS cityid, city_name AS cityname, rtPlu AS pluid, salesAmount AS salevalue, commodityType AS commoditytype, CAST(_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS memnum, busi_grp_id AS groupid, busi_grp_name AS groupname, div_id AS categoryid, goodsname, (((retailQuantity * cost_price) * (1 + (pur_tax_rate / 100))) ROUND 2) AS cost, retailQuantity AS amount, CAST(channelFlag) AS channelflag, CAST(returnFlag) AS isreturn, firm_id AS firmid, firm_name AS firmname, dept_id AS deptid, dept_name AS deptname, brand, brand_name AS brandname, operation_type_id AS operationtypeid, operation_type_name AS operationtypename, efct_sign_id AS efctsignid, efct_sign_name AS efctsignname, shop_goods_sts_id AS shopgoodsstsid, shop_goods_sts_name AS shopgoodsstsname, vendor_id AS vendorid, vendor_name AS vendorname, vendor_id_new AS vendoridnew, vendor_name_new AS vendornamenew, catg_l_id AS catglid, catg_l_name AS catglname, catg_m_id AS catgmid, catg_m_name AS catgmname, catg_s_id AS catgsid, catg_s_name AS catgsname, zone_id AS zoneid, zone_name AS zonename, bd_id AS bdid, bd_name AS bdname, firm_g1_id AS firmg1id, firm_g1_name AS firmg1name, firm_g2_id AS firmg2id, firm_g2_name AS firmg2name, ((cost_price * (1 + (pur_tax_rate / 100))) ROUND 2) AS cycleunitprice, businessDayDate AS saledate]) (1/10). at org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.handleExecutionException(AsyncCheckpointRunnable.java:191) at org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.run(AsyncCheckpointRunnable.java:138) ... 3 moreCaused by: java.util.concurrent.CancellationException at java.util.concurrent.FutureTask.report(FutureTask.java:121) at java.util.concurrent.FutureTask.get(FutureTask.java:192) at org.apache.flink.runtime.concurrent.FutureUtils.runIfNotDoneAndGet(FutureUtils.java:479) at org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer.<init>(OperatorSnapshotFinalizer.java:50) at org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable.run(AsyncCheckpointRunnable.java:102) ... 3 more



--
This message was sent by Atlassian Jira
(v8.3.4#803005)