StreamingRuntimeContext extends AbstractRuntimeUDFContext { @Override public <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); stateProperties.initializeSerializerUnlessSet(getExecutionConfig()); return keyedStateStore.getListState(stateProperties); } StreamingRuntimeContext(AbstractStreamOperator<?> operator,
									Environment env, Map<String, Accumulator<?, ?>> accumulators); InputSplitProvider getInputSplitProvider(); ProcessingTimeService getProcessingTimeService(); @Override boolean hasBroadcastVariable(String name); @Override List<RT> getBroadcastVariable(String name); @Override C getBroadcastVariableWithInitializer(String name, BroadcastVariableInitializer<T, C> initializer); @Override List<T> getSideInput(SideInput<T> handle); @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();  }
@Test public void testListStateInstantiation() 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()); ListStateDescriptor<TaskInfo> descr = new ListStateDescriptor<>("name", TaskInfo.class); context.getListState(descr); ListStateDescriptor<?> descrIntercepted = (ListStateDescriptor<?>) descriptorCapture.get(); TypeSerializer<?> serializer = descrIntercepted.getSerializer(); assertTrue(serializer instanceof ListSerializer); TypeSerializer<?> elementSerializer = descrIntercepted.getElementSerializer(); assertTrue(elementSerializer instanceof KryoSerializer); assertTrue(((KryoSerializer<?>) elementSerializer).getKryo().getRegistration(Path.class).getId() > 0); }
@Test public void testListStateReturnsEmptyListByDefault() throws Exception { StreamingRuntimeContext context = new StreamingRuntimeContext( createListPlainMockOp(), createMockEnvironment(), Collections.<String, Accumulator<?, ?>>emptyMap()); ListStateDescriptor<String> descr = new ListStateDescriptor<>("name", String.class); ListState<String> state = context.getListState(descr); Iterable<String> value = state.get(); assertNotNull(value); assertFalse(value.iterator().hasNext()); }