@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 localWindowStoreShouldNotAllowInitOrClose() {
        foreachSetUp();

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

        final WindowStore<String, Long> windowStore = mock(WindowStore.class);
        when(stateManager.getStore("LocalWindowStore")).thenAnswer(answer -> windowStoreMock(windowStore));
        mockStateStoreFlush(windowStore);

        doAnswer(answer -> {
            putExecuted = true;
            return null;
        }).when(windowStore).put(anyString(), anyLong(), anyLong());

        context = buildProcessorContextImpl(streamsConfig, stateManager);

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

        mockProcessorNodeWithLocalKeyValueStore();

        doTest("LocalWindowStore", (Consumer<WindowStore<String, Long>>) store -> {
            verifyStoreCannotBeInitializedOrClosed(store);

            store.flush();
            assertTrue(flushExecuted);

            store.put("1", 1L, 1L);
            assertTrue(putExecuted);

            assertEquals(iters.get(0), store.fetchAll(0L, 0L));
            assertEquals(windowStoreIter, store.fetch(KEY, 0L, 1L));
            assertEquals(iters.get(1), store.fetch(KEY, KEY, 0L, 1L));
            assertEquals((Long) VALUE, store.fetch(KEY, 1L));
            assertEquals(iters.get(2), store.all());
        });
    }