@Override
    public void onLeave(Class<? extends State> newState) {
        if (!StateWithExecutionGraph.class.isAssignableFrom(newState)) {
            // we are leaving the StateWithExecutionGraph --> we need to dispose temporary services
            operatorCoordinatorHandler.disposeAllOperatorCoordinators();
        }
    }
@Test
    void testOperatorCoordinatorShutdownOnLeave() throws Exception {
        try (MockStateWithExecutionGraphContext context =
                new MockStateWithExecutionGraphContext()) {

            final TestingOperatorCoordinatorHandler testingOperatorCoordinatorHandler =
                    new TestingOperatorCoordinatorHandler();
            final TestingStateWithExecutionGraph stateWithExecutionGraph =
                    createStateWithExecutionGraph(context, testingOperatorCoordinatorHandler);

            stateWithExecutionGraph.onLeave(AdaptiveSchedulerTest.DummyState.class);

            assertThat(testingOperatorCoordinatorHandler.isDisposed()).isTrue();
        }
    }