[
https://issues.apache.org/jira/browse/FLINK-10735?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Fei Feng updated FLINK-10735:
-----------------------------
Environment:
Hadoop 2.7
flink 1.6.2
Component/s: TaskManager
ResourceManager
> flink on yarn close container exception
> ---------------------------------------
>
> Key: FLINK-10735
> URL: https://issues.apache.org/jira/browse/FLINK-10735
> Project: Flink
> Issue Type: Bug
> Components: ResourceManager, TaskManager, YARN
> Affects Versions: 1.6.2
> Environment: Hadoop 2.7
> flink 1.6.2
> Reporter: Fei Feng
> Priority: Critical
>
> flink on yarn with detached mode, when cancle flink job，yarn resource release
> very slow
> Log：
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Job
> 32F01A0FC50EFE8F4794AD0C45678EC4: xxx switched from state RUNNING to
> CANCELLING.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Source:
> Kafka09TableSource(SS_SOLD_DATE_SK, SS_SOLD_TIME_SK, SS_ITEM_SK,
> SS_CUSTOMER_SK, SS_CDEMO_SK, SS_HDEMO_SK, SS_ADDR_SK, SS_STORE_SK,
> SS_PROMO_SK, SS_TICKET_NUMBER, SS_QUANTITY, SS_WHOLESALE_COST, SS_LIST_PRICE,
> SS_SALES_PRICE, SS_EXT_DISCOUNT_AMT, SS_EXT_SALES_PRICE,
> SS_EXT_WHOLESALE_COST, SS_EXT_LIST_PRICE, SS_EXT_TAX, SS_COUPON_AMT,
> SS_NET_PAID, SS_NET_PAID_INC_TAX, SS_NET_PROFIT, ROWTIME) -> from:
> (SS_WHOLESALE_COST, SS_SALES_PRICE, SS_COUPON_AMT, SS_NET_PAID,
> SS_NET_PROFIT, ROWTIME) -> Timestamps/Watermarks -> where: (>(SS_COUPON_AMT,
> 0)), select: (ROWTIME, SS_WHOLESALE_COST, SS_SALES_PRICE, SS_NET_PAID,
> SS_NET_PROFIT) -> time attribute: (ROWTIME) (1/3)
> (0807b5f291f897ac4545dbfdb8ec3448) switched from RUNNING to CANCELING.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Source:
> Kafka09TableSource(SS_SOLD_DATE_SK, SS_SOLD_TIME_SK, SS_ITEM_SK,
> SS_CUSTOMER_SK, SS_CDEMO_SK, SS_HDEMO_SK, SS_ADDR_SK, SS_STORE_SK,
> SS_PROMO_SK, SS_TICKET_NUMBER, SS_QUANTITY, SS_WHOLESALE_COST, SS_LIST_PRICE,
> SS_SALES_PRICE, SS_EXT_DISCOUNT_AMT, SS_EXT_SALES_PRICE,
> SS_EXT_WHOLESALE_COST, SS_EXT_LIST_PRICE, SS_EXT_TAX, SS_COUPON_AMT,
> SS_NET_PAID, SS_NET_PAID_INC_TAX, SS_NET_PROFIT, ROWTIME) -> from:
> (SS_WHOLESALE_COST, SS_SALES_PRICE, SS_COUPON_AMT, SS_NET_PAID,
> SS_NET_PROFIT, ROWTIME) -> Timestamps/Watermarks -> where: (>(SS_COUPON_AMT,
> 0)), select: (ROWTIME, SS_WHOLESALE_COST, SS_SALES_PRICE, SS_NET_PAID,
> SS_NET_PROFIT) -> time attribute: (ROWTIME) (2/3)
> (a56a70eb6807dacf18fbf272ee6160e2) switched from RUNNING to CANCELING.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Source:
> Kafka09TableSource(SS_SOLD_DATE_SK, SS_SOLD_TIME_SK, SS_ITEM_SK,
> SS_CUSTOMER_SK, SS_CDEMO_SK, SS_HDEMO_SK, SS_ADDR_SK, SS_STORE_SK,
> SS_PROMO_SK, SS_TICKET_NUMBER, SS_QUANTITY, SS_WHOLESALE_COST, SS_LIST_PRICE,
> SS_SALES_PRICE, SS_EXT_DISCOUNT_AMT, SS_EXT_SALES_PRICE,
> SS_EXT_WHOLESALE_COST, SS_EXT_LIST_PRICE, SS_EXT_TAX, SS_COUPON_AMT,
> SS_NET_PAID, SS_NET_PAID_INC_TAX, SS_NET_PROFIT, ROWTIME) -> from:
> (SS_WHOLESALE_COST, SS_SALES_PRICE, SS_COUPON_AMT, SS_NET_PAID,
> SS_NET_PROFIT, ROWTIME) -> Timestamps/Watermarks -> where: (>(SS_COUPON_AMT,
> 0)), select: (ROWTIME, SS_WHOLESALE_COST, SS_SALES_PRICE, SS_NET_PAID,
> SS_NET_PROFIT) -> time attribute: (ROWTIME) (3/3)
> (a2eca3dc06087dfdaf3fd0200e545cc4) switched from RUNNING to CANCELING.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: window:
> (TumblingGroupWindow('w$, 'ROWTIME, 3600000.millis)), select:
> (SUM(SS_WHOLESALE_COST) AS EXPR$1, SUM(SS_SALES_PRICE) AS EXPR$2,
> SUM(SS_NET_PAID) AS EXPR$3, SUM(SS_NET_PROFIT) AS EXPR$4, start('w$) AS
> w$start, end('w$) AS w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS
> w$proctime) -> where: (>(EXPR$4, 1000)), select: (CAST(w$start) AS WSTART,
> EXPR$1, EXPR$2, EXPR$3, EXPR$4) -> to: Row (1/1)
> (9fa3592c74eda97124a033b6afea6c87) switched from RUNNING to CANCELING.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Sink:
> JDBCAppendTableSink(RS_DAY, RS_WHOLESALE_COST, RS_SALES_PRICE, RS_NET_PAID,
> RS_NET_PROFIT) (1/3) (6d4f413ace1206714566b610e5bf47b6) switched from RUNNING
> to CANCELING.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Sink:
> JDBCAppendTableSink(RS_DAY, RS_WHOLESALE_COST, RS_SALES_PRICE, RS_NET_PAID,
> RS_NET_PROFIT) (2/3) (8871efc7c19abae7f833e02aa1d2107d) switched from RUNNING
> to CANCELING.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Sink:
> JDBCAppendTableSink(RS_DAY, RS_WHOLESALE_COST, RS_SALES_PRICE, RS_NET_PAID,
> RS_NET_PROFIT) (3/3) (fcaf0b9247d60a31e3381d70871d929f) switched from RUNNING
> to CANCELING.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Source:
> Kafka09TableSource(SS_SOLD_DATE_SK, SS_SOLD_TIME_SK, SS_ITEM_SK,
> SS_CUSTOMER_SK, SS_CDEMO_SK, SS_HDEMO_SK, SS_ADDR_SK, SS_STORE_SK,
> SS_PROMO_SK, SS_TICKET_NUMBER, SS_QUANTITY, SS_WHOLESALE_COST, SS_LIST_PRICE,
> SS_SALES_PRICE, SS_EXT_DISCOUNT_AMT, SS_EXT_SALES_PRICE,
> SS_EXT_WHOLESALE_COST, SS_EXT_LIST_PRICE, SS_EXT_TAX, SS_COUPON_AMT,
> SS_NET_PAID, SS_NET_PAID_INC_TAX, SS_NET_PROFIT, ROWTIME) -> from:
> (SS_WHOLESALE_COST, SS_SALES_PRICE, SS_COUPON_AMT, SS_NET_PAID,
> SS_NET_PROFIT, ROWTIME) -> Timestamps/Watermarks -> where: (>(SS_COUPON_AMT,
> 0)), select: (ROWTIME, SS_WHOLESALE_COST, SS_SALES_PRICE, SS_NET_PAID,
> SS_NET_PROFIT) -> time attribute: (ROWTIME) (3/3)
> (a2eca3dc06087dfdaf3fd0200e545cc4) switched from CANCELING to CANCELED.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Sink:
> JDBCAppendTableSink(RS_DAY, RS_WHOLESALE_COST, RS_SALES_PRICE, RS_NET_PAID,
> RS_NET_PROFIT) (1/3) (6d4f413ace1206714566b610e5bf47b6) switched from
> CANCELING to CANCELED.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Source:
> Kafka09TableSource(SS_SOLD_DATE_SK, SS_SOLD_TIME_SK, SS_ITEM_SK,
> SS_CUSTOMER_SK, SS_CDEMO_SK, SS_HDEMO_SK, SS_ADDR_SK, SS_STORE_SK,
> SS_PROMO_SK, SS_TICKET_NUMBER, SS_QUANTITY, SS_WHOLESALE_COST, SS_LIST_PRICE,
> SS_SALES_PRICE, SS_EXT_DISCOUNT_AMT, SS_EXT_SALES_PRICE,
> SS_EXT_WHOLESALE_COST, SS_EXT_LIST_PRICE, SS_EXT_TAX, SS_COUPON_AMT,
> SS_NET_PAID, SS_NET_PAID_INC_TAX, SS_NET_PROFIT, ROWTIME) -> from:
> (SS_WHOLESALE_COST, SS_SALES_PRICE, SS_COUPON_AMT, SS_NET_PAID,
> SS_NET_PROFIT, ROWTIME) -> Timestamps/Watermarks -> where: (>(SS_COUPON_AMT,
> 0)), select: (ROWTIME, SS_WHOLESALE_COST, SS_SALES_PRICE, SS_NET_PAID,
> SS_NET_PROFIT) -> time attribute: (ROWTIME) (1/3)
> (0807b5f291f897ac4545dbfdb8ec3448) switched from CANCELING to CANCELED.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Source:
> Kafka09TableSource(SS_SOLD_DATE_SK, SS_SOLD_TIME_SK, SS_ITEM_SK,
> SS_CUSTOMER_SK, SS_CDEMO_SK, SS_HDEMO_SK, SS_ADDR_SK, SS_STORE_SK,
> SS_PROMO_SK, SS_TICKET_NUMBER, SS_QUANTITY, SS_WHOLESALE_COST, SS_LIST_PRICE,
> SS_SALES_PRICE, SS_EXT_DISCOUNT_AMT, SS_EXT_SALES_PRICE,
> SS_EXT_WHOLESALE_COST, SS_EXT_LIST_PRICE, SS_EXT_TAX, SS_COUPON_AMT,
> SS_NET_PAID, SS_NET_PAID_INC_TAX, SS_NET_PROFIT, ROWTIME) -> from:
> (SS_WHOLESALE_COST, SS_SALES_PRICE, SS_COUPON_AMT, SS_NET_PAID,
> SS_NET_PROFIT, ROWTIME) -> Timestamps/Watermarks -> where: (>(SS_COUPON_AMT,
> 0)), select: (ROWTIME, SS_WHOLESALE_COST, SS_SALES_PRICE, SS_NET_PAID,
> SS_NET_PROFIT) -> time attribute: (ROWTIME) (2/3)
> (a56a70eb6807dacf18fbf272ee6160e2) switched from CANCELING to CANCELED.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Sink:
> JDBCAppendTableSink(RS_DAY, RS_WHOLESALE_COST, RS_SALES_PRICE, RS_NET_PAID,
> RS_NET_PROFIT) (2/3) (8871efc7c19abae7f833e02aa1d2107d) switched from
> CANCELING to CANCELED.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: window:
> (TumblingGroupWindow('w$, 'ROWTIME, 3600000.millis)), select:
> (SUM(SS_WHOLESALE_COST) AS EXPR$1, SUM(SS_SALES_PRICE) AS EXPR$2,
> SUM(SS_NET_PAID) AS EXPR$3, SUM(SS_NET_PROFIT) AS EXPR$4, start('w$) AS
> w$start, end('w$) AS w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS
> w$proctime) -> where: (>(EXPR$4, 1000)), select: (CAST(w$start) AS WSTART,
> EXPR$1, EXPR$2, EXPR$3, EXPR$4) -> to: Row (1/1)
> (9fa3592c74eda97124a033b6afea6c87) switched from CANCELING to CANCELED.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Sink:
> JDBCAppendTableSink(RS_DAY, RS_WHOLESALE_COST, RS_SALES_PRICE, RS_NET_PAID,
> RS_NET_PROFIT) (3/3) (fcaf0b9247d60a31e3381d70871d929f) switched from
> CANCELING to CANCELED.
> 18/10/31 19:43:59 INFO executiongraph.ExecutionGraph: Job
> 32F01A0FC50EFE8F4794AD0C45678EC4: insert into Result2
> select cast(tumble_start(rowtime, interval '1' hour) as varchar) as wstart,
> sum(ss_wholesale_cost), sum(ss_sales_price), sum(ss_net_paid),
> sum(ss_net_profit)
> from store_sales
> where ss_coupon_amt > 0
> group by tumble(rowtime, interval '1' hour)
> having sum(ss_net_profit) > 1000 (f0086980bd43a077dcfabd76e996f422) switched
> from state CANCELLING to CANCELED.
> 18/10/31 19:43:59 INFO checkpoint.CheckpointCoordinator: Stopping checkpoint
> coordinator for job f0086980bd43a077dcfabd76e996f422.
> 18/10/31 19:43:59 INFO checkpoint.StandaloneCompletedCheckpointStore:
> Shutting down
> 18/10/31 19:43:59 INFO dispatcher.StandaloneDispatcher: Job
> f0086980bd43a077dcfabd76e996f422 reached globally terminal state CANCELED.
> 18/10/31 19:43:59 INFO jobmaster.JobMaster: Stopping the JobMaster for job
> 32F01A0FC50EFE8F4794AD0C45678EC4: insert into Result2
> select cast(tumble_start(rowtime, interval '1' hour) as varchar) as wstart,
> sum(ss_wholesale_cost), sum(ss_sales_price), sum(ss_net_paid),
> sum(ss_net_profit)
> from store_sales
> where ss_coupon_amt > 0
> group by tumble(rowtime, interval '1' hour)
> having sum(ss_net_profit) > 1000(f0086980bd43a077dcfabd76e996f422).
> 18/10/31 19:43:59 INFO jobmaster.JobMaster: Close ResourceManager connection
> 39b6f62c8f9ef7681eb9f284d893ad0c: JobManager is shutting down..
> 18/10/31 19:43:59 INFO yarn.YarnResourceManager: Disconnect job manager
> [email protected]://flink@node2:43683/user/jobmanager_13
> for job f0086980bd43a077dcfabd76e996f422 from the resource manager.
> 18/10/31 19:44:00 INFO slotpool.SlotPool: Suspending SlotPool.
> 18/10/31 19:44:00 INFO slotpool.SlotPool: Stopping SlotPool.
> 18/10/31 19:44:00 INFO jobmaster.JobManagerRunner: JobManagerRunner already
> shutdown.
> 18/10/31 19:44:01 INFO yarn.YarnResourceManager: Stopping container
> container_1538963842459_0857_01_000017.
> 18/10/31 19:44:01 WARN yarn.YarnResourceManager: Error while calling YARN
> Node Manager to stop container
> org.apache.hadoop.yarn.exceptions.YarnException: Container
> container_1538963842459_0857_01_000017 is not handled by this NodeManager
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
> at
> org.apache.hadoop.yarn.api.records.impl.pb.SerializedExceptionPBImpl.instantiateException(SerializedExceptionPBImpl.java:152)
> at
> org.apache.hadoop.yarn.api.records.impl.pb.SerializedExceptionPBImpl.deSerialize(SerializedExceptionPBImpl.java:106)
> at
> org.apache.hadoop.yarn.client.api.impl.NMClientImpl.stopContainerInternal(NMClientImpl.java:297)
> at
> org.apache.hadoop.yarn.client.api.impl.NMClientImpl.stopContainer(NMClientImpl.java:247)
> at
> org.apache.flink.yarn.YarnResourceManager.stopWorker(YarnResourceManager.java:307)
> at
> org.apache.flink.yarn.YarnResourceManager.stopWorker(YarnResourceManager.java:74)
> at
> org.apache.flink.runtime.resourcemanager.ResourceManager.releaseResource(ResourceManager.java:872)
> at
> org.apache.flink.runtime.resourcemanager.ResourceManager$ResourceActionsImpl.releaseResource(ResourceManager.java:1077)
> at
> org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.checkTaskManagerTimeouts(SlotManager.java:914)
> at
> org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.lambda$null$0(SlotManager.java:195)
> at
> org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:332)
> at
> org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:158)
> at
> org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:70)
> at
> org.apache.flink.runtime.rpc.akka.AkkaRpcActor.onReceive(AkkaRpcActor.java:142)
> at
> org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.onReceive(FencedAkkaRpcActor.java:40)
> at
> akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:165)
> at akka.actor.Actor$class.aroundReceive(Actor.scala:502)
> at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:95)
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
> at akka.actor.ActorCell.invoke(ActorCell.scala:495)
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
> at akka.dispatch.Mailbox.run(Mailbox.scala:224)
> at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> at
> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> 18/10/31 19:44:01 INFO yarn.YarnResourceManager: Closing TaskExecutor
> connection container_1538963842459_0857_01_000017 because: TaskExecutor
> exceeded the idle timeout.
> 18/10/31 19:44:01 WARN yarn.YarnResourceManager: Discard registration from
> TaskExecutor container_1538963842459_0857_01_000017 at
> (akka.tcp://flink@node9:45391/user/taskmanager_0) because the framework did
> not recognize it
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
