RecordWriter { public void broadcastEvent(AbstractEvent event) throws IOException, InterruptedException { final Buffer eventBuffer = EventSerializer.toBuffer(event); try { for (int targetChannel = 0; targetChannel < numChannels; targetChannel++) { RecordSerializer<T> serializer = serializers[targetChannel]; synchronized (serializer) { Buffer buffer = serializer.getCurrentBuffer(); if (buffer != null) { numBytesOut.inc(buffer.getSize()); writeAndClearBuffer(buffer, targetChannel, serializer); } else if (serializer.hasData()) { throw new IllegalStateException("No buffer, but serializer has buffered data."); } eventBuffer.retain(); targetPartition.writeBuffer(eventBuffer, targetChannel); } } } finally { eventBuffer.recycle(); } } RecordWriter(ResultPartitionWriter writer); @SuppressWarnings("unchecked")  RecordWriter(ResultPartitionWriter writer, ChannelSelector<T> channelSelector); void emit(T record); void broadcastEmit(T record); void randomEmit(T record); void broadcastEvent(AbstractEvent event); void flush(); void clearBuffers(); void setMetricGroup(TaskIOMetricGroup metrics);  }
@Test public void testBroadcastEventNoRecords() throws Exception { int numChannels = 4; int bufferSize = 32; @SuppressWarnings("unchecked") Queue<BufferOrEvent>[] queues = new Queue[numChannels]; for (int i = 0; i < numChannels; i++) { queues[i] = new ArrayDeque<>(); } BufferProvider bufferProvider = createBufferProvider(bufferSize); ResultPartitionWriter partitionWriter = createCollectingPartitionWriter(queues, bufferProvider); RecordWriter<ByteArrayIO> writer = new RecordWriter<>(partitionWriter, new RoundRobin<ByteArrayIO>()); CheckpointBarrier barrier = new CheckpointBarrier(Integer.MAX_VALUE + 919192L, Integer.MAX_VALUE + 18828228L, CheckpointOptions.forCheckpoint()); writer.broadcastEvent(barrier); verify(bufferProvider, times(0)).requestBufferBlocking(); for (Queue<BufferOrEvent> queue : queues) { assertEquals(1, queue.size()); BufferOrEvent boe = queue.remove(); assertTrue(boe.isEvent()); assertEquals(barrier, boe.getEvent()); } }
@Test public void testBroadcastEventBufferReferenceCounting() throws Exception { Buffer buffer = EventSerializer.toBuffer(EndOfPartitionEvent.INSTANCE); PowerMockito .stub(PowerMockito.method(EventSerializer.class, "toBuffer")) .toReturn(buffer); @SuppressWarnings("unchecked") ArrayDeque<BufferOrEvent>[] queues = new ArrayDeque[]{new ArrayDeque(), new ArrayDeque()}; ResultPartitionWriter partition = createCollectingPartitionWriter(queues, new TestInfiniteBufferProvider()); RecordWriter<?> writer = new RecordWriter<>(partition); writer.broadcastEvent(EndOfPartitionEvent.INSTANCE); assertEquals(1, queues[0].size()); assertEquals(1, queues[1].size()); assertTrue(buffer.isRecycled()); }