### Question:
RecordWriter { public void broadcastEvent(AbstractEvent event) throws IOException { try (BufferConsumer eventBufferConsumer = EventSerializer.toBufferConsumer(event)) { for (int targetChannel = 0; targetChannel < numChannels; targetChannel++) { RecordSerializer<T> serializer = serializers[targetChannel]; tryFinishCurrentBufferBuilder(targetChannel, serializer); targetPartition.addBufferConsumer(eventBufferConsumer.copy(), targetChannel); } if (flushAlways) { flushAll(); } } } RecordWriter(ResultPartitionWriter writer); @SuppressWarnings("unchecked")  RecordWriter(ResultPartitionWriter writer, ChannelSelector<T> channelSelector);  RecordWriter(ResultPartitionWriter writer, ChannelSelector<T> channelSelector, boolean flushAlways); void emit(T record); void broadcastEmit(T record); void randomEmit(T record); void broadcastEvent(AbstractEvent event); void flushAll(); void clearBuffers(); void setMetricGroup(TaskIOMetricGroup metrics);  }### Answer:
@Test public void testBroadcastEventNoRecords() throws Exception { int numChannels = 4; int bufferSize = 32; @SuppressWarnings("unchecked") Queue<BufferConsumer>[] queues = new Queue[numChannels]; for (int i = 0; i < numChannels; i++) { queues[i] = new ArrayDeque<>(); } TestPooledBufferProvider bufferProvider = new TestPooledBufferProvider(Integer.MAX_VALUE, bufferSize); ResultPartitionWriter partitionWriter = new CollectingPartitionWriter(queues, bufferProvider); RecordWriter<ByteArrayIO> writer = new RecordWriter<>(partitionWriter, new RoundRobin<ByteArrayIO>()); CheckpointBarrier barrier = new CheckpointBarrier(Integer.MAX_VALUE + 919192L, Integer.MAX_VALUE + 18828228L, CheckpointOptions.forCheckpointWithDefaultLocation()); writer.broadcastEvent(barrier); assertEquals(0, bufferProvider.getNumberOfCreatedBuffers()); for (int i = 0; i < numChannels; i++) { assertEquals(1, queues[i].size()); BufferOrEvent boe = parseBuffer(queues[i].remove(), i); assertTrue(boe.isEvent()); assertEquals(barrier, boe.getEvent()); assertEquals(0, queues[i].size()); } }
@Test public void testBroadcastEventBufferReferenceCounting() throws Exception { @SuppressWarnings("unchecked") ArrayDeque<BufferConsumer>[] queues = new ArrayDeque[] { new ArrayDeque(), new ArrayDeque() }; ResultPartitionWriter partition = new CollectingPartitionWriter(queues, new TestPooledBufferProvider(Integer.MAX_VALUE)); RecordWriter<?> writer = new RecordWriter<>(partition); writer.broadcastEvent(EndOfPartitionEvent.INSTANCE); assertEquals(1, queues[0].size()); assertEquals(1, queues[1].size()); BufferConsumer bufferConsumer1 = queues[0].getFirst(); BufferConsumer bufferConsumer2 = queues[1].getFirst(); for (int i = 0; i < queues.length; i++) { assertTrue(parseBuffer(queues[i].remove(), i).isEvent()); } assertTrue(bufferConsumer1.isRecycled()); assertTrue(bufferConsumer2.isRecycled()); }
@Test public void testBroadcastEventBufferIndependence() throws Exception { @SuppressWarnings("unchecked") ArrayDeque<BufferConsumer>[] queues = new ArrayDeque[]{new ArrayDeque(), new ArrayDeque()}; ResultPartitionWriter partition = new CollectingPartitionWriter(queues, new TestPooledBufferProvider(Integer.MAX_VALUE)); RecordWriter<?> writer = new RecordWriter<>(partition); writer.broadcastEvent(EndOfPartitionEvent.INSTANCE); assertEquals(1, queues[0].size()); assertEquals(1, queues[1].size()); Buffer buffer1 = buildSingleBuffer(queues[0].remove()); Buffer buffer2 = buildSingleBuffer(queues[1].remove()); assertEquals(0, buffer1.getReaderIndex()); assertEquals(0, buffer2.getReaderIndex()); buffer1.setReaderIndex(1); assertEquals("Buffer 2 shares the same reader index as buffer 1", 0, buffer2.getReaderIndex()); }