@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()); }
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);  }