You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ya...@apache.org on 2014/09/23 08:18:02 UTC
git commit: SAMZA-308: Allow the container JVM to be customized
Repository: incubator-samza
Updated Branches:
refs/heads/master fd603c37f -> f371718c5
SAMZA-308: Allow the container JVM to be customized
Project: http://git-wip-us.apache.org/repos/asf/incubator-samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-samza/commit/f371718c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-samza/tree/f371718c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-samza/diff/f371718c
Branch: refs/heads/master
Commit: f371718c5f8b9f2bf97437f5e6cb1aa89eec8be3
Parents: fd603c3
Author: Yan Fang <ya...@gmail.com>
Authored: Mon Sep 22 23:04:22 2014 -0700
Committer: Yan Fang <ya...@gmail.com>
Committed: Mon Sep 22 23:04:22 2014 -0700
----------------------------------------------------------------------
.../versioned/jobs/configuration-table.html | 37 ++++++++++++++++----
.../samza/config/ShellCommandConfig.scala | 8 +++++
.../apache/samza/job/ShellCommandBuilder.scala | 1 +
.../org/apache/samza/config/YarnConfig.scala | 3 ++
.../org/apache/samza/job/yarn/YarnJob.scala | 7 ++--
5 files changed, 46 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f371718c/docs/learn/documentation/versioned/jobs/configuration-table.html
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html
index 069babe..3875145 100644
--- a/docs/learn/documentation/versioned/jobs/configuration-table.html
+++ b/docs/learn/documentation/versioned/jobs/configuration-table.html
@@ -311,13 +311,24 @@
<td class="description">
Any JVM options to include in the command line when executing Samza containers. For example,
this can be used to set the JVM heap size, to tune the garbage collector, or to enable
- <a href="/learn/tutorials/{{site.version}}/remote-debugging-samza.html">remote debugging</a>. Note
- there are some issues with the current implementation of <code>task.opts</code>:
- <ul>
- <li>If you set this property, the log configuration is disrupted. Please see
- <a href="https://issues.apache.org/jira/browse/SAMZA-109">SAMZA-109</a> for a workaround.</li>
- <li>This cannot be used when running with <code>ThreadJobFactory</code></li>
- </ul>
+ <a href="/learn/tutorials/{{site.version}}/remote-debugging-samza.html">remote debugging</a>.
+ This cannot be used when running with <code>ThreadJobFactory</code>. Anything you put in
+ <code>task.opts</code> gets forwarded directly to the commandline as part of the JVM invocation.
+ <dl>
+ <dt>Example: <code>task.opts=-XX:+HeapDumpOnOutOfMemoryError -XX:+UseConcMarkSweepGC</code></dt>
+ </dl>
+ </td>
+ </tr>
+
+ <tr>
+ <td class="property" id="task-java-home">task.java.home</td>
+ <td class="default"></td>
+ <td class="description">
+ The JAVA_HOME path for Samza containers. By setting this property, you can use a java version that is
+ different from your cluster's java version. Remember to set the <code>yarn.am.java.home</code> as well.
+ <dl>
+ <dt>Example: <code>task.java.home=/usr/java/jdk1.8.0_05</code></dt>
+ </dl>
</td>
</tr>
@@ -1099,6 +1110,18 @@
</tr>
<tr>
+ <td class="property" id="yarn-am-java-home">yarn.am.java.home</td>
+ <td class="default"></td>
+ <td class="description">
+ The JAVA_HOME path for Samza AM. By setting this property, you can use a java version that is
+ different from your cluster's java version. Remember to set the <code>task.java.home</code> as well.
+ <dl>
+ <dt>Example: <code>yarn.am.java.home=/usr/java/jdk1.8.0_05</code></dt>
+ </dl>
+ </td>
+ </tr>
+
+ <tr>
<td class="property" id="yarn-am-poll-interval-ms">yarn.am.poll.interval.ms</td>
<td class="default">1000</td>
<td class="description">
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f371718c/samza-core/src/main/scala/org/apache/samza/config/ShellCommandConfig.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/config/ShellCommandConfig.scala b/samza-core/src/main/scala/org/apache/samza/config/ShellCommandConfig.scala
index e4197fa..69cea2c 100644
--- a/samza-core/src/main/scala/org/apache/samza/config/ShellCommandConfig.scala
+++ b/samza-core/src/main/scala/org/apache/samza/config/ShellCommandConfig.scala
@@ -47,12 +47,18 @@ object ShellCommandConfig {
val ENV_JAVA_OPTS = "JAVA_OPTS"
/**
+ * The JAVA_HOME path for running the task
+ */
+ val ENV_JAVA_HOME = "JAVA_HOME"
+
+ /**
* Specifies whether the config for ENV_CONFIG and ENV_SYSTEM_STREAMS are compressed or not.
*/
val ENV_COMPRESS_CONFIG = "SAMZA_COMPRESS_CONFIG"
val COMMAND_SHELL_EXECUTE = "task.execute"
val TASK_JVM_OPTS = "task.opts"
+ val TASK_JAVA_HOME = "task.java.home"
val COMPRESS_ENV_CONFIG = "task.config.compress"
implicit def Config2ShellCommand(config: Config) = new ShellCommandConfig(config)
@@ -63,5 +69,7 @@ class ShellCommandConfig(config: Config) extends ScalaMapConfig(config) {
def getTaskOpts = getOption(ShellCommandConfig.TASK_JVM_OPTS)
+ def getJavaHome = getOption(ShellCommandConfig.TASK_JAVA_HOME)
+
def isEnvConfigCompressed = getBoolean(ShellCommandConfig.COMPRESS_ENV_CONFIG, false)
}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f371718c/samza-core/src/main/scala/org/apache/samza/job/ShellCommandBuilder.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/job/ShellCommandBuilder.scala b/samza-core/src/main/scala/org/apache/samza/job/ShellCommandBuilder.scala
index 8c5533c..7519117 100644
--- a/samza-core/src/main/scala/org/apache/samza/job/ShellCommandBuilder.scala
+++ b/samza-core/src/main/scala/org/apache/samza/job/ShellCommandBuilder.scala
@@ -109,6 +109,7 @@ class ShellCommandBuilder extends CommandBuilder {
ShellCommandConfig.ENV_TASK_NAME_TO_CHANGELOG_PARTITION_MAPPING -> taskNameToChangeLogPartitionMappingString,
ShellCommandConfig.ENV_CONFIG -> envConfig,
ShellCommandConfig.ENV_JAVA_OPTS -> config.getTaskOpts.getOrElse(""),
+ ShellCommandConfig.ENV_JAVA_HOME -> config.getJavaHome.getOrElse(""),
ShellCommandConfig.ENV_COMPRESS_CONFIG -> isCompressed)
}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f371718c/samza-yarn/src/main/scala/org/apache/samza/config/YarnConfig.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/config/YarnConfig.scala b/samza-yarn/src/main/scala/org/apache/samza/config/YarnConfig.scala
index 0cda102..03395e2 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/config/YarnConfig.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/config/YarnConfig.scala
@@ -31,6 +31,7 @@ object YarnConfig {
val AM_JMX_ENABLED = "yarn.am.jmx.enabled"
val AM_CONTAINER_MAX_MEMORY_MB = "yarn.am.container.memory.mb"
val AM_POLL_INTERVAL_MS = "yarn.am.poll.interval.ms"
+ val AM_JAVA_HOME = "yarn.am.java.home"
implicit def Config2Yarn(config: Config) = new YarnConfig(config)
}
@@ -55,4 +56,6 @@ class YarnConfig(config: Config) extends ScalaMapConfig(config) {
def getAMPollIntervalMs: Option[Int] = getOption(YarnConfig.AM_POLL_INTERVAL_MS).map(_.toInt)
def getJmxServerEnabled = getBoolean(YarnConfig.AM_JMX_ENABLED, true)
+
+ def getAMJavaHome = getOption(YarnConfig.AM_JAVA_HOME)
}
http://git-wip-us.apache.org/repos/asf/incubator-samza/blob/f371718c/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 34af73b..03986b1 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
@@ -47,10 +47,10 @@ object YarnJob {
*/
class YarnJob(config: Config, hadoopConfig: Configuration) extends StreamJob {
import YarnJob._
-
+
val client = new ClientHelper(hadoopConfig)
var appId: Option[ApplicationId] = None
-
+
def submit: YarnJob = {
appId = client.submitApplication(
new Path(config.getPackagePath.getOrElse(throw new SamzaException("No YARN package path defined in config."))),
@@ -62,7 +62,8 @@ class YarnJob(config: Config, hadoopConfig: Configuration) extends StreamJob {
Some(Map(
ShellCommandConfig.ENV_CONFIG -> Util.envVarEscape(JsonConfigSerializer.toJson(config)),
ShellCommandConfig.ENV_CONTAINER_NAME -> Util.envVarEscape("application-master"),
- ShellCommandConfig.ENV_JAVA_OPTS -> Util.envVarEscape(config.getAmOpts.getOrElse("")))),
+ ShellCommandConfig.ENV_JAVA_OPTS -> Util.envVarEscape(config.getAmOpts.getOrElse("")),
+ ShellCommandConfig.ENV_JAVA_HOME -> Util.envVarEscape(config.getAMJavaHome.getOrElse("")))),
Some("%s_%s" format (config.getName.get, config.getJobId.getOrElse(1))))
this