flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From u..@apache.org
Subject flink git commit: [FLINK-2412] [runtime] Check if parent released before querying in-memory buffer in SpillableSubpartitionView
Date Fri, 31 Jul 2015 09:02:52 GMT
Repository: flink
Updated Branches:
  refs/heads/master 3c236b2c8 -> a56aad74e


[FLINK-2412] [runtime] Check if parent released before querying in-memory buffer in SpillableSubpartitionView


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a56aad74
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a56aad74
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a56aad74

Branch: refs/heads/master
Commit: a56aad74eb35a19623d58b6dd0228ec5bc1bb422
Parents: 3c236b2c
Author: Ufuk Celebi <uce@apache.org>
Authored: Thu Jul 30 13:49:52 2015 +0200
Committer: Ufuk Celebi <uce@apache.org>
Committed: Fri Jul 31 11:02:42 2015 +0200

----------------------------------------------------------------------
 .../partition/SpillableSubpartition.java        |  2 +-
 .../partition/SpillableSubpartitionView.java    |  4 +++
 .../partition/SpillableSubpartitionTest.java    | 33 ++++++++++++++++++--
 3 files changed, 36 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/a56aad74/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartition.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartition.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartition.java
index 91f2042..4a18691 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartition.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartition.java
@@ -59,7 +59,7 @@ class SpillableSubpartition extends ResultSubpartition {
 	private boolean isFinished;
 
 	/** Flag indicating whether the subpartition has been released. */
-	private boolean isReleased;
+	boolean isReleased;
 
 	/** The read view to consume this subpartition. */
 	private ResultSubpartitionView readView;

http://git-wip-us.apache.org/repos/asf/flink/blob/a56aad74/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
index d37f042..972e34b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionView.java
@@ -73,6 +73,10 @@ class SpillableSubpartitionView implements ResultSubpartitionView {
 
 		// 1) In-memory
 		synchronized (parent.buffers) {
+			if (parent.isReleased) {
+				return null;
+			}
+
 			if (parent.spillWriter == null) {
 				if (currentQueuePosition < numberOfBuffers) {
 					Buffer buffer = parent.buffers.get(currentQueuePosition);

http://git-wip-us.apache.org/repos/asf/flink/blob/a56aad74/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
index c530eff..d7e56c8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.runtime.io.disk.iomanager.AsynchronousBufferFileWriter;
 import org.apache.flink.runtime.io.disk.iomanager.FileIOChannel;
 import org.apache.flink.runtime.io.disk.iomanager.IOManager;
 import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
-import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.util.TestInfiniteBufferProvider;
 import org.junit.AfterClass;
 import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
@@ -35,9 +35,12 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 import static org.apache.flink.runtime.io.disk.iomanager.IOManager.IOMode.SYNC;
+import static org.junit.Assert.assertNull;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
 
 public class SpillableSubpartitionTest extends SubpartitionTestBase {
@@ -59,7 +62,6 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
 		return new SpillableSubpartition(0, mock(ResultPartition.class), ioManager, SYNC);
 	}
 
-
 	/**
 	 * Tests a fix for FLINK-2384.
 	 *
@@ -118,4 +120,31 @@ public class SpillableSubpartitionTest extends SubpartitionTestBase {
 		doneLatch.countDown();
 		blockingFinish.get();
 	}
+
+	/**
+	 * Tests a fix for FLINK-2412.
+	 *
+	 * @see <a href="https://issues.apache.org/jira/browse/FLINK-2412">FLINK-2412</a>
+	 */
+	@Test
+	public void testReleasePartitionAndGetNext() throws Exception {
+		// Create partition and add some buffers
+		SpillableSubpartition partition = new SpillableSubpartition(
+				0, mock(ResultPartition.class), ioManager, SYNC);
+
+		partition.finish();
+
+		// Create the read view
+		ResultSubpartitionView readView = spy(partition
+				.createReadView(new TestInfiniteBufferProvider()));
+
+		// The released state check (of the parent) needs to be independent
+		// of the released state of the view.
+		doNothing().when(readView).releaseAllResources();
+
+		// Release the partition, but the view does not notice yet.
+		partition.release();
+
+		assertNull(readView.getNextBuffer());
+	}
 }


Mime
View raw message