PipelinedSubpartition extends ResultSubpartition { boolean registerListener(NotificationListener listener) { synchronized (buffers) { if (!buffers.isEmpty() || isReleased) { return false; } if (registeredListener == null) { registeredListener = listener; return true; } throw new IllegalStateException("Already registered listener."); } } PipelinedSubpartition(int index, ResultPartition parent); @Override boolean add(Buffer buffer); @Override void finish(); @Override void release(); @Override int releaseMemory(); @Override boolean isReleased(); @Override PipelinedSubpartitionView createReadView(BufferProvider bufferProvider); @Override String toString();  }
@Test public void testRegisterListener() throws Exception { final PipelinedSubpartition subpartition = createSubpartition(); final TestNotificationListener listener = new TestNotificationListener(); assertTrue(subpartition.registerListener(listener)); try { subpartition.registerListener(listener); fail("Did not throw expected exception after duplicate listener registration."); } catch (IllegalStateException expected) { } }