[FLINK-9087] [runtime] change the method signature of RecordWriter.br…

…oadcastEvent() from BufferConsumer to void

This closes apache#5802.

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java

@@ -150,7 +150,7 @@ private void sendToTarget(T record, int targetChannel) throws IOException, Inter
  }
  }
 
- public BufferConsumer broadcastEvent(AbstractEvent event) throws IOException {
+ 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];
@@ -164,7 +164,6 @@ public BufferConsumer broadcastEvent(AbstractEvent event) throws IOException {
  if (flushAlways) {
  flushAll();
  }
- return eventBufferConsumer;
  }
  }
 

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordWriterTest.java

@@ -299,18 +299,23 @@ public void testBroadcastEventBufferReferenceCounting() throws Exception {
  new CollectingPartitionWriter(queues, new TestPooledBufferProvider(Integer.MAX_VALUE));
  RecordWriter<?> writer = new RecordWriter<>(partition);
 
- BufferConsumer bufferConsumer = writer.broadcastEvent(EndOfPartitionEvent.INSTANCE);
+ writer.broadcastEvent(EndOfPartitionEvent.INSTANCE);
 
  // Verify added to all queues
  assertEquals(1, queues[0].size());
  assertEquals(1, queues[1].size());
 
+ // get references to buffer consumers (copies from the original event buffer consumer)
+ BufferConsumer bufferConsumer1 = queues[0].getFirst();
+ BufferConsumer bufferConsumer2 = queues[1].getFirst();
+
  // process all collected events (recycles the buffer)
  for (int i = 0; i < queues.length; i++) {
  assertTrue(parseBuffer(queues[i].remove(), i).isEvent());
  }
 
- assertTrue(bufferConsumer.isRecycled());
+ assertTrue(bufferConsumer1.isRecycled());
+ assertTrue(bufferConsumer2.isRecycled());
  }
 
  /**