"No operators defined in streaming topology" error when Flink app still starts successfully

[Shane Bishop]

Hi all,

My team has started seeing the error "java.lang.IllegalStateException: No 
operators defined in streaming topology. Cannot execute." However, even with 
this error, the Flink application starts and runs fine, and the Flink job 
renders fine in the Flink Dashboard.

Attached is the full stacktrace.

This error comes from when we call 
StreamExecutionEnvironment#getExecutionPlan(). In the code snippet below, we 
call this method on the last line of the snippet.

>From poking around online, I found 
>https://stackoverflow.com/questions/54977290/flink-no-operators-defined-in-streaming-topology-cannot-execute,
> which suggests the problem could be that we don't set a sink, but in the code 
>below you will see we do set a sink (just maybe not in a way that 
>getExecutionPlan() expects).

Can this be safely ignored? Is there something we can do so that 
getExecutionPlan() will work properly, or otherwise fix/suppress this error?

Below is the code (some portions have been redacted):

final StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<RedactedType> source =
    env.addSource(new RedactedType())
        .uid(<redacted>)
        .name(<redacted>)
        .shuffle();

DataStream<Tuple3<Long, String, byte[]>> stream =
    AsyncDataStream.unorderedWait(source, new RedactedType(), 10000, 
TimeUnit.MILLISECONDS)
        .uid(<redacted>)
        .name(<redacted>);

stream
    .flatMap(new RedactedType())
    .uid(<redacted>)
    .name(<redacted>)
    .flatMap(new RedactedType())
    .uid(<redacted>)
    .name(<redacted>)
    .shuffle()
    .addSink(new RedactedType()) // Set sink
    .uid(<redacted>)
    .name(<redacted>);

env.execute("<Redacted job name>");
logger.info("Started job; executionPlan={}", env.getExecutionPlan()); // line 66

Thanks,
Shane

org.apache.flink.runtime.rest.handler.RestHandlerException: Could not execute 
application.
        at 
org.apache.flink.runtime.webmonitor.handlers.JarRunOverrideHandler.lambda$handleRequest$4(JarRunOverrideHandler.java:247)
        at 
java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:930)
        at 
java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:907)
        at 
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
        at 
java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1705)
        at 
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at 
java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at 
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at 
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.util.concurrent.CompletionException: 
org.apache.flink.util.FlinkRuntimeException: Could not execute application.
        at 
java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)
        at 
java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)
        at 
java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1702)
        ... 6 more
Caused by: org.apache.flink.util.FlinkRuntimeException: Could not execute 
application.
        at 
org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:88)
        at 
org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:70)
        at 
org.apache.flink.runtime.webmonitor.handlers.JarRunOverrideHandler.lambda$handleRequest$3(JarRunOverrideHandler.java:238)
        at 
java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
        ... 6 more
Caused by: org.apache.flink.client.program.ProgramInvocationException: The main 
method caused an error: No operators defined in streaming topology. Cannot 
execute.
        at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
        at 
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
        at 
org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
        at 
org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84)
        ... 9 more
Caused by: java.lang.IllegalStateException: No operators defined in streaming 
topology. Cannot execute.
        at 
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraphGenerator(StreamExecutionEnvironment.java:2019)
        at 
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:2010)
        at 
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionPlan(StreamExecutionEnvironment.java:2043)
        at com.trendmicro.c1ws.tdax.App.main(App.java:66)
        at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:566)
        at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
        ... 12 more