@Test
public void testReadMultipleFilesWithSingleBufferPool() throws Exception {

	ExecutorService executor = null;
	BufferFileWriter[] writers = null;
	ResultSubpartitionView[] readers = null;
	try {
		executor = Executors.newCachedThreadPool();
		// Setup
		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);
		// Wait for writers to finish
		for (BufferFileWriter writer : writers) {
			writer.close();
		}
		// Create the views depending on the test configuration
		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, // +1 for end of partition event
				consumers[i]);
			consumers[i].setSubpartitionView(readers[i]);
		}
		final List<Future<Boolean>> results = Lists.newArrayList();
		// Submit the consuming tasks
		for (TestSubpartitionConsumer consumer : consumers) {
			results.add(executor.submit(consumer));
		}
		// Wait for the results
		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();
		}
	}
}