flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
/**
	 * Verify that an empty {@code ReduceState} yields {@code null}.
	 */
Verify that an empty  ReduceState yields  null.
Verify that an empty ReduceState yields null .
@Test
	public void testReducingStateDefaultValue() throws Exception {
		AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE);
		ReducingStateDescriptor<String> kvId = new ReducingStateDescriptor<>("id", new AppendingReduce(), String.class);
		ReducingState<String> state = backend.getPartitionedState(
				VoidNamespace.INSTANCE,
				VoidNamespaceSerializer.INSTANCE, kvId);
		backend.setCurrentKey(1);
		assertNull(state.get());
		state.add("Ciao");
		assertEquals("Ciao", state.get());
		state.clear();
		assertNull(state.get());
		backend.dispose();
	}
@ Test public void testReducingStateDefaultValue ( ) throws Exception { AbstractKeyedStateBackend < Integer > backend = createKeyedBackend ( IntSerializer . INSTANCE ) ; ReducingStateDescriptor < String > kvId = new ReducingStateDescriptor < > ( "id" , new AppendingReduce ( ) , String . class ) ; ReducingState < String > state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , kvId ) ; backend . setCurrentKey ( 1 ) ; assertNull ( state . get ( ) ) ; state . add ( "Ciao" ) ; assertEquals ( "Ciao" , state . get ( ) ) ; state . clear ( ) ; assertNull ( state . get ( ) ) ; backend . dispose ( ) ; }
Verify that an empty {@code ReduceState} yields {@code null}.
Verify that an empty { @code ReduceState } yields { @code null } .