[FLINK-1607] [runtime] Moves PartialPartitionInfo from ExecutionVerte…

…x to Execution to automatically clear them in case of restart

This closes #436.

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java

@@ -111,6 +111,7 @@ public class Execution implements Serializable {
 
 	private final FiniteDuration timeout;
 
+	private ConcurrentLinkedQueue<PartialPartitionInfo> partialPartitionInfos;
 
 	private volatile ExecutionState state = CREATED;
 
@@ -134,6 +135,8 @@ public Execution(ExecutionVertex vertex, int attemptNumber, long startTimestamp,
 		markTimestamp(ExecutionState.CREATED, startTimestamp);
 
 		this.timeout = timeout;
+
+		this.partialPartitionInfos = new ConcurrentLinkedQueue<PartialPartitionInfo>();
 	}
 
 	// --------------------------------------------------------------------------------------------
@@ -188,6 +191,9 @@ public void prepareForArchiving() {
 			throw new IllegalStateException("Cannot archive Execution while the assigned resource is still running.");
 		}
 		assignedResource = null;
+
+		partialPartitionInfos.clear();
+		partialPartitionInfos = null;
 	}
 
 	// --------------------------------------------------------------------------------------------
@@ -595,10 +601,11 @@ else if (current == CANCELING || current == RUNNING || current == DEPLOYING) {
 		}
 	}
 
-	void sendPartitionInfos() {
-		ConcurrentLinkedQueue<PartialPartitionInfo> partialPartitionInfos =
-				vertex.getPartialPartitionInfos();
+	void cachePartitionInfo(PartialPartitionInfo partitionInfo) {
+		partialPartitionInfos.add(partitionInfo);
+	}
 
+	void sendPartitionInfos() {
 		// check if the ExecutionVertex has already been archived and thus cleared the
 		// partial partition infos queue
 		if(partialPartitionInfos != null && !partialPartitionInfos.isEmpty()) {

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java

@@ -47,7 +47,6 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 import static com.google.common.base.Preconditions.checkElementIndex;
@@ -75,8 +74,6 @@ public class ExecutionVertex implements Serializable {
 
 	private ExecutionEdge[][] inputEdges;
 
-	private ConcurrentLinkedQueue<PartialPartitionInfo> partialPartitionInfos;
-
 	private final int subTaskIndex;
 
 	private final List<Execution> priorExecutions;
@@ -114,8 +111,6 @@ public ExecutionVertex(ExecutionJobVertex jobVertex, int subTaskIndex,
 
 		this.inputEdges = new ExecutionEdge[jobVertex.getJobVertex().getInputs().size()][];
 
-		this.partialPartitionInfos = new ConcurrentLinkedQueue<PartialPartitionInfo>();
-
 		this.priorExecutions = new CopyOnWriteArrayList<Execution>();
 
 		this.currentExecution = new Execution(this, 0, createTimestamp, timeout);
@@ -204,10 +199,6 @@ public ExecutionGraph getExecutionGraph() {
 		return this.jobVertex.getGraph();
 	}
 
-	public ConcurrentLinkedQueue<PartialPartitionInfo> getPartialPartitionInfos() {
-		return partialPartitionInfos;
-	}
-
 	// --------------------------------------------------------------------------------------------
 	//  Graph building
 	// --------------------------------------------------------------------------------------------
@@ -451,12 +442,10 @@ public void prepareForArchiving() throws IllegalStateException {
 		this.resultPartitions = null;
 		this.inputEdges = null;
 		this.locationConstraintInstances = null;
-		this.partialPartitionInfos.clear();
-		this.partialPartitionInfos = null;
 	}
 
 	public void cachePartitionInfo(PartialPartitionInfo partitionInfo){
-		this.partialPartitionInfos.add(partitionInfo);
+		getCurrentExecutionAttempt().cachePartitionInfo(partitionInfo);
 	}
 
 	void sendPartitionInfos() {

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java

@@ -21,7 +21,6 @@
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,7 +33,6 @@
  * This test starts a MiniYARNCluster with a CapacityScheduler.
  * Is has, by default a queue called "default". The configuration here adds another queue: "qa-team".
  */
-@Ignore("Failing as well :-(")
 public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
 	private static final Logger LOG = LoggerFactory.getLogger(YARNSessionCapacitySchedulerITCase.class);
 

flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java

@@ -30,7 +30,6 @@
 import org.apache.log4j.spi.LoggingEvent;
 import org.junit.Assert;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -44,7 +43,6 @@
  * This test starts a MiniYARNCluster with a FIFO scheudler.
  * There are no queues for that scheduler.
  */
-@Ignore("Because if fails :-(")
 public class YARNSessionFIFOITCase extends YarnTestBase {
 	private static final Logger LOG = LoggerFactory.getLogger(YARNSessionFIFOITCase.class);
 

tools/log4j-travis.properties

@@ -16,7 +16,7 @@
 # limitations under the License.
 ################################################################################
 
-log4j.rootLogger=DEBUG, file
+log4j.rootLogger=INFO, file
 
 # -----------------------------------------------------------------------------
 # Console (use 'console')