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

        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);

        context = buildProcessorContextImpl(streamsConfig, stateManager);

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

        mockProcessorNodeWithLocalKeyValueStore();

        context.addProcessorMetadataKeyValue("key1", 100L);
        final Long value = context.processorMetadataForKey("key1");
        assertEquals(100L, value.longValue());

        final Long noValue = context.processorMetadataForKey("nokey");
        assertNull(noValue);
    }