diff --git a/pom.xml b/pom.xml
index 71b131f..af8fe98 100644
--- a/pom.xml
+++ b/pom.xml
@@ -137,7 +137,7 @@
         <miniball.version>1.0.3</miniball.version>
         <s4.version>0.6.0-incubating</s4.version>
         <samza.version>0.7.0</samza.version>
-        <flink.version>0.9.0</flink.version>
+        <flink.version>0.10.1</flink.version>
         <slf4j-log4j12.version>1.7.2</slf4j-log4j12.version>
         <slf4j-simple.version>1.7.5</slf4j-simple.version>
         <maven-surefire-plugin.version>2.18</maven-surefire-plugin.version>
diff --git a/samoa-flink/pom.xml b/samoa-flink/pom.xml
index f0266fa..5575643 100644
--- a/samoa-flink/pom.xml
+++ b/samoa-flink/pom.xml
@@ -70,7 +70,7 @@
         </dependency>
   		<dependency>
 			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-streaming-core</artifactId>
+			<artifactId>flink-streaming-java</artifactId>
 			<version>${flink.version}</version>
             <!--<scope>provided</scope>-->
 		</dependency>
diff --git a/samoa-flink/src/main/java/org/apache/samoa/flink/FlinkDoTask.java b/samoa-flink/src/main/java/org/apache/samoa/flink/FlinkDoTask.java
index cd0b82c..7805371 100644
--- a/samoa-flink/src/main/java/org/apache/samoa/flink/FlinkDoTask.java
+++ b/samoa-flink/src/main/java/org/apache/samoa/flink/FlinkDoTask.java
@@ -21,16 +21,9 @@
  */
 
 import com.github.javacliparser.ClassOption;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
 
-import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.samoa.flink.helpers.CircleDetection;
-import org.apache.samoa.flink.helpers.Utils;
 import org.apache.samoa.flink.topology.impl.FlinkComponentFactory;
-import org.apache.samoa.flink.topology.impl.FlinkProcessingItem;
-import org.apache.samoa.flink.topology.impl.FlinkStream;
 import org.apache.samoa.flink.topology.impl.FlinkTopology;
 import org.apache.samoa.tasks.Task;
 import org.slf4j.Logger;
diff --git a/samoa-flink/src/main/java/org/apache/samoa/flink/helpers/CircleDetection.java b/samoa-flink/src/main/java/org/apache/samoa/flink/helpers/CycleDetection.java
similarity index 85%
rename from samoa-flink/src/main/java/org/apache/samoa/flink/helpers/CircleDetection.java
rename to samoa-flink/src/main/java/org/apache/samoa/flink/helpers/CycleDetection.java
index 400e49c..9aedb25 100644
--- a/samoa-flink/src/main/java/org/apache/samoa/flink/helpers/CircleDetection.java
+++ b/samoa-flink/src/main/java/org/apache/samoa/flink/helpers/CycleDetection.java
@@ -26,12 +26,12 @@
 import java.util.Stack;
 
 /**
- * This class contains all logic needed in order to mark circles in job graphs explicitly such as 
- * in the case of Apache Flink. A circle is defined as a list of node ids ordered in topological 
+ * This class contains all logic needed in order to mark cycles in job graphs explicitly such as 
+ * in the case of Apache Flink. A cycle is defined as a list of node ids ordered in topological 
  * (DFS) order.
  * 
  */
