[FLINK-1930] [runtime] Improve exception when bufferpools have been s…

…hut down.
VonRosenchild · Apr 23, 2015 · 4dbf030 · 4dbf030
1 parent 4672e95
commit 4dbf030
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 5 deletions.
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
@@ -141,7 +141,7 @@ private Buffer requestBuffer(boolean isBlocking) throws InterruptedException, IO
 
  while (availableMemorySegments.isEmpty()) {
  if (isDestroyed) {
- return null;
+ throw new IllegalStateException("Buffer pool is destroyed.");
  }
 
  if (numberOfRequestedMemorySegments < currentPoolSize) {

diff --git a/...runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java b/...runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java
@@ -113,7 +113,13 @@ public void testRequestMoreThanAvailable() throws IOException {
  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 List<Buffer> call() throws Exception {
 
  // 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;
  }

diff --git a/...ntime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java b/...ntime/src/test/java/org/apache/flink/runtime/io/network/buffer/NetworkBufferPoolTest.java
@@ -120,8 +120,21 @@ public void testDestroyAll() {
  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));