You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/04/23 15:50:55 UTC
flink git commit: [FLINK-1930] [runtime] Improve exception when
bufferpools have been shut down.
Repository: flink
Updated Branches:
refs/heads/master 4672e95ef -> 4dbf030a6
[FLINK-1930] [runtime] Improve exception when bufferpools have been shut down.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4dbf030a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4dbf030a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4dbf030a
Branch: refs/heads/master
Commit: 4dbf030a6b0415832862c3fd0c3fe7403878a998
Parents: 4672e95
Author: Stephan Ewen <se...@apache.org>
Authored: Wed Apr 22 17:52:21 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Thu Apr 23 15:15:49 2015 +0200
----------------------------------------------------------------------
.../runtime/io/network/buffer/LocalBufferPool.java | 2 +-
.../io/network/buffer/LocalBufferPoolTest.java | 16 ++++++++++++++--
.../io/network/buffer/NetworkBufferPoolTest.java | 17 +++++++++++++++--
3 files changed, 30 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4dbf030a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
index 1da2b8b..4cb1521 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java
@@ -141,7 +141,7 @@ class LocalBufferPool implements BufferPool {
while (availableMemorySegments.isEmpty()) {
if (isDestroyed) {
- return null;
+ throw new IllegalStateException("Buffer pool is destroyed.");
}
if (numberOfRequestedMemorySegments < currentPoolSize) {
http://git-wip-us.apache.org/repos/asf/flink/blob/4dbf030a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
index 9f04d94..e8e9ec8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
@@ -113,7 +113,13 @@ public class LocalBufferPoolTest {
public void testRequestAfterDestroy() throws IOException {
localBufferPool.lazyDestroy();
- assertNull(localBufferPool.requestBuffer());
+ try {
+ localBufferPool.requestBuffer();
+ fail("Call should have failed with an IllegalStateException");
+ }
+ catch (IllegalStateException e) {
+ // we expect exactly that
+ }
}
@Test
@@ -292,7 +298,13 @@ public class LocalBufferPoolTest {
// Try to request the next buffer (but pool should be destroyed either right before
// the request or more likely during the request).
- assertNull(localBufferPool.requestBufferBlocking());
+ try {
+ localBufferPool.requestBufferBlocking();
+ fail("Call should have failed with an IllegalStateException");
+ }
+ catch (IllegalStateException e) {
+ // we expect exactly that
+ }
return requested;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/4dbf030a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
index b24082e..6b22cd9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
@@ -120,8 +120,21 @@ public class NetworkBufferPoolTest {
assertEquals(globalPool.getTotalNumberOfMemorySegments(), globalPool.getNumberOfAvailableMemorySegments());
// can request no more buffers
- assertNull(fixedPool.requestBuffer());
- assertNull(nonFixedPool.requestBuffer());
+ try {
+ fixedPool.requestBuffer();
+ fail("Should fail with an IllegalStateException");
+ }
+ catch (IllegalStateException e) {
+ // that's the way we like it, aha, aha
+ }
+
+ try {
+ nonFixedPool.requestBuffer();
+ fail("Should fail with an IllegalStateException");
+ }
+ catch (IllegalStateException e) {
+ // stayin' alive
+ }
// can create a new pool now
assertNotNull(globalPool.createBufferPool(10, false));