diff --git a/engine/src/main/java/com/datatorrent/stram/StreamingContainerAgent.java b/engine/src/main/java/com/datatorrent/stram/StreamingContainerAgent.java
index 81dc96e..a5fc3a5 100644
--- a/engine/src/main/java/com/datatorrent/stram/StreamingContainerAgent.java
+++ b/engine/src/main/java/com/datatorrent/stram/StreamingContainerAgent.java
@@ -185,7 +185,11 @@
         if (ndi.type == OperatorDeployInfo.OperatorType.UNIFIER) {
           // input attributes of the downstream operator
           for (InputPortMeta sink : streamMeta.getSinks()) {
-            portInfo.contextAttributes = sink.getAttributes();
+            try {
+              portInfo.contextAttributes = sink.getAttributes().clone();
+            } catch (CloneNotSupportedException e) {
+              throw new RuntimeException("Cannot clone attributes", e);
+            }
             break;
           }
         }
@@ -199,7 +203,7 @@
             // Create mappings for all non-inline operators
             if (input.target.getContainer() != out.source.getContainer()) {
               InputPortMeta inputPortMeta = getIdentifyingInputPortMeta(input);
-              StreamCodec<?> streamCodecInfo = getStreamCodec(inputPortMeta);
+              StreamCodec<?> streamCodecInfo = inputPortMeta.getStreamCodec();
               Integer id = physicalPlan.getStreamCodecIdentifier(streamCodecInfo);
               if (!portInfo.streamCodecs.containsKey(id)) {
                 portInfo.streamCodecs.put(id, streamCodecInfo);
@@ -231,11 +235,19 @@
         InputPortMeta inputPortMeta = getInputPortMeta(oper.getOperatorMeta(), streamMeta);
 
         if (inputPortMeta != null) {
-          inputInfo.contextAttributes = inputPortMeta.getAttributes();
+          try {
+            inputInfo.contextAttributes = inputPortMeta.getAttributes().clone();
+          } catch (CloneNotSupportedException e) {
+            throw new RuntimeException("Cannot clone attributes", e);
+          }
         }
 
         if (inputInfo.contextAttributes == null && ndi.type == OperatorDeployInfo.OperatorType.UNIFIER) {
-          inputInfo.contextAttributes = in.source.logicalStream.getSource().getAttributes();
+          try {
+            inputInfo.contextAttributes = in.source.logicalStream.getSource().getAttributes().clone();
+          } catch (CloneNotSupportedException e) {
+            throw new RuntimeException("Cannot clone attributes", e);
+          }
         }
 
         inputInfo.sourceNodeId = sourceOutput.source.getId();
@@ -272,7 +284,7 @@
         // On the input side there is a unlikely scenario of partitions even for inline stream that is being
         // handled. Always specifying a stream codec configuration in case that scenario happens.
         InputPortMeta idInputPortMeta = getIdentifyingInputPortMeta(in);
-        StreamCodec<?> streamCodecInfo = getStreamCodec(idInputPortMeta);
+        StreamCodec<?> streamCodecInfo = idInputPortMeta.getStreamCodec();
         Integer id = physicalPlan.getStreamCodecIdentifier(streamCodecInfo);
         inputInfo.streamCodecs.put(id, streamCodecInfo);
         ndi.inputs.add(inputInfo);
@@ -327,23 +339,6 @@
     return operator;
   }
 
-  public static StreamCodec<?> getStreamCodec(InputPortMeta inputPortMeta)
-  {
-    if (inputPortMeta != null) {
-      StreamCodec<?> codec = inputPortMeta.getValue(PortContext.STREAM_CODEC);
-      if (codec == null) {
-        // it cannot be this object that gets returned. Depending on this value is dangerous 
-        codec = inputPortMeta.getPortObject().getStreamCodec();
-        if (codec != null) {
-          // don't create codec multiple times - it will assign a new identifier
-          inputPortMeta.getAttributes().put(PortContext.STREAM_CODEC, codec);
-        }
-      }
-      return codec;
-    }
-    return null;
-  }
-
   /**
    * Create deploy info for operator.
    * <p>
diff --git a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
index aa79243..2fe8107 100644
--- a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
+++ b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java
@@ -2146,7 +2146,7 @@
           }
 
           for (InputPortMeta ipm : out.logicalStream.getSinks()) {
-            StreamCodec<?> streamCodecInfo = StreamingContainerAgent.getStreamCodec(ipm);
+            StreamCodec<?> streamCodecInfo = ipm.getStreamCodec();
             Integer codecId = plan.getStreamCodecIdentifier(streamCodecInfo);
             // following needs to match the concat logic in StreamingContainer
             String sourceIdentifier = Integer.toString(operator.getId()).concat(Component.CONCAT_SEPARATOR).concat(out.portName).concat(Component.CONCAT_SEPARATOR).concat(codecId.toString());
@@ -2243,7 +2243,7 @@
             for (PTOperator.PTOutput out : operator.getOutputs()) {
               if (!out.isDownStreamInline()) {
                 for (InputPortMeta ipm : out.logicalStream.getSinks()) {
-                  StreamCodec<?> streamCodecInfo = StreamingContainerAgent.getStreamCodec(ipm);
+                  StreamCodec<?> streamCodecInfo = ipm.getStreamCodec();
                   Integer codecId = plan.getStreamCodecIdentifier(streamCodecInfo);
                   // following needs to match the concat logic in StreamingContainer
                   String sourceIdentifier = Integer.toString(operator.getId()).concat(Component.CONCAT_SEPARATOR).concat(out.portName).concat(Component.CONCAT_SEPARATOR).concat(codecId.toString());
diff --git a/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java b/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java
index 3825505..01d9a1c 100644
--- a/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java
+++ b/engine/src/main/java/com/datatorrent/stram/plan/logical/LogicalPlan.java
@@ -61,6 +61,8 @@
 import com.datatorrent.stram.engine.DefaultUnifier;
 import com.datatorrent.stram.engine.Slider;
 
+import static com.datatorrent.api.Context.PortContext.STREAM_CODEC;
+
 /**
  * DAG contains the logical declarations of operators and streams.
  * <p>
@@ -270,6 +272,20 @@
       throw new UnsupportedOperationException("Not supported yet.");
     }
 
+    public StreamCodec<?> getStreamCodec()
+    {
+      return attributes.get(STREAM_CODEC);
+    }
+
+    void setStreamCodec(StreamCodec<?> streamCodec)
+    {
+      if (streamCodec != null) {
+        StreamCodec<?> oldStreamCodec = attributes.put(STREAM_CODEC, streamCodec);
+        if (oldStreamCodec != null && oldStreamCodec != streamCodec) {
+          LOG.warn("Input port {} stream codec was changed from {} to {}", this.getPortName(), oldStreamCodec, streamCodec);
+        }
+      }
+    }
   }
 
   public final class OutputPortMeta implements DAG.OutputPortMeta, Serializable
@@ -681,14 +697,14 @@
 
     private void addStreamCodec(InputPortMeta sinkToPersistPortMeta, InputPort<?> port)
     {
-      StreamCodec<Object> inputStreamCodec = sinkToPersistPortMeta.getValue(PortContext.STREAM_CODEC) != null ? (StreamCodec<Object>) sinkToPersistPortMeta.getValue(PortContext.STREAM_CODEC) : (StreamCodec<Object>) sinkToPersistPortMeta.getPortObject().getStreamCodec();
+      StreamCodec<Object> inputStreamCodec = sinkToPersistPortMeta.getValue(STREAM_CODEC) != null ? (StreamCodec<Object>) sinkToPersistPortMeta.getValue(STREAM_CODEC) : (StreamCodec<Object>) sinkToPersistPortMeta.getPortObject().getStreamCodec();
       if (inputStreamCodec != null) {
         Map<InputPortMeta, StreamCodec<Object>> codecs = new HashMap<InputPortMeta, StreamCodec<Object>>();
         codecs.put(sinkToPersistPortMeta, inputStreamCodec);
         InputPortMeta persistOperatorPortMeta = assertGetPortMeta(port);
-        StreamCodec<Object> specifiedCodecForPersistOperator = (persistOperatorPortMeta.getValue(PortContext.STREAM_CODEC) != null) ? (StreamCodec<Object>) persistOperatorPortMeta.getValue(PortContext.STREAM_CODEC) : (StreamCodec<Object>) port.getStreamCodec();
+        StreamCodec<Object> specifiedCodecForPersistOperator = (persistOperatorPortMeta.getValue(STREAM_CODEC) != null) ? (StreamCodec<Object>) persistOperatorPortMeta.getValue(STREAM_CODEC) : (StreamCodec<Object>) port.getStreamCodec();
         StreamCodecWrapperForPersistance<Object> codec = new StreamCodecWrapperForPersistance<Object>(codecs, specifiedCodecForPersistOperator);
-        setInputPortAttribute(port, PortContext.STREAM_CODEC, codec);
+        setInputPortAttribute(port, STREAM_CODEC, codec);
       }
     }
 
@@ -980,6 +996,12 @@
         metaPort.adqAnnotation = adqAnnotation;
         inPortMap.put(portObject, metaPort);
         markInputPortIfHidden(metaPort.getPortName(), metaPort, field.getDeclaringClass());
+        if (metaPort.getStreamCodec() == null) {
+          metaPort.setStreamCodec(portObject.getStreamCodec());
+        } else if (portObject.getStreamCodec() != null) {
+          LOG.info("Input port {} attribute {} overrides codec {} with {} codec", metaPort.getPortName(), STREAM_CODEC.getSimpleName(),
+              portObject.getStreamCodec(), metaPort.getStreamCodec());
+        }
       }
 
       @Override
@@ -1612,7 +1634,7 @@
     }
     for (StreamMeta n: this.streams.values()) {
       for (InputPortMeta sink : n.getSinks()) {
-        StreamCodec<?> streamCodec = sink.getValue(PortContext.STREAM_CODEC);
+        StreamCodec<?> streamCodec = sink.getValue(STREAM_CODEC);
         if (streamCodec != null) {
           classNames.add(streamCodec.getClass().getName());
         } else {
diff --git a/engine/src/main/java/com/datatorrent/stram/plan/physical/StreamMapping.java b/engine/src/main/java/com/datatorrent/stram/plan/physical/StreamMapping.java
index f30ceb6..7efce6c 100644
--- a/engine/src/main/java/com/datatorrent/stram/plan/physical/StreamMapping.java
+++ b/engine/src/main/java/com/datatorrent/stram/plan/physical/StreamMapping.java
@@ -38,7 +38,6 @@
 import com.datatorrent.api.StreamCodec;
 
 import com.datatorrent.common.util.Pair;
-import com.datatorrent.stram.StreamingContainerAgent;
 import com.datatorrent.stram.plan.logical.LogicalPlan;
 import com.datatorrent.stram.plan.logical.LogicalPlan.InputPortMeta;
 import com.datatorrent.stram.plan.logical.LogicalPlan.OperatorMeta;
@@ -230,7 +229,7 @@
       boolean separateUnifiers = false;
       Integer lastId = null;
       for (InputPortMeta ipm : streamMeta.getSinks()) {
-        StreamCodec<?> streamCodecInfo = StreamingContainerAgent.getStreamCodec(ipm);
+        StreamCodec<?> streamCodecInfo = ipm.getStreamCodec();
         Integer id = plan.getStreamCodecIdentifier(streamCodecInfo);
         if (lastId == null) {
           lastId = id;
@@ -249,7 +248,7 @@
           unifierSources = setupCascadingUnifiers(this.upstream, currentUnifiers, limit, 0);
         } else {
           for (InputPortMeta ipm : streamMeta.getSinks()) {
-            StreamCodec<?> streamCodecInfo = StreamingContainerAgent.getStreamCodec(ipm);
+            StreamCodec<?> streamCodecInfo = ipm.getStreamCodec();
             if (!cascadeUnifierSourcesMap.containsKey(streamCodecInfo)) {
               unifierSources = setupCascadingUnifiers(this.upstream, currentUnifiers, limit, 0);
               cascadeUnifierSourcesMap.put(streamCodecInfo, unifierSources);
@@ -303,7 +302,7 @@
             unifier.inputs.clear();
             List<PTOutput> doperUnifierSources = unifierSources;
             if (separateUnifiers) {
-              StreamCodec<?> streamCodecInfo = StreamingContainerAgent.getStreamCodec(doperEntry.second);
+              StreamCodec<?> streamCodecInfo = doperEntry.second.getStreamCodec();
               List<PTOutput> cascadeSources = cascadeUnifierSourcesMap.get(streamCodecInfo);
               if (cascadeSources != null) {
                 doperUnifierSources = cascadeSources;
diff --git a/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java b/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
index ddf3448..da5456e 100644
--- a/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/StreamCodecTest.java
@@ -1152,7 +1152,7 @@
                                        Map<Integer, StreamCodec<?>> streamCodecs,
                                        String id, PhysicalPlan plan )
   {
-    StreamCodec<?> streamCodecInfo = StreamingContainerAgent.getStreamCodec(operatorMeta.getMeta(inputPort));
+    StreamCodec<?> streamCodecInfo = operatorMeta.getMeta(inputPort).getStreamCodec();
     Assert.assertTrue("stream codec identifier not present" + id, isStrCodecPresent(streamCodecInfo, plan));
     Integer streamCodecIdentifier = plan.getStreamCodecIdentifier(streamCodecInfo);
     checkPresentStreamCodecInfo(streamCodecs, id, streamCodecIdentifier, streamCodecInfo);