You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by be...@apache.org on 2013/08/03 00:21:22 UTC
[5/5] git commit: Removed Hadoop from repository.
Removed Hadoop from repository.
Review: https://reviews.apache.org/r/13245
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/c621ae05
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/c621ae05
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/c621ae05
Branch: refs/heads/master
Commit: c621ae052f15ce6f18d6e08a7c9064b6a7a74711
Parents: 3a7b926
Author: Benjamin Hindman <be...@gmail.com>
Authored: Fri Aug 2 14:30:12 2013 -0700
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Fri Aug 2 15:02:31 2013 -0700
----------------------------------------------------------------------
Makefile.am | 2 +-
README | 12 -
configure.ac | 1 -
hadoop/README.md | 128 ---
hadoop/pom.xml | 110 ---
.../org/apache/hadoop/mapred/MesosExecutor.java | 145 ----
.../apache/hadoop/mapred/MesosScheduler.java | 845 -------------------
7 files changed, 1 insertion(+), 1242 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/c621ae05/Makefile.am
----------------------------------------------------------------------
diff --git a/Makefile.am b/Makefile.am
index fc1c1d0..260486b 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -18,7 +18,7 @@ ACLOCAL_AMFLAGS = -I m4
AUTOMAKE_OPTIONS = foreign
-SUBDIRS = . 3rdparty src ec2 hadoop jenkins
+SUBDIRS = . 3rdparty src ec2 jenkins
EXTRA_DIST =
http://git-wip-us.apache.org/repos/asf/mesos/blob/c621ae05/README
----------------------------------------------------------------------
diff --git a/README b/README
index 7fb6cba..e94531d 100644
--- a/README
+++ b/README
@@ -95,18 +95,6 @@ addition, see [build]/bin for scripts that can run the tests and also
launch the tests in GDB.
-Hadoop
-======
-
-Included in the distribution is a runnable tutorial on using Hadoop on
-Mesos (./hadoop/TUTORIAL.sh). Try it out!
-
-You can also "build" a self-contained distribution of Hadoop with the
-necessary Mesos components by doing 'make hadoop-0.20.205.0' or 'make
-hadoop-0.20.2-cdh3u3' from within [build]/hadoop (this uses the
-tutorial mentioned above).
-
-
Installing
==========
http://git-wip-us.apache.org/repos/asf/mesos/blob/c621ae05/configure.ac
----------------------------------------------------------------------
diff --git a/configure.ac b/configure.ac
index 4dc1cc7..7c9adac 100644
--- a/configure.ac
+++ b/configure.ac
@@ -78,7 +78,6 @@ AC_CONFIG_SUBDIRS([3rdparty/libprocess])
AC_CONFIG_FILES([Makefile])
AC_CONFIG_FILES([ec2/Makefile])
-AC_CONFIG_FILES([hadoop/Makefile])
AC_CONFIG_FILES([jenkins/Makefile])
AC_CONFIG_FILES([jenkins/pom.xml])
AC_CONFIG_FILES([src/Makefile])
http://git-wip-us.apache.org/repos/asf/mesos/blob/c621ae05/hadoop/README.md
----------------------------------------------------------------------
diff --git a/hadoop/README.md b/hadoop/README.md
deleted file mode 100644
index 36d7eb4..0000000
--- a/hadoop/README.md
+++ /dev/null
@@ -1,128 +0,0 @@
-Hadoop on Mesos
----------------
-
-#### Overview ####
-
-To run _Hadoop on Mesos_ you need to add the `hadoop-mesos-0.0.1.jar`
-library to your Hadoop distribution (any distribution that supports
-`hadoop-core-1.2.0` should work) and set some new configuration
-properties. Read on for details.
-
-#### Build ####
-
-You can build `hadoop-mesos-0.0.1.jar` using Maven:
-
-```
-$ mvn package
-```
-
-If successful, the JAR will be at `target/hadoop-mesos-0.0.1.jar`.
-
-> NOTE: If you want to build against a different version of Mesos than
-> the default you'll need to update `mesos-version` in `pom.xml`.
-
-We plan to provide already built JARs at http://repository.apache.org
-in the near future!
-
-#### Package ####
-
-You'll need to download an existing Hadoop distribution. For this
-guide, we'll use [CDH4.2.1][CDH4.2.1]. First grab the tar archive and
-extract it.
-
-```
-$ wget http://archive.cloudera.com/cdh4/cdh/4/mr1-2.0.0-mr1-cdh4.2.1.tar.gz
-...
-$ tar zxf mr1-2.0.0-mr1-cdh4.2.1.tar.gz
-```
-
-> **Take note**, the extracted directory is `hadoop-2.0.0-mr1-cdh4.2.1`.
-
-Now copy `hadoop-mesos-0.0.1.jar` into the `lib` folder.
-
-```
-$ cp /path/to/hadoop-mesos-0.0.1.jar hadoop-2.0.0-mr1-cdh4.2.1/lib/
-```
-
-_That's it!_ You now have a _Hadoop on Mesos_ distribution!
-
-[CDH4.2.1]: http://www.cloudera.com/content/support/en/documentation/cdh4-documentation/cdh4-documentation-v4-2-1.html
-
-#### Upload ####
-
-You'll want to upload your _Hadoop on Mesos_ distribution somewhere
-that Mesos can access in order to launch each `TaskTracker`. For
-example, if you're already running HDFS:
-
-```
-$ tar czf hadoop-2.0.0-mr1-cdh4.2.1.tar.gz hadoop-2.0.0-mr1-cdh4.2.1
-$ hadoop fs -put hadoop-2.0.0-mr1-cdh4.2.1.tar.gz /hadoop-2.0.0-mr1-cdh4.2.1.tar.gz
-```
-
-> **Consider** any permissions issues with your uploaded location
-> (i.e., on HDFS you'll probably want to make the file world
-> readable).
-
-Now you'll need to configure your `JobTracker` to launch each
-`TaskTracker` on Mesos!
-
-#### Configure ####
-
-Along with the normal configuration properties you might want to set
-to launch a `JobTracker`, you'll need to set some Mesos specific ones
-too.
-
-Here are the mandatory configuration properties for
-`conf/mapred-site.xml` (initialized to values representative of
-running in [pseudo distributed
-operation](http://hadoop.apache.org/docs/stable/single_node_setup.html#PseudoDistributed):
-
-```
-<property>
- <name>mapred.job.tracker</name>
- <value>localhost:9001</value>
-</property>
-<property>
- <name>mapred.jobtracker.taskScheduler</name>
- <value>org.apache.hadoop.mapred.MesosScheduler</value>
-</property>
-<property>
- <name>mapred.mesos.taskScheduler</name>
- <value>org.apache.hadoop.mapred.JobQueueTaskScheduler</value>
-</property>
-<property>
- <name>mapred.mesos.master</name>
- <value>localhost:5050</value>
-</property>
-<property>
- <name>mapred.mesos.executor</name>
- <value>hdfs://localhost:9000/hadoop-2.0.0-mr1-cdh4.2.1.tar.gz</value>
-</property>
-```
-
-#### Start ####
-
-Now you can start the `JobTracker` but you'll need to include the path
-to the Mesos native library.
-
-On Linux:
-
-```
-$ MESOS_NATIVE_LIBRARY=/path/to/libmesos.so hadoop jobtracker
-```
-
-And on OS X:
-
-```
-$ MESOS_NATIVE_LIBRARY=/path/to/libmesos.dylib hadoop jobtracker
-```
-
-> **NOTE: You do not need to worry about distributing your Hadoop
-> configuration! All of the configuration properties read by the**
-> `JobTracker` **along with any necessary** `TaskTracker` **specific
-> _overrides_ will get serialized and passed to each** `TaskTracker`
-> **on startup.**
-
-_Please email user@mesos.apache.org with questions!_
-
-----------
http://git-wip-us.apache.org/repos/asf/mesos/blob/c621ae05/hadoop/pom.xml
----------------------------------------------------------------------
diff --git a/hadoop/pom.xml b/hadoop/pom.xml
deleted file mode 100644
index baf3e5b..0000000
--- a/hadoop/pom.xml
+++ /dev/null
@@ -1,110 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <groupId>org.apache.mesos</groupId>
- <artifactId>hadoop-mesos</artifactId>
- <version>0.0.1</version>
-
- <properties>
- <encoding>UTF-8</encoding>
-
- <!-- language versions -->
- <java.abi>1.6</java.abi>
-
- <!-- runtime deps versions -->
- <commons-logging.version>1.1.1</commons-logging.version>
- <hadoop-core.version>1.2.0</hadoop-core.version>
- <mesos.version>0.14.0</mesos.version>
- <protobuf.version>2.4.1</protobuf.version>
-
- <!-- test deps versions -->
- <junit.version>4.11</junit.version>
- <mockito.version>1.9.5</mockito.version>
- </properties>
-
- <!-- TODO(benh): Remove repository once we've published a JDK6 JAR. -->
- <repositories>
- <repository>
- <id>mesosphere-public-repo</id>
- <name>Mesosphere Public Repo</name>
- <url>http://s3.amazonaws.com/mesosphere-maven-public/</url>
- </repository>
- </repositories>
-
- <dependencies>
- <dependency>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging</artifactId>
- <version>${commons-logging.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-core</artifactId>
- <version>${hadoop-core.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.mesos</groupId>
- <!-- TODO(benh): Use 'mesos' after we've published a JDK6 JAR. -->
- <artifactId>mesos_jdk6</artifactId>
- <version>${mesos.version}</version>
- </dependency>
- <dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- <version>${protobuf.version}</version>
- </dependency>
-
- <!-- Test scope -->
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <version>${junit.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- <version>3.1</version>
- <configuration>
- <source>${java.abi}</source>
- <target>${java.abi}</target>
- <showDeprecation>true</showDeprecation>
- <showWarnings>true</showWarnings>
- </configuration>
- </plugin>
-
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <version>2.0</version>
- <configuration>
- <filters>
- <filter>
- <artifact>*:*</artifact>
- <excludes>
- <exclude>META-INF/*.SF</exclude>
- <exclude>META-INF/*.DSA</exclude>
- <exclude>META-INF/*.RSA</exclude>
- </excludes>
- </filter>
- </filters>
- </configuration>
- <executions>
- <execution>
- <phase>package</phase>
- <goals>
- <goal>shade</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
-
- </plugins>
- </build>
-
-</project>
http://git-wip-us.apache.org/repos/asf/mesos/blob/c621ae05/hadoop/src/main/java/org/apache/hadoop/mapred/MesosExecutor.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/hadoop/mapred/MesosExecutor.java b/hadoop/src/main/java/org/apache/hadoop/mapred/MesosExecutor.java
deleted file mode 100644
index ccf0090..0000000
--- a/hadoop/src/main/java/org/apache/hadoop/mapred/MesosExecutor.java
+++ /dev/null
@@ -1,145 +0,0 @@
-package org.apache.hadoop.mapred;
-
-import java.io.IOException;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.mesos.Executor;
-import org.apache.mesos.ExecutorDriver;
-import org.apache.mesos.MesosExecutorDriver;
-import org.apache.mesos.Protos.Environment.Variable;
-import org.apache.mesos.Protos.ExecutorInfo;
-import org.apache.mesos.Protos.FrameworkInfo;
-import org.apache.mesos.Protos.SlaveInfo;
-import org.apache.mesos.Protos.Status;
-import org.apache.mesos.Protos.TaskID;
-import org.apache.mesos.Protos.TaskInfo;
-import org.apache.mesos.Protos.TaskState;
-import org.apache.mesos.Protos.TaskStatus;
-
-public class MesosExecutor implements Executor {
- public static final Log LOG = LogFactory.getLog(MesosExecutor.class);
-
- private JobConf conf;
- private TaskTracker taskTracker;
-
- @Override
- public void registered(ExecutorDriver driver, ExecutorInfo executorInfo,
- FrameworkInfo frameworkInfo, SlaveInfo slaveInfo) {
- LOG.info("Executor registered with the slave");
-
- conf = new JobConf();
-
- // Get TaskTracker's config options from environment variables set by the
- // JobTracker.
- if (executorInfo.getCommand().hasEnvironment()) {
- for (Variable variable : executorInfo.getCommand().getEnvironment()
- .getVariablesList()) {
- LOG.info("Setting config option : " + variable.getName() + " to "
- + variable.getValue());
- conf.set(variable.getName(), variable.getValue());
- }
- }
-
- // Get hostname from Mesos to make sure we match what it reports
- // to the JobTracker.
- conf.set("slave.host.name", slaveInfo.getHostname());
-
- // Set the mapred.local directory inside the executor sandbox, so that
- // different TaskTrackers on the same host do not step on each other.
- conf.set("mapred.local.dir", System.getProperty("user.dir") + "/mapred");
- }
-
- @Override
- public void launchTask(final ExecutorDriver driver, final TaskInfo task) {
- LOG.info("Launching task : " + task.getTaskId().getValue());
-
- // NOTE: We need to manually set the context class loader here because,
- // the TaskTracker is unable to find LoginModule class otherwise.
- Thread.currentThread().setContextClassLoader(
- TaskTracker.class.getClassLoader());
-
- try {
- taskTracker = new TaskTracker(conf);
- } catch (IOException e) {
- LOG.fatal("Failed to start TaskTracker", e);
- System.exit(1);
- } catch (InterruptedException e) {
- LOG.fatal("Failed to start TaskTracker", e);
- System.exit(1);
- }
-
- // Spin up a TaskTracker in a new thread.
- new Thread("TaskTracker Run Thread") {
- @Override
- public void run() {
- taskTracker.run();
-
- // Send a TASK_FINISHED status update.
- // We do this here because we want to send it in a separate thread
- // than was used to call killTask().
- driver.sendStatusUpdate(TaskStatus.newBuilder()
- .setTaskId(task.getTaskId())
- .setState(TaskState.TASK_FINISHED)
- .build());
-
- // Give some time for the update to reach the slave.
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- LOG.error("Failed to sleep TaskTracker thread", e);
- }
-
- // Stop the executor.
- driver.stop();
- }
- }.start();
-
- driver.sendStatusUpdate(TaskStatus.newBuilder()
- .setTaskId(task.getTaskId())
- .setState(TaskState.TASK_RUNNING).build());
- }
-
- @Override
- public void killTask(ExecutorDriver driver, TaskID taskId) {
- LOG.info("Killing task : " + taskId.getValue());
- try {
- taskTracker.shutdown();
- } catch (IOException e) {
- LOG.error("Failed to shutdown TaskTracker", e);
- } catch (InterruptedException e) {
- LOG.error("Failed to shutdown TaskTracker", e);
- }
- }
-
- @Override
- public void reregistered(ExecutorDriver driver, SlaveInfo slaveInfo) {
- LOG.info("Executor reregistered with the slave");
- }
-
- @Override
- public void disconnected(ExecutorDriver driver) {
- LOG.info("Executor disconnected from the slave");
- }
-
- @Override
- public void frameworkMessage(ExecutorDriver d, byte[] msg) {
- LOG.info("Executor received framework message of length: " + msg.length
- + " bytes");
- }
-
- @Override
- public void error(ExecutorDriver d, String message) {
- LOG.error("MesosExecutor.error: " + message);
- }
-
- @Override
- public void shutdown(ExecutorDriver d) {
- LOG.info("Executor asked to shutdown");
- }
-
- public static void main(String[] args) {
- MesosExecutorDriver driver = new MesosExecutorDriver(new MesosExecutor());
- System.exit(driver.run() == Status.DRIVER_STOPPED ? 0 : 1);
- }
-}
http://git-wip-us.apache.org/repos/asf/mesos/blob/c621ae05/hadoop/src/main/java/org/apache/hadoop/mapred/MesosScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/hadoop/mapred/MesosScheduler.java b/hadoop/src/main/java/org/apache/hadoop/mapred/MesosScheduler.java
deleted file mode 100644
index 0332dea..0000000
--- a/hadoop/src/main/java/org/apache/hadoop/mapred/MesosScheduler.java
+++ /dev/null
@@ -1,845 +0,0 @@
-package org.apache.hadoop.mapred;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.Iterator;
-
-import org.apache.commons.httpclient.HttpHost;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.server.jobtracker.TaskTracker;
-import org.apache.hadoop.mapreduce.TaskType;
-import org.apache.mesos.MesosSchedulerDriver;
-import org.apache.mesos.Protos;
-import org.apache.mesos.Protos.CommandInfo;
-import org.apache.mesos.Protos.ExecutorID;
-import org.apache.mesos.Protos.ExecutorInfo;
-import org.apache.mesos.Protos.FrameworkID;
-import org.apache.mesos.Protos.FrameworkInfo;
-import org.apache.mesos.Protos.MasterInfo;
-import org.apache.mesos.Protos.Offer;
-import org.apache.mesos.Protos.OfferID;
-import org.apache.mesos.Protos.Resource;
-import org.apache.mesos.Protos.SlaveID;
-import org.apache.mesos.Protos.TaskID;
-import org.apache.mesos.Protos.TaskInfo;
-import org.apache.mesos.Protos.TaskState;
-import org.apache.mesos.Protos.Value;
-import org.apache.mesos.Scheduler;
-import org.apache.mesos.SchedulerDriver;
-
-import static org.apache.hadoop.util.StringUtils.join;
-
-public class MesosScheduler extends TaskScheduler implements Scheduler {
- public static final Log LOG = LogFactory.getLog(MesosScheduler.class);
-
- private SchedulerDriver driver;
- private TaskScheduler taskScheduler;
- private JobTracker jobTracker;
- private Configuration conf;
-
- // This is the memory overhead for a jvm process. This needs to be added
- // to a jvm process's resource requirement, in addition to its heap size.
- private static final double JVM_MEM_OVERHEAD_PERCENT_DEFAULT = 0.1; // 10%.
-
- // TODO(vinod): Consider parsing the slot memory from the configuration jvm
- // heap options (e.g: mapred.child.java.opts).
-
- // NOTE: It appears that there's no real resource requirements for a
- // map / reduce slot. We therefore define a default slot as:
- // 0.2 cores.
- // 512 MB memory.
- // 1 GB of disk space.
- private static final double SLOT_CPUS_DEFAULT = 0.2; // 0.2 cores.
- private static final int SLOT_DISK_DEFAULT = 1024; // 1 GB.
- private static final int SLOT_JVM_HEAP_DEFAULT = 256; // 256MB.
-
- private static final double TASKTRACKER_CPUS = 1.0; // 1 core.
- private static final int TASKTRACKER_MEM_DEFAULT = 1024; // 1 GB.
-
- // The default behavior in Hadoop is to use 4 slots per TaskTracker:
- private static final int MAP_SLOTS_DEFAULT = 2;
- private static final int REDUCE_SLOTS_DEFAULT = 2;
-
- // The amount of time to wait for task trackers to launch before
- // giving up.
- private static final long LAUNCH_TIMEOUT_MS = 300000; // 5 minutes
-
- // Count of the launched trackers for TaskID generation.
- private long launchedTrackers = 0;
-
- // Maintains a mapping from {tracker host:port -> MesosTracker}.
- // Used for tracking the slots of each TaskTracker and the corresponding
- // Mesos TaskID.
- private Map<HttpHost, MesosTracker> mesosTrackers =
- new HashMap<HttpHost, MesosTracker>();
-
- private JobInProgressListener jobListener = new JobInProgressListener() {
- @Override
- public void jobAdded(JobInProgress job) throws IOException {
- LOG.info("Added job " + job.getJobID());
- }
-
- @Override
- public void jobRemoved(JobInProgress job) {
- LOG.info("Removed job " + job.getJobID());
- }
-
- @Override
- public void jobUpdated(JobChangeEvent event) {
- synchronized (MesosScheduler.this) {
- JobInProgress job = event.getJobInProgress();
-
- // If the job is complete, kill all the corresponding idle TaskTrackers.
- if (!job.isComplete()) {
- return;
- }
-
- LOG.info("Completed job : " + job.getJobID());
-
- List<TaskInProgress> completed = new ArrayList<TaskInProgress>();
-
- // Map tasks.
- completed.addAll(job.reportTasksInProgress(true, true));
-
- // Reduce tasks.
- completed.addAll(job.reportTasksInProgress(false, true));
-
- for (TaskInProgress task : completed) {
- // Check that this task actually belongs to this job
- if (task.getJob().getJobID() != job.getJobID()) {
- continue;
- }
-
- for (TaskStatus status : task.getTaskStatuses()) {
- // Make a copy to iterate over keys and delete values.
- Set<HttpHost> trackers = new HashSet<HttpHost>(
- mesosTrackers.keySet());
-
- // Remove the task from the map.
- for (HttpHost tracker : trackers) {
- MesosTracker mesosTracker = mesosTrackers.get(tracker);
-
- if (!mesosTracker.active) {
- LOG.warn("Ignoring TaskTracker: " + tracker
- + " because it might not have sent a hearbeat");
- continue;
- }
-
- LOG.info("Removing completed task : " + status.getTaskID()
- + " of tracker " + status.getTaskTracker());
-
- mesosTracker.hadoopJobs.remove(job.getJobID());
-
- // If the TaskTracker doesn't have any running tasks, kill it.
- if (mesosTracker.hadoopJobs.isEmpty()) {
- LOG.info("Killing Mesos task: " + mesosTracker.taskId + " on host "
- + mesosTracker.host);
-
- driver.killTask(mesosTracker.taskId);
- mesosTracker.timer.cancel();
- mesosTrackers.remove(tracker);
- }
- }
- }
- }
- }
- }
- };
-
- public MesosScheduler() {}
-
- // TaskScheduler methods.
- @Override
- public synchronized void start() throws IOException {
- conf = getConf();
- String taskTrackerClass = conf.get("mapred.mesos.taskScheduler",
- "org.apache.hadoop.mapred.JobQueueTaskScheduler");
-
- try {
- taskScheduler =
- (TaskScheduler) Class.forName(taskTrackerClass).newInstance();
- taskScheduler.setConf(conf);
- taskScheduler.setTaskTrackerManager(taskTrackerManager);
- } catch (ClassNotFoundException e) {
- LOG.fatal("Failed to initialize the TaskScheduler", e);
- System.exit(1);
- } catch (InstantiationException e) {
- LOG.fatal("Failed to initialize the TaskScheduler", e);
- System.exit(1);
- } catch (IllegalAccessException e) {
- LOG.fatal("Failed to initialize the TaskScheduler", e);
- System.exit(1);
- }
-
- // Add the job listener to get job related updates.
- taskTrackerManager.addJobInProgressListener(jobListener);
-
- LOG.info("Starting MesosScheduler");
- jobTracker = (JobTracker) super.taskTrackerManager;
-
- String master = conf.get("mapred.mesos.master", "local");
-
- try {
- FrameworkInfo frameworkInfo = FrameworkInfo
- .newBuilder()
- .setUser("")
- .setCheckpoint(conf.getBoolean("mapred.mesos.checkpoint", false))
- .setName("Hadoop: (RPC port: " + jobTracker.port + ","
- + " WebUI port: " + jobTracker.infoPort + ")").build();
-
- driver = new MesosSchedulerDriver(this, frameworkInfo, master);
- driver.start();
- } catch (Exception e) {
- // If the MesosScheduler can't be loaded, the JobTracker won't be useful
- // at all, so crash it now so that the user notices.
- LOG.fatal("Failed to start MesosScheduler", e);
- System.exit(1);
- }
-
- taskScheduler.start();
- }
-
- @Override
- public synchronized void terminate() throws IOException {
- try {
- LOG.info("Stopping MesosScheduler");
- driver.stop();
- } catch (Exception e) {
- LOG.error("Failed to stop Mesos scheduler", e);
- }
-
- taskScheduler.terminate();
- }
-
- @Override
- public synchronized List<Task> assignTasks(TaskTracker taskTracker)
- throws IOException {
- HttpHost tracker = new HttpHost(taskTracker.getStatus().getHost(),
- taskTracker.getStatus().getHttpPort());
-
- if (!mesosTrackers.containsKey(tracker)) {
- // TODO(bmahler): Consider allowing non-Mesos TaskTrackers.
- LOG.info("Unknown/exited TaskTracker: " + tracker + ". ");
- return null;
- }
-
- // Let the underlying task scheduler do the actual task scheduling.
- List<Task> tasks = taskScheduler.assignTasks(taskTracker);
-
- // The Hadoop Fair Scheduler is known to return null.
- if (tasks != null) {
- // Keep track of which TaskTracker contains which tasks.
- for (Task task : tasks) {
- mesosTrackers.get(tracker).hadoopJobs.add(task.getJobID());
- }
- }
-
- return tasks;
- }
-
- @Override
- public synchronized Collection<JobInProgress> getJobs(String queueName) {
- return taskScheduler.getJobs(queueName);
- }
-
- @Override
- public synchronized void refresh() throws IOException {
- taskScheduler.refresh();
- }
-
- // Mesos Scheduler methods.
- // These are synchronized, where possible. Some of these methods need to access the
- // JobTracker, which can lead to a deadlock:
- // See: https://issues.apache.org/jira/browse/MESOS-429
- // The workaround employed here is to unsynchronize those methods needing access to
- // the JobTracker state and use explicit synchronization instead as appropriate.
- // TODO(bmahler): Provide a cleaner solution to this issue. One solution is to
- // split up the Scheduler and TaskScheduler implementations in order to break the
- // locking cycle. This would require a synchronized class to store the shared
- // state across our Scheduler and TaskScheduler implementations, and provide
- // atomic operations as needed.
- @Override
- public synchronized void registered(SchedulerDriver schedulerDriver,
- FrameworkID frameworkID, MasterInfo masterInfo) {
- LOG.info("Registered as " + frameworkID.getValue()
- + " with master " + masterInfo);
- }
-
- @Override
- public synchronized void reregistered(SchedulerDriver schedulerDriver,
- MasterInfo masterInfo) {
- LOG.info("Re-registered with master " + masterInfo);
- }
-
- public synchronized void killTracker(MesosTracker tracker) {
- driver.killTask(tracker.taskId);
- mesosTrackers.remove(tracker.host);
- }
-
- // For some reason, pendingMaps() and pendingReduces() doesn't return the
- // values we expect. We observed negative values, which may be related to
- // https://issues.apache.org/jira/browse/MAPREDUCE-1238. Below is the
- // algorithm that is used to calculate the pending tasks within the Hadoop
- // JobTracker sources (see 'printTaskSummary' in
- // src/org/apache/hadoop/mapred/jobdetails_jsp.java).
- private int getPendingTasks(TaskInProgress[] tasks) {
- int totalTasks = tasks.length;
- int runningTasks = 0;
- int finishedTasks = 0;
- int killedTasks = 0;
- for (int i = 0; i < totalTasks; ++i) {
- TaskInProgress task = tasks[i];
- if (task == null) {
- continue;
- }
- if (task.isComplete()) {
- finishedTasks += 1;
- } else if (task.isRunning()) {
- runningTasks += 1;
- } else if (task.wasKilled()) {
- killedTasks += 1;
- }
- }
- int pendingTasks = totalTasks - runningTasks - killedTasks - finishedTasks;
- return pendingTasks;
- }
-
- // This method uses explicit synchronization in order to avoid deadlocks when
- // accessing the JobTracker.
- @Override
- public void resourceOffers(SchedulerDriver schedulerDriver,
- List<Offer> offers) {
- // Before synchronizing, we pull all needed information from the JobTracker.
- final HttpHost jobTrackerAddress =
- new HttpHost(jobTracker.getHostname(), jobTracker.getTrackerPort());
-
- final Collection<TaskTrackerStatus> taskTrackers = jobTracker.taskTrackers();
-
- final List<JobInProgress> jobsInProgress = new ArrayList<JobInProgress>();
- for (JobStatus status : jobTracker.jobsToComplete()) {
- jobsInProgress.add(jobTracker.getJob(status.getJobID()));
- }
-
- synchronized (this) {
- // Compute the number of pending maps and reduces.
- int pendingMaps = 0;
- int pendingReduces = 0;
- for (JobInProgress progress : jobsInProgress) {
- // JobStatus.pendingMaps/Reduces may return the wrong value on
- // occasion. This seems to be safer.
- pendingMaps += getPendingTasks(progress.getTasks(TaskType.MAP));
- pendingReduces += getPendingTasks(progress.getTasks(TaskType.REDUCE));
- }
-
- // Mark active (heartbeated) TaskTrackers and compute idle slots.
- int idleMapSlots = 0;
- int idleReduceSlots = 0;
- int unhealthyTrackers = 0;
-
- for (TaskTrackerStatus status : taskTrackers) {
- if (!status.getHealthStatus().isNodeHealthy()) {
- // Skip this node if it's unhealthy.
- ++unhealthyTrackers;
- continue;
- }
-
- HttpHost host = new HttpHost(status.getHost(), status.getHttpPort());
- if (mesosTrackers.containsKey(host)) {
- mesosTrackers.get(host).active = true;
- mesosTrackers.get(host).timer.cancel();
- idleMapSlots += status.getAvailableMapSlots();
- idleReduceSlots += status.getAvailableReduceSlots();
- }
- }
-
- // Consider the TaskTrackers that have yet to become active as being idle,
- // otherwise we will launch excessive TaskTrackers.
- int inactiveMapSlots = 0;
- int inactiveReduceSlots = 0;
- for (MesosTracker tracker : mesosTrackers.values()) {
- if (!tracker.active) {
- inactiveMapSlots += tracker.mapSlots;
- inactiveReduceSlots += tracker.reduceSlots;
- }
- }
-
- // To ensure Hadoop jobs begin promptly, we can specify a minimum number
- // of 'hot slots' to be available for use. This addresses the
- // TaskTracker spin up delay that exists with Hadoop on Mesos. This can
- // be a nuisance with lower latency applications, such as ad-hoc Hive
- // queries.
- int minimumMapSlots = conf.getInt("mapred.mesos.total.map.slots.minimum", 0);
- int minimumReduceSlots =
- conf.getInt("mapred.mesos.total.reduce.slots.minimum", 0);
-
- // Compute how many slots we need to allocate.
- int neededMapSlots = Math.max(
- minimumMapSlots - (idleMapSlots + inactiveMapSlots),
- pendingMaps - (idleMapSlots + inactiveMapSlots));
- int neededReduceSlots = Math.max(
- minimumReduceSlots - (idleReduceSlots + inactiveReduceSlots),
- pendingReduces - (idleReduceSlots + inactiveReduceSlots));
-
- LOG.info(join("\n", Arrays.asList(
- "JobTracker Status",
- " Pending Map Tasks: " + pendingMaps,
- " Pending Reduce Tasks: " + pendingReduces,
- " Idle Map Slots: " + idleMapSlots,
- " Idle Reduce Slots: " + idleReduceSlots,
- " Inactive Map Slots: " + inactiveMapSlots
- + " (launched but no hearbeat yet)",
- " Inactive Reduce Slots: " + inactiveReduceSlots
- + " (launched but no hearbeat yet)",
- " Needed Map Slots: " + neededMapSlots,
- " Needed Reduce Slots: " + neededReduceSlots,
- " Unhealthy Trackers: " + unhealthyTrackers)));
-
- // Launch TaskTrackers to satisfy the slot requirements.
- // TODO(bmahler): Consider slotting intelligently.
- // Ex: If more map slots are needed, but no reduce slots are needed,
- // launch a map-only TaskTracker to better satisfy the slot needs.
- for (Offer offer : offers) {
- if (neededMapSlots <= 0 && neededReduceSlots <= 0) {
- driver.declineOffer(offer.getId());
- continue;
- }
-
- // Ensure these values aren't < 0.
- neededMapSlots = Math.max(0, neededMapSlots);
- neededReduceSlots = Math.max(0, neededReduceSlots);
-
- double cpus = -1.0;
- double mem = -1.0;
- double disk = -1.0;
- Set<Integer> ports = new HashSet<Integer>();
-
- // Pull out the cpus, memory, disk, and 2 ports from the offer.
- for (Resource resource : offer.getResourcesList()) {
- if (resource.getName().equals("cpus")
- && resource.getType() == Value.Type.SCALAR) {
- cpus = resource.getScalar().getValue();
- } else if (resource.getName().equals("mem")
- && resource.getType() == Value.Type.SCALAR) {
- mem = resource.getScalar().getValue();
- } else if (resource.getName().equals("disk")
- && resource.getType() == Value.Type.SCALAR) {
- disk = resource.getScalar().getValue();
- } else if (resource.getName().equals("ports")
- && resource.getType() == Value.Type.RANGES) {
- for (Value.Range range : resource.getRanges().getRangeList()) {
- Integer begin = (int)range.getBegin();
- Integer end = (int)range.getEnd();
- if (end < begin) {
- LOG.warn("Ignoring invalid port range: begin=" + begin + " end=" + end);
- continue;
- }
- while (begin <= end && ports.size() < 2) {
- ports.add(begin);
- begin += 1;
- }
- }
- }
- }
-
- int mapSlotsMax = conf.getInt("mapred.tasktracker.map.tasks.maximum",
- MAP_SLOTS_DEFAULT);
- int reduceSlotsMax =
- conf.getInt("mapred.tasktracker.reduce.tasks.maximum",
- REDUCE_SLOTS_DEFAULT);
-
- // What's the minimum number of map and reduce slots we should try to
- // launch?
- long mapSlots = 0;
- long reduceSlots = 0;
-
- double slotCpus = conf.getFloat("mapred.mesos.slot.cpus",
- (float) SLOT_CPUS_DEFAULT);
- double slotDisk = conf.getInt("mapred.mesos.slot.disk",
- SLOT_DISK_DEFAULT);
-
- int slotMem = conf.getInt("mapred.mesos.slot.mem",
- SLOT_JVM_HEAP_DEFAULT);
- long slotJVMHeap = Math.round((double)slotMem -
- (JVM_MEM_OVERHEAD_PERCENT_DEFAULT * slotMem));
-
- int tasktrackerMem = conf.getInt("mapred.mesos.tasktracker.mem",
- TASKTRACKER_MEM_DEFAULT);
- long tasktrackerJVMHeap = Math.round((double)tasktrackerMem -
- (JVM_MEM_OVERHEAD_PERCENT_DEFAULT * tasktrackerMem));
-
- // Minimum resource requirements for the container (TaskTracker + map/red
- // tasks).
- double containerCpus = TASKTRACKER_CPUS;
- double containerMem = tasktrackerMem;
- double containerDisk = 0;
-
- // Determine how many slots we can allocate.
- int slots = mapSlotsMax + reduceSlotsMax;
- slots = (int)Math.min(slots, (cpus - containerCpus) / slotCpus);
- slots = (int)Math.min(slots, (mem - containerMem) / slotMem);
- slots = (int)Math.min(slots, (disk - containerDisk) / slotDisk);
-
- // Is this offer too small for even the minimum slots?
- if (slots < 1 || ports.size() < 2) {
- LOG.info(join("\n", Arrays.asList(
- "Declining offer with insufficient resources for a TaskTracker: ",
- " cpus: offered " + cpus + " needed " + containerCpus,
- " mem : offered " + mem + " needed " + containerMem,
- " disk: offered " + disk + " needed " + containerDisk,
- " ports: " + (ports.size() < 2
- ? " less than 2 offered"
- : " at least 2 (sufficient)"))));
-
- driver.declineOffer(offer.getId());
- continue;
- }
-
- // Is the number of slots we need sufficiently small? If so, we can
- // allocate exactly the number we need.
- if (slots >= neededMapSlots + neededReduceSlots && neededMapSlots <
- mapSlotsMax && neededReduceSlots < reduceSlotsMax) {
- mapSlots = neededMapSlots;
- reduceSlots = neededReduceSlots;
- } else {
- // Allocate slots fairly for this resource offer.
- double mapFactor = (double)neededMapSlots / (neededMapSlots + neededReduceSlots);
- double reduceFactor = (double)neededReduceSlots / (neededMapSlots + neededReduceSlots);
- // To avoid map/reduce slot starvation, don't allow more than 50%
- // spread between map/reduce slots when we need both mappers and
- // reducers.
- if (neededMapSlots > 0 && neededReduceSlots > 0) {
- if (mapFactor < 0.25) {
- mapFactor = 0.25;
- } else if (mapFactor > 0.75) {
- mapFactor = 0.75;
- }
- if (reduceFactor < 0.25) {
- reduceFactor = 0.25;
- } else if (reduceFactor > 0.75) {
- reduceFactor = 0.75;
- }
- }
- mapSlots = Math.min(Math.min((long)(mapFactor * slots), mapSlotsMax), neededMapSlots);
- // The remaining slots are allocated for reduces.
- slots -= mapSlots;
- reduceSlots = Math.min(Math.min(slots, reduceSlotsMax), neededReduceSlots);
- }
-
- Iterator<Integer> portIter = ports.iterator();
- HttpHost httpAddress = new HttpHost(offer.getHostname(), portIter.next());
- HttpHost reportAddress = new HttpHost(offer.getHostname(), portIter.next());
-
- // Check that this tracker is not already launched. This problem was
- // observed on a few occasions, but not reliably. The main symptom was
- // that entries in `mesosTrackers` were being lost, and task trackers
- // would be 'lost' mysteriously (probably because the ports were in
- // use). This problem has since gone away with a rewrite of the port
- // selection code, but the check + logging is left here.
- // TODO(brenden): Diagnose this to determine root cause.
-
- if (mesosTrackers.containsKey(httpAddress)) {
- LOG.info(join("\n", Arrays.asList(
- "Declining offer because host/port combination is in use: ",
- " cpus: offered " + cpus + " needed " + containerCpus,
- " mem : offered " + mem + " needed " + containerMem,
- " disk: offered " + disk + " needed " + containerDisk,
- " ports: " + ports)));
-
- driver.declineOffer(offer.getId());
- continue;
- }
-
- TaskID taskId = TaskID.newBuilder()
- .setValue("Task_Tracker_" + launchedTrackers++).build();
-
- LOG.info("Launching task " + taskId.getValue() + " on "
- + httpAddress.toString() + " with mapSlots=" + mapSlots + " reduceSlots=" + reduceSlots);
-
- // Add this tracker to Mesos tasks.
- mesosTrackers.put(httpAddress, new MesosTracker(httpAddress, taskId,
- mapSlots, reduceSlots, this));
-
- // Create the environment depending on whether the executor is going to be
- // run locally.
- // TODO(vinod): Do not pass the mapred config options as environment
- // variables.
- Protos.Environment.Builder envBuilder = Protos.Environment
- .newBuilder()
- .addVariables(
- Protos.Environment.Variable
- .newBuilder()
- .setName("mapred.job.tracker")
- .setValue(jobTrackerAddress.getHostName() + ':'
- + jobTrackerAddress.getPort()))
- .addVariables(
- Protos.Environment.Variable
- .newBuilder()
- .setName("mapred.task.tracker.http.address")
- .setValue(
- httpAddress.getHostName() + ':' + httpAddress.getPort()))
- .addVariables(
- Protos.Environment.Variable
- .newBuilder()
- .setName("mapred.task.tracker.report.address")
- .setValue(reportAddress.getHostName() + ':'
- + reportAddress.getPort()))
- .addVariables(
- Protos.Environment.Variable.newBuilder()
- .setName("mapred.map.child.java.opts")
- .setValue("-Xmx" + slotJVMHeap + "m"))
- .addVariables(
- Protos.Environment.Variable.newBuilder()
- .setName("mapred.reduce.child.java.opts")
- .setValue("-Xmx" + slotJVMHeap + "m"))
- .addVariables(
- Protos.Environment.Variable.newBuilder()
- .setName("HADOOP_HEAPSIZE")
- .setValue("" + tasktrackerJVMHeap))
- .addVariables(
- Protos.Environment.Variable.newBuilder()
- .setName("mapred.tasktracker.map.tasks.maximum")
- .setValue("" + mapSlots))
- .addVariables(
- Protos.Environment.Variable.newBuilder()
- .setName("mapred.tasktracker.reduce.tasks.maximum")
- .setValue("" + reduceSlots));
-
- // Set java specific environment, appropriately.
- Map<String, String> env = System.getenv();
- if (env.containsKey("JAVA_HOME")) {
- envBuilder.addVariables(Protos.Environment.Variable.newBuilder()
- .setName("JAVA_HOME")
- .setValue(env.get("JAVA_HOME")));
- }
-
- if (env.containsKey("JAVA_LIBRARY_PATH")) {
- envBuilder.addVariables(Protos.Environment.Variable.newBuilder()
- .setName("JAVA_LIBRARY_PATH")
- .setValue(env.get("JAVA_LIBRARY_PATH")));
- }
-
- // Command info differs when performing a local run.
- CommandInfo commandInfo = null;
- String master = conf.get("mapred.mesos.master", "local");
-
- if (master.equals("local")) {
- commandInfo = CommandInfo.newBuilder()
- .setEnvironment(envBuilder)
- .setValue(new File("bin/hadoop").getCanonicalPath() +
- " org.apache.hadoop.mapred.MesosExecutor")
- .build();
- } else {
- String uri = conf.get("mapred.mesos.executor");
- commandInfo = CommandInfo.newBuilder()
- .setEnvironment(envBuilder)
- .setValue("cd hadoop-* && " +
- "./bin/hadoop org.apache.hadoop.mapred.MesosExecutor")
- .addUris(CommandInfo.URI.newBuilder().setValue(uri)).build();
- }
-
- TaskInfo info = TaskInfo
- .newBuilder()
- .setName(taskId.getValue())
- .setTaskId(taskId)
- .setSlaveId(offer.getSlaveId())
- .addResources(
- Resource
- .newBuilder()
- .setName("cpus")
- .setType(Value.Type.SCALAR)
- .setScalar(Value.Scalar.newBuilder().setValue(
- (mapSlots + reduceSlots) * slotCpus)))
- .addResources(
- Resource
- .newBuilder()
- .setName("mem")
- .setType(Value.Type.SCALAR)
- .setScalar(Value.Scalar.newBuilder().setValue(
- (mapSlots + reduceSlots) * slotMem)))
- .addResources(
- Resource
- .newBuilder()
- .setName("disk")
- .setType(Value.Type.SCALAR)
- .setScalar(Value.Scalar.newBuilder().setValue(
- (mapSlots + reduceSlots) * slotDisk)))
- .addResources(
- Resource
- .newBuilder()
- .setName("ports")
- .setType(Value.Type.RANGES)
- .setRanges(
- Value.Ranges
- .newBuilder()
- .addRange(Value.Range.newBuilder()
- .setBegin(httpAddress.getPort())
- .setEnd(httpAddress.getPort()))
- .addRange(Value.Range.newBuilder()
- .setBegin(reportAddress.getPort())
- .setEnd(reportAddress.getPort()))))
- .setExecutor(
- ExecutorInfo
- .newBuilder()
- .setExecutorId(ExecutorID.newBuilder().setValue(
- "executor_" + taskId.getValue()))
- .setName("Hadoop TaskTracker")
- .setSource(taskId.getValue())
- .addResources(
- Resource
- .newBuilder()
- .setName("cpus")
- .setType(Value.Type.SCALAR)
- .setScalar(Value.Scalar.newBuilder().setValue(
- (TASKTRACKER_CPUS))))
- .addResources(
- Resource
- .newBuilder()
- .setName("mem")
- .setType(Value.Type.SCALAR)
- .setScalar(Value.Scalar.newBuilder().setValue(
- (tasktrackerMem)))).setCommand(commandInfo))
- .build();
-
- driver.launchTasks(offer.getId(), Arrays.asList(info));
-
- neededMapSlots -= mapSlots;
- neededReduceSlots -= reduceSlots;
- }
-
- if (neededMapSlots <= 0 && neededReduceSlots <= 0) {
- LOG.info("Satisfied map and reduce slots needed.");
- } else {
- LOG.info("Unable to fully satisfy needed map/reduce slots: "
- + (neededMapSlots > 0 ? neededMapSlots + " map slots " : "")
- + (neededReduceSlots > 0 ? neededReduceSlots + " reduce slots " : "")
- + "remaining");
- }
- }
- }
-
- @Override
- public synchronized void offerRescinded(SchedulerDriver schedulerDriver,
- OfferID offerID) {
- LOG.warn("Rescinded offer: " + offerID.getValue());
- }
-
- @Override
- public synchronized void statusUpdate(SchedulerDriver schedulerDriver,
- Protos.TaskStatus taskStatus) {
- LOG.info("Status update of " + taskStatus.getTaskId().getValue()
- + " to " + taskStatus.getState().name()
- + " with message " + taskStatus.getMessage());
-
- // Remove the TaskTracker if the corresponding Mesos task has reached a
- // terminal state.
- switch (taskStatus.getState()) {
- case TASK_FINISHED:
- case TASK_FAILED:
- case TASK_KILLED:
- case TASK_LOST:
- // Make a copy to iterate over keys and delete values.
- Set<HttpHost> trackers = new HashSet<HttpHost>(mesosTrackers.keySet());
-
- // Remove the task from the map.
- for (HttpHost tracker : trackers) {
- if (mesosTrackers.get(tracker).taskId.equals(taskStatus.getTaskId())) {
- LOG.info("Removing terminated TaskTracker: " + tracker);
- mesosTrackers.get(tracker).timer.cancel();
- mesosTrackers.remove(tracker);
- }
- }
- break;
- case TASK_STAGING:
- case TASK_STARTING:
- case TASK_RUNNING:
- break;
- default:
- LOG.error("Unexpected TaskStatus: " + taskStatus.getState().name());
- break;
- }
- }
-
- @Override
- public synchronized void frameworkMessage(SchedulerDriver schedulerDriver,
- ExecutorID executorID, SlaveID slaveID, byte[] bytes) {
- LOG.info("Framework Message of " + bytes.length + " bytes"
- + " from executor " + executorID.getValue()
- + " on slave " + slaveID.getValue());
- }
-
- @Override
- public synchronized void disconnected(SchedulerDriver schedulerDriver) {
- LOG.warn("Disconnected from Mesos master.");
- }
-
- @Override
- public synchronized void slaveLost(SchedulerDriver schedulerDriver,
- SlaveID slaveID) {
- LOG.warn("Slave lost: " + slaveID.getValue());
- }
-
- @Override
- public synchronized void executorLost(SchedulerDriver schedulerDriver,
- ExecutorID executorID, SlaveID slaveID, int status) {
- LOG.warn("Executor " + executorID.getValue()
- + " lost with status " + status + " on slave " + slaveID);
- }
-
- @Override
- public synchronized void error(SchedulerDriver schedulerDriver, String s) {
- LOG.error("Error from scheduler driver: " + s);
- }
-
- /**
- * Used to track the our launched TaskTrackers.
- */
- private class MesosTracker {
- public volatile HttpHost host;
- public TaskID taskId;
- public long mapSlots;
- public long reduceSlots;
- public volatile boolean active = false; // Set once tracked by the JobTracker.
- public Timer timer;
- public volatile MesosScheduler scheduler;
-
- // Tracks Hadoop job tasks running on the tracker.
- public Set<JobID> hadoopJobs = new HashSet<JobID>();
-
- public MesosTracker(HttpHost host, TaskID taskId, long mapSlots,
- long reduceSlots, MesosScheduler scheduler) {
- this.host = host;
- this.taskId = taskId;
- this.mapSlots = mapSlots;
- this.reduceSlots = reduceSlots;
- this.scheduler = scheduler;
-
- this.timer = new Timer();
- timer.schedule(new TimerTask() {
- @Override
- public void run() {
- synchronized (MesosTracker.this.scheduler) {
- // If the tracker activated while we were awaiting to acquire the
- // lock, return.
- if (MesosTracker.this.active) return;
-
- LOG.warn("Tracker " + MesosTracker.this.host + " failed to launch within " +
- LAUNCH_TIMEOUT_MS / 1000 + " seconds, killing it");
- MesosTracker.this.scheduler.killTracker(MesosTracker.this);
- }
- }
- }, LAUNCH_TIMEOUT_MS);
- }
- }
-}