@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());
	// No records emitted yet, broadcast should not request a buffer
	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());
	}
}