diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/rules/InsertBrokerNotifierForChannelRule.java b/asterix-bad/src/main/java/org/apache/asterix/bad/rules/InsertBrokerNotifierForChannelRule.java
index 9ead7f0..258aaf1 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/rules/InsertBrokerNotifierForChannelRule.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/rules/InsertBrokerNotifierForChannelRule.java
@@ -31,7 +31,6 @@
 import org.apache.asterix.om.base.AString;
 import org.apache.asterix.om.constants.AsterixConstantValue;
 import org.apache.asterix.om.functions.BuiltinFunctions;
-import org.apache.asterix.om.types.IAType;
 import org.apache.commons.lang3.mutable.Mutable;
 import org.apache.commons.lang3.mutable.MutableObject;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -44,7 +43,6 @@
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
 import org.apache.hyracks.algebricks.core.algebra.expressions.AggregateFunctionCallExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
-import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
@@ -182,10 +180,9 @@
     }
 
     private DelegateOperator createBrokerOp(LogicalVariable brokerEndpointVar, LogicalVariable sendVar,
-            LogicalVariable channelExecutionVar, String channelDataverse, String channelName, boolean push,
-            IAType resultType) {
+            LogicalVariable channelExecutionVar, String channelDataverse, String channelName, boolean push) {
         NotifyBrokerOperator notifyBrokerOp =
-                new NotifyBrokerOperator(brokerEndpointVar, sendVar, channelExecutionVar, push, resultType);
+                new NotifyBrokerOperator(brokerEndpointVar, sendVar, channelExecutionVar, push);
         EntityId activeId = new EntityId(BADConstants.CHANNEL_EXTENSION_NAME, channelDataverse, channelName);
         NotifyBrokerPOperator notifyBrokerPOp = new NotifyBrokerPOperator(activeId);
         notifyBrokerOp.setPhysicalOperator(notifyBrokerPOp);
@@ -203,12 +200,10 @@
         while (assign.getOperatorTag() != LogicalOperatorTag.ASSIGN) {
             assign = (AbstractLogicalOperator) assign.getInputs().get(0).getValue();
         }
-        IVariableTypeEnvironment env = assign.computeOutputTypeEnvironment(context);
-        IAType resultType = (IAType) env.getVarType(sendVar);
 
         //Create the NotifyBrokerOperator
         DelegateOperator extensionOp = createBrokerOp(brokerEndpointVar, sendVar, channelExecutionVar, channelDataverse,
-                channelName, true, resultType);
+                channelName, true);
 
         extensionOp.getInputs().add(new MutableObject<>(eOp));
         context.computeAndSetTypeEnvironmentForOperator(extensionOp);
@@ -260,7 +255,7 @@
 
         //Create the NotifyBrokerOperator
         DelegateOperator extensionOp = createBrokerOp(brokerEndpointVar, sendListVar, channelExecutionVar,
-                channelDataverse, channelName, false, null);
+                channelDataverse, channelName, false);
 
         //Set the input for the distinct as the old top
         extensionOp.getInputs().add(new MutableObject<>(groupbyOp));
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/NotifyBrokerOperator.java b/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/NotifyBrokerOperator.java
index df0f0f4..5fc9b22 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/NotifyBrokerOperator.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/NotifyBrokerOperator.java
@@ -20,7 +20,6 @@
 
 import java.util.Collection;
 
-import org.apache.asterix.om.types.IAType;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractDelegatedLogicalOperator;
@@ -35,15 +34,13 @@
     private final LogicalVariable channelExecutionVar;
     private final LogicalVariable pushListVar;
     private final boolean push;
-    private final IAType recordType;
 
     public NotifyBrokerOperator(LogicalVariable brokerEndpointVar, LogicalVariable pushListVar,
-            LogicalVariable resultSetVar, boolean push, IAType recordType) {
+            LogicalVariable resultSetVar, boolean push) {
         this.brokerEndpointVar = brokerEndpointVar;
         this.channelExecutionVar = resultSetVar;
         this.pushListVar = pushListVar;
         this.push = push;
-        this.recordType = recordType;
     }
 
     public LogicalVariable getPushListVar() {
@@ -58,10 +55,6 @@
         return channelExecutionVar;
     }
 
-    public IAType getRecordType() {
-        return recordType;
-    }
-
     public boolean getPush() {
         return push;
     }
@@ -79,7 +72,7 @@
 
     @Override
     public IOperatorDelegate newInstance() {
-        return new NotifyBrokerOperator(brokerEndpointVar, pushListVar, channelExecutionVar, push, recordType);
+        return new NotifyBrokerOperator(brokerEndpointVar, pushListVar, channelExecutionVar, push);
     }
 
     @Override
diff --git a/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/NotifyBrokerPOperator.java b/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/NotifyBrokerPOperator.java
index b9cfbfd..264a994 100644
--- a/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/NotifyBrokerPOperator.java
+++ b/asterix-bad/src/main/java/org/apache/asterix/bad/runtime/NotifyBrokerPOperator.java
@@ -27,6 +27,7 @@
 import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
 import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.DelegateOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema;
@@ -78,7 +79,10 @@
         LogicalVariable pushListVar = ((NotifyBrokerOperator) notify.getDelegate()).getPushListVar();
         LogicalVariable brokerVar = ((NotifyBrokerOperator) notify.getDelegate()).getBrokerEndpointVariable();
         LogicalVariable executionVar = ((NotifyBrokerOperator) notify.getDelegate()).getChannelExecutionVariable();
-        IAType recordType = ((NotifyBrokerOperator) notify.getDelegate()).getRecordType();
+
+        IVariableTypeEnvironment env = context.getTypeEnvironment(op.getInputs().get(0).getValue());
+        IAType recordType = (IAType) env.getVarType(pushListVar);
+
         boolean push = ((NotifyBrokerOperator) notify.getDelegate()).getPush();
 
         int brokerColumn = inputSchemas[0].findVariable(brokerVar);