### Question:
PipelinedSubpartition extends ResultSubpartition { @Override public PipelinedSubpartitionView createReadView(BufferAvailabilityListener availabilityListener) throws IOException { synchronized (buffers) { checkState(!isReleased); checkState(readView == null, "Subpartition %s of is being (or already has been) consumed, " + "but pipelined subpartitions can only be consumed once.", index, parent.getPartitionId()); LOG.debug("Creating read view for subpartition {} of partition {}.", index, parent.getPartitionId()); readView = new PipelinedSubpartitionView(this, availabilityListener); if (!buffers.isEmpty()) { notifyDataAvailable(); } } return readView; } PipelinedSubpartition(int index, ResultPartition parent); @Override boolean add(BufferConsumer bufferConsumer); @Override void flush(); @Override void finish(); @Override void release(); @Override int releaseMemory(); @Override boolean isReleased(); @Override PipelinedSubpartitionView createReadView(BufferAvailabilityListener availabilityListener); boolean isAvailable(); @Override String toString(); @Override int unsynchronizedGetNumberOfQueuedBuffers();  }### Answer:
@Test public void testIllegalReadViewRequest() throws Exception { final PipelinedSubpartition subpartition = createSubpartition(); assertNotNull(subpartition.createReadView(new NoOpBufferAvailablityListener())); try { subpartition.createReadView(new NoOpBufferAvailablityListener()); fail("Did not throw expected exception after duplicate notifyNonEmpty view request."); } catch (IllegalStateException expected) { } }