@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) { } }
@Override public PipelinedSubpartitionView createReadView(BufferAvailabilityListener availabilityListener) throws IOException { final boolean notifyDataAvailable; 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 {}.", parent.getOwningTaskName(), index, parent.getPartitionId()); readView = new PipelinedSubpartitionView(this, availabilityListener); notifyDataAvailable = !buffers.isEmpty(); } if (notifyDataAvailable) { notifyDataAvailable(); } return readView; }
PipelinedSubpartition extends ResultSubpartition { @Override public PipelinedSubpartitionView createReadView(BufferAvailabilityListener availabilityListener) throws IOException { final boolean notifyDataAvailable; 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 {}.", parent.getOwningTaskName(), index, parent.getPartitionId()); readView = new PipelinedSubpartitionView(this, availabilityListener); notifyDataAvailable = !buffers.isEmpty(); } if (notifyDataAvailable) { notifyDataAvailable(); } return readView; } }
PipelinedSubpartition extends ResultSubpartition { @Override public PipelinedSubpartitionView createReadView(BufferAvailabilityListener availabilityListener) throws IOException { final boolean notifyDataAvailable; 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 {}.", parent.getOwningTaskName(), index, parent.getPartitionId()); readView = new PipelinedSubpartitionView(this, availabilityListener); notifyDataAvailable = !buffers.isEmpty(); } if (notifyDataAvailable) { notifyDataAvailable(); } return readView; } PipelinedSubpartition(int index, ResultPartition parent); }
PipelinedSubpartition extends ResultSubpartition { @Override public PipelinedSubpartitionView createReadView(BufferAvailabilityListener availabilityListener) throws IOException { final boolean notifyDataAvailable; 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 {}.", parent.getOwningTaskName(), index, parent.getPartitionId()); readView = new PipelinedSubpartitionView(this, availabilityListener); notifyDataAvailable = !buffers.isEmpty(); } if (notifyDataAvailable) { notifyDataAvailable(); } return readView; } PipelinedSubpartition(int index, ResultPartition parent); @Override boolean add(BufferConsumer bufferConsumer); @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(); @Override void flush(); }
PipelinedSubpartition extends ResultSubpartition { @Override public PipelinedSubpartitionView createReadView(BufferAvailabilityListener availabilityListener) throws IOException { final boolean notifyDataAvailable; 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 {}.", parent.getOwningTaskName(), index, parent.getPartitionId()); readView = new PipelinedSubpartitionView(this, availabilityListener); notifyDataAvailable = !buffers.isEmpty(); } if (notifyDataAvailable) { notifyDataAvailable(); } return readView; } PipelinedSubpartition(int index, ResultPartition parent); @Override boolean add(BufferConsumer bufferConsumer); @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(); @Override void flush();  }