[jira] [Updated] (BEAM-12477) pipeline.run() blocks when using FlinkRunner

["Stefan Wachter (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/BEAM-12477?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Stefan Wachter updated BEAM-12477:
----------------------------------
    Description: 
{{pipeline.run()}} is documented to be asynchronous (cf. [create-your-pipeline|https://beam.apache.org/documentation/pipelines/create-your-pipeline/]). It seems that when using FlinkRunner (embedded or remote) the call blocks until the pipeline finishes.

Digging into Flink code I found that both, {{LocalStreamEnvironment}} and {{RemoteStreamEnvironment}} set {{execution.attached}} to true. This causes that {{StreamExecutionEnvironment.execute}} blocks later on:

{code}
    public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
        final JobClient jobClient = executeAsync(streamGraph);

        try {
            final JobExecutionResult jobExecutionResult;

            if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
                jobExecutionResult = jobClient.getJobExecutionResult().get(); // <==== execution is blocked here
            } else {
                jobExecutionResult = new DetachedJobExecutionResult(jobClient.getJobID());
            }

            jobListeners.forEach(
                    jobListener -> jobListener.onJobExecuted(jobExecutionResult, null));

            return jobExecutionResult;
        } catch (Throwable t) {
            // get() on the JobExecutionResult Future will throw an ExecutionException. This
            // behaviour was largely not there in Flink versions before the PipelineExecutor
            // refactoring so we should strip that exception.
            Throwable strippedException = ExceptionUtils.stripExecutionException(t);

            jobListeners.forEach(
                    jobListener -> {
                        jobListener.onJobExecuted(null, strippedException);
                    });
            ExceptionUtils.rethrowException(strippedException);

            // never reached, only make javac happy
            return null;
        }
    }
{code}

  was:{{pipeline.run()}} is documented to be asynchronous (cf. [create-your-pipeline|https://beam.apache.org/documentation/pipelines/create-your-pipeline/]). It seems that when using FlinkRunner (embedded or remote) the call blocks until the pipeline finishes.


> pipeline.run() blocks when using FlinkRunner
> --------------------------------------------
>
>                 Key: BEAM-12477
>                 URL: https://issues.apache.org/jira/browse/BEAM-12477
>             Project: Beam
>          Issue Type: Bug
>          Components: runner-flink
>    Affects Versions: 2.29.0
>            Reporter: Stefan Wachter
>            Priority: P2
>
> {{pipeline.run()}} is documented to be asynchronous (cf. [create-your-pipeline|https://beam.apache.org/documentation/pipelines/create-your-pipeline/]). It seems that when using FlinkRunner (embedded or remote) the call blocks until the pipeline finishes.
> Digging into Flink code I found that both, {{LocalStreamEnvironment}} and {{RemoteStreamEnvironment}} set {{execution.attached}} to true. This causes that {{StreamExecutionEnvironment.execute}} blocks later on:
> {code}
>     public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
>         final JobClient jobClient = executeAsync(streamGraph);
>         try {
>             final JobExecutionResult jobExecutionResult;
>             if (configuration.getBoolean(DeploymentOptions.ATTACHED)) {
>                 jobExecutionResult = jobClient.getJobExecutionResult().get(); // <==== execution is blocked here
>             } else {
>                 jobExecutionResult = new DetachedJobExecutionResult(jobClient.getJobID());
>             }
>             jobListeners.forEach(
>                     jobListener -> jobListener.onJobExecuted(jobExecutionResult, null));
>             return jobExecutionResult;
>         } catch (Throwable t) {
>             // get() on the JobExecutionResult Future will throw an ExecutionException. This
>             // behaviour was largely not there in Flink versions before the PipelineExecutor
>             // refactoring so we should strip that exception.
>             Throwable strippedException = ExceptionUtils.stripExecutionException(t);
>             jobListeners.forEach(
>                     jobListener -> {
>                         jobListener.onJobExecuted(null, strippedException);
>                     });
>             ExceptionUtils.rethrowException(strippedException);
>             // never reached, only make javac happy
>             return null;
>         }
>     }
> {code}



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