private void handleSourceEvent(int subtask, int attemptNumber, SourceEvent event) {
        LOG.debug(
                "Source {} received custom event from parallel task {} (#{}): {}",
                operatorName,
                subtask,
                attemptNumber,
                event);

        if (context.isConcurrentExecutionAttemptsSupported()) {
            checkState(
                    enumerator instanceof SupportsHandleExecutionAttemptSourceEvent,
                    "The split enumerator %s must implement SupportsHandleExecutionAttemptSourceEvent "
                            + "to be used in concurrent execution attempts scenario (e.g. if "
                            + "speculative execution is enabled).",
                    enumerator.getClass().getCanonicalName());
            ((SupportsHandleExecutionAttemptSourceEvent) enumerator)
                    .handleSourceEvent(subtask, attemptNumber, event);
        } else {
            enumerator.handleSourceEvent(subtask, event);
        }
    }
@Test
    void testHandleSourceEvent() throws Exception {
        sourceReady();

        SourceEvent sourceEvent = new SourceEvent() {};
        sourceCoordinator.handleEventFromOperator(0, 0, new SourceEventWrapper(sourceEvent));
        waitForCoordinatorToProcessActions();

        assertThat(getEnumerator().getHandledSourceEvent()).hasSize(1);
        assertThat(getEnumerator().getHandledSourceEvent().get(0)).isEqualTo(sourceEvent);
    }