@Test public void testReducingStateInstantiation() throws Exception { final ExecutionConfig config = new ExecutionConfig(); config.registerKryoType(Path.class); final AtomicReference<Object> descriptorCapture = new AtomicReference<>(); StreamingRuntimeContext context = new StreamingRuntimeContext( createDescriptorCapturingMockOp(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); }
@Override public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); stateProperties.initializeSerializerUnlessSet(getExecutionConfig()); return keyedStateStore.getReducingState(stateProperties); }
StreamingRuntimeContext extends AbstractRuntimeUDFContext { @Override public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); stateProperties.initializeSerializerUnlessSet(getExecutionConfig()); return keyedStateStore.getReducingState(stateProperties); } }
StreamingRuntimeContext extends AbstractRuntimeUDFContext { @Override public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); stateProperties.initializeSerializerUnlessSet(getExecutionConfig()); return keyedStateStore.getReducingState(stateProperties); } StreamingRuntimeContext(AbstractStreamOperator<?> operator,
									Environment env, Map<String, Accumulator<?, ?>> accumulators); }
StreamingRuntimeContext extends AbstractRuntimeUDFContext { @Override public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); stateProperties.initializeSerializerUnlessSet(getExecutionConfig()); return keyedStateStore.getReducingState(stateProperties); } StreamingRuntimeContext(AbstractStreamOperator<?> operator,
									Environment env, Map<String, Accumulator<?, ?>> accumulators); InputSplitProvider getInputSplitProvider(); ProcessingTimeService getProcessingTimeService(); GlobalAggregateManager getGlobalAggregateManager(); String getOperatorUniqueID(); @Override boolean hasBroadcastVariable(String name); @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 AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT> stateProperties); @Override FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC> stateProperties); @Override MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties); boolean isCheckpointingEnabled(); CheckpointingMode getCheckpointMode(); long getBufferTimeout(); }
StreamingRuntimeContext extends AbstractRuntimeUDFContext { @Override public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); stateProperties.initializeSerializerUnlessSet(getExecutionConfig()); return keyedStateStore.getReducingState(stateProperties); } StreamingRuntimeContext(AbstractStreamOperator<?> operator,
									Environment env, Map<String, Accumulator<?, ?>> accumulators); InputSplitProvider getInputSplitProvider(); ProcessingTimeService getProcessingTimeService(); GlobalAggregateManager getGlobalAggregateManager(); String getOperatorUniqueID(); @Override boolean hasBroadcastVariable(String name); @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 AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT> stateProperties); @Override FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC> stateProperties); @Override MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties); boolean isCheckpointingEnabled(); CheckpointingMode getCheckpointMode(); long getBufferTimeout();  }