@Test public void testCepRuntimeContext() { final String taskName = "foobarTask"; final MetricGroup metricGroup = new UnregisteredMetricsGroup(); final int numberOfParallelSubtasks = 42; final int indexOfSubtask = 43; final int attemptNumber = 1337; final String taskNameWithSubtask = "barfoo"; final ExecutionConfig executionConfig = mock(ExecutionConfig.class); final ClassLoader userCodeClassLoader = mock(ClassLoader.class); final DistributedCache distributedCache = mock(DistributedCache.class); RuntimeContext mockedRuntimeContext = mock(RuntimeContext.class); when(mockedRuntimeContext.getTaskName()).thenReturn(taskName); when(mockedRuntimeContext.getMetricGroup()).thenReturn(metricGroup); when(mockedRuntimeContext.getNumberOfParallelSubtasks()).thenReturn(numberOfParallelSubtasks); when(mockedRuntimeContext.getIndexOfThisSubtask()).thenReturn(indexOfSubtask); when(mockedRuntimeContext.getAttemptNumber()).thenReturn(attemptNumber); when(mockedRuntimeContext.getTaskNameWithSubtasks()).thenReturn(taskNameWithSubtask); when(mockedRuntimeContext.getExecutionConfig()).thenReturn(executionConfig); when(mockedRuntimeContext.getUserCodeClassLoader()).thenReturn(userCodeClassLoader); when(mockedRuntimeContext.getDistributedCache()).thenReturn(distributedCache); RuntimeContext runtimeContext = new CepRuntimeContext(mockedRuntimeContext); assertEquals(taskName, runtimeContext.getTaskName()); assertEquals(metricGroup, runtimeContext.getMetricGroup()); assertEquals(numberOfParallelSubtasks, runtimeContext.getNumberOfParallelSubtasks()); assertEquals(indexOfSubtask, runtimeContext.getIndexOfThisSubtask()); assertEquals(attemptNumber, runtimeContext.getAttemptNumber()); assertEquals(taskNameWithSubtask, runtimeContext.getTaskNameWithSubtasks()); assertEquals(executionConfig, runtimeContext.getExecutionConfig()); assertEquals(userCodeClassLoader, runtimeContext.getUserCodeClassLoader()); assertEquals(distributedCache, runtimeContext.getDistributedCache()); try { runtimeContext.getState(new ValueStateDescriptor<>("foobar", Integer.class, 42)); fail("Expected getState to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getListState(new ListStateDescriptor<>("foobar", Integer.class)); fail("Expected getListState to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getReducingState(new ReducingStateDescriptor<>( "foobar", mock(ReduceFunction.class), Integer.class)); fail("Expected getReducingState to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getAggregatingState(new AggregatingStateDescriptor<>( "foobar", mock(AggregateFunction.class), Integer.class)); fail("Expected getAggregatingState to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getFoldingState(new FoldingStateDescriptor<>( "foobar", 0, mock(FoldFunction.class), Integer.class)); fail("Expected getFoldingState to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getMapState(new MapStateDescriptor<>("foobar", Integer.class, String.class)); fail("Expected getMapState to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.addAccumulator("foobar", mock(Accumulator.class)); fail("Expected addAccumulator to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getAccumulator("foobar"); fail("Expected getAccumulator to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getAllAccumulators(); fail("Expected getAllAccumulators to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getIntCounter("foobar"); fail("Expected getIntCounter to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getLongCounter("foobar"); fail("Expected getLongCounter to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getDoubleCounter("foobar"); fail("Expected getDoubleCounter to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getHistogram("foobar"); fail("Expected getHistogram to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.hasBroadcastVariable("foobar"); fail("Expected hasBroadcastVariable to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getBroadcastVariable("foobar"); fail("Expected getBroadcastVariable to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } try { runtimeContext.getBroadcastVariableWithInitializer( "foobar", mock(BroadcastVariableInitializer.class)); fail("Expected getBroadcastVariableWithInitializer to fail with unsupported operation exception."); } catch (UnsupportedOperationException e) { } }
CepRuntimeContext(final RuntimeContext runtimeContext) { this.runtimeContext = checkNotNull(runtimeContext); }
CepRuntimeContext implements RuntimeContext { CepRuntimeContext(final RuntimeContext runtimeContext) { this.runtimeContext = checkNotNull(runtimeContext); } }
CepRuntimeContext implements RuntimeContext { CepRuntimeContext(final RuntimeContext runtimeContext) { this.runtimeContext = checkNotNull(runtimeContext); } CepRuntimeContext(final RuntimeContext runtimeContext); }
CepRuntimeContext implements RuntimeContext { CepRuntimeContext(final RuntimeContext runtimeContext) { this.runtimeContext = checkNotNull(runtimeContext); } CepRuntimeContext(final RuntimeContext runtimeContext); @Override String getTaskName(); @Override MetricGroup getMetricGroup(); @Override int getNumberOfParallelSubtasks(); @Override int getMaxNumberOfParallelSubtasks(); @Override int getIndexOfThisSubtask(); @Override int getAttemptNumber(); @Override String getTaskNameWithSubtasks(); @Override ExecutionConfig getExecutionConfig(); @Override ClassLoader getUserCodeClassLoader(); @Override DistributedCache getDistributedCache(); @Override void addAccumulator(
			final String name,
			final Accumulator<V, A> accumulator); @Override Accumulator<V, A> getAccumulator(final String name); @Override Map<String, Accumulator<?, ?>> getAllAccumulators(); @Override IntCounter getIntCounter(final String name); @Override LongCounter getLongCounter(final String name); @Override DoubleCounter getDoubleCounter(final String name); @Override Histogram getHistogram(final String name); @Override boolean hasBroadcastVariable(final String name); @Override List<RT> getBroadcastVariable(final String name); @Override C getBroadcastVariableWithInitializer(
			final String name,
			final BroadcastVariableInitializer<T, C> initializer); @Override ValueState<T> getState(final ValueStateDescriptor<T> stateProperties); @Override ListState<T> getListState(final ListStateDescriptor<T> stateProperties); @Override ReducingState<T> getReducingState(final ReducingStateDescriptor<T> stateProperties); @Override AggregatingState<IN, OUT> getAggregatingState(
			final AggregatingStateDescriptor<IN, ACC, OUT> stateProperties); @Override FoldingState<T, ACC> getFoldingState(final FoldingStateDescriptor<T, ACC> stateProperties); @Override MapState<UK, UV> getMapState(final MapStateDescriptor<UK, UV> stateProperties); }
CepRuntimeContext implements RuntimeContext { CepRuntimeContext(final RuntimeContext runtimeContext) { this.runtimeContext = checkNotNull(runtimeContext); } CepRuntimeContext(final RuntimeContext runtimeContext); @Override String getTaskName(); @Override MetricGroup getMetricGroup(); @Override int getNumberOfParallelSubtasks(); @Override int getMaxNumberOfParallelSubtasks(); @Override int getIndexOfThisSubtask(); @Override int getAttemptNumber(); @Override String getTaskNameWithSubtasks(); @Override ExecutionConfig getExecutionConfig(); @Override ClassLoader getUserCodeClassLoader(); @Override DistributedCache getDistributedCache(); @Override void addAccumulator(
			final String name,
			final Accumulator<V, A> accumulator); @Override Accumulator<V, A> getAccumulator(final String name); @Override Map<String, Accumulator<?, ?>> getAllAccumulators(); @Override IntCounter getIntCounter(final String name); @Override LongCounter getLongCounter(final String name); @Override DoubleCounter getDoubleCounter(final String name); @Override Histogram getHistogram(final String name); @Override boolean hasBroadcastVariable(final String name); @Override List<RT> getBroadcastVariable(final String name); @Override C getBroadcastVariableWithInitializer(
			final String name,
			final BroadcastVariableInitializer<T, C> initializer); @Override ValueState<T> getState(final ValueStateDescriptor<T> stateProperties); @Override ListState<T> getListState(final ListStateDescriptor<T> stateProperties); @Override ReducingState<T> getReducingState(final ReducingStateDescriptor<T> stateProperties); @Override AggregatingState<IN, OUT> getAggregatingState(
			final AggregatingStateDescriptor<IN, ACC, OUT> stateProperties); @Override FoldingState<T, ACC> getFoldingState(final FoldingStateDescriptor<T, ACC> stateProperties); @Override MapState<UK, UV> getMapState(final MapStateDescriptor<UK, UV> stateProperties);  }