Flink读取GCP PubSub报错：流拓扑未定义算子无法执行

问题描述 #

执行flink run Flink.jar启动Flink作业时，出现以下错误：

Starting execution of program

------------------------------------------------------------
 The program finished with the following exception:

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:621)
    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:466)
    at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:274)
    at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:746)
    at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:273)
    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:205)
    at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1008)
    at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1081)
    at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1081)
Caused by: java.lang.IllegalStateException: No operators defined in streaming topology. Cannot execute.
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraphGenerator(StreamExecutionEnvironment.java:1545)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:1540)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1507)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1489)
    at org.flink.ReadFromPubsub.main(ReadFromPubsub.java:30)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:604)
    ... 9 more

用户使用的代码如下：

package org.flink;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;


public class ReadFromPubsub
{

    public static void main(String args[]) throws Exception 
    {
        System.out.println("Flink Pubsub Code Read 1");
        StreamExecutionEnvironment streamExecEnv= StreamExecutionEnvironment.getExecutionEnvironment();
        
        DeserializationSchema<String> deserializer = new SimpleStringSchema();
       
        SourceFunction<String> pubsubSource = PubSubSource.newBuilder()
                                                            .withDeserializationSchema(deserializer)
                                                            .withProjectName("vz-it-np-gudv-dev-vzntdo-0")
                                                            .withSubscriptionName("subscription1")
                                                            .build();
        
        streamExecEnv.addSource(pubsubSource);
        streamExecEnv.execute();
    }
}

问题原因 #

Flink的流处理拓扑要求必须构建完整的数据处理链路：从Source读取数据后，必须有后续的算子对数据进行处理或输出（比如Sink、打印、转换等）。当前代码仅添加了PubSub Source，但没有任何消费数据的算子，Flink会判定拓扑无效，因此抛出该错误。

解决方案 #

在Source之后添加至少一个算子，比如用于测试的print()，或者生产环境的实际Sink（如写入GCS、BigQuery等）。修改后的代码如下：

package org.flink;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;


public class ReadFromPubsub
{

    public static void main(String args[]) throws Exception 
    {
        System.out.println("Flink Pubsub Code Read 1");
        StreamExecutionEnvironment streamExecEnv= StreamExecutionEnvironment.getExecutionEnvironment();
        
        DeserializationSchema<String> deserializer = new SimpleStringSchema();
       
        SourceFunction<String> pubsubSource = PubSubSource.newBuilder()
                                                            .withDeserializationSchema(deserializer)
                                                            .withProjectName("vz-it-np-gudv-dev-vzntdo-0")
                                                            .withSubscriptionName("subscription1")
                                                            .build();
        
        // 添加print()算子消费Source的数据，构建完整拓扑
        streamExecEnv.addSource(pubsubSource).print();
        streamExecEnv.execute();
    }
}

额外注意事项 #

1. 若为生产环境，建议将print()替换为实际的Sink算子，避免仅打印数据。
2. 确认GCP项目名称、PubSub订阅名称正确，且Flink作业拥有访问该PubSub订阅的权限（可通过GCP服务账号密钥配置）。

内容的提问来源于stack exchange，提问作者Nagesh B Viswanadham