TEZ-3726. Clean up DeletionTracker's reflection instantiation and provide ContainerLauncher with dagComplete() functionality (kshukla)
diff --git a/TEZ-3334-CHANGES.txt b/TEZ-3334-CHANGES.txt
index a2afb15..8924915 100644
--- a/TEZ-3334-CHANGES.txt
+++ b/TEZ-3334-CHANGES.txt
@@ -4,6 +4,7 @@
INCOMPATIBLE CHANGES:
ALL CHANGES:
+ TEZ-3726. Clean up DeletionTracker's reflection instantiation and provide ContainerLauncher with dagComplete() functionality
TEZ-3725. Cleanup http connections and other unnecessary fields in DAG Deletion tracker classes.
TEZ-3705. Modify DeletionTracker and deletion threads to be initialized only if enabled for tez_shuffle
TEZ-3685. ShuffleHandler completedInputSet off-by-one error
diff --git a/tez-common/src/main/java/org/apache/tez/common/DagContainerLauncher.java b/tez-common/src/main/java/org/apache/tez/common/DagContainerLauncher.java
new file mode 100644
index 0000000..e3bd385
--- /dev/null
+++ b/tez-common/src/main/java/org/apache/tez/common/DagContainerLauncher.java
@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tez.common;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.tez.common.security.JobTokenSecretManager;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.serviceplugins.api.ContainerLauncher;
+import org.apache.tez.serviceplugins.api.ContainerLauncherContext;
+
+/**
+ * Plugin to allow custom container launchers to be written to launch containers that want to
+ * support cleanup of DAG level directories upon DAG completion in session mode. The directories are created by
+ * the Tez Shuffle Handler (tez_shuffle). A typical implementation of dagComplete() method would contain logic to send
+ * http request(s) for dag deletion to the nodes that support this auxiliary service.
+ */
+@Public
+@Unstable
+public abstract class DagContainerLauncher extends ContainerLauncher {
+
+ public DagContainerLauncher(ContainerLauncherContext containerLauncherContext) {
+ super(containerLauncherContext);
+ }
+
+ public abstract void dagComplete(TezDAGID dag, JobTokenSecretManager jobTokenSecretManager);
+}
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/ContainerLauncherWrapper.java b/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/ContainerLauncherWrapper.java
index c70ab10..8ecac14 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/ContainerLauncherWrapper.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/ContainerLauncherWrapper.java
@@ -14,6 +14,7 @@
package org.apache.tez.dag.app.launcher;
+import org.apache.tez.common.DagContainerLauncher;
import org.apache.tez.common.security.JobTokenSecretManager;
import org.apache.tez.dag.records.TezDAGID;
import org.apache.tez.serviceplugins.api.ContainerLaunchRequest;
@@ -41,11 +42,8 @@
}
public void dagComplete(TezDAGID dag, JobTokenSecretManager jobTokenSecretManager) {
- if (real instanceof TezContainerLauncherImpl) {
- ((TezContainerLauncherImpl)real).dagComplete(dag, jobTokenSecretManager);
- }
- if (real instanceof LocalContainerLauncher) {
- ((LocalContainerLauncher)real).dagComplete(dag, jobTokenSecretManager);
+ if (real instanceof DagContainerLauncher) {
+ ((DagContainerLauncher)real).dagComplete(dag, jobTokenSecretManager);
}
}
}
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/DeletionTrackerImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/DeletionTrackerImpl.java
index b7583ae..52b6347 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/DeletionTrackerImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/DeletionTrackerImpl.java
@@ -19,6 +19,7 @@
package org.apache.tez.dag.app.launcher;
+import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
@@ -38,12 +39,11 @@
public class DeletionTrackerImpl extends DeletionTracker {
private static final Logger LOG = LoggerFactory.getLogger(DeletionTrackerImpl.class);
- private Map<NodeId, Integer> nodeIdShufflePortMap;
+ private Map<NodeId, Integer> nodeIdShufflePortMap = new HashMap<NodeId, Integer>();
private ExecutorService dagCleanupService;
- public DeletionTrackerImpl(Map<NodeId, Integer> nodeIdShufflePortMap, Configuration conf) {
+ public DeletionTrackerImpl(Configuration conf) {
super(conf);
- this.nodeIdShufflePortMap = nodeIdShufflePortMap;
this.dagCleanupService = new ThreadPoolExecutor(0, conf.getInt(TezConfiguration.TEZ_AM_DAG_CLEANUP_THREAD_COUNT_LIMIT,
TezConfiguration.TEZ_AM_DAG_CLEANUP_THREAD_COUNT_LIMIT_DEFAULT), 10,
TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/LocalContainerLauncher.java b/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/LocalContainerLauncher.java
index 4793bd7..d50b49e 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/LocalContainerLauncher.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/LocalContainerLauncher.java
@@ -44,7 +44,7 @@
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.tez.common.DagContainerLauncher;
import org.apache.tez.common.ReflectionUtils;
import org.apache.tez.common.TezUtils;
import org.apache.tez.dag.records.TezDAGID;
@@ -53,7 +53,6 @@
import org.apache.tez.runtime.library.common.TezRuntimeUtils;
import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
import org.apache.tez.serviceplugins.api.ContainerLaunchRequest;
-import org.apache.tez.serviceplugins.api.ContainerLauncher;
import org.apache.tez.serviceplugins.api.ContainerLauncherContext;
import org.apache.tez.serviceplugins.api.ContainerStopRequest;
import org.apache.tez.serviceplugins.api.TaskAttemptEndReason;
@@ -82,7 +81,7 @@
* Since all (sub)tasks share the same local directory, they must be executed
* sequentially in order to avoid creating/deleting the same files/dirs.
*/
-public class LocalContainerLauncher extends ContainerLauncher {
+public class LocalContainerLauncher extends DagContainerLauncher {
private static final Logger LOG = LoggerFactory.getLogger(LocalContainerLauncher.class);
@@ -162,9 +161,7 @@
String deletionTrackerClassName = conf.get(TezConfiguration.TEZ_AM_DELETION_TRACKER_CLASS,
TezConfiguration.TEZ_AM_DELETION_TRACKER_CLASS_DEFAULT);
deletionTracker = ReflectionUtils.createClazzInstance(
- deletionTrackerClassName, new Class[]{
- Map.class, Configuration.class},
- new Object[]{new HashMap<NodeId, Integer>(), conf});
+ deletionTrackerClassName, new Class[]{Configuration.class}, new Object[]{conf});
}
}
@@ -408,6 +405,7 @@
}
}
+ @Override
public void dagComplete(TezDAGID dag, JobTokenSecretManager jobTokenSecretManager) {
if (deletionTracker != null) {
deletionTracker.dagComplete(dag, jobTokenSecretManager);
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/TezContainerLauncherImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/TezContainerLauncherImpl.java
index 922575f..67fc4ed 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/TezContainerLauncherImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/launcher/TezContainerLauncherImpl.java
@@ -34,7 +34,7 @@
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.hadoop.io.DataInputByteBuffer;
-import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.tez.common.DagContainerLauncher;
import org.apache.tez.common.ReflectionUtils;
import org.apache.tez.common.TezUtils;
import org.apache.tez.common.security.JobTokenSecretManager;
@@ -44,7 +44,6 @@
import org.apache.tez.runtime.library.common.TezRuntimeUtils;
import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
import org.apache.tez.serviceplugins.api.ContainerLaunchRequest;
-import org.apache.tez.serviceplugins.api.ContainerLauncher;
import org.apache.tez.serviceplugins.api.ContainerLauncherContext;
import org.apache.tez.serviceplugins.api.ContainerStopRequest;
import org.slf4j.Logger;
@@ -75,7 +74,7 @@
/**
* This class is responsible for launching of containers.
*/
-public class TezContainerLauncherImpl extends ContainerLauncher {
+public class TezContainerLauncherImpl extends DagContainerLauncher {
// TODO Ensure the same thread is used to launch / stop the same container. Or - ensure event ordering.
static final Logger LOG = LoggerFactory.getLogger(TezContainerLauncherImpl.class);
@@ -340,9 +339,7 @@
String deletionTrackerClassName = conf.get(TezConfiguration.TEZ_AM_DELETION_TRACKER_CLASS,
TezConfiguration.TEZ_AM_DELETION_TRACKER_CLASS_DEFAULT);
deletionTracker = ReflectionUtils.createClazzInstance(
- deletionTrackerClassName, new Class[]{
- Map.class, Configuration.class},
- new Object[]{new HashMap<NodeId, Integer>(), conf});
+ deletionTrackerClassName, new Class[]{Configuration.class}, new Object[]{conf});
}
}
@@ -444,6 +441,7 @@
}
}
+ @Override
public void dagComplete(TezDAGID dag, JobTokenSecretManager jobTokenSecretManager) {
if (deletionTracker != null) {
deletionTracker.dagComplete(dag, jobTokenSecretManager);
