SpilledSubpartitionView implements ResultSubpartitionView, NotificationListener { @Override public void releaseAllResources() throws IOException { if (isReleased.compareAndSet(false, true)) { spillWriter.closeAndDelete(); fileReader.close(); bufferPool.destroy(); } } SpilledSubpartitionView(
		ResultSubpartition parent,
		int memorySegmentSize,
		BufferFileWriter spillWriter,
		long numberOfSpilledBuffers,
		BufferAvailabilityListener availabilityListener); @Override void onNotification(); @Override Buffer getNextBuffer(); @Override void notifyBuffersAvailable(long buffers); @Override void notifySubpartitionConsumed(); @Override void releaseAllResources(); @Override boolean isReleased(); @Override Throwable getFailureCause(); @Override String toString();  }
@Test public void testReadMultipleFilesWithSingleBufferPool() throws Exception { ExecutorService executor = null; BufferFileWriter[] writers = null; ResultSubpartitionView[] readers = null; try { executor = Executors.newCachedThreadPool(); writers = new BufferFileWriter[]{ createWriterAndWriteBuffers(IO_MANAGER, writerBufferPool, 512), createWriterAndWriteBuffers(IO_MANAGER, writerBufferPool, 512) }; readers = new ResultSubpartitionView[writers.length]; TestSubpartitionConsumer[] consumers = new TestSubpartitionConsumer[writers.length]; BufferProvider inputBuffers = new TestPooledBufferProvider(2); ResultSubpartition parent = mock(ResultSubpartition.class); for (BufferFileWriter writer : writers) { writer.close(); } for (int i = 0; i < readers.length; i++) { consumers[i] = new TestSubpartitionConsumer( false, new TestConsumerCallback.RecyclingCallback()); readers[i] = new SpilledSubpartitionView( parent, inputBuffers.getMemorySegmentSize(), writers[i], 512 + 1, consumers[i]); consumers[i].setSubpartitionView(readers[i]); } final List<Future<Boolean>> results = Lists.newArrayList(); for (TestSubpartitionConsumer consumer : consumers) { results.add(executor.submit(consumer)); } for (Future<Boolean> res : results) { try { res.get(2, TimeUnit.MINUTES); } catch (TimeoutException e) { throw new TimeoutException("There has been a timeout in the test. This " + "indicates that there is a bug/deadlock in the tested subpartition " + "view."); } } } finally { if (writers != null) { for (BufferFileWriter writer : writers) { if (writer != null) { writer.deleteChannel(); } } } if (readers != null) { for (ResultSubpartitionView reader : readers) { if (reader != null) { reader.releaseAllResources(); } } } if (executor != null) { executor.shutdown(); } } }