Repository: tez Updated Branches: refs/heads/branch-0.7 eec72121a -> 9a8cd4239
TEZ-2445. Disable the object cleanup in local mode in LogicalIOProcessorRuntimeTask. Contributed by Jeff Zhang. (cherry picked from commit c048de1c54413076339e2f0339b1fc40f7d1f24d) Project: http://git-wip-us.apache.org/repos/asf/tez/repo Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/9a8cd423 Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/9a8cd423 Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/9a8cd423 Branch: refs/heads/branch-0.7 Commit: 9a8cd4239762779dbb1f52cd3561a4aacf3e69c7 Parents: eec7212 Author: Siddharth Seth <[email protected]> Authored: Wed May 13 11:03:25 2015 -0700 Committer: Siddharth Seth <[email protected]> Committed: Wed May 13 11:04:13 2015 -0700 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../runtime/LogicalIOProcessorRuntimeTask.java | 15 +++++++++++---- .../TestLogicalIOProcessorRuntimeTask.java | 19 ++++++++++++++----- 3 files changed, 26 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tez/blob/9a8cd423/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index cb2dd3e..b3d2872 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -11,6 +11,7 @@ INCOMPATIBLE CHANGES Default max limit increased. Should not affect existing users. ALL CHANGES: + TEZ-2445. Disable the object cleanup in local mode in LogicalIOProcessorRuntimeTask. TEZ-2057. tez-dag/pom.xml contains versions for dependencies. TEZ-1529. ATS and TezClient integration in secure kerberos enabled cluster. TEZ-2435. Add public key to KEYS http://git-wip-us.apache.org/repos/asf/tez/blob/9a8cd423/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java ---------------------------------------------------------------------- diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java index e419a8d..24f62a0 100644 --- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java +++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java @@ -53,6 +53,7 @@ import org.apache.tez.common.RunnableWithNdc; import org.apache.tez.dag.api.InputDescriptor; import org.apache.tez.dag.api.OutputDescriptor; import org.apache.tez.dag.api.ProcessorDescriptor; +import org.apache.tez.dag.api.TezConfiguration; import org.apache.tez.dag.api.TezUncheckedException; import org.apache.tez.dag.records.TezTaskAttemptID; import org.apache.tez.runtime.api.AbstractLogicalIOProcessor; @@ -763,8 +764,7 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask { inputsMap.clear(); outputsMap.clear(); - inputSpecs.clear(); - outputSpecs.clear(); + inputsMap.clear(); outputsMap.clear(); @@ -772,8 +772,15 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask { inputContextMap.clear(); outputContextMap.clear(); - if (groupInputSpecs != null) { - groupInputSpecs.clear(); + // only clean up objects in non-local mode, because local mode share the same + // taskSpec in AM rather than getting it through RPC in non-local mode + /** Put other objects here when they are shared between AM & TezChild in local mode **/ + if (!tezConf.getBoolean(TezConfiguration.TEZ_LOCAL_MODE, TezConfiguration.TEZ_LOCAL_MODE_DEFAULT)) { + inputSpecs.clear(); + outputSpecs.clear(); + if (groupInputSpecs != null) { + groupInputSpecs.clear(); + } } if (groupInputsMap != null) { groupInputsMap.clear(); http://git-wip-us.apache.org/repos/asf/tez/blob/9a8cd423/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java ---------------------------------------------------------------------- diff --git a/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java b/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java index b337bc7..0acb7b8 100644 --- a/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java +++ b/tez-runtime-internals/src/test/java/org/apache/tez/runtime/TestLogicalIOProcessorRuntimeTask.java @@ -109,7 +109,8 @@ public class TestLogicalIOProcessorRuntimeTask { - + // local mode + tezConf.setBoolean(TezConfiguration.TEZ_LOCAL_MODE, true); LogicalIOProcessorRuntimeTask lio2 = new LogicalIOProcessorRuntimeTask(task2, 0, tezConf, null, umbilical, serviceConsumerMetadata, new HashMap<String, String>(), startedInputsMap, null, "", new ExecutionContextImpl("localhost"), Runtime.getRuntime().maxMemory()); @@ -158,14 +159,22 @@ public class TestLogicalIOProcessorRuntimeTask { assertTrue(outputContext.getUserPayload() == null); assertTrue(outputContext.getObjectRegistry() == null); } - - assertEquals(0, lio.inputSpecs.size()); + boolean localMode = lio.tezConf.getBoolean(TezConfiguration.TEZ_LOCAL_MODE, + TezConfiguration.TEZ_LOCAL_MODE_DEFAULT); + if (localMode) { + assertEquals(1, lio.inputSpecs.size()); + assertEquals(1, lio.outputSpecs.size()); + assertTrue(lio.groupInputSpecs == null || lio.groupInputSpecs.size() == 0); + } else { + assertEquals(0, lio.inputSpecs.size()); + assertEquals(0, lio.outputSpecs.size()); + assertTrue(lio.groupInputSpecs == null || lio.groupInputSpecs.size() == 0); + } + assertEquals(0, lio.inputsMap.size()); assertEquals(0, lio.inputContextMap.size()); - assertEquals(0, lio.outputSpecs.size()); assertEquals(0, lio.outputsMap.size()); assertEquals(0, lio.outputContextMap.size()); - assertTrue(lio.groupInputSpecs == null || lio.groupInputSpecs.size() == 0); assertNull(lio.groupInputsMap); assertNull(lio.processor); assertNull(lio.processorContext);
