@Test
@SuppressWarnings("unchecked,rawtypes")
public void testListState() {

	File tempDir = new File(ConfigConstants.DEFAULT_TASK_MANAGER_TMP_PATH, UUID.randomUUID().toString());
	try {
		FsStateBackend fileBackend = new FsStateBackend(localFileUri(tempDir));
		DbStateBackend backend = new DbStateBackend(conf, fileBackend);
		Environment env = new DummyEnvironment("test", 2, 0);
		backend.initializeForJob(env, "dummy_test_kv_list", IntSerializer.INSTANCE);
		ListStateDescriptor<String> kvId = new ListStateDescriptor<>("id", StringSerializer.INSTANCE);
		ListState<String> state = backend.getPartitionedState(null, VoidSerializer.INSTANCE, kvId);
		@SuppressWarnings("unchecked")
		KvState<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend> kv =
			(KvState<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend>) state;
		Joiner joiner = Joiner.on(",");
		// some modifications to the state
		kv.setCurrentKey(1);
		assertEquals("", joiner.join(state.get()));
		state.add("1");
		kv.setCurrentKey(2);
		assertEquals("", joiner.join(state.get()));
		state.add("2");
		kv.setCurrentKey(1);
		assertEquals("1", joiner.join(state.get()));
		// draw a snapshot
		KvStateSnapshot<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend> snapshot1 =
			kv.snapshot(682375462378L, 2);
		// make some more modifications
		kv.setCurrentKey(1);
		state.add("u1");
		kv.setCurrentKey(2);
		state.add("u2");
		kv.setCurrentKey(3);
		state.add("u3");
		// draw another snapshot
		KvStateSnapshot<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend> snapshot2 =
			kv.snapshot(682375462379L, 4);
		// validate the original state
		kv.setCurrentKey(1);
		assertEquals("1,u1", joiner.join(state.get()));
		kv.setCurrentKey(2);
		assertEquals("2,u2", joiner.join(state.get()));
		kv.setCurrentKey(3);
		assertEquals("u3", joiner.join(state.get()));
		kv.dispose();
		// restore the second snapshot and validate it
		KvState<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend> restored2 = snapshot2.restoreState(
			backend,
			IntSerializer.INSTANCE,
			this.getClass().getClassLoader(), 20);
		@SuppressWarnings("unchecked")
		ListState<String> restored2State = (ListState<String>) restored2;
		restored2.setCurrentKey(1);
		assertEquals("1,u1", joiner.join(restored2State.get()));
		restored2.setCurrentKey(2);
		assertEquals("2,u2", joiner.join(restored2State.get()));
		restored2.setCurrentKey(3);
		assertEquals("u3", joiner.join(restored2State.get()));
	}
	catch (Exception e) {
		e.printStackTrace();
		fail(e.getMessage());
	}
}