@Override
    public void start() throws Exception {
        LOG.info("Starting split enumerator for source {}.", operatorName);

        // we mark this as started first, so that we can later distinguish the cases where
        // 'start()' wasn't called and where 'start()' failed.
        started = true;

        // there are two ways the SplitEnumerator can get created:
        //  (1) Source.restoreEnumerator(), in which case the 'resetToCheckpoint()' method creates
        // it
        //  (2) Source.createEnumerator, in which case it has not been created, yet, and we create
        // it here
        if (enumerator == null) {
            final ClassLoader userCodeClassLoader =
                    context.getCoordinatorContext().getUserCodeClassloader();
            try (TemporaryClassLoaderContext ignored =
                    TemporaryClassLoaderContext.of(userCodeClassLoader)) {
                enumerator = source.createEnumerator(context);
            } catch (Throwable t) {
                ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
                LOG.error("Failed to create Source Enumerator for source {}", operatorName, t);
                context.failJob(t);
                return;
            }
        }

        // The start sequence is the first task in the coordinator executor.
        // We rely on the single-threaded coordinator executor to guarantee
        // the other methods are invoked after the enumerator has started.
        runInEventLoop(() -> enumerator.start(), "starting the SplitEnumerator.");

        if (coordinatorListeningID != null) {
            coordinatorStore.compute(
                    coordinatorListeningID,
                    (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 OperatorCoordinator) {
                            // The coordinator has not registered or needs to be recreated after
                            // global failover.
                            return this;
                        } else {
                            checkState(
                                    oldValue instanceof OperatorEvent,
                                    "The existing value for "
                                            + coordinatorStore
                                            + "is expected to be an operator event, but it is in fact "
                                            + oldValue);
                            LOG.info(
                                    "Handling event {} received before the source coordinator with ID {} is registered",
                                    oldValue,
                                    coordinatorListeningID);
                            handleEventFromOperator(0, 0, (OperatorEvent) oldValue);

                            // Since for non-global failover the coordinator will not be recreated
                            // and for global failover both the sender and receiver need to restart,
                            // the coordinator will receive the event only once.
                            // As the event has been processed, it can be removed safely and there's
                            // no need to register the coordinator for further events as well.
                            return null;
                        }
                    });
        }

        if (watermarkAlignmentParams.isEnabled()) {
            LOG.info("Starting schedule the period announceCombinedWatermark task");
            coordinatorStore.putIfAbsent(
                    watermarkAlignmentParams.getWatermarkGroup(), new WatermarkAggregator<>());
            context.schedulePeriodTask(
                    this::announceCombinedWatermark,
                    watermarkAlignmentParams.getUpdateInterval(),
                    watermarkAlignmentParams.getUpdateInterval(),
                    TimeUnit.MILLISECONDS);
        }
    }
@Test
    void testStart() throws Exception {
        sourceCoordinator.start();
        waitForCoordinatorToProcessActions();

        assertThat(getEnumerator().isStarted()).isTrue();
    }