diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java index 4d724f8..d892391 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java @@ -83,8 +83,8 @@ void init(JobConf jconf, TezProcessorContext processorContext, MRTaskReporter mr // Start all the Outputs. for (Entry outputEntry : outputs.entrySet()) { l4j.info("Starting Output: " + outputEntry.getKey()); - ((TezKVOutputCollector) outMap.get(outputEntry.getKey())).initialize(); outputEntry.getValue().start(); + ((TezKVOutputCollector) outMap.get(outputEntry.getKey())).initialize(); } ObjectCache cache = ObjectCacheFactory.getCache(jconf); diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java index 800725d..0e4bdff 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java @@ -200,8 +200,8 @@ void run() throws Exception { for (Entry outputEntry : outputs.entrySet()) { l4j.info("Starting Output: " + outputEntry.getKey()); - ((TezKVOutputCollector) outMap.get(outputEntry.getKey())).initialize(); outputEntry.getValue().start(); + ((TezKVOutputCollector) outMap.get(outputEntry.getKey())).initialize(); } KeyValuesReader kvsReader;