@Test public void testBroadcastEventNoRecords() throws Exception { int numberOfChannels = 4; int bufferSize = 32; @SuppressWarnings("unchecked") Queue<BufferConsumer>[] queues = new Queue[numberOfChannels]; for (int i = 0; i < numberOfChannels; 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); 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 < numberOfChannels; 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()); } }
public void broadcastEvent(AbstractEvent event) throws IOException { try (BufferConsumer eventBufferConsumer = EventSerializer.toBufferConsumer(event)) { for (int targetChannel = 0; targetChannel < numberOfChannels; targetChannel++) { tryFinishCurrentBufferBuilder(targetChannel); targetPartition.addBufferConsumer(eventBufferConsumer.copy(), targetChannel); } if (flushAlways) { flushAll(); } } }
RecordWriter { public void broadcastEvent(AbstractEvent event) throws IOException { try (BufferConsumer eventBufferConsumer = EventSerializer.toBufferConsumer(event)) { for (int targetChannel = 0; targetChannel < numberOfChannels; targetChannel++) { tryFinishCurrentBufferBuilder(targetChannel); targetPartition.addBufferConsumer(eventBufferConsumer.copy(), targetChannel); } if (flushAlways) { flushAll(); } } } }
RecordWriter { public void broadcastEvent(AbstractEvent event) throws IOException { try (BufferConsumer eventBufferConsumer = EventSerializer.toBufferConsumer(event)) { for (int targetChannel = 0; targetChannel < numberOfChannels; targetChannel++) { tryFinishCurrentBufferBuilder(targetChannel); targetPartition.addBufferConsumer(eventBufferConsumer.copy(), targetChannel); } if (flushAlways) { flushAll(); } } } RecordWriter(ResultPartitionWriter writer);  RecordWriter(
			ResultPartitionWriter writer,
			ChannelSelector<T> channelSelector,
			long timeout,
			String taskName); }
RecordWriter { public void broadcastEvent(AbstractEvent event) throws IOException { try (BufferConsumer eventBufferConsumer = EventSerializer.toBufferConsumer(event)) { for (int targetChannel = 0; targetChannel < numberOfChannels; targetChannel++) { tryFinishCurrentBufferBuilder(targetChannel); targetPartition.addBufferConsumer(eventBufferConsumer.copy(), targetChannel); } if (flushAlways) { flushAll(); } } } RecordWriter(ResultPartitionWriter writer);  RecordWriter(
			ResultPartitionWriter writer,
			ChannelSelector<T> channelSelector,
			long timeout,
			String taskName); void emit(T record); void broadcastEmit(T record); void randomEmit(T record); void broadcastEvent(AbstractEvent event); void flushAll(); void clearBuffers(); void setMetricGroup(TaskIOMetricGroup metrics); void close(); static RecordWriter createRecordWriter(
			ResultPartitionWriter writer,
			ChannelSelector channelSelector,
			long timeout,
			String taskName); static RecordWriter createRecordWriter(
			ResultPartitionWriter writer,
			ChannelSelector channelSelector,
			String taskName); }
RecordWriter { public void broadcastEvent(AbstractEvent event) throws IOException { try (BufferConsumer eventBufferConsumer = EventSerializer.toBufferConsumer(event)) { for (int targetChannel = 0; targetChannel < numberOfChannels; targetChannel++) { tryFinishCurrentBufferBuilder(targetChannel); targetPartition.addBufferConsumer(eventBufferConsumer.copy(), targetChannel); } if (flushAlways) { flushAll(); } } } RecordWriter(ResultPartitionWriter writer);  RecordWriter(
			ResultPartitionWriter writer,
			ChannelSelector<T> channelSelector,
			long timeout,
			String taskName); void emit(T record); void broadcastEmit(T record); void randomEmit(T record); void broadcastEvent(AbstractEvent event); void flushAll(); void clearBuffers(); void setMetricGroup(TaskIOMetricGroup metrics); void close(); static RecordWriter createRecordWriter(
			ResultPartitionWriter writer,
			ChannelSelector channelSelector,
			long timeout,
			String taskName); static RecordWriter createRecordWriter(
			ResultPartitionWriter writer,
			ChannelSelector channelSelector,
			String taskName);  }