[FLINK-1515] [gelly] Splitted runVertexCentricIteration into...

[FLINK-1515] [gelly] Splitted runVertexCentricIteration into createVertexIteration and runVertexIteration to make VertexCentricIteration object accessible to developers.

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java

@@ -1120,17 +1120,30 @@ public class Graph<K extends Comparable<K> & Serializable, VV extends Serializab
 	}
 
 	/**
-	 * Runs a Vertex-Centric iteration on the graph.
+	 * Create a Vertex-Centric iteration on the graph.
 	 * 
 	 * @param vertexUpdateFunction the vertex update function
 	 * @param messagingFunction the messaging function
 	 * @param maximumNumberOfIterations maximum number of iterations to perform
 	 * @return
 	 */
-	public <M> Graph<K, VV, EV> runVertexCentricIteration(VertexUpdateFunction<K, VV, M> vertexUpdateFunction,
-			MessagingFunction<K, VV, M, EV> messagingFunction,	int maximumNumberOfIterations) {
-		DataSet<Vertex<K, VV>> newVertices = vertices.runOperation(VertexCentricIteration
-				.withEdges(edges, vertexUpdateFunction, messagingFunction, maximumNumberOfIterations));
+	public <M> VertexCentricIteration<K, VV, M, EV> createVertexCentricIteration(
+			VertexUpdateFunction<K, VV, M> vertexUpdateFunction,
+			MessagingFunction<K, VV, M, EV> messagingFunction,
+			int maximumNumberOfIterations) {
+		return VertexCentricIteration.withEdges(edges, vertexUpdateFunction,
+				messagingFunction, maximumNumberOfIterations);
+	}
+    
+	/**
+	 * Runs a Vertex-Centric iteration on the graph.
+	 * 
+	 * @param iteration the Vertex-Centric iteration to run
+	 * @return
+	 */
+	public <M> Graph<K, VV, EV> runVertexCentricIteration(
+			VertexCentricIteration<K, VV, M, EV> iteration) {
+		DataSet<Vertex<K, VV>> newVertices = vertices.runOperation(iteration);
 		return new Graph<K, VV, EV>(newVertices, this.edges, this.context);
 	}
 

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/LabelPropagation.java

@@ -22,6 +22,7 @@ import org.apache.flink.graph.Graph;
 import org.apache.flink.graph.GraphAlgorithm;
 import org.apache.flink.graph.spargel.MessageIterator;
 import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
 import org.apache.flink.types.NullValue;
 
@@ -55,8 +56,9 @@ public class LabelPropagation<K extends Comparable<K> & Serializable>
 
 		// iteratively adopt the most frequent label among the neighbors
 		// of each vertex
-		return input.runVertexCentricIteration(new UpdateVertexLabel<K>(),
-				new SendNewLabelToNeighbors<K>(), maxIterations);
+		VertexCentricIteration<K, Long, Long, NullValue> iteration = input.createVertexCentricIteration(
+				new UpdateVertexLabel<K>(), new SendNewLabelToNeighbors<K>(), maxIterations);
+		return input.runVertexCentricIteration(iteration);
 	}
 
 	/**

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/PageRank.java

@@ -25,6 +25,7 @@ import org.apache.flink.graph.Graph;
 import org.apache.flink.graph.GraphAlgorithm;
 import org.apache.flink.graph.spargel.MessageIterator;
 import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
 
 public class PageRank<K extends Comparable<K> & Serializable> implements
@@ -42,8 +43,10 @@ public class PageRank<K extends Comparable<K> & Serializable> implements
 
 	@Override
 	public Graph<K, Double, Double> run(Graph<K, Double, Double> network) {
-		return network.runVertexCentricIteration(new VertexRankUpdater<K>(
-				numVertices, beta), new RankMessenger<K>(), maxIterations);
+		VertexCentricIteration<K, Double, Double, Double> iteration = network.createVertexCentricIteration(
+				new VertexRankUpdater<K>(numVertices, beta), new RankMessenger<K>(), maxIterations);
+
+		return network.runVertexCentricIteration(iteration);
 	}
 
 	/**

flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java

@@ -25,6 +25,7 @@ import org.apache.flink.graph.GraphAlgorithm;
 import org.apache.flink.graph.Vertex;
 import org.apache.flink.graph.spargel.MessageIterator;
 import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
 
 import java.io.Serializable;
@@ -44,9 +45,11 @@ public class SingleSourceShortestPaths<K extends Comparable<K> & Serializable>
 	@Override
 	public Graph<K, Double, Double> run(Graph<K, Double, Double> input) {
 
-		return input.mapVertices(new InitVerticesMapper<K>(srcVertexId))
-				.runVertexCentricIteration(new VertexDistanceUpdater<K>(),
-						new MinDistanceMessenger<K>(), maxIterations);
+		VertexCentricIteration<K, Double, Double, Double> iteration = input.mapVertices(
+				new InitVerticesMapper<K>(srcVertexId)).createVertexCentricIteration(new VertexDistanceUpdater<K>(),
+				new MinDistanceMessenger<K>(), maxIterations);
+
+		return input.runVertexCentricIteration(iteration);
 	}
 
 	public static final class InitVerticesMapper<K extends Comparable<K> & Serializable>

flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConnectedComponentsITCase.java

@@ -30,6 +30,7 @@ import org.apache.flink.graph.Graph;
 import org.apache.flink.graph.Vertex;
 import org.apache.flink.graph.spargel.MessageIterator;
 import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexCentricIteration;
 import org.apache.flink.graph.spargel.VertexUpdateFunction;
 import org.apache.flink.test.testdata.ConnectedComponentsData;
 import org.apache.flink.test.util.JavaProgramTestBase;
@@ -64,7 +65,8 @@ public class VertexCentricConnectedComponentsITCase extends JavaProgramTestBase
 		DataSet<Vertex<Long, Long>> initialVertices = vertexIds.map(new IdAssigner());
 		Graph<Long, Long, NullValue> graph = Graph.fromDataSet(initialVertices, edges, env); 
 		
-		Graph<Long, Long, NullValue> result = graph.runVertexCentricIteration(new CCUpdater(), new CCMessager(), 100);
+		VertexCentricIteration<Long, Long, Long, NullValue> iteration = graph.createVertexCentricIteration(new CCUpdater(), new CCMessager(), 100);
+		Graph<Long, Long, NullValue> result = graph.runVertexCentricIteration(iteration);
 		
 		result.getVertices().writeAsCsv(resultPath, "\n", " ");
 		env.execute();