StreamingRuntimeContext extends AbstractRuntimeUDFContext { @Override public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) { requireNonNull(stateProperties, "The state properties must not be null"); try { stateProperties.initializeSerializerUnlessSet(getExecutionConfig()); return operator.getPartitionedState(stateProperties); } catch (Exception e) { throw new RuntimeException("Error while getting state", e); } } StreamingRuntimeContext(AbstractStreamOperator<?> operator,
									Environment env, Map<String, Accumulator<?, ?>> accumulators); InputSplitProvider getInputSplitProvider(); void registerTimer(long time, Triggerable target); @Override List<RT> getBroadcastVariable(String name); @Override C getBroadcastVariableWithInitializer(String name, BroadcastVariableInitializer<T, C> initializer); @Override ValueState<T> getState(ValueStateDescriptor<T> stateProperties); @Override ListState<T> getListState(ListStateDescriptor<T> stateProperties); @Override ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties); @Override @Deprecated OperatorState<S> getKeyValueState(String name, Class<S> stateType, S defaultState); @Override @Deprecated OperatorState<S> getKeyValueState(String name, TypeInformation<S> stateType, S defaultState); boolean isCheckpointingEnabled(); CheckpointingMode getCheckpointMode(); long getBufferTimeout();  }
@Test public void testReduceingStateInstantiation() throws Exception { final ExecutionConfig config = new ExecutionConfig(); config.registerKryoType(Path.class); final AtomicReference<Object> descriptorCapture = new AtomicReference<>(); StreamingRuntimeContext context = new StreamingRuntimeContext( createMockOp(descriptorCapture, config), createMockEnvironment(), Collections.<String, Accumulator<?, ?>>emptyMap()); @SuppressWarnings("unchecked") ReduceFunction<TaskInfo> reducer = (ReduceFunction<TaskInfo>) mock(ReduceFunction.class); ReducingStateDescriptor<TaskInfo> descr = new ReducingStateDescriptor<>("name", reducer, TaskInfo.class); context.getReducingState(descr); StateDescriptor<?, ?> descrIntercepted = (StateDescriptor<?, ?>) descriptorCapture.get(); TypeSerializer<?> serializer = descrIntercepted.getSerializer(); assertTrue(serializer instanceof KryoSerializer); assertTrue(((KryoSerializer<?>) serializer).getKryo().getRegistration(Path.class).getId() > 0); }