@Override
    public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event) {
        DynamicFilteringData currentData =
                ((DynamicFilteringEvent) ((SourceEventWrapper) event).getSourceEvent()).getData();
        if (receivedFilteringData == null) {
            receivedFilteringData = currentData;
        } else {
            // Since there might be speculative execution or failover, we may receive multiple
            // notifications, and we can't tell for sure which one is valid for further processing.
            if (DynamicFilteringData.isEqual(receivedFilteringData, currentData)) {
                // If the notifications contain exactly the same data, everything is alright, and
                // we don't need to send the event again.
                return;
            } else {
                // In case the mismatching of the source filtering result and the dim data, which
                // may leads to incorrect result, trigger global failover for fully recomputing.
                throw new IllegalStateException(
                        "DynamicFilteringData is recomputed but not equal. "
                                + "Triggering global failover in case the result is incorrect. "
                                + " It's recommended to re-run the job with dynamic filtering disabled.");
            }
        }

        for (String listenerID : dynamicFilteringDataListenerIDs) {
            coordinatorStore.compute(
                    listenerID,
                    (key, oldValue) -> {
                        // The value for a listener ID can be a source coordinator listening to an
                        // event, or an event waiting to be retrieved
                        if (oldValue == null || oldValue instanceof OperatorEvent) {
                            // If the listener has not been registered, or after a global failover
                            // without cleanup the store, we simply update it to the latest value.
                            // The listener coordinator would retrieve the event once it's started.
                            LOG.info(
                                    "Updating event {} before the source coordinator with ID {} is registered",
                                    event,
                                    listenerID);
                            return event;
                        } else {
                            checkState(
                                    oldValue instanceof OperatorCoordinator,
                                    "The existing value for "
                                            + listenerID
                                            + "is expected to be an operator coordinator, but it is in fact "
                                            + oldValue);
                            LOG.info(
                                    "Distributing event {} to source coordinator with ID {}",
                                    event,
                                    listenerID);
                            try {
                                // Subtask index and attempt number is not necessary for handling
                                // DynamicFilteringEvent.
                                ((OperatorCoordinator) oldValue)
                                        .handleEventFromOperator(0, 0, event);
                            } catch (Exception e) {
                                ExceptionUtils.rethrow(e);
                            }

                            // Dynamic filtering event is expected to be sent only once. So after
                            // the coordinator is notified, it can be removed from the store.
                            return null;
                        }
                    });
        }
    }
@Test
    void testRedistributeData() throws Exception {
        MockOperatorCoordinatorContext context =
                new MockOperatorCoordinatorContext(new OperatorID(), 1);
        String listenerID1 = "test-listener-1";
        String listenerID2 = "test-listener-2";

        TestingOperatorCoordinator listener1 = new TestingOperatorCoordinator(context);
        TestingOperatorCoordinator listener2 = new TestingOperatorCoordinator(context);
        context.getCoordinatorStore().putIfAbsent(listenerID1, listener1);
        context.getCoordinatorStore().putIfAbsent(listenerID2, listener2);

        RowType rowType = RowType.of(new IntType());
        OperatorEvent testEvent = dynamicFilteringEvent(rowType, Collections.emptyList());
        try (DynamicFilteringDataCollectorOperatorCoordinator coordinator =
                new DynamicFilteringDataCollectorOperatorCoordinator(
                        context, Arrays.asList(listenerID1, listenerID2))) {
            coordinator.handleEventFromOperator(0, 1, testEvent);
        }

        assertThat(listener1.getNextReceivedOperatorEvent()).isSameAs(testEvent);
        assertThat(listener1.getNextReceivedOperatorEvent()).isNull();
        assertThat(listener2.getNextReceivedOperatorEvent()).isSameAs(testEvent);
        assertThat(listener2.getNextReceivedOperatorEvent()).isNull();
    }