diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
index 69f4e8aa3b774..b01dbc1d05667 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
@@ -17,12 +17,9 @@
 
 package org.apache.kafka.streams.processor.internals;
 
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.StreamingConfig;
@@ -34,7 +31,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -63,43 +59,26 @@ public ProcessorContextImpl(int id,
                                 StreamTask task,
                                 StreamingConfig config,
                                 RecordCollector collector,
-                                Metrics metrics) throws IOException {
+                                ProcessorStateManager stateMgr,
+                                Metrics metrics) {
         this.id = id;
         this.task = task;
         this.metrics = metrics;
         this.collector = collector;
+        this.stateMgr = stateMgr;
 
         this.keySerializer = config.getConfiguredInstance(StreamingConfig.KEY_SERIALIZER_CLASS_CONFIG, Serializer.class);
         this.valSerializer = config.getConfiguredInstance(StreamingConfig.VALUE_SERIALIZER_CLASS_CONFIG, Serializer.class);
         this.keyDeserializer = config.getConfiguredInstance(StreamingConfig.KEY_DESERIALIZER_CLASS_CONFIG, Deserializer.class);
         this.valDeserializer = config.getConfiguredInstance(StreamingConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Deserializer.class);
 
-        File stateFile = new File(config.getString(StreamingConfig.STATE_DIR_CONFIG), Integer.toString(id));
-
-        log.info("Creating restoration consumer client for stream task [" + task.id() + "]");
-
-        Consumer restoreConsumer = new KafkaConsumer<>(
-            config.getConsumerConfigs(),
-            new ByteArrayDeserializer(),
-            new ByteArrayDeserializer());
-
-        this.stateMgr = new ProcessorStateManager(id, stateFile, restoreConsumer);
-
         this.initialized = false;
     }
 
-    public ProcessorStateManager stateManager() {
-        return this.stateMgr;
-    }
-
     public RecordCollector recordCollector() {
         return this.collector;
     }
 
-    public StreamTask task() {
-        return this.task;
-    }
-
     @Override
     public boolean joinable() {
         Set<TopicPartition> partitions = this.task.partitions();
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index 39ec04904eae9..be155186054de 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -19,16 +19,19 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.streams.StreamingConfig;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.HashMap;
@@ -55,6 +58,7 @@ public class StreamTask implements Punctuator {
 
     private final Map<TopicPartition, Long> consumedOffsets;
     private final RecordCollector recordCollector;
+    private final ProcessorStateManager stateMgr;
 
     private boolean commitRequested = false;
     private boolean commitOffsetNeeded = false;
@@ -104,13 +108,24 @@ public StreamTask(int id,
         // create the record recordCollector that maintains the produced offsets
         this.recordCollector = new RecordCollector(producer);
 
-        // initialize the topology with its own context
+        log.info("Creating restoration consumer client for stream task [" + id + "]");
+
+        Consumer restoreConsumer = new KafkaConsumer<>(
+                config.getConsumerConfigs(),
+                new ByteArrayDeserializer(),
+                new ByteArrayDeserializer());
+
+        // create the processor state manager
         try {
-            this.processorContext = new ProcessorContextImpl(id, this, config, recordCollector, new Metrics());
+            File stateFile = new File(config.getString(StreamingConfig.STATE_DIR_CONFIG), Integer.toString(id));
+            this.stateMgr = new ProcessorStateManager(id, stateFile, restoreConsumer);
         } catch (IOException e) {
-            throw new KafkaException("Error while creating the state manager in processor context", e);
+            throw new KafkaException("Error while creating the state manager", e);
         }
 
+        // initialize the topology with its own context
+        this.processorContext = new ProcessorContextImpl(id, this, config, recordCollector, stateMgr, new Metrics());
+
         // initialize the task by initializing all its processor nodes in the topology
         for (ProcessorNode node : this.topology.processors()) {
             this.currNode = node;
@@ -232,7 +247,7 @@ public ProcessorTopology topology() {
      */
     public void commit() {
         // 1) flush local state
-        ((ProcessorContextImpl) processorContext).stateManager().flush();
+        stateMgr.flush();
 
         // 2) flush produced records in the downstream
         // TODO: this will actually block on all produced records across the tasks
@@ -287,7 +302,7 @@ public void close() {
             throw exception;
 
         try {
-            ((ProcessorContextImpl) processorContext).stateManager().close(recordCollector.offsets());
+            stateMgr.close(recordCollector.offsets());
         } catch (IOException e) {
             throw new KafkaException("Error while closing the state manager in processor context", e);
         }