title: use collect() and env.execute() in one flink job
body: <p>I'm trying to write a computation in Flink which requires two phases.</p>
  
  <p>In the first phase, I create a Graph and get its vertex ids:</p>
  
  <pre><code>List&lt;String&gt; ids = graph.getVertexIds().collect();
  </code></pre>
  
  <p>In the second phase, I'd like to use these ids to run SingleSourceShortestPath for each vertex.</p>
  
  <pre><code>for (String id: ids){
          System.out.println("Source Id: "+id);
          graph.run( new SingleSourceShortestPaths&lt;String, String&gt;(id, 10)).print();
      }
  </code></pre>
  
  <p>It works locally (in IntelliJ IDE and command lines using <code>./bin/flink run ...</code>), but when I submit the job on Flink using its WebUI, the program just execute till <code>collect()</code> method and does not run the remaining of the program (for statement and <code>print()</code>). </p>
  
  <p>What is the problem?</p>
  
  <p>Here is my code:</p>
  
  <pre><code>import org.apache.flink.api.common.functions.MapFunction;
  import org.apache.flink.api.java.ExecutionEnvironment;
  import org.apache.flink.graph.Edge;
  import org.apache.flink.graph.Graph;
  import org.apache.flink.graph.library.SingleSourceShortestPaths;
  
  import java.util.ArrayList;
  import java.util.List;
  
  public class Main {
      public static void main(String[] args) throws Exception {
  
          ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  
          Edge&lt;String, Double&gt; e1 = new Edge&lt;String, Double&gt;("1", "2", 0.5);
          Edge&lt;String, Double&gt; e2 = new Edge&lt;String, Double&gt;("2", "3", 0.5);
          Edge&lt;String, Double&gt; e3 = new Edge&lt;String, Double&gt;("4", "5", 0.5);
          Edge&lt;String, Double&gt; e4 = new Edge&lt;String, Double&gt;("5", "6", 0.5);
          Edge&lt;String, Double&gt; e5 = new Edge&lt;String, Double&gt;("7", "8", 0.5);
  
  
          List&lt;Edge&lt;String, Double&gt;&gt; edgeList = new ArrayList&lt;Edge&lt;String, Double&gt;&gt;();
          edgeList.add(e1);
          edgeList.add(e2);
          edgeList.add(e3);
          edgeList.add(e4);
          edgeList.add(e5);
  
  
          Graph&lt;String, String, Double&gt; graph = Graph.fromCollection(edgeList,
                  new MapFunction&lt;String, String&gt;() {
                      public String map(String value) {
                          return value;
                      }
                  }, env);
  
          List&lt;String&gt; ids = graph.getVertexIds().collect();
  
          for (String id: ids){
              System.out.println("Source Id: "+id);
              graph.run( new SingleSourceShortestPaths&lt;String, String&gt;(id, 10)).print();
          }
      }
  }
  </code></pre>