Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/4581#discussion_r148537550
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/SpillableSubpartitionTest.java
---
@@ -231,31 +255,48 @@ public void
testConsumeSpillablePartitionSpilledDuringConsume() throws Exception
// Initial notification
assertEquals(1, listener.getNumNotifiedBuffers());
+ assertFalse(buffer.isRecycled());
Buffer read = reader.getNextBuffer();
- assertNotNull(read);
+ assertSame(buffer, read);
read.recycle();
assertEquals(2, listener.getNumNotifiedBuffers());
+ assertFalse(buffer.isRecycled());
// Spill now
assertEquals(2, partition.releaseMemory());
+ assertFalse(buffer.isRecycled()); // still one in the reader!
listener.awaitNotifications(4, 30_000);
assertEquals(4, listener.getNumNotifiedBuffers());
read = reader.getNextBuffer();
- assertNotNull(read);
+ assertSame(buffer, read);
read.recycle();
+ // now the buffer may be freed, depending on the timing of the
write operation
+ // -> let's do this check at the end of the test (to save some
time)
read = reader.getNextBuffer();
assertNotNull(read);
+ assertNotSame(buffer, read);
+ assertFalse(read.isRecycled());
read.recycle();
+ assertTrue(read.isRecycled());
// End of partition
read = reader.getNextBuffer();
assertNotNull(read);
assertEquals(EndOfPartitionEvent.class,
EventSerializer.fromBuffer(read,
ClassLoader.getSystemClassLoader()).getClass());
+ assertFalse(read.isRecycled());
read.recycle();
+ assertTrue(read.isRecycled());
+
+ // finally check that the buffer has been freed after a
successful (or failed) write
--- End diff --
ditto failed write case and does this test have anything to do with fixed
bug?
---
