CepRuntimeContext(final RuntimeContext runtimeContext) {
        this.runtimeContext = checkNotNull(runtimeContext);
    }
@Test
    public void testCepRuntimeContext() {
        final String taskName = "foobarTask";
        final OperatorMetricGroup metricGroup =
                UnregisteredMetricsGroup.createOperatorMetricGroup();
        final int numberOfParallelSubtasks = 43;
        final int indexOfSubtask = 42;
        final int attemptNumber = 1337;
        final String taskNameWithSubtask = "foobarTask (43/43)#1337";
        final Map<String, String> globalJobParameters = new HashMap<>();
        globalJobParameters.put("k1", "v1");
        final ClassLoader userCodeClassLoader = mock(ClassLoader.class);
        final DistributedCache distributedCache = mock(DistributedCache.class);
        final boolean isObjectReused = true;

        RuntimeContext mockedRuntimeContext = mock(RuntimeContext.class);
        TaskInfoImpl taskInfo =
                new TaskInfoImpl(
                        taskName,
                        numberOfParallelSubtasks,
                        indexOfSubtask,
                        numberOfParallelSubtasks,
                        attemptNumber);
        when(mockedRuntimeContext.getTaskInfo()).thenReturn(taskInfo);
        when(mockedRuntimeContext.getMetricGroup()).thenReturn(metricGroup);
        when(mockedRuntimeContext.getGlobalJobParameters()).thenReturn(globalJobParameters);
        when(mockedRuntimeContext.isObjectReuseEnabled()).thenReturn(isObjectReused);
        when(mockedRuntimeContext.getUserCodeClassLoader()).thenReturn(userCodeClassLoader);
        when(mockedRuntimeContext.getDistributedCache()).thenReturn(distributedCache);

        RuntimeContext runtimeContext = new CepRuntimeContext(mockedRuntimeContext);

        assertEquals(taskName, runtimeContext.getTaskInfo().getTaskName());
        assertEquals(metricGroup, runtimeContext.getMetricGroup());
        assertEquals(
                numberOfParallelSubtasks,
                runtimeContext.getTaskInfo().getNumberOfParallelSubtasks());
        assertEquals(indexOfSubtask, runtimeContext.getTaskInfo().getIndexOfThisSubtask());
        assertEquals(attemptNumber, runtimeContext.getTaskInfo().getAttemptNumber());
        assertEquals(taskNameWithSubtask, runtimeContext.getTaskInfo().getTaskNameWithSubtasks());
        assertEquals(globalJobParameters, runtimeContext.getGlobalJobParameters());
        assertEquals(isObjectReused, runtimeContext.isObjectReuseEnabled());
        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) {
            // expected
        }

        try {
            runtimeContext.getListState(new ListStateDescriptor<>("foobar", Integer.class));
            fail("Expected getListState to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getReducingState(
                    new ReducingStateDescriptor<>(
                            "foobar", mock(ReduceFunction.class), Integer.class));
            fail("Expected getReducingState to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getAggregatingState(
                    new AggregatingStateDescriptor<>(
                            "foobar", mock(AggregateFunction.class), Integer.class));
            fail("Expected getAggregatingState to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getMapState(
                    new MapStateDescriptor<>("foobar", Integer.class, String.class));
            fail("Expected getMapState to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.addAccumulator("foobar", mock(Accumulator.class));
            fail("Expected addAccumulator to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getAccumulator("foobar");
            fail("Expected getAccumulator to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getIntCounter("foobar");
            fail("Expected getIntCounter to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getLongCounter("foobar");
            fail("Expected getLongCounter to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getDoubleCounter("foobar");
            fail("Expected getDoubleCounter to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getHistogram("foobar");
            fail("Expected getHistogram to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.hasBroadcastVariable("foobar");
            fail("Expected hasBroadcastVariable to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getBroadcastVariable("foobar");
            fail("Expected getBroadcastVariable to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }

        try {
            runtimeContext.getBroadcastVariableWithInitializer(
                    "foobar", mock(BroadcastVariableInitializer.class));
            fail(
                    "Expected getBroadcastVariableWithInitializer to fail with unsupported operation exception.");
        } catch (UnsupportedOperationException e) {
            // expected
        }
    }