@Override
    public void registerAndStartNewCoordinators(
            Collection<OperatorCoordinatorHolder> coordinators,
            ComponentMainThreadExecutor mainThreadExecutor,
            final int parallelism) {

        for (OperatorCoordinatorHolder coordinator : coordinators) {
            coordinatorMap.put(coordinator.operatorId(), coordinator);
            coordinator.lazyInitialize(
                    globalFailureHandler,
                    mainThreadExecutor,
                    executionGraph.getCheckpointCoordinator(),
                    parallelism);
        }
        startOperatorCoordinators(coordinators);
    }
@Test
    void testRegisterAndStartNewCoordinators() throws Exception {

        final JobVertex[] jobVertices = createJobVertices(BLOCKING);
        OperatorID operatorId1 = OperatorID.fromJobVertexID(jobVertices[0].getID());
        OperatorID operatorId2 = OperatorID.fromJobVertexID(jobVertices[1].getID());

        ExecutionGraph executionGraph = createDynamicGraph(jobVertices);
        ExecutionJobVertex ejv1 = executionGraph.getJobVertex(jobVertices[0].getID());
        ExecutionJobVertex ejv2 = executionGraph.getJobVertex(jobVertices[1].getID());
        executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread());

        executionGraph.initializeJobVertex(ejv1, 0L);

        DefaultOperatorCoordinatorHandler handler =
                new DefaultOperatorCoordinatorHandler(executionGraph, throwable -> {});
        assertThat(handler.getCoordinatorMap().keySet()).contains(operatorId1);

        executionGraph.initializeJobVertex(ejv2, 0L);
        handler.registerAndStartNewCoordinators(
                ejv2.getOperatorCoordinators(),
                executionGraph.getJobMasterMainThreadExecutor(),
                ejv2.getParallelism());

        assertThat(handler.getCoordinatorMap().keySet()).contains(operatorId1, operatorId2);
    }