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

	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_reduce", IntSerializer.INSTANCE);
		ReducingStateDescriptor<String> kvId = new ReducingStateDescriptor<>("id",
			new ReduceFunction<String>() {
				private static final long serialVersionUID = 1L;
				@Override
				public String reduce(String value1, String value2) throws Exception {
					return value1 + "," + value2;
				}
			},
			StringSerializer.INSTANCE);
		ReducingState<String> state = backend.getPartitionedState(null, VoidSerializer.INSTANCE, kvId);
		@SuppressWarnings("unchecked")
		KvState<Integer, Void, ReducingState<String>, ReducingStateDescriptor<String>, DbStateBackend> kv =
			(KvState<Integer, Void, ReducingState<String>, ReducingStateDescriptor<String>, DbStateBackend>) state;
		Joiner joiner = Joiner.on(",");
		// some modifications to the state
		kv.setCurrentKey(1);
		assertEquals(null, state.get());
		state.add("1");
		kv.setCurrentKey(2);
		assertEquals(null, state.get());
		state.add("2");
		kv.setCurrentKey(1);
		assertEquals("1", state.get());
		// draw a snapshot
		KvStateSnapshot<Integer, Void, ReducingState<String>, ReducingStateDescriptor<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, ReducingState<String>, ReducingStateDescriptor<String>, DbStateBackend> snapshot2 =
			kv.snapshot(682375462379L, 4);
		// validate the original state
		kv.setCurrentKey(1);
		assertEquals("1,u1", state.get());
		kv.setCurrentKey(2);
		assertEquals("2,u2", state.get());
		kv.setCurrentKey(3);
		assertEquals("u3", state.get());
		kv.dispose();
		// restore the second snapshot and validate it
		KvState<Integer, Void, ReducingState<String>, ReducingStateDescriptor<String>, DbStateBackend> restored2 = snapshot2.restoreState(
			backend,
			IntSerializer.INSTANCE,
			this.getClass().getClassLoader(), 20);
		@SuppressWarnings("unchecked")
		ReducingState<String> restored2State = (ReducingState<String>) restored2;
		restored2.setCurrentKey(1);
		assertEquals("1,u1", restored2State.get());
		restored2.setCurrentKey(2);
		assertEquals("2,u2", restored2State.get());
		restored2.setCurrentKey(3);
		assertEquals("u3", restored2State.get());
	}
	catch (Exception e) {
		e.printStackTrace();
		fail(e.getMessage());
	}
}