@Override
    public void transitionToActive(final StreamTask streamTask, final RecordCollector recordCollector, final ThreadCache newCache) {
        if (stateManager.taskType() != TaskType.ACTIVE) {
            throw new IllegalStateException("Tried to transition processor context to active but the state manager's " +
                                                "type was " + stateManager.taskType());
        }
        this.streamTask = streamTask;
        this.collector = recordCollector;
        this.cache = newCache;
        addAllFlushListenersToNewCache();
    }
@Test
    public void globalTimestampedKeyValueStoreShouldBeReadOnly() {
        foreachSetUp();

        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
        when(stateManager.getGlobalStore(anyString())).thenReturn(null);

        final TimestampedKeyValueStore<String, Long> timestampedKeyValueStoreMock = mock(TimestampedKeyValueStore.class);
        when(stateManager.getGlobalStore("GlobalTimestampedKeyValueStore")).thenAnswer(answer -> timestampedKeyValueStoreMock(timestampedKeyValueStoreMock));

        context = buildProcessorContextImpl(streamsConfig, stateManager);

        final StreamTask task = mock(StreamTask.class);
        context.transitionToActive(task, null, null);

        mockProcessorNodeWithLocalKeyValueStore();

        doTest("GlobalTimestampedKeyValueStore", (Consumer<TimestampedKeyValueStore<String, Long>>) store -> {
            verifyStoreCannotBeInitializedOrClosed(store);

            checkThrowsUnsupportedOperation(store::flush, "flush()");
            checkThrowsUnsupportedOperation(() -> store.put("1", ValueAndTimestamp.make(1L, 2L)), "put()");
            checkThrowsUnsupportedOperation(() -> store.putIfAbsent("1", ValueAndTimestamp.make(1L, 2L)), "putIfAbsent()");
            checkThrowsUnsupportedOperation(() -> store.putAll(Collections.emptyList()), "putAll()");
            checkThrowsUnsupportedOperation(() -> store.delete("1"), "delete()");

            assertEquals(VALUE_AND_TIMESTAMP, store.get(KEY));
            assertEquals(timestampedRangeIter, store.range("one", "two"));
            assertEquals(timestampedAllIter, store.all());
            assertEquals(VALUE, store.approximateNumEntries());
        });
    }