@Override
    public void broadcastEvent(AbstractEvent event, boolean isPriorityEvent) throws IOException {
        Buffer buffer = EventSerializer.toBuffer(event, isPriorityEvent);
        try {
            ByteBuffer serializedEvent = buffer.getNioBufferReadable();
            broadcast(serializedEvent, buffer.getDataType());
        } finally {
            buffer.recycleBuffer();
        }
    }
@Test
    void testBroadcastEvent() throws Exception {
        final int numBuffers = 1;
        BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
        try (HsResultPartition resultPartition = createHsResultPartition(2, bufferPool)) {
            resultPartition.broadcastEvent(EndOfPartitionEvent.INSTANCE, false);
            // broadcast event does not request buffer
            assertThat(bufferPool.getNumberOfAvailableMemorySegments()).isEqualTo(1);

            Tuple2[] viewAndListeners = createSubpartitionViews(resultPartition, 2);

            boolean[] receivedEvent = new boolean[2];
            readData(
                    viewAndListeners,
                    (buffer, subpartition) -> {
                        assertThat(buffer.getDataType().isEvent()).isTrue();
                        try {
                            AbstractEvent event =
                                    EventSerializer.fromSerializedEvent(
                                            buffer.readOnlySlice().getNioBufferReadable(),
                                            HsResultPartitionTest.class.getClassLoader());
                            assertThat(event).isInstanceOf(EndOfPartitionEvent.class);
                            receivedEvent[subpartition] = true;
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    });

            assertThat(receivedEvent).containsExactly(true, true);
        }
    }