diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
index 55be728647f2e9aef33be0c28f8df878e2ead9b7..1b79cf52ad7b0e5df9e195d80ed0a5ec0f103509 100644
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/Graph.java
+++ b/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);
 	}
 
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/LabelPropagation.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/LabelPropagation.java
index 69d7713c6c4ccac54e9674e5843a3b3367096d96..33a04e71fe9c2bfbfae42dd4dc10711dab701492 100644
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/LabelPropagation.java
+++ b/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);
 	}
 
 	/**
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/PageRank.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/PageRank.java
index 39b8ef1a88554e17cb6f0a98bdfcc3d95030dd74..46f725fa9f98ac4c63a72bf6e3f425f1e4acf352 100644
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/PageRank.java
+++ b/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);
 	}
 
 	/**
diff --git a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java b/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java
index 2f575e78a891ffe654e2154797f2a4ceff1b14ae..f0690000535616a72282a014499557638783dbcc 100644
--- a/flink-staging/flink-gelly/src/main/java/org/apache/flink/graph/library/SingleSourceShortestPaths.java
+++ b/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>
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConnectedComponentsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConnectedComponentsITCase.java
index 380e027a153ccfdd356ba5f1754c4b6f411d1f8c..8fb9a1162ba8a90a8d762d82fec4fa177aab717c 100644
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConnectedComponentsITCase.java
+++ b/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();