You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Yun Gao (Jira)" <ji...@apache.org> on 2021/11/09 16:03:00 UTC
[jira] [Commented] (FLINK-24846) AsyncWaitOperator fails during
stop-with-savepoint
[ https://issues.apache.org/jira/browse/FLINK-24846?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17441253#comment-17441253 ]
Yun Gao commented on FLINK-24846:
---------------------------------
The direct reason for this problem is currently on stopping the StreamTask, the mailbox would be first preClose to forbid adding new mails, then it would drain all the pending mails. However, in this case for the AsyncWaitOperator to produce the last piece of the async task result, it seems to use a recursive way to output the pending completed records: namely for each mail it outputs one record, if there are new records, it would submit more mails, which conflict with the action of preClose before.
Perhaps after we finish reverting the process of stop-with-savepoint to first endInput and emit all the completed elements, and then taking a savepoint and terminate, the issue would be solved?
> AsyncWaitOperator fails during stop-with-savepoint
> --------------------------------------------------
>
> Key: FLINK-24846
> URL: https://issues.apache.org/jira/browse/FLINK-24846
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Task
> Affects Versions: 1.14.0
> Reporter: Piotr Nowojski
> Priority: Critical
> Attachments: log-jm.txt
>
>
> {noformat}
> Caused by: org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailbox$MailboxClosedException: Mailbox is in state QUIESCED, but is required to be in state OPEN for put operations.
> at org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl.checkPutStateConditions(TaskMailboxImpl.java:269) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl.put(TaskMailboxImpl.java:197) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl.execute(MailboxExecutorImpl.java:74) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.api.common.operators.MailboxExecutor.execute(MailboxExecutor.java:103) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.outputCompletedElement(AsyncWaitOperator.java:304) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.access$100(AsyncWaitOperator.java:78) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.processResults(AsyncWaitOperator.java:370) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.lambda$processInMailbox$0(AsyncWaitOperator.java:351) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.drain(MailboxProcessor.java:177) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:854) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:767) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) ~[flink-dist_2.11-1.14.0.jar:1.14.0]
> at java.lang.Thread.run(Thread.java:829) ~[?:?]
> {noformat}
> As reported by a user on [the mailing list:|https://mail-archives.apache.org/mod_mbox/flink-user/202111.mbox/%3CCAO6dnLwtLNxkr9qXG202ysrnse18Wgvph4hqHZe3ar8cuXAfDw%40mail.gmail.com%3E]
> {quote}
> I failed to stop a job with savepoint with the following message:
> Inconsistent execution state after stopping with savepoint. At least one execution is still in one of the following states: FAILED, CANCELED. A global fail-over is triggered to recover the job 452594f3ec5797f399e07f95c884a44b.
> The job manager said
> A savepoint was created at hdfs://mobdata-flink-hdfs/driving-habits/svpts/savepoint-452594-f60305755d0e but the corresponding job 452594f3ec5797f399e07f95c884a44b didn't terminate successfully.
> while complaining about
> Mailbox is in state QUIESCED, but is required to be in state OPEN for put operations.
> Is it okay to ignore this kind of error?
> Please see the attached files for the detailed context.
> FYI,
> - I used the latest 1.14.0
> - I started the job with "$FLINK_HOME"/bin/flink run --target yarn-per-job
> - I couldn't reproduce the exception using the same jar so I might not able to provide DUBUG messages
> {quote}
--
This message was sent by Atlassian Jira
(v8.20.1#820001)