-public class CircleDetection {
+public class CycleDetection {
 	private int[] index;
 	private int[] lowLink;
 	private int counter;
@@ -40,12 +40,12 @@
 	List<Integer>[] graph;
 
 
-	public CircleDetection() {
-		stack = new Stack<Integer>();
+	public CycleDetection() {
+		stack = new Stack<>();
 		scc = new ArrayList<>();
 	}
 
-	public List<List<Integer>> getCircles(List<Integer>[] adjacencyList) {
+	public List<List<Integer>> getCycles(List<Integer>[] adjacencyList) {
 		graph = adjacencyList;
 		index = new int[adjacencyList.length];
 		lowLink = new int[adjacencyList.length];
@@ -82,7 +82,7 @@
 					w = stack.pop();
 					sccComponent.add(w);
 				} while (neighbor != w);
-				//add neighbor again, just in case it is a member of another circle 
+				//add neighbor again, just in case it is a member of another cycle 
 				stack.add(neighbor); 
 				scc.add(sccComponent);
 			}
diff --git a/samoa-flink/src/main/java/org/apache/samoa/flink/helpers/Utils.java b/samoa-flink/src/main/java/org/apache/samoa/flink/helpers/Utils.java
index 38b4bdc..ce01567 100644
--- a/samoa-flink/src/main/java/org/apache/samoa/flink/helpers/Utils.java
+++ b/samoa-flink/src/main/java/org/apache/samoa/flink/helpers/Utils.java
@@ -32,8 +32,6 @@
 import org.apache.samoa.flink.topology.impl.SamoaType;
 import org.apache.samoa.utils.PartitioningScheme;
 
-import java.util.List;
-
 import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;
 
 public class Utils {
@@ -45,7 +43,7 @@
 			case BROADCAST:
 				return stream.broadcast();
 			case GROUP_BY_KEY:
-				return stream.groupBy(new KeySelector<SamoaType, String>() {
+				return stream.keyBy(new KeySelector<SamoaType, String>() {
 					@Override
 					public String getKey(SamoaType samoaType) throws Exception {
 						return samoaType.f0;
diff --git a/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkProcessingItem.java b/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkProcessingItem.java
index 28701df..9e3c880 100644
--- a/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkProcessingItem.java
+++ b/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkProcessingItem.java
@@ -22,20 +22,20 @@
 
 
 import com.google.common.collect.Lists;
-
 import org.apache.flink.api.common.functions.Function;
 import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.samoa.core.ContentEvent;
 import org.apache.samoa.core.Processor;
 import org.apache.samoa.flink.helpers.Utils;
 import org.apache.samoa.topology.ProcessingItem;
 import org.apache.samoa.topology.Stream;
 import org.apache.samoa.utils.PartitioningScheme;
-import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -60,9 +60,8 @@
 	private int parallelism;
 	private static int numberOfPIs = 0;
 	private int piID;
-	private List<Integer> circleId; //check if we can refactor this
+	private List<Integer> cycleId; //check if we can refactor this
 	private boolean onIteration;
-	//private int circleId; //check if we can refactor this
 
 	public FlinkProcessingItem(StreamExecutionEnvironment env, Processor proc) {
 		this(env, proc, 1);
@@ -79,8 +78,8 @@
 		this.processor = proc;
 		this.parallelism = parallelism;
 		this.piID = numberOfPIs++;
-		this.circleId = new ArrayList<Integer>() {
-		}; // if size equals 0, then it is part of no circle
+		this.cycleId = new ArrayList<Integer>() {
+		}; // if size equals 0, then it is part of no cycle
 	}
 
 	public Stream createStream() {
@@ -90,12 +89,12 @@
 	}
 
 	public void putToStream(ContentEvent data, Stream targetStream) {
-		output.collect(SamoaType.of(data, targetStream.getStreamId()));
+		output.collect(new StreamRecord<>(SamoaType.of(data, targetStream.getStreamId())));
 	}
-
+	
 	@Override
-	public void open(Configuration parameters) throws Exception {
-		super.open(parameters);
+	public void open() throws Exception {
+		super.open();
 		this.processor.onCreate(getComponentId());
 	}
 
@@ -148,8 +147,13 @@
 	}
 
 	@Override
-	public void processElement(SamoaType samoaType) throws Exception {
-		fun.processEvent(samoaType.f1);
+	public void processElement(StreamRecord<SamoaType> streamRecord) throws Exception {
+		fun.processEvent(streamRecord.getValue().f1);
+	}
+
+	@Override
+	public void processWatermark(Watermark watermark) throws Exception {
+
 	}
 
 	@Override
@@ -175,10 +179,6 @@
 		return parallelism;
 	}
 
-	public void setParallelism(int parallelism) {
-		this.parallelism = parallelism;
-	}
-
 	public List<FlinkStream> getOutputStreams() {
 		return outputStreams;
 	}
@@ -187,28 +187,24 @@
 		return this.outStream;
 	}
 
-	public void setOutStream(DataStream outStream) {
-		this.outStream = outStream;
-	}
-
 	@Override
 	public int getComponentId() {
 		return piID;
 	}
 
-	public boolean isPartOfCircle() {
-		return this.circleId.size() > 0;
+	public boolean isPartOfCycle() {
+		return this.cycleId.size() > 0;
 	}
 
-	public List<Integer> getCircleIds() {
-		return circleId;
+	public List<Integer> getCycleIds() {
+		return cycleId;
 	}
 
-	public void addPItoLoop(int piId) {
-		this.circleId.add(piId);
+	public void addPItoCycle(int piId) {
+		this.cycleId.add(piId);
 	}
 
-	public DataStream<SamoaType> getInStream() {
+	public DataStream<SamoaType> getDataStream() {
 		return inStream;
 	}
 
@@ -219,6 +215,7 @@
 	public void setOnIteration(boolean onIteration) {
 		this.onIteration = onIteration;
 	}
+	
 
 	static class SamoaDelegateFunction implements Function, Serializable {
 		private final Processor proc;
diff --git a/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkStream.java b/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkStream.java
index 31617a7..286802c 100644
--- a/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkStream.java
+++ b/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkStream.java
@@ -36,46 +36,44 @@
 public class FlinkStream extends AbstractStream implements FlinkComponent, Serializable {
 
 	private static int outputCounter = 0;
-	private FlinkComponent procItem;
-	private transient DataStream<SamoaType> dataStream;
-	private int sourcePiId;
-	private String flinkStreamId;
+	private FlinkComponent sourceComponent;
+	private transient DataStream<SamoaType> filteredStream;
+	private String filterID;
 
 	public FlinkStream(FlinkComponent sourcePi) {
-		this.procItem = sourcePi;
-		this.sourcePiId = sourcePi.getComponentId();
+		this.sourceComponent = sourcePi;
 		setStreamId("stream-" + Integer.toString(outputCounter));
-		flinkStreamId = "stream-" + Integer.toString(outputCounter);
+		filterID = "stream-" + Integer.toString(outputCounter);
 		outputCounter++;
 	}
 
 	@Override
 	public void initialise() {
-		if (procItem instanceof FlinkProcessingItem) {
-			dataStream = procItem.getOutStream().filter(Utils.getFilter(getStreamId()))
-			.setParallelism(((FlinkProcessingItem) procItem).getParallelism());
+		if (sourceComponent instanceof FlinkProcessingItem) {
+			filteredStream = sourceComponent.getOutStream().filter(Utils.getFilter(getStreamId()))
+			.setParallelism(((FlinkProcessingItem) sourceComponent).getParallelism());
 		} else
-			dataStream = procItem.getOutStream();
+			filteredStream = sourceComponent.getOutStream();
 	}
 
 	@Override
 	public boolean canBeInitialised() {
-		return procItem.isInitialised();
+		return sourceComponent.isInitialised();
 	}
 
 	@Override
 	public boolean isInitialised() {
-		return dataStream != null;
+		return filteredStream != null;
 	}
 
 	@Override
 	public DataStream getOutStream() {
-		return dataStream;
+		return filteredStream;
 	}
 
 	@Override
 	public void put(ContentEvent event) {
-		((FlinkProcessingItem) procItem).putToStream(event, this);
+		((FlinkProcessingItem) sourceComponent).putToStream(event, this);
 	}
 
 	@Override
@@ -84,11 +82,11 @@
 	}
 
 	public int getSourcePiId() {
-		return sourcePiId;
+		return sourceComponent.getComponentId();
 	}
 
 	@Override
 	public String getStreamId() {
-		return flinkStreamId;
+		return filterID;
 	}
 }
diff --git a/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkTopology.java b/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkTopology.java
index 65c52c6..a09ba71 100644
--- a/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkTopology.java
+++ b/samoa-flink/src/main/java/org/apache/samoa/flink/topology/impl/FlinkTopology.java
@@ -21,17 +21,15 @@
  */
 
 
-
 import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
-
 import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.streaming.api.datastream.IterativeDataStream;
+import org.apache.flink.streaming.api.datastream.IterativeStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.samoa.flink.helpers.CircleDetection;
-import org.apache.samoa.flink.helpers.Utils;
+import org.apache.samoa.flink.helpers.CycleDetection;
 import org.apache.samoa.topology.AbstractTopology;
 import org.apache.samoa.topology.EntranceProcessingItem;
 import org.apache.samoa.utils.PartitioningScheme;
@@ -43,144 +41,148 @@
 
 /**
  * A SAMOA topology on Apache Flink
- * 
+ * <p/>
  * A Samoa-Flink Streaming Topology is DAG of ProcessingItems encapsulated within custom operators.
- * Streams are tagged and filtered in each operator's output so they can be routed to the right 
+ * Streams are tagged and filtered in each operator's output so they can be routed to the right
  * operator respectively. Building a Flink topology from a Samoa task involves invoking all these
- * stream transformations and finally, marking and initiating loops in the graph. We have to do that
- * since Flink only allows explicit loops in the topology started with 'iterate()' and closed with
- * 'closeWith()'. Thus, when we build a flink topology we have to do it incrementally from the 
- * sources, mark loops and initialize them with explicit iterations.
- * 
+ * stream transformations and finally, marking and initiating cycles in the graph. We have to do that
+ * since Flink only allows explicit cycles in the topology started with 'iterate()' and closed with
+ * 'closeWith()'. Thus, when we build a flink topology we have to do it incrementally from the
+ * sources, mark cycles and initialize them with explicit iterations.
  */
 public class FlinkTopology extends AbstractTopology {
 
-	private static final Logger logger = LoggerFactory.getLogger(FlinkTopology.class);
-	public static StreamExecutionEnvironment env;
-	public List<List<FlinkProcessingItem>> topologyLoops = new ArrayList<>();
-	public  List<Integer> backEdges = new ArrayList<Integer>();
+    private static final Logger logger = LoggerFactory.getLogger(FlinkTopology.class);
+    public static StreamExecutionEnvironment env;
+    public List<List<FlinkProcessingItem>> cycles = new ArrayList<>();
+    public List<Integer> backEdges = new ArrayList<Integer>(); 
 
-	public FlinkTopology(String name, StreamExecutionEnvironment env) {
-		super(name);
-		this.env = env;
-	}
+    public FlinkTopology(String name, StreamExecutionEnvironment env) {
+        super(name);
+        this.env = env;
+    }
 
-	public StreamExecutionEnvironment getEnvironment() {
-		return env;
-	}
-	
-	public void build() {
-		markCircles();
-		for (EntranceProcessingItem src : getEntranceProcessingItems()) {
-			((FlinkEntranceProcessingItem) src).initialise();
-		}
-		initComponents(ImmutableList.copyOf(Iterables.filter(getProcessingItems(), FlinkProcessingItem.class)));
-	}
+    public StreamExecutionEnvironment getEnvironment() {
+        return env;
+    }
 
-	private void initComponents(ImmutableList<FlinkProcessingItem> flinkComponents) {
-		if (flinkComponents.isEmpty()) return;
+    public void build() {
+        markCycles();
+        for (EntranceProcessingItem src : getEntranceProcessingItems()) {
+            ((FlinkEntranceProcessingItem) src).initialise();
+        }
+        initComponents(ImmutableList.copyOf(Iterables.filter(getProcessingItems(), FlinkProcessingItem.class)));
+    }
 
-		for (FlinkProcessingItem comp : flinkComponents) {
-			if (comp.canBeInitialised() && !comp.isInitialised() && !comp.isPartOfCircle()) {
-				comp.initialise();
-				comp.initialiseStreams();
+    private void initComponents(ImmutableList<FlinkProcessingItem> flinkComponents) {
+        if (flinkComponents.isEmpty()) return;
 
-			}//if component is part of one or more circle
-			else if (comp.isPartOfCircle() && !comp.isInitialised()) {
-				for (Integer circle : comp.getCircleIds()) {
-					//check if circle can be initialized
-					if (checkCircleReady(circle)) {
-						logger.debug("Circle: " + circle + " can be initialised");
-						initialiseCircle(circle);
-					} else {
-						logger.debug("Circle cannot be initialised");
-					}
-				}
-			}
+        for (FlinkProcessingItem comp : flinkComponents) {
+            if (comp.canBeInitialised() && !comp.isInitialised() && !comp.isPartOfCycle()) {
+                comp.initialise();
+                comp.initialiseStreams();
 
-		}
-		initComponents(ImmutableList.copyOf(Iterables.filter(flinkComponents, new Predicate<FlinkProcessingItem>() {
-			@Override
-			public boolean apply(FlinkProcessingItem flinkComponent) {
-				return !flinkComponent.isInitialised();
-			}
-		})));
-	}
+            }//if component is part of one or more cycle
+            else if (comp.isPartOfCycle() && !comp.isInitialised()) {
+                for (Integer cycle : comp.getCycleIds()) {
+                    //check if cycle can be initialized
+                    if (completenessCheck(cycle)) {
+                        logger.debug("Cycle: " + cycle + " can be initialised");
+                        initializeCycle(cycle);
+                    } else {
+                        logger.debug("Cycle cannot be initialised");
+                    }
+                }
+            }
+        }
+        initComponents(ImmutableList.copyOf(Iterables.filter(flinkComponents, new Predicate<FlinkProcessingItem>() {
+            @Override
+            public boolean apply(FlinkProcessingItem flinkComponent) {
+                return !flinkComponent.isInitialised();
+            }
+        })));
+    }
 
-	private void markCircles(){
-		List<FlinkProcessingItem> pis = Lists.newArrayList(Iterables.filter(getProcessingItems(), FlinkProcessingItem.class));
-		List<Integer>[] graph = new List[pis.size()];
-		FlinkProcessingItem[] processingItems = new FlinkProcessingItem[pis.size()];
+    /**
+     * Detects and marks all cycles and backedges needed to construct a Flink topology
+     */
+    private void markCycles() {
+        List<FlinkProcessingItem> pis = Lists.newArrayList(Iterables.filter(getProcessingItems(), FlinkProcessingItem.class));
+        List<Integer>[] graph = new List[pis.size()];
+        FlinkProcessingItem[] processingItems = new FlinkProcessingItem[pis.size()];
 
 
-		for (int i=0;i<pis.size();i++) {
-			graph[i] = new ArrayList<Integer>();
-		}
-		//construct the graph of the topology for the Processing Items (No entrance pi is included)
-		for (FlinkProcessingItem pi: pis) {
-			processingItems[pi.getComponentId()] = pi;
-			for (Tuple3<FlinkStream, PartitioningScheme, Integer> is : pi.getInputStreams()) {
-				if (is.f2 != -1) graph[is.f2].add(pi.getComponentId());
-			}
-		}
-		for (int g=0;g<graph.length;g++)
-			logger.debug(graph[g].toString());
+        for (int i = 0; i < pis.size(); i++) {
+            graph[i] = new ArrayList<>();
+        }
+        //construct the graph of the topology for the Processing Items (No entrance pi is included)
+        for (FlinkProcessingItem pi : pis) {
+            processingItems[pi.getComponentId()] = pi;
+            for (Tuple3<FlinkStream, PartitioningScheme, Integer> is : pi.getInputStreams()) {
+                if (is.f2 != -1) graph[is.f2].add(pi.getComponentId());
+            }
+        }
+        for (int g = 0; g < graph.length; g++)
+            logger.debug(graph[g].toString());
 
-		CircleDetection detCircles = new CircleDetection();
-		List<List<Integer>> circles = detCircles.getCircles(graph);
+        CycleDetection detCycles = new CycleDetection();
+        List<List<Integer>> graphCycles = detCycles.getCycles(graph);
 
-		//update PIs, regarding being part of a circle.
-		for (List<Integer> c : circles){
-			List<FlinkProcessingItem> circle = new ArrayList<>();
-			for (Integer it : c){
-				circle.add(processingItems[it]);
-				processingItems[it].addPItoLoop(topologyLoops.size());
-			}
-			topologyLoops.add(circle);
-			backEdges.add(circle.get(0).getComponentId());
-		}
-		logger.debug("Circles detected in the topology: " + circles);
-	}
-	
+        //update PIs, regarding being part of a cycle.
+        for (List<Integer> c : graphCycles) {
+            List<FlinkProcessingItem> cycle = new ArrayList<>();
+            for (Integer it : c) {
+                cycle.add(processingItems[it]);
+                processingItems[it].addPItoCycle(cycles.size());
+            }
+            cycles.add(cycle);
+            backEdges.add(cycle.get(0).getComponentId());
+        }
+        logger.debug("Cycles detected in the topology: " + graphCycles);
+    }
 
-	private boolean checkCircleReady(int circleId) {
 
-		List<Integer> circleIds = new ArrayList<>();
+    private boolean completenessCheck(int cycleId) {
 
-		for (FlinkProcessingItem pi : topologyLoops.get(circleId)) {
-			circleIds.add(pi.getComponentId());
-		}
-		//check that all incoming to the circle streams are initialised
-		for (FlinkProcessingItem procItem : topologyLoops.get(circleId)) {
-			for (Tuple3<FlinkStream, PartitioningScheme, Integer> inputStream : procItem.getInputStreams()) {
-				//if a inputStream is not initialized AND source of inputStream is not in the circle or a tail of other circle
-				if ((!inputStream.f0.isInitialised()) && (!circleIds.contains(inputStream.f2)) && (!backEdges.contains(inputStream.f2)))
-					return false;
-			}
-		}
-		return true;
-	}
+        List<Integer> cycleIDs = new ArrayList<>();
 
-	private void initialiseCircle(int circleId) {
-		//get the head and tail of circle
-		FlinkProcessingItem tail = topologyLoops.get(circleId).get(0);
-		FlinkProcessingItem head = topologyLoops.get(circleId).get(topologyLoops.get(circleId).size() - 1);
+        for (FlinkProcessingItem pi : cycles.get(cycleId)) {
+            cycleIDs.add(pi.getComponentId());
+        }
+        //check that all incoming to the cycle streams are initialised
+        for (FlinkProcessingItem procItem : cycles.get(cycleId)) {
+            for (Tuple3<FlinkStream, PartitioningScheme, Integer> inputStream : procItem.getInputStreams()) {
+                //if a inputStream is not initialized AND source of inputStream is not in the cycle or a tail of other cycle
+                if ((!inputStream.f0.isInitialised()) && (!cycleIDs.contains(inputStream.f2)) && (!backEdges.contains(inputStream.f2)))
+                    return false;
+            }
+        }
+        return true;
+    }
 
-		//initialise source stream of the iteration, so as to use it for the iteration starting point
-		if (!head.isInitialised()) {
-			head.setOnIteration(true);
-			head.initialise();
-			head.initialiseStreams();
-		}
+    private void initializeCycle(int cycleID) {
+        //get the head and tail of cycle
+        FlinkProcessingItem tail = cycles.get(cycleID).get(0);
+        FlinkProcessingItem head = cycles.get(cycleID).get(cycles.get(cycleID).size() - 1);
 
-		//initialise all nodes after head
-		for (int node = topologyLoops.get(circleId).size() - 2; node >= 0; node--) {
-			topologyLoops.get(circleId).get(node).initialise();
-			topologyLoops.get(circleId).get(node).initialiseStreams();
-		}
+        //initialise source stream of the iteration, so as to use it for the iteration starting point
+        if (!head.isInitialised()) {
+            head.setOnIteration(true);
+            head.initialise();
+            head.initialiseStreams();
+        }
 
-		((IterativeDataStream) head.getInStream()).closeWith(head.getInputStreamBySourceID(tail.getComponentId()).getOutStream());
-	}
+        //initialise all nodes after head
+        for (int node = cycles.get(cycleID).size() - 2; node >= 0; node--) {
+            FlinkProcessingItem processingItem = cycles.get(cycleID).get(node);
+            processingItem.initialise();
+            processingItem.initialiseStreams();
+        }
+
+        SingleOutputStreamOperator backedge = (SingleOutputStreamOperator) head.getInputStreamBySourceID(tail.getComponentId()).getOutStream();
+        backedge.setParallelism(head.getParallelism());
+        ((IterativeStream) head.getDataStream()).closeWith(backedge);
+    }
 
 
 }