@Test
public void testStateBackendLoadingAndClosing() throws Exception {

	Configuration taskManagerConfig = new Configuration();
	taskManagerConfig.setString(CoreOptions.STATE_BACKEND, MockStateBackend.class.getName());
	StreamConfig cfg = new StreamConfig(new Configuration());
	cfg.setOperatorID(new OperatorID(4711L, 42L));
	cfg.setStreamOperator(new StreamSource<>(new MockSourceFunction()));
	cfg.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);
	Task task = createTask(StateBackendTestSource.class, cfg, taskManagerConfig);
	StateBackendTestSource.fail = false;
	task.startTaskThread();
	// wait for clean termination
	task.getExecutingThread().join();
	// ensure that the state backends are closed
	verify(StateBackendTestSource.operatorStateBackend).close();
	verify(StateBackendTestSource.keyedStateBackend).close();
	assertEquals(ExecutionState.FINISHED, task.getExecutionState());
}