You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ni...@apache.org on 2016/07/12 18:28:22 UTC
[2/3] samza git commit: SAMZA-903: Refactor UI state variables
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java b/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
index 7778a38..1fac7f4 100644
--- a/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
+++ b/samza-yarn/src/main/java/org/apache/samza/job/yarn/YarnClusterResourceManager.java
@@ -19,6 +19,8 @@
package org.apache.samza.job.yarn;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.AMRMClient;
@@ -40,6 +42,7 @@ import org.apache.samza.util.hadoop.HttpFileSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -90,6 +93,7 @@ public class YarnClusterResourceManager extends ClusterResourceManager implement
*/
private final SamzaYarnAppMasterService service;
+ private final YarnConfig yarnConfig;
/**
* State variables to map Yarn specific callbacks into Samza specific callbacks.
@@ -126,18 +130,19 @@ public class YarnClusterResourceManager extends ClusterResourceManager implement
int nodePort = Integer.parseInt(nodePortString);
int nodeHttpPort = Integer.parseInt(nodeHttpPortString);
YarnConfig yarnConfig = new YarnConfig(config);
+ this.yarnConfig = yarnConfig;
int interval = yarnConfig.getAMPollIntervalMs();
//Instantiate the AM Client.
this.amClient = AMRMClientAsync.createAMRMClientAsync(interval, this);
- this.state = new YarnAppState(jobModelManager, -1, containerId, nodeHostString, nodePort, nodeHttpPort, samzaAppState);
+ this.state = new YarnAppState(-1, containerId, nodeHostString, nodePort, nodeHttpPort);
log.info("Initialized YarnAppState: {}", state.toString());
- this.service = new SamzaYarnAppMasterService(config, this.state, registry);
+ this.service = new SamzaYarnAppMasterService(config, samzaAppState, this.state, registry, hConfig);
log.info("ContainerID str {}, Nodehost {} , Nodeport {} , NodeHttpport {}", new Object [] {containerIdStr, nodeHostString, nodePort, nodeHttpPort});
- this.lifecycle = new SamzaYarnAppMasterLifecycle(yarnConfig.getContainerMaxMemoryMb(), yarnConfig.getContainerMaxCpuCores(), state, amClient );
+ this.lifecycle = new SamzaYarnAppMasterLifecycle(yarnConfig.getContainerMaxMemoryMb(), yarnConfig.getContainerMaxCpuCores(), samzaAppState, state, amClient );
yarnContainerRunner = new YarnContainerRunner(config, hConfig);
}
@@ -312,6 +317,33 @@ public class YarnClusterResourceManager extends ClusterResourceManager implement
amClient.stop();
log.info("Stopping the AM service " );
service.onShutdown();
+
+ if(status != SamzaApplicationState.SamzaAppStatus.UNDEFINED) {
+ cleanupStagingDir();
+ }
+ }
+
+ /**
+ * Cleans up the staging directory of the job. All exceptions during the cleanup
+ * are swallowed.
+ */
+ private void cleanupStagingDir() {
+ String yarnJobStagingDirectory = yarnConfig.getYarnJobStagingDirectory();
+ if(yarnJobStagingDirectory != null) {
+ JobContext context = new JobContext();
+ context.setAppStagingDir(new Path(yarnJobStagingDirectory));
+
+ FileSystem fs = null;
+ try {
+ fs = FileSystem.get(hConfig);
+ } catch (IOException e) {
+ log.error("Unable to clean up file system: {}", e);
+ return;
+ }
+ if(fs != null) {
+ YarnJobUtil.cleanupStagingDir(context, fs);
+ }
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/resources/scalate/WEB-INF/views/index.scaml
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/resources/scalate/WEB-INF/views/index.scaml b/samza-yarn/src/main/resources/scalate/WEB-INF/views/index.scaml
index 93660c7..93176ff 100644
--- a/samza-yarn/src/main/resources/scalate/WEB-INF/views/index.scaml
+++ b/samza-yarn/src/main/resources/scalate/WEB-INF/views/index.scaml
@@ -15,7 +15,8 @@
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
--@ val state: org.apache.samza.job.yarn.SamzaAppState
+-@ val state: org.apache.samza.job.yarn.YarnAppState
+-@ val samzaAppState: org.apache.samza.clustermanager.SamzaApplicationState
-@ val config: scala.collection.immutable.TreeMap[String, String]
-@ val rmHttpAddress: String
-@ val jobName: String = config.get("job.name").getOrElse("MISSING JOB NAME")
@@ -91,10 +92,10 @@
%a(target="_blank" href="http://#{state.nodeHost}:#{state.nodeHttpPort.toString}/node/containerlogs/#{state.amContainerId.toString}/#{username}")= state.amContainerId.toString
%tr
%td.key JMX server url
- %td= state.jmxUrl
+ %td= samzaAppState.jmxUrl
%tr
%td.key JMX server tunneling url
- %td= state.jmxTunnelingUrl
+ %td= samzaAppState.jmxTunnelingUrl
%div.tab-pane#containers
%h2 Containers
@@ -102,16 +103,16 @@
%tr
%tr
%td.key Completed
- %td= state.completedContainers.toString
+ %td= samzaAppState.completedContainers.toString
%tr
%td.key Needed
- %td= state.neededContainers.toString
+ %td= samzaAppState.neededContainers.toString
%tr
%td.key Failed
- %td= state.failedContainers.toString
+ %td= samzaAppState.failedContainers.toString
%tr
%td.key Released
- %td= state.releasedContainers.toString
+ %td= samzaAppState.releasedContainers.toString
%h2 Running Containers
%table.table.table-striped.table-bordered.tablesorter#containers-table
@@ -124,7 +125,7 @@
%th Up Time
%th JMX access
%tbody
- - for((containerId, container) <- state.runningContainers)
+ - for((containerId, container) <- state.runningYarnContainers)
%tr
%td #{containerId.toString}
%td
@@ -136,8 +137,8 @@
%td
Up time: #{container.upTimeStr()}
%td
- Ordinary: #{state.jobCoordinator.jobModel.getContainerToHostValue(containerId, org.apache.samza.coordinator.stream.messages.SetContainerHostMapping.JMX_URL_KEY)}
- Tunneling: #{state.jobCoordinator.jobModel.getContainerToHostValue(containerId, org.apache.samza.coordinator.stream.messages.SetContainerHostMapping.JMX_TUNNELING_URL_KEY)}
+ Ordinary: #{samzaAppState.jobModelManager.jobModel.getContainerToHostValue(containerId, org.apache.samza.coordinator.stream.messages.SetContainerHostMapping.JMX_URL_KEY)}
+ Tunneling: #{samzaAppState.jobModelManager.jobModel.getContainerToHostValue(containerId, org.apache.samza.coordinator.stream.messages.SetContainerHostMapping.JMX_TUNNELING_URL_KEY)}
%h2 Failed Containers
%table.table.table-striped.table-bordered.tablesorter#containers-table
@@ -162,10 +163,10 @@
%tbody
%tr
%td.key Total
- %td= state.containerCount.toString
+ %td= samzaAppState.containerCount.toString
%tr
%td.key Finished
- %td= state.finishedContainers.size.toString
+ %td= samzaAppState.finishedContainers.toString
%h3 TaskName Assignment
%table.table.table-striped.table-bordered.tablesorter#taskids-table
@@ -176,8 +177,8 @@
%th SystemStreamPartitions
%th Container
%tbody
- - for((containerId, container) <- state.runningContainers)
- - val containerModel = state.jobCoordinator.jobModel.getContainers.get(containerId)
+ - for((containerId, container) <- state.runningYarnContainers)
+ - val containerModel = samzaAppState.jobModelManager.jobModel.getContainers.get(containerId)
- for((taskName, taskModel) <- containerModel.getTasks)
%tr
%td= containerId
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMaster.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMaster.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMaster.scala
deleted file mode 100644
index 1fb18be..0000000
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMaster.scala
+++ /dev/null
@@ -1,172 +0,0 @@
-/*
- * 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.samza.job.yarn
-
-import java.io.IOException
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{Path, FileSystem}
-
-import scala.collection.JavaConversions.asScalaBuffer
-import org.apache.hadoop.yarn.api.ApplicationConstants
-import org.apache.hadoop.yarn.api.records.{FinalApplicationStatus, Container, ContainerStatus, NodeReport}
-import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest
-import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync
-import org.apache.hadoop.yarn.conf.YarnConfiguration
-import org.apache.hadoop.yarn.util.ConverterUtils
-import org.apache.samza.config.MapConfig
-import org.apache.samza.config.Config
-import org.apache.samza.config.ShellCommandConfig
-import org.apache.samza.config.JobConfig.Config2Job
-import org.apache.samza.config.YarnConfig
-import org.apache.samza.metrics.JmxServer
-import org.apache.samza.metrics.MetricsRegistryMap
-import org.apache.samza.util.hadoop.HttpFileSystem
-import org.apache.samza.util.Logging
-import org.apache.samza.serializers.model.SamzaObjectMapper
-import org.apache.samza.coordinator.JobModelManager
-import org.apache.samza.SamzaException
-
-/**
- * When YARN executes an application master, it needs a bash command to
- * execute. For Samza, YARN will execute this main method when starting Samza's
- * YARN application master.
- *
- * <br/><br/>
- *
- * The main method gets all of the environment variables (passed by Samza's
- * YARN client, and YARN itself), and wires up everything to run Samza's
- * application master.
- */
-object SamzaAppMaster extends Logging with AMRMClientAsync.CallbackHandler {
- val DEFAULT_POLL_INTERVAL_MS: Int = 1000
- var listeners: List[YarnAppMasterListener] = null
- var storedException: Throwable = null
-
- def main(args: Array[String]) {
- putMDC("containerName", "samza-application-master")
- val containerIdStr = System.getenv(ApplicationConstants.Environment.CONTAINER_ID.toString)
- info("got container id: %s" format containerIdStr)
- val containerId = ConverterUtils.toContainerId(containerIdStr)
- val applicationAttemptId = containerId.getApplicationAttemptId
- info("got app attempt id: %s" format applicationAttemptId)
- val nodeHostString = System.getenv(ApplicationConstants.Environment.NM_HOST.toString)
- info("got node manager host: %s" format nodeHostString)
- val nodePortString = System.getenv(ApplicationConstants.Environment.NM_PORT.toString)
- info("got node manager port: %s" format nodePortString)
- val nodeHttpPortString = System.getenv(ApplicationConstants.Environment.NM_HTTP_PORT.toString)
- info("got node manager http port: %s" format nodeHttpPortString)
- val coordinatorSystemConfig = new MapConfig(SamzaObjectMapper.getObjectMapper.readValue(System.getenv(ShellCommandConfig.ENV_COORDINATOR_SYSTEM_CONFIG), classOf[Config]))
- info("got coordinator system config: %s" format coordinatorSystemConfig)
- val registry = new MetricsRegistryMap
- val jobCoordinator = JobModelManager(coordinatorSystemConfig, registry)
- val config = jobCoordinator.jobModel.getConfig
- val yarnConfig = new YarnConfig(config)
- info("got config: %s" format coordinatorSystemConfig)
- putMDC("jobName", config.getName.getOrElse(throw new SamzaException("can not find the job name")))
- putMDC("jobId", config.getJobId.getOrElse("1"))
- val hConfig = new YarnConfiguration
- hConfig.set("fs.http.impl", classOf[HttpFileSystem].getName)
- val interval = yarnConfig.getAMPollIntervalMs
- val amClient = AMRMClientAsync.createAMRMClientAsync[ContainerRequest](interval, this)
- val clientHelper = new ClientHelper(hConfig)
- val containerMem = yarnConfig.getContainerMaxMemoryMb
- val containerCpu = yarnConfig.getContainerMaxCpuCores
- val jmxServer = if (yarnConfig.getJmxServerEnabled) Some(new JmxServer()) else None
- val jobContext = new JobContext
- Option(yarnConfig.getYarnJobStagingDirectory).map {
- jobStagingDirectory => jobContext.setAppStagingDir(new Path(jobStagingDirectory))
- }
-
- // wire up all of the yarn event listeners
- val state = new SamzaAppState(jobCoordinator, -1, containerId, nodeHostString, nodePortString.toInt, nodeHttpPortString.toInt)
- try {
- if (jmxServer.isDefined) {
- state.jmxUrl = jmxServer.get.getJmxUrl
- state.jmxTunnelingUrl = jmxServer.get.getTunnelingJmxUrl
- }
-
- val service = new SamzaAppMasterService(config, state, registry, clientHelper, hConfig)
- val lifecycle = new SamzaAppMasterLifecycle(containerMem, containerCpu, state, amClient)
- val metrics = new SamzaAppMasterMetrics(config, state, registry)
- val taskManager = new SamzaTaskManager(config, state, amClient, hConfig)
-
- listeners = List(service, lifecycle, metrics, taskManager)
- run(amClient, listeners, hConfig, interval)
- } finally {
- if (state.status != FinalApplicationStatus.UNDEFINED) {
- YarnJobUtil.cleanupStagingDir(jobContext, FileSystem.get(hConfig))
- }
-
- // jmxServer has to be stopped or will prevent process from exiting.
- if (jmxServer.isDefined) {
- jmxServer.get.stop
- }
- }
- }
-
- def run(amClient: AMRMClientAsync[ContainerRequest], listeners: List[YarnAppMasterListener], hConfig: YarnConfiguration, interval: Int): Unit = {
- try {
- amClient.init(hConfig)
- amClient.start
- listeners.foreach(_.onInit)
- var isShutdown: Boolean = false
- // have the loop to prevent the process from exiting until the job is to shutdown or error occurs on amClient
- while (!isShutdown && !listeners.map(_.shouldShutdown).reduceLeft(_ || _) && storedException == null) {
- try {
- Thread.sleep(interval)
- } catch {
- case e: InterruptedException => {
- isShutdown = true
- info("got interrupt in app master thread, so shutting down")
- }
- }
- }
- } finally {
- // listeners has to be stopped
- listeners.foreach(listener => try {
- listener.onShutdown
- } catch {
- case e: Exception => warn("Listener %s failed to shutdown." format listener, e)
- })
-
- // amClient has to be stopped
- amClient.stop
- }
- }
-
- override def onContainersCompleted(statuses: java.util.List[ContainerStatus]): Unit =
- statuses.foreach(containerStatus => listeners.foreach(_.onContainerCompleted(containerStatus)))
-
- override def onContainersAllocated(containers: java.util.List[Container]): Unit =
- containers.foreach(container => listeners.foreach(_.onContainerAllocated(container)))
-
- override def onShutdownRequest: Unit = listeners.foreach(_.onReboot)
-
- override def onNodesUpdated(updatedNodes: java.util.List[NodeReport]): Unit = Unit
-
- // TODO need to think about meaningful SAMZA's progress
- override def getProgress: Float = 0.0F
-
- override def onError(e: Throwable): Unit = {
- error("Error occured in amClient's callback", e)
- storedException = e
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterLifecycle.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterLifecycle.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterLifecycle.scala
deleted file mode 100644
index 2a5c0d8..0000000
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterLifecycle.scala
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.samza.job.yarn
-
-import org.apache.hadoop.yarn.api.records.FinalApplicationStatus
-import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest
-import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync
-import org.apache.samza.SamzaException
-import org.apache.samza.util.Logging
-
-/**
- * Responsible for managing the lifecycle of the application master. Mostly,
- * this means registering and unregistering with the RM, and shutting down
- * when the RM tells us to Reboot.
- */
-class SamzaAppMasterLifecycle(containerMem: Int, containerCpu: Int, state: SamzaAppState, amClient: AMRMClientAsync[ContainerRequest]) extends YarnAppMasterListener with Logging {
- var validResourceRequest = true
- var shutdownMessage: String = null
-
- override def onInit() {
- val host = state.nodeHost
- val response = amClient.registerApplicationMaster(host, state.rpcUrl.getPort, "%s:%d" format (host, state.trackingUrl.getPort))
-
- // validate that the YARN cluster can handle our container resource requirements
- val maxCapability = response.getMaximumResourceCapability
- val maxMem = maxCapability.getMemory
- val maxCpu = maxCapability.getVirtualCores
-
- info("Got AM register response. The YARN RM supports container requests with max-mem: %s, max-cpu: %s" format (maxMem, maxCpu))
-
- if (containerMem > maxMem || containerCpu > maxCpu) {
- shutdownMessage = "The YARN cluster is unable to run your job due to unsatisfiable resource requirements. You asked for mem: %s, and cpu: %s." format (containerMem, containerCpu)
- error(shutdownMessage)
- validResourceRequest = false
- state.status = FinalApplicationStatus.FAILED
- state.jobHealthy.set(false)
- }
- }
-
- override def onReboot() {
- throw new SamzaException("Received a reboot signal from the RM, so throwing an exception to reboot the AM.")
- }
-
- override def onShutdown() {
- info("Shutting down.")
- amClient.unregisterApplicationMaster(state.status, shutdownMessage, null)
- }
-
- override def shouldShutdown = !validResourceRequest
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala
deleted file mode 100644
index 054d8b6..0000000
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterMetrics.scala
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * 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.samza.job.yarn
-import org.apache.samza.metrics.MetricsRegistryMap
-import org.apache.samza.metrics.JvmMetrics
-import org.apache.samza.config.Config
-import org.apache.samza.task.TaskContext
-import org.apache.samza.Partition
-import org.apache.samza.metrics.MetricsRegistry
-import org.apache.samza.config.StreamConfig.Config2Stream
-import org.apache.samza.config.MetricsConfig.Config2Metrics
-import org.apache.samza.metrics.MetricsReporterFactory
-import org.apache.samza.util.Util
-import org.apache.samza.metrics.ReadableMetricsRegistry
-import org.apache.samza.util.Logging
-import org.apache.samza.SamzaException
-import java.util.Timer
-import java.util.TimerTask
-import org.apache.samza.metrics.MetricsHelper
-
-object SamzaAppMasterMetrics {
- val sourceName = "ApplicationMaster"
-}
-
-/**
- * Responsible for wiring up Samza's metrics. Given that Samza has a metric
- * registry, we might as well use it. This class takes Samza's application
- * master state, and converts it to metrics.
- */
-class SamzaAppMasterMetrics(
- val config: Config,
- val state: SamzaAppState,
- val registry: ReadableMetricsRegistry) extends MetricsHelper with YarnAppMasterListener with Logging {
-
- val jvm = new JvmMetrics(registry)
- val reporters = config.getMetricReporterNames.map(reporterName => {
- val metricsFactoryClassName = config
- .getMetricsFactoryClass(reporterName)
- .getOrElse(throw new SamzaException("Metrics reporter %s missing .class config" format reporterName))
-
- val reporter =
- Util
- .getObj[MetricsReporterFactory](metricsFactoryClassName)
- .getMetricsReporter(reporterName, SamzaAppMasterMetrics.sourceName, config)
-
- reporter.register(SamzaAppMasterMetrics.sourceName, registry)
- (reporterName, reporter)
- }).toMap
-
- override def onInit() {
- val mRunningContainers = newGauge("running-containers", () => state.runningContainers.size)
- val mNeededContainers = newGauge("needed-containers", () => state.neededContainers.get())
- val mCompletedContainers = newGauge("completed-containers", () => state.completedContainers.get())
- val mFailedContainers = newGauge("failed-containers", () => state.failedContainers.get())
- val mReleasedContainers = newGauge("released-containers", () => state.releasedContainers.get())
- val mContainers = newGauge("container-count", () => state.containerCount)
- val mHost = newGauge("http-host", () => state.nodeHost)
- val mTrackingPort = newGauge("http-port", () => state.trackingUrl.getPort)
- val mRpcPort = newGauge("rpc-port", () => state.rpcUrl.getPort)
- val mAppAttemptId = newGauge("app-attempt-id", () => state.appAttemptId.toString)
- val mJobHealthy = newGauge("job-healthy", () => if (state.jobHealthy.get()) 1 else 0)
- val mLocalityMatchedRequests = newGauge(
- "locality-matched",
- () => {
- if (state.containerRequests.get() != 0) {
- state.matchedContainerRequests.get() / state.containerRequests.get()
- } else {
- 0L
- }
- })
-
- jvm.start
- reporters.values.foreach(_.start)
- }
-
- override def onShutdown() {
- reporters.values.foreach(_.stop)
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterService.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterService.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterService.scala
deleted file mode 100644
index 979d81d..0000000
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaAppMasterService.scala
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * 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.samza.job.yarn
-
-import org.apache.hadoop.security.UserGroupInformation
-import org.apache.hadoop.yarn.conf.YarnConfiguration
-import org.apache.samza.coordinator.stream.CoordinatorStreamWriter
-import org.apache.samza.coordinator.stream.messages.SetConfig
-import org.apache.samza.util.Logging
-import org.apache.samza.config.Config
-import org.apache.samza.metrics.ReadableMetricsRegistry
-import org.apache.samza.SamzaException
-import org.apache.samza.coordinator.server.HttpServer
-import org.apache.samza.coordinator.server.JobServlet
-import org.apache.samza.webapp.ApplicationMasterRestServlet
-import org.apache.samza.webapp.ApplicationMasterWebServlet
-
-/**
- * Samza's application master runs a very basic HTTP/JSON service to allow
- * dashboards to check on the status of a job. SamzaAppMasterService starts
- * up the web service when initialized.
- * <p />
- * Besides the HTTP/JSON service endpoints, it also starts an optional
- * SecurityManager which takes care of the security needs when running in
- * a secure environment.
- */
-class SamzaAppMasterService(config: Config, state: SamzaAppState, registry: ReadableMetricsRegistry, clientHelper: ClientHelper, yarnConfiguration: YarnConfiguration) extends YarnAppMasterListener with Logging {
- var rpcApp: HttpServer = null
- var webApp: HttpServer = null
- val SERVER_URL_OPT: String = "samza.autoscaling.server.url"
- var securityManager: Option[SamzaAppMasterSecurityManager] = None
-
- override def onInit() {
- // try starting the samza AM dashboard at a random rpc and tracking port
- info("Starting webapp at a random rpc and tracking port")
-
- rpcApp = new HttpServer(resourceBasePath = "scalate")
- rpcApp.addServlet("/*", new ApplicationMasterRestServlet(config, state, registry))
- rpcApp.start
-
- webApp = new HttpServer(resourceBasePath = "scalate")
- webApp.addServlet("/*", new ApplicationMasterWebServlet(config, state))
- webApp.start
-
- state.jobCoordinator.start
- state.rpcUrl = rpcApp.getUrl
- state.trackingUrl = webApp.getUrl
- state.coordinatorUrl = state.jobCoordinator.server.getUrl
-
- //write server url to coordinator stream
- val coordinatorStreamWriter: CoordinatorStreamWriter = new CoordinatorStreamWriter(config)
- coordinatorStreamWriter.start()
- coordinatorStreamWriter.sendMessage(SetConfig.TYPE, SERVER_URL_OPT, state.coordinatorUrl.toString)
- coordinatorStreamWriter.stop()
- debug("sent server url message with value: %s " format state.coordinatorUrl.toString)
-
- info("Webapp is started at (rpc %s, tracking %s, coordinator %s)" format(state.rpcUrl, state.trackingUrl, state.coordinatorUrl))
-
- // start YarnSecurityManger for a secure cluster
- if (UserGroupInformation.isSecurityEnabled) {
- securityManager = Option {
- val securityManager = new SamzaAppMasterSecurityManager(config, yarnConfiguration)
- securityManager.start
- securityManager
- }
- }
- }
-
- override def onShutdown() {
- if (rpcApp != null) {
- rpcApp.stop
- }
-
- if (webApp != null) {
- webApp.stop
- }
-
- state.jobCoordinator.stop
-
- securityManager.map {
- securityManager => securityManager.stop
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterLifecycle.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterLifecycle.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterLifecycle.scala
index 2ed9baf..c9c1e18 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterLifecycle.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterLifecycle.scala
@@ -33,7 +33,7 @@ import org.apache.samza.util.Logging
* when the RM tells us to Reboot.
*/
//This class is used in the refactored code path as called by run-jc.sh
-class SamzaYarnAppMasterLifecycle(containerMem: Int, containerCpu: Int, state: YarnAppState, amClient: AMRMClientAsync[ContainerRequest]) extends Logging {
+class SamzaYarnAppMasterLifecycle(containerMem: Int, containerCpu: Int, samzaAppState: SamzaApplicationState, state: YarnAppState, amClient: AMRMClientAsync[ContainerRequest]) extends Logging {
var validResourceRequest = true
var shutdownMessage: String = null
var webApp: SamzaYarnAppMasterService = null
@@ -51,8 +51,8 @@ class SamzaYarnAppMasterLifecycle(containerMem: Int, containerCpu: Int, state: Y
shutdownMessage = "The YARN cluster is unable to run your job due to unsatisfiable resource requirements. You asked for mem: %s, and cpu: %s." format (containerMem, containerCpu)
error(shutdownMessage)
validResourceRequest = false
- state.samzaAppState.status = SamzaAppStatus.FAILED;
- state.samzaAppState.jobHealthy.set(false)
+ samzaAppState.status = SamzaAppStatus.FAILED;
+ samzaAppState.jobHealthy.set(false)
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterService.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterService.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterService.scala
index f62bec1..5f2bfc5 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterService.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/SamzaYarnAppMasterService.scala
@@ -19,42 +19,47 @@
package org.apache.samza.job.yarn
+import org.apache.hadoop.security.UserGroupInformation
+import org.apache.hadoop.yarn.conf.YarnConfiguration
+import org.apache.samza.clustermanager.SamzaApplicationState
import org.apache.samza.config.Config
import org.apache.samza.coordinator.server.HttpServer
import org.apache.samza.coordinator.stream.CoordinatorStreamWriter
import org.apache.samza.coordinator.stream.messages.SetConfig
import org.apache.samza.metrics.ReadableMetricsRegistry
import org.apache.samza.util.Logging
+import org.apache.samza.webapp.{ApplicationMasterWebServlet, ApplicationMasterRestServlet}
/**
- * Samza's application master runs a very basic HTTP/JSON service to allow
- * dashboards to check on the status of a job. SamzaAppMasterService starts
- * up the web service when initialized.
- */
+ * Samza's application master runs a very basic HTTP/JSON service to allow
+ * dashboards to check on the status of a job. SamzaAppMasterService starts
+ * up the web service when initialized.
+ */
//This class is used in the refactored code path as called by run-jc.sh
-class SamzaYarnAppMasterService(config: Config, state: YarnAppState, registry: ReadableMetricsRegistry) extends Logging {
+class SamzaYarnAppMasterService(config: Config, samzaAppState: SamzaApplicationState, state: YarnAppState, registry: ReadableMetricsRegistry, yarnConfiguration: YarnConfiguration) extends Logging {
var rpcApp: HttpServer = null
var webApp: HttpServer = null
val SERVER_URL_OPT: String = "samza.autoscaling.server.url"
+ var securityManager: Option[SamzaAppMasterSecurityManager] = None
- def onInit() {
+ def onInit() {
// try starting the samza AM dashboard at a random rpc and tracking port
info("Starting webapp at a random rpc and tracking port")
rpcApp = new HttpServer(resourceBasePath = "scalate")
- //TODO: Since the state has changed into Samza specific and Yarn specific states, this UI has to be refactored too.
- //rpcApp.addServlet("/*", refactor ApplicationMasterRestServlet(config, state, registry))
+
+ rpcApp.addServlet("/*", new ApplicationMasterRestServlet(config, samzaAppState, state, registry))
rpcApp.start
webApp = new HttpServer(resourceBasePath = "scalate")
- //webApp.addServlet("/*", refactor ApplicationMasterWebServlet(config, state))
+ webApp.addServlet("/*", new ApplicationMasterWebServlet(config, samzaAppState, state))
webApp.start
- state.jobModelManager.start
+ samzaAppState.jobModelManager.start
state.rpcUrl = rpcApp.getUrl
state.trackingUrl = webApp.getUrl
- state.coordinatorUrl = state.jobModelManager.server.getUrl
+ state.coordinatorUrl = samzaAppState.jobModelManager.server.getUrl
//write server url to coordinator stream
val coordinatorStreamWriter: CoordinatorStreamWriter = new CoordinatorStreamWriter(config)
@@ -64,9 +69,19 @@ class SamzaYarnAppMasterService(config: Config, state: YarnAppState, registry: R
debug("Sent server url message with value: %s " format state.coordinatorUrl.toString)
info("Webapp is started at (rpc %s, tracking %s, coordinator %s)" format(state.rpcUrl, state.trackingUrl, state.coordinatorUrl))
+
+ // start YarnSecurityManger for a secure cluster
+ if (UserGroupInformation.isSecurityEnabled) {
+ securityManager = Option {
+ val securityManager = new SamzaAppMasterSecurityManager(config, yarnConfiguration)
+ securityManager.start
+ securityManager
+ }
+ }
+
}
- def onShutdown() {
+ def onShutdown() {
if (rpcApp != null) {
rpcApp.stop
}
@@ -75,6 +90,11 @@ class SamzaYarnAppMasterService(config: Config, state: YarnAppState, registry: R
webApp.stop
}
- state.jobModelManager.stop
+ samzaAppState.jobModelManager.stop
+
+ securityManager.map {
+ securityManager => securityManager.stop
+ }
+
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala
index 3ca1f0d..46dc4d1 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/job/yarn/YarnJob.scala
@@ -86,11 +86,11 @@ class YarnJob(config: Config, hadoopConfig: Configuration) extends StreamJob {
}
logger.info("Inside YarnJob: fwk_path is %s, ver is %s use it directly " format(fwkPath, fwkVersion))
- var cmdExec = "./__package/bin/run-am.sh" // default location
+ var cmdExec = "./__package/bin/run-jc.sh" // default location
if (!fwkPath.isEmpty()) {
// if we have framework installed as a separate package - use it
- cmdExec = fwkPath + "/" + fwkVersion + "/bin/run-am.sh"
+ cmdExec = fwkPath + "/" + fwkVersion + "/bin/run-jc.sh"
logger.info("Using FWK path: " + "export SAMZA_LOG_DIR=%s && ln -sfn %s logs && exec %s 1>logs/%s 2>logs/%s".
format(ApplicationConstants.LOG_DIR_EXPANSION_VAR, ApplicationConstants.LOG_DIR_EXPANSION_VAR, cmdExec,
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala b/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala
index a40ab72..cdd389c 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala
@@ -19,17 +19,18 @@
package org.apache.samza.webapp
+import org.apache.samza.clustermanager.SamzaApplicationState
import org.scalatra._
import scalate.ScalateSupport
import org.apache.samza.config.Config
-import org.apache.samza.job.yarn.{SamzaAppState, ClientHelper}
+import org.apache.samza.job.yarn.{YarnAppState, ClientHelper}
import org.apache.samza.metrics._
import scala.collection.JavaConversions._
import org.apache.hadoop.yarn.conf.YarnConfiguration
import java.util.HashMap
import org.apache.samza.serializers.model.SamzaObjectMapper
-class ApplicationMasterRestServlet(config: Config, state: SamzaAppState, registry: ReadableMetricsRegistry) extends ScalatraServlet with ScalateSupport {
+class ApplicationMasterRestServlet(config: Config, samzaAppState: SamzaApplicationState, state: YarnAppState, registry: ReadableMetricsRegistry) extends ScalatraServlet with ScalateSupport {
val yarnConfig = new YarnConfiguration
val client = new ClientHelper(yarnConfig)
val jsonMapper = SamzaObjectMapper.getObjectMapper
@@ -78,11 +79,11 @@ class ApplicationMasterRestServlet(config: Config, state: SamzaAppState, registr
get("/am") {
val containers = new HashMap[String, HashMap[String, Object]]
- state.runningContainers.foreach {
+ state.runningYarnContainers.foreach {
case (containerId, container) =>
val yarnContainerId = container.id.toString
val containerMap = new HashMap[String, Object]
- val taskModels = state.jobCoordinator.jobModel.getContainers.get(containerId).getTasks
+ val taskModels = samzaAppState.jobModelManager.jobModel.getContainers.get(containerId).getTasks
containerMap.put("yarn-address", container.nodeHttpAddress)
containerMap.put("start-time", container.startTime.toString)
containerMap.put("up-time", container.upTime.toString)
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala b/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala
index 605332a..a32cd65 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala
@@ -19,16 +19,17 @@
package org.apache.samza.webapp
+import org.apache.samza.clustermanager.SamzaApplicationState
import org.scalatra._
import scalate.ScalateSupport
-import org.apache.samza.job.yarn.{SamzaAppState}
+import org.apache.samza.job.yarn.YarnAppState
import org.apache.samza.config.Config
import scala.collection.JavaConversions._
import scala.collection.immutable.TreeMap
import org.apache.hadoop.yarn.conf.YarnConfiguration
import org.apache.hadoop.yarn.webapp.util.WebAppUtils
-class ApplicationMasterWebServlet(config: Config, state: SamzaAppState) extends ScalatraServlet with ScalateSupport {
+class ApplicationMasterWebServlet(config: Config, samzaAppState: SamzaApplicationState, state: YarnAppState) extends ScalatraServlet with ScalateSupport {
val yarnConfig = new YarnConfiguration
before() {
@@ -39,6 +40,7 @@ class ApplicationMasterWebServlet(config: Config, state: SamzaAppState) extends
layoutTemplate("/WEB-INF/views/index.scaml",
"config" -> TreeMap(config.sanitize.toMap.toArray: _*),
"state" -> state,
+ "samzaAppState" -> samzaAppState,
"rmHttpAddress" -> WebAppUtils.getRMWebAppURLWithScheme(yarnConfig))
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerAllocator.java
----------------------------------------------------------------------
diff --git a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerAllocator.java b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerAllocator.java
deleted file mode 100644
index e21aded..0000000
--- a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerAllocator.java
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * 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.samza.job.yarn;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.hadoop.yarn.client.api.AMRMClient;
-import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
-import org.apache.hadoop.yarn.util.ConverterUtils;
-import org.apache.samza.config.Config;
-import org.apache.samza.config.MapConfig;
-import org.apache.samza.job.yarn.util.MockContainerRequestState;
-import org.apache.samza.job.yarn.util.TestUtil;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-public class TestContainerAllocator extends TestContainerAllocatorCommon {
-
- private final Config config = new MapConfig(new HashMap<String, String>() {
- {
- put("yarn.container.count", "1");
- put("systems.test-system.samza.factory", "org.apache.samza.job.yarn.MockSystemFactory");
- put("yarn.container.memory.mb", "512");
- put("yarn.package.path", "/foo");
- put("task.inputs", "test-system.test-stream");
- put("systems.test-system.samza.key.serde", "org.apache.samza.serializers.JsonSerde");
- put("systems.test-system.samza.msg.serde", "org.apache.samza.serializers.JsonSerde");
- put("yarn.container.retry.count", "1");
- put("yarn.container.retry.window.ms", "1999999999");
- put("yarn.container.request.timeout.ms", "3");
- put("yarn.allocator.sleep.ms", "1");
- }
- });
-
- @Override
- protected Config getConfig() {
- return config;
- }
-
- @Override
- protected MockContainerRequestState createContainerRequestState(
- AMRMClientAsync<AMRMClient.ContainerRequest> amClient) {
- return new MockContainerRequestState(amClient, false);
- }
-
- /**
- * Test request containers with no containerToHostMapping makes the right number of requests
- */
- @Test
- public void testRequestContainersWithNoMapping() throws Exception {
- int containerCount = 4;
- Map<Integer, String> containersToHostMapping = new HashMap<Integer, String>();
- for (int i = 0; i < containerCount; i++) {
- containersToHostMapping.put(i, null);
- }
- allocatorThread.start();
-
- containerAllocator.requestContainers(containersToHostMapping);
-
- assertNotNull(requestState);
-
- assertNotNull(requestState.getRequestsQueue());
- assertTrue(requestState.getRequestsQueue().size() == 4);
-
- // If host-affinty is not enabled, it doesn't update the requestMap
- assertNotNull(requestState.getRequestsToCountMap());
- assertTrue(requestState.getRequestsToCountMap().keySet().size() == 0);
- }
- /**
- * Adds all containers returned to ANY_HOST only
- */
- @Test
- public void testAddContainer() throws Exception {
- assertNull(requestState.getContainersOnAHost("host1"));
- assertNull(requestState.getContainersOnAHost(ANY_HOST));
-
- containerAllocator.addContainer(TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000002"), "host1", 123));
- containerAllocator.addContainer(TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000003"), "host3", 123));
-
- assertNull(requestState.getContainersOnAHost("host1"));
- assertNotNull(requestState.getContainersOnAHost(ANY_HOST));
- assertTrue(requestState.getContainersOnAHost(ANY_HOST).size() == 2);
- }
-
- /**
- * Test requestContainers
- */
- @Test
- public void testRequestContainers() throws Exception {
- Map<Integer, String> containersToHostMapping = new HashMap<Integer, String>() {
- {
- put(0, "host1");
- put(1, "host2");
- put(2, null);
- put(3, "host1");
- }
- };
-
- allocatorThread.start();
-
- containerAllocator.requestContainers(containersToHostMapping);
-
- assertNotNull(testAMRMClient.requests);
- assertEquals(4, testAMRMClient.requests.size());
-
- assertNotNull(requestState);
-
- assertNotNull(requestState.getRequestsQueue());
- assertTrue(requestState.getRequestsQueue().size() == 4);
-
- // If host-affinty is not enabled, it doesn't update the requestMap
- assertNotNull(requestState.getRequestsToCountMap());
- assertTrue(requestState.getRequestsToCountMap().keySet().size() == 0);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerAllocatorCommon.java
----------------------------------------------------------------------
diff --git a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerAllocatorCommon.java b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerAllocatorCommon.java
deleted file mode 100644
index 0bbd48d..0000000
--- a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerAllocatorCommon.java
+++ /dev/null
@@ -1,225 +0,0 @@
-/*
- * 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.samza.job.yarn;
-
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.hadoop.yarn.client.api.AMRMClient;
-import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
-import org.apache.hadoop.yarn.client.api.async.impl.AMRMClientAsyncImpl;
-import org.apache.hadoop.yarn.util.ConverterUtils;
-import org.apache.samza.config.Config;
-import org.apache.samza.config.YarnConfig;
-import org.apache.samza.container.TaskName;
-import org.apache.samza.coordinator.JobModelManager;
-import org.apache.samza.coordinator.server.HttpServer;
-import org.apache.samza.job.model.ContainerModel;
-import org.apache.samza.job.model.JobModel;
-import org.apache.samza.job.model.TaskModel;
-import org.apache.samza.job.yarn.util.MockContainerListener;
-import org.apache.samza.job.yarn.util.MockContainerRequestState;
-import org.apache.samza.job.yarn.util.MockContainerUtil;
-import org.apache.samza.job.yarn.util.MockHttpServer;
-import org.apache.samza.job.yarn.util.TestAMRMClientImpl;
-import org.apache.samza.job.yarn.util.TestUtil;
-import org.eclipse.jetty.servlet.DefaultServlet;
-import org.eclipse.jetty.servlet.ServletHolder;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-
-/**
- * Handles all common fields/tests for ContainerAllocators.
- */
-public abstract class TestContainerAllocatorCommon {
- protected static final String ANY_HOST = ContainerRequestState.ANY_HOST;
-
- protected final HttpServer server = new MockHttpServer("/", 7777, null, new ServletHolder(DefaultServlet.class));
-
- protected AMRMClientAsyncImpl amRmClientAsync;
- protected TestAMRMClientImpl testAMRMClient;
- protected MockContainerRequestState requestState;
- protected AbstractContainerAllocator containerAllocator;
- protected Thread allocatorThread;
- protected ContainerUtil containerUtil;
-
- protected SamzaAppState state = new SamzaAppState(getCoordinator(1), -1, ConverterUtils.toContainerId("container_1350670447861_0003_01_000001"), "", 1, 2);
-
- protected abstract Config getConfig();
- protected abstract MockContainerRequestState createContainerRequestState(AMRMClientAsync<AMRMClient.ContainerRequest> amClient);
-
- private JobModelManager getCoordinator(int containerCount) {
- Map<Integer, ContainerModel> containers = new java.util.HashMap<>();
- for (int i = 0; i < containerCount; i++) {
- ContainerModel container = new ContainerModel(i, new HashMap<TaskName, TaskModel>());
- containers.put(i, container);
- }
- JobModel jobModel = new JobModel(getConfig(), containers);
- return new JobModelManager(jobModel, server, null);
- }
-
-
- @Before
- public void setup() throws Exception {
- // Create AMRMClient
- testAMRMClient = new TestAMRMClientImpl(
- TestUtil.getAppMasterResponse(
- false,
- new ArrayList<Container>(),
- new ArrayList<ContainerStatus>()
- ));
- amRmClientAsync = TestUtil.getAMClient(testAMRMClient);
-
- // Initialize certain state variables
- state.coordinatorUrl = new URL("http://localhost:7778/");
-
- containerUtil = TestUtil.getContainerUtil(getConfig(), state);
-
- requestState = createContainerRequestState(amRmClientAsync);
- containerAllocator = new HostAwareContainerAllocator(
- amRmClientAsync,
- containerUtil,
- new YarnConfig(getConfig())
- );
- Field requestStateField = containerAllocator.getClass().getSuperclass().getDeclaredField("containerRequestState");
- requestStateField.setAccessible(true);
- requestStateField.set(containerAllocator, requestState);
-
- allocatorThread = new Thread(containerAllocator);
- }
-
- @After
- public void teardown() throws Exception {
- containerAllocator.setIsRunning(false);
- allocatorThread.join();
- }
-
- /**
- * If the container fails to start e.g because it fails to connect to a NM on a host that
- * is down, the allocator should request a new container on a different host.
- */
- @Test
- public void testRerequestOnAnyHostIfContainerStartFails() throws Exception {
- final Container container = TestUtil
- .getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000001"), "host2", 123);
- final Container container1 = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000002"), "host1", 123);
-
- ((MockContainerUtil) containerUtil).containerStartException = new IOException("You shall not... connect to the NM!");
-
- Runnable releasedContainerAssertions = new Runnable() {
- @Override
- public void run() {
- // The failed container should be released. The successful one should not.
- assertNotNull(testAMRMClient.getRelease());
- assertEquals(1, testAMRMClient.getRelease().size());
- assertTrue(testAMRMClient.getRelease().contains(container.getId()));
- }
- };
-
- Runnable assignedContainerAssertions = new Runnable() {
- @Override
- public void run() {
- // Test that the first request assignment had a preferred host and the retry didn't
- assertEquals(2, requestState.assignedRequests.size());
-
- SamzaContainerRequest request = requestState.assignedRequests.remove();
- assertEquals(0, request.expectedContainerId);
- assertEquals("host2", request.getPreferredHost());
-
- request = requestState.assignedRequests.remove();
- assertEquals(0, request.expectedContainerId);
- assertEquals("ANY_HOST", request.getPreferredHost());
-
- // This routine should be called after the retry is assigned, but before it's started.
- // So there should still be 1 container needed because neededContainers should not be decremented for a failed start.
- assertEquals(1, state.neededContainers.get());
- }
- };
-
- // Set up our final asserts before starting the allocator thread
- MockContainerListener listener = new MockContainerListener(2, 1, 2, 0, null, releasedContainerAssertions, assignedContainerAssertions, null);
- requestState.registerContainerListener(listener);
- state.neededContainers.set(1); // Normally this would be done in the SamzaTaskManager
-
- // Only request 1 container and we should see 2 assignments in the assertions above (because of the retry)
- containerAllocator.requestContainer(0, "host2");
- containerAllocator.addContainer(container);
- containerAllocator.addContainer(container1);
-
- allocatorThread.start();
-
- listener.verify();
- }
-
-
- /**
- * Extra allocated containers that are returned by the RM and unused by the AM should be released.
- * Containers are considered "extra" only when there are no more pending requests to fulfill
- * @throws Exception
- */
- @Test
- public void testAllocatorReleasesExtraContainers() throws Exception {
- final Container container = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000001"), "host1", 123);
- final Container container1 = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000002"), "host1", 123);
- final Container container2 = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000003"), "host2", 123);
-
- Runnable releasedContainerAssertions = new Runnable() {
- @Override
- public void run() {
- assertNotNull(testAMRMClient.getRelease());
- assertEquals(2, testAMRMClient.getRelease().size());
- assertTrue(testAMRMClient.getRelease().contains(container1.getId()));
- assertTrue(testAMRMClient.getRelease().contains(container2.getId()));
-
- // Test that state is cleaned up
- assertEquals(0, requestState.getRequestsQueue().size());
- assertEquals(0, requestState.getRequestsToCountMap().size());
- assertNull(requestState.getContainersOnAHost("host1"));
- assertNull(requestState.getContainersOnAHost("host2"));
- }
- };
-
- // Set up our final asserts before starting the allocator thread
- MockContainerListener listener = new MockContainerListener(3, 2, 0, 0, null, releasedContainerAssertions, null, null);
- requestState.registerContainerListener(listener);
-
- containerAllocator.requestContainer(0, "host1");
-
- containerAllocator.addContainer(container);
- containerAllocator.addContainer(container1);
- containerAllocator.addContainer(container2);
-
- allocatorThread.start();
-
- listener.verify();
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerRequestState.java
----------------------------------------------------------------------
diff --git a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerRequestState.java b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerRequestState.java
deleted file mode 100644
index 402fe78..0000000
--- a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestContainerRequestState.java
+++ /dev/null
@@ -1,221 +0,0 @@
-/*
- * 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.samza.job.yarn;
-
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.hadoop.yarn.client.api.AMRMClient;
-import org.apache.hadoop.yarn.client.api.async.impl.AMRMClientAsyncImpl;
-import org.apache.hadoop.yarn.util.ConverterUtils;
-import org.apache.samza.job.yarn.util.TestAMRMClientImpl;
-import org.apache.samza.job.yarn.util.TestUtil;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.util.ArrayList;
-
-import static org.junit.Assert.*;
-
-public class TestContainerRequestState {
- private AMRMClientAsyncImpl amRmClientAsync;
- private TestAMRMClientImpl testAMRMClient;
- private static final String ANY_HOST = ContainerRequestState.ANY_HOST;
-
- @Before
- public void setup() {
- // Create AMRMClient
- testAMRMClient = new TestAMRMClientImpl(
- TestUtil.getAppMasterResponse(
- false,
- new ArrayList<Container>(),
- new ArrayList<ContainerStatus>()
- ));
- amRmClientAsync = TestUtil.getAMClient(testAMRMClient);
- }
-
- /**
- * Test state after a request is submitted
- */
- @Test
- public void testUpdateRequestState() {
- // Host-affinity is enabled
- ContainerRequestState state = new ContainerRequestState(amRmClientAsync, true);
- SamzaContainerRequest request = new SamzaContainerRequest(0, "abc");
- state.updateRequestState(request);
-
- assertNotNull(testAMRMClient.requests);
- assertEquals(1, testAMRMClient.requests.size());
- assertEquals(request.getIssuedRequest(), testAMRMClient.requests.get(0));
-
- assertNotNull(state.getRequestsQueue());
- assertTrue(state.getRequestsQueue().size() == 1);
-
- assertNotNull(state.getRequestsToCountMap());
- assertNotNull(state.getRequestsToCountMap().get("abc"));
- assertEquals(1, state.getRequestsToCountMap().get("abc").get());
-
- assertNotNull(state.getContainersOnAHost("abc"));
- assertEquals(0, state.getContainersOnAHost("abc").size());
-
- // Host-affinity is not enabled
- ContainerRequestState state1 = new ContainerRequestState(amRmClientAsync, false);
- SamzaContainerRequest request1 = new SamzaContainerRequest(1, null);
- state1.updateRequestState(request1);
-
- assertNotNull(testAMRMClient.requests);
- assertEquals(2, testAMRMClient.requests.size());
-
- AMRMClient.ContainerRequest expectedContainerRequest = request1.getIssuedRequest();
- AMRMClient.ContainerRequest actualContainerRequest = testAMRMClient.requests.get(1);
-
- assertEquals(expectedContainerRequest.getCapability(), actualContainerRequest.getCapability());
- assertEquals(expectedContainerRequest.getPriority(), actualContainerRequest.getPriority());
- assertNull(actualContainerRequest.getNodes());
-
- assertNotNull(state1.getRequestsQueue());
- assertTrue(state1.getRequestsQueue().size() == 1);
-
- assertNotNull(state1.getRequestsToCountMap());
- assertNull(state1.getRequestsToCountMap().get(ANY_HOST));
-
- }
-
- /**
- * Test addContainer() updates the state correctly
- */
- @Test
- public void testAddContainer() {
- // Add container to ANY_LIST when host-affinity is not enabled
- ContainerRequestState state = new ContainerRequestState(amRmClientAsync, false);
- Container container = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000002"), "abc", 123);
- state.addContainer(container);
-
- assertNotNull(state.getRequestsQueue());
- assertNotNull(state.getRequestsToCountMap());
- assertNotNull(state.getContainersOnAHost(ANY_HOST));
-
- assertEquals(1, state.getContainersOnAHost(ANY_HOST).size());
- assertEquals(container, state.getContainersOnAHost(ANY_HOST).get(0));
-
- // Container Allocated when there is no request in queue
- ContainerRequestState state1 = new ContainerRequestState(amRmClientAsync, true);
- Container container1 = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000003"), "zzz", 123);
- state1.addContainer(container1);
-
- assertNotNull(state1.getRequestsQueue());
- assertEquals(0, state1.getRequestsQueue().size());
-
- assertNull(state1.getContainersOnAHost("zzz"));
- assertNotNull(state1.getContainersOnAHost(ANY_HOST));
- assertEquals(1, state1.getContainersOnAHost(ANY_HOST).size());
- assertEquals(container1, state1.getContainersOnAHost(ANY_HOST).get(0));
-
- // Container Allocated on a Requested Host
- state1.updateRequestState(new SamzaContainerRequest(0, "abc"));
-
- assertNotNull(state1.getRequestsQueue());
- assertEquals(1, state1.getRequestsQueue().size());
-
- assertNotNull(state1.getRequestsToCountMap());
- assertNotNull(state1.getRequestsToCountMap().get("abc"));
- assertEquals(1, state1.getRequestsToCountMap().get("abc").get());
-
- state1.addContainer(container);
-
- assertNotNull(state1.getContainersOnAHost("abc"));
- assertEquals(1, state1.getContainersOnAHost("abc").size());
- assertEquals(container, state1.getContainersOnAHost("abc").get(0));
-
- // Container Allocated on host that was not requested
- Container container2 = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000004"), "xyz", 123);
- state1.addContainer(container2);
-
- assertNull(state1.getContainersOnAHost("xyz"));
- assertNotNull(state1.getContainersOnAHost(ANY_HOST));
- assertEquals(2, state1.getContainersOnAHost(ANY_HOST).size());
- assertEquals(container2, state1.getContainersOnAHost(ANY_HOST).get(1));
-
- // Extra containers were allocated on a host that was requested
- Container container3 = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000005"), "abc", 123);
- state1.addContainer(container3);
-
- assertEquals(3, state1.getContainersOnAHost(ANY_HOST).size());
- assertEquals(container3, state1.getContainersOnAHost(ANY_HOST).get(2));
- }
-
- /**
- * Test request state after container is assigned to a host
- * * Assigned on requested host
- * * Assigned on any host
- */
- @Test
- public void testContainerAssignment() throws Exception {
- // Host-affinity enabled
- ContainerRequestState state = new ContainerRequestState(amRmClientAsync, true);
- SamzaContainerRequest request = new SamzaContainerRequest(0, "abc");
- SamzaContainerRequest request1 = new SamzaContainerRequest(0, "def");
-
- state.updateRequestState(request);
- state.updateRequestState(request1);
-
- Container container = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000001"), "abc", 123);
- Container container1 = TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000001"), "zzz", 123);
- state.addContainer(container);
- state.addContainer(container1);
-
- assertEquals(2, state.getRequestsQueue().size());
- assertEquals(2, state.getRequestsToCountMap().size());
-
- assertNotNull(state.getContainersOnAHost("abc"));
- assertEquals(1, state.getContainersOnAHost("abc").size());
- assertEquals(container, state.getContainersOnAHost("abc").get(0));
-
- assertNotNull(state.getContainersOnAHost("def"));
- assertEquals(0, state.getContainersOnAHost("def").size());
-
- assertNotNull(state.getContainersOnAHost(ANY_HOST));
- assertEquals(1, state.getContainersOnAHost(ANY_HOST).size());
- assertEquals(container1, state.getContainersOnAHost(ANY_HOST).get(0));
-
- // Container assigned on the requested host
- state.updateStateAfterAssignment(request, "abc", container);
-
- assertEquals(1, state.getRequestsQueue().size());
- assertEquals(request1, state.getRequestsQueue().peek());
-
- assertNotNull(state.getRequestsToCountMap().get("abc"));
- assertEquals(0, state.getRequestsToCountMap().get("abc").get());
-
- assertNotNull(state.getContainersOnAHost("abc"));
- assertEquals(0, state.getContainersOnAHost("abc").size());
-
- // Container assigned on any host
- state.updateStateAfterAssignment(request1, ANY_HOST, container1);
-
- assertEquals(0, state.getRequestsQueue().size());
-
- assertNotNull(state.getRequestsToCountMap().get("def"));
- assertEquals(0, state.getRequestsToCountMap().get("def").get());
-
- assertNotNull(state.getContainersOnAHost(ANY_HOST));
- assertEquals(0, state.getContainersOnAHost(ANY_HOST).size());
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestHostAwareContainerAllocator.java
----------------------------------------------------------------------
diff --git a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestHostAwareContainerAllocator.java b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestHostAwareContainerAllocator.java
deleted file mode 100644
index ead7200..0000000
--- a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestHostAwareContainerAllocator.java
+++ /dev/null
@@ -1,253 +0,0 @@
-/*
- * 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.samza.job.yarn;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.client.api.AMRMClient;
-import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
-import org.apache.hadoop.yarn.util.ConverterUtils;
-import org.apache.samza.config.Config;
-import org.apache.samza.config.MapConfig;
-import org.apache.samza.job.yarn.util.MockContainerListener;
-import org.apache.samza.job.yarn.util.MockContainerRequestState;
-import org.apache.samza.job.yarn.util.MockContainerUtil;
-import org.apache.samza.job.yarn.util.TestUtil;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-public class TestHostAwareContainerAllocator extends TestContainerAllocatorCommon {
-
- private final Config config = new MapConfig(new HashMap<String, String>() {
- {
- put("yarn.container.count", "1");
- put("systems.test-system.samza.factory", "org.apache.samza.job.yarn.MockSystemFactory");
- put("yarn.container.memory.mb", "512");
- put("yarn.package.path", "/foo");
- put("task.inputs", "test-system.test-stream");
- put("systems.test-system.samza.key.serde", "org.apache.samza.serializers.JsonSerde");
- put("systems.test-system.samza.msg.serde", "org.apache.samza.serializers.JsonSerde");
- put("yarn.container.retry.count", "1");
- put("yarn.container.retry.window.ms", "1999999999");
- put("yarn.samza.host-affinity.enabled", "true");
- put("yarn.container.request.timeout.ms", "3");
- put("yarn.allocator.sleep.ms", "1");
- }
- });
-
- @Override
- protected Config getConfig() {
- return config;
- }
-
- @Override
- protected MockContainerRequestState createContainerRequestState(AMRMClientAsync<AMRMClient.ContainerRequest> amClient) {
- return new MockContainerRequestState(amClient, true);
- }
-
- /**
- * Test request containers with no containerToHostMapping makes the right number of requests
- */
- @Test
- public void testRequestContainersWithNoMapping() throws Exception {
- int containerCount = 4;
- Map<Integer, String> containersToHostMapping = new HashMap<Integer, String>();
- for (int i = 0; i < containerCount; i++) {
- containersToHostMapping.put(i, null);
- }
-
- allocatorThread.start();
-
- containerAllocator.requestContainers(containersToHostMapping);
-
- assertNotNull(requestState);
-
- assertNotNull(requestState.getRequestsQueue());
- assertEquals(4, requestState.getRequestsQueue().size());
-
- assertNotNull(requestState.getRequestsToCountMap());
- assertEquals(1, requestState.getRequestsToCountMap().keySet().size());
- assertTrue(requestState.getRequestsToCountMap().keySet().contains(ANY_HOST));
- }
-
- /**
- * Add containers to the correct host in the request state
- */
- @Test
- public void testAddContainerWithHostAffinity() throws Exception {
- containerAllocator.requestContainers(new HashMap<Integer, String>() {
- {
- put(0, "host1");
- put(1, "host3");
- }
- });
-
- assertNotNull(requestState.getContainersOnAHost("host1"));
- assertEquals(0, requestState.getContainersOnAHost("host1").size());
-
- assertNotNull(requestState.getContainersOnAHost("host3"));
- assertEquals(0, requestState.getContainersOnAHost("host3").size());
-
- assertNull(requestState.getContainersOnAHost(ANY_HOST));
-
- containerAllocator.addContainer(TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000002"), "host1", 123));
- containerAllocator.addContainer(TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000004"), "host2", 123));
- containerAllocator.addContainer(TestUtil.getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000003"), "host3", 123));
-
- assertNotNull(requestState.getContainersOnAHost("host1"));
- assertEquals(1, requestState.getContainersOnAHost("host1").size());
-
- assertNotNull(requestState.getContainersOnAHost("host3"));
- assertEquals(1, requestState.getContainersOnAHost("host3").size());
-
- assertNotNull(requestState.getContainersOnAHost(ANY_HOST));
- assertTrue(requestState.getContainersOnAHost(ANY_HOST).size() == 1);
- assertEquals(ConverterUtils.toContainerId("container_1350670447861_0003_01_000004"),
- requestState.getContainersOnAHost(ANY_HOST).get(0).getId());
- }
-
-
- @Test
- public void testRequestContainers() throws Exception {
- Map<Integer, String> containersToHostMapping = new HashMap<Integer, String>() {
- {
- put(0, "host1");
- put(1, "host2");
- put(2, null);
- put(3, "host1");
- }
- };
- allocatorThread.start();
-
- containerAllocator.requestContainers(containersToHostMapping);
-
- assertNotNull(testAMRMClient.requests);
- assertEquals(4, testAMRMClient.requests.size());
-
- assertNotNull(requestState);
-
- assertNotNull(requestState.getRequestsQueue());
- assertTrue(requestState.getRequestsQueue().size() == 4);
-
- assertNotNull(requestState.getRequestsToCountMap());
- Map<String, AtomicInteger> requestsMap = requestState.getRequestsToCountMap();
-
- assertNotNull(requestsMap.get("host1"));
- assertEquals(2, requestsMap.get("host1").get());
-
- assertNotNull(requestsMap.get("host2"));
- assertEquals(1, requestsMap.get("host2").get());
-
- assertNotNull(requestsMap.get(ANY_HOST));
- assertEquals(1, requestsMap.get(ANY_HOST).get());
- }
-
- /**
- * Handles expired requests correctly and assigns ANY_HOST
- */
- @Test
- public void testExpiredRequestHandling() throws Exception {
- Map<Integer, String> containersToHostMapping = new HashMap<Integer, String>() {
- {
- put(0, "requestedHost1");
- put(1, "requestedHost2");
- }
- };
- containerAllocator.requestContainers(containersToHostMapping);
-
- assertNotNull(requestState.getRequestsQueue());
- assertTrue(requestState.getRequestsQueue().size() == 2);
-
- assertNotNull(requestState.getRequestsToCountMap());
- assertNotNull(requestState.getRequestsToCountMap().get("requestedHost1"));
- assertTrue(requestState.getRequestsToCountMap().get("requestedHost1").get() == 1);
-
- assertNotNull(requestState.getRequestsToCountMap().get("requestedHost2"));
- assertTrue(requestState.getRequestsToCountMap().get("requestedHost2").get() == 1);
-
- final Container container0 = TestUtil
- .getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000002"), "availableHost1", 123);
- final Container container1 = TestUtil
- .getContainer(ConverterUtils.toContainerId("container_1350670447861_0003_01_000003"), "availableHost2", 123);
-
- Runnable addedContainerAssertions = new Runnable() {
- @Override
- public void run() {
- assertNotNull(requestState.getRequestsToCountMap());
- assertNull(requestState.getContainersOnAHost("availableHost1"));
- assertNull(requestState.getContainersOnAHost("availableHost2"));
- assertNotNull(requestState.getContainersOnAHost(ANY_HOST));
- assertTrue(requestState.getContainersOnAHost(ANY_HOST).size() == 2);
- }
- };
-
- Runnable assignedContainerAssertions = new Runnable() {
- @Override
- public void run() {
- List<Container> anyHostContainers = requestState.getContainersOnAHost(ANY_HOST);
- assertTrue(anyHostContainers == null || anyHostContainers.isEmpty());
-
- assertNotNull(requestState.getRequestsQueue());
- assertTrue(requestState.getRequestsQueue().size() == 0);
-
- assertNotNull(requestState.getRequestsToCountMap());
- assertNotNull(requestState.getRequestsToCountMap().get("requestedHost1"));
- assertNotNull(requestState.getRequestsToCountMap().get("requestedHost2"));
- }
- };
-
- Runnable runningContainerAssertions = new Runnable() {
- @Override
- public void run() {
- MockContainerUtil mockContainerUtil = (MockContainerUtil) containerUtil;
-
- assertNotNull(mockContainerUtil.runningContainerList.get("availableHost1"));
- assertTrue(mockContainerUtil.runningContainerList.get("availableHost1").contains(container0));
-
- assertNotNull(mockContainerUtil.runningContainerList.get("availableHost2"));
- assertTrue(mockContainerUtil.runningContainerList.get("availableHost2").contains(container1));
- }
- };
- // Set up our final asserts before starting the allocator thread
- MockContainerListener listener = new MockContainerListener(
- 2, 0, 2, 2,
- addedContainerAssertions,
- null,
- assignedContainerAssertions,
- runningContainerAssertions);
- requestState.registerContainerListener(listener);
- ((MockContainerUtil) containerUtil).registerContainerListener(listener);
-
- containerAllocator.addContainer(container0);
- containerAllocator.addContainer(container1);
-
- // Start after adding containers to avoid a race condition between the allocator thread
- // using the containers and the assertions after the containers are added.
- allocatorThread.start();
-
- listener.verify();
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/9396ee5c/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestSamzaContainerRequest.java
----------------------------------------------------------------------
diff --git a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestSamzaContainerRequest.java b/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestSamzaContainerRequest.java
deleted file mode 100644
index ad0f4d3..0000000
--- a/samza-yarn/src/test/java/org/apache/samza/job/yarn/TestSamzaContainerRequest.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.samza.job.yarn;
-
-import org.junit.Test;
-
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-public class TestSamzaContainerRequest {
- private static final String ANY_HOST = ContainerRequestState.ANY_HOST;
-
- @Test
- public void testPreferredHostIsNeverNull() {
- SamzaContainerRequest request = new SamzaContainerRequest(0, null);
-
- assertNotNull(request.getPreferredHost());
-
- // preferredHost is null, it should automatically default to ANY_HOST
- assertTrue(request.getPreferredHost().equals(ANY_HOST));
-
- SamzaContainerRequest request1 = new SamzaContainerRequest(1, "abc");
- assertNotNull(request1.getPreferredHost());
- assertTrue(request1.getPreferredHost().equals("abc"));
- }
-
- @Test
- public void testAnyHostIsNotPassedToYarnRequest() {
- SamzaContainerRequest request = new SamzaContainerRequest(0, null);
- assertNull(request.getIssuedRequest().getNodes());
-
- SamzaContainerRequest request1 = new SamzaContainerRequest(1, ANY_HOST);
- assertNull(request1.getIssuedRequest().getNodes());
- }
-}