[FLINK-2406] [streaming] Abstract and improve stream alignment via th…

…e BarrierBuffer

 - Add an interface for the functionaliy of the barrier buffer (for later addition of other implementatiions)
 - Add broader tests for the BarrierBuffer, inluding trailing data and barrier races.
 - Checkpoint barriers are handled by the buffer directly, rather than being returned and re-injected.
 - Simplify logic in the BarrierBuffer and fix certain corner cases.
 - Give access to spill directories properly via I/O manager, rather than via GlobalConfiguration singleton.
 - Rename the "BarrierBufferIOTest" to "BarrierBufferMassiveRandomTest"
 - A lot of code style/robustness fixes (proplery define constants, visibility, exception signatures)

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/IOManager.java

@@ -301,9 +301,19 @@ public abstract BulkBlockChannelReader createBulkBlockChannelReader(FileIOChanne
 	 * 
 	 * @return The number of temporary file directories.
 	 */
-	public int getNumberOfTempDirs() {
+	public int getNumberOfSpillingDirectories() {
 		return this.paths.length;
 	}
+
+	/**
+	 * Gets the directories that the I/O manager spills to.
+	 * 
+	 * @return The directories that the I/O manager spills to.
+	 */
+	public File[] getSpillingDirectories() {
+		return this.paths;
+	}
+
 
 	protected int getNextPathNum() {
 		final int next = this.nextPath;

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/EndOfPartitionEvent.java

@@ -22,19 +22,34 @@
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.runtime.event.task.RuntimeEvent;
 
-import java.io.IOException;
 
 public class EndOfPartitionEvent extends RuntimeEvent {
 
 	public static final EndOfPartitionEvent INSTANCE = new EndOfPartitionEvent();
-
+
+
 	@Override
-	public void read(DataInputView in) throws IOException {
+	public void read(DataInputView in) {
 		// Nothing to do here
 	}
 
 	@Override
-	public void write(DataOutputView out) throws IOException {
+	public void write(DataOutputView out) {
 		// Nothing to do here
 	}
+
+	@Override
+	public int hashCode() {
+		return 1965146673;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		return obj != null && obj.getClass() == EndOfPartitionEvent.class;
+	}
+
+	@Override
+	public String toString() {
+		return getClass().getSimpleName();
+	}
 }

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java

@@ -408,7 +408,7 @@ public BufferOrEvent getNextBufferOrEvent() throws IOException, InterruptedExcep
 
 		// Sanity check that notifications only happen when data is available
 		if (buffer == null) {
-			throw new IllegalStateException("Bug in input gate/channel logic: input gate got" +
+			throw new IllegalStateException("Bug in input gate/channel logic: input gate got " +
 					"notified by channel about available data, but none was available.");
 		}