我为ApacheFlink创建了一个简单的作业，它使用gelly提供的pagerank实现。
在ide内部运行，一切正常。但是，我尝试使用jobmanagerweb界面将包含作业的jar提交给在我的机器上运行的flink示例。但是，flink没有得到正确的作业计划并执行pagerank，而是提出并执行了一个非常奇怪的计划，只计算图的顶点数。
我做了一些研究和调试，发现gelly提供的pagerank的实现开始计算图的顶点数，但它没有作为算法的参数提供：

if (numberOfVertices == 0) {
    numberOfVertices = network.numberOfVertices();
}

此计算意味着一个嵌入式作业。由于运算符是惰性的，因此不会触发任何计算。在flink服务器中，首先要做的是获取作业计划。这是在一个特殊的环境下完成的， OptimizerPlanEnvironment ，它提供了以下内容 result 方法：

public JobExecutionResult execute(String jobName) throws Exception {
    Plan plan = createProgramPlan(jobName);
    this.optimizerPlan = compiler.compile(plan);

    // do not go on with anything now!
    throw new ProgramAbortException();
}

问题来自这里。一旦 ProgramAbortException 抛出时，程序返回到目前为止计算的计划。但是只计算了内部作业计划，因此这样就不会计算或执行主作业计划。
这是我使用的代码：

public class Job {
    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        Graph<Long, Double, Double> graph = Graph.fromDataSet(
            PageRankData.getDefaultEdgeDataSet(env), new VertexInit(), env);
        graph.run(new PageRank<Long>(0.85, 10)).print();
    }

    private static class VertexInit implements MapFunction<Long, Double> {
        @Override
        public Double map(Long value) throws Exception { return 1.0; }
    }
}

如果提供了顶点数，则执行例如。 graph.run(new PageRank<Long>(0.85, 5, 10)) ，没有问题，正确计算了计划并计算了pagerank。
我的问题是：我做错了什么？或者这真的是Flink的虫子？