@Override
    public void resetToCheckpoint(final long checkpointId, @Nullable final byte[] checkpointData) {
        // First bump up the coordinator epoch to fence out the active coordinator.
        LOG.info("Resetting coordinator to checkpoint.");
        // Replace the coordinator variable with a new DeferrableCoordinator instance.
        // At this point the internal coordinator of the new coordinator has not been created.
        // After this point all the subsequent calls will be made to the new coordinator.
        final DeferrableCoordinator oldCoordinator = coordinator;
        final DeferrableCoordinator newCoordinator =
                new DeferrableCoordinator(context.getOperatorId());
        coordinator = newCoordinator;
        // Close the old coordinator asynchronously in a separate closing thread.
        // The future will be completed when the old coordinator closes.
        CompletableFuture<Void> closingFuture = oldCoordinator.closeAsync(closingTimeoutMs);

        // Create and possibly start the coordinator and apply all meanwhile deferred calls
        // capture the status whether the coordinator was started when this method was called
        final boolean wasStarted = this.started;

        closingFuture.whenComplete(
                (ignored, e) -> {
                    if (e != null) {
                        LOG.warn(
                                String.format(
                                        "Received exception when closing "
                                                + "operator coordinator for %s.",
                                        oldCoordinator.operatorId),
                                e);
                    }
                    if (!closed) {
                        // The previous coordinator has closed. Create a new one.
                        newCoordinator.createNewInternalCoordinator(context, provider);
                        newCoordinator.resetAndStart(checkpointId, checkpointData, wasStarted);
                        newCoordinator.processPendingCalls();
                    }
                });
    }
@Test
    void testResetToCheckpointTimeout() throws Exception {
        final long closingTimeoutMs = 1L;
        // Let the user coordinator block on close.
        TestingCoordinatorProvider provider = new TestingCoordinatorProvider(new CountDownLatch(1));
        MockOperatorCoordinatorContext context =
                new MockOperatorCoordinatorContext(OPERATOR_ID, NUM_SUBTASKS);
        RecreateOnResetOperatorCoordinator coordinator =
                (RecreateOnResetOperatorCoordinator) provider.create(context, closingTimeoutMs);

        coordinator.resetToCheckpoint(2L, new byte[0]);
        CommonTestUtils.waitUtil(
                context::isJobFailed,
                Duration.ofSeconds(5),
                "The job should fail due to resetToCheckpoint() timeout.");
    }