ResultPartitionWriter implements EventListener<TaskEvent> { public void writeBufferToAllChannels(final Buffer eventBuffer) throws IOException { try { for (int targetChannel = 0; targetChannel < partition.getNumberOfSubpartitions(); targetChannel++) { eventBuffer.retain(); writeBuffer(eventBuffer, targetChannel); } } finally { eventBuffer.recycle(); } } ResultPartitionWriter(ResultPartition partition); ResultPartitionID getPartitionId(); BufferProvider getBufferProvider(); int getNumberOfOutputChannels(); int getNumTargetKeyGroups(); void writeBuffer(Buffer buffer, int targetChannel); void writeBufferToAllChannels(final Buffer eventBuffer); void subscribeToEvent(EventListener<TaskEvent> eventListener, Class<? extends TaskEvent> eventType); @Override void onEvent(TaskEvent event);  }
@Test public void testWriteBufferToAllChannelsReferenceCounting() throws Exception { Buffer buffer = EventSerializer.toBuffer(EndOfPartitionEvent.INSTANCE); ResultPartition partition = new ResultPartition( "TestTask", mock(TaskActions.class), new JobID(), new ResultPartitionID(), ResultPartitionType.PIPELINED, 2, 2, mock(ResultPartitionManager.class), mock(ResultPartitionConsumableNotifier.class), mock(IOManager.class), false); ResultPartitionWriter partitionWriter = new ResultPartitionWriter( partition); partitionWriter.writeBufferToAllChannels(buffer); assertEquals(2, partition.getTotalNumberOfBuffers()); partition.release(); assertTrue(buffer.isRecycled()); }