{"action":"created","comment":{"url":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/comments/73197157","id":73197157,"diff_hunk":"@@ -62,47 +86,142 @@ public static void main(final String[] args) throws Exception {\n       proxyUser.doAs(new PrivilegedExceptionAction<Void>() {\n         @Override\n         public Void run() throws Exception {\n-          runSpark(args, new Configuration());\n+          runSpark(args);\n           return null;\n         }\n       });\n     } else {\n-      runSpark(args, new Configuration());\n+      runSpark(args);\n     }   \n   }\n \n   /**\n-   * Actually calling the spark-submit command\n+   * Actually adjusts cmd args based on execution environment and calls the spark-submit command\n    * \n    * @param args\n    */\n-  private static void runSpark(String[] args, Configuration conf) {\n+  private static void runSpark(String[] args) {\n \n     if (args.length == 0) {\n       throw new RuntimeException(\"SparkSubmit cannot run with zero args\");\n     }\n \n+    // Arg String passed to here are long strings delimited by SparkJobArg.delimiter\n     // munge everything together and repartition based by our ^Z character, instead of by the\n     // default \"space\" character\n     StringBuilder concat = new StringBuilder();\n     concat.append(args[0]);\n     for (int i = 1; i < args.length; i++) {\n       concat.append(\" \" + args[i]);\n     }\n+    String[] newArgs = concat.toString().split(SparkJobArg.delimiter);\n+\n+    // Sample: [--driver-java-options, , --master, yarn-cluster, --class, myclass,\n+    // --conf, queue=default, --executor-memory, 1g, --num-executors, 15, my.jar, myparams]\n+    logger.info(\"Args before adjusting driver java opts: \" + Arrays.toString(newArgs));\n+\n+    // Adjust driver java opts param\n+    handleDriverJavaOpts(newArgs);\n+\n+    // If dynamic allocation policy for this jobtype is turned on, adjust related param\n+    handleDynamicResourceAllocation(newArgs);\n+\n+    // If yarn cluster enables node labeling, adjust related param\n+    handleNodeLabeling(newArgs);\n \n-    final String[] newArgs = concat.toString().split(SparkJobArg.delimiter);\n-    logger.info(\"newArgs: \" + Arrays.toString(newArgs));\n+    // Realign params after adjustment\n+    newArgs = removeNullsFromArgArray(newArgs);\n+    logger.info(\"Args after adjusting driver java opts: \" + Arrays.toString(newArgs));\n+\n+    org.apache.spark.deploy.SparkSubmit$.MODULE$.main(newArgs);\n+  }\n+\n+  private static void handleDriverJavaOpts(String[] argArray) {\n+    Configuration conf = new Configuration();\n+    // Driver java opts is always the first elem(param name) and second elem(value) in the argArray\n+    // Get current driver java opts here\n+    StringBuilder driverJavaOptions = new StringBuilder(argArray[1]);\n+    // In spark-submit, when both --driver-java-options and conf spark.driver.extraJavaOptions is used,\n+    // spark-submit will only pick --driver-java-options, an arg we always have\n+    // So if user gives --conf spark.driver.extraJavaOptions=XX, we append the value in --driver-java-options\n+    for (int i = 0; i < argArray.length; i++) {\n+      if (argArray[i].equals(SparkJobArg.SPARK_CONF_PREFIX.sparkParamName) \n+        && argArray[i+1].startsWith(SPARK_CONF_EXTRA_DRIVER_OPTIONS)) {\n+        driverJavaOptions.append(argArray[++i].substring(SPARK_CONF_EXTRA_DRIVER_OPTIONS.length() + 1));\n+      }\n+    }\n     \n-    StringBuilder driverJavaOptions = new StringBuilder(newArgs[1]);    \n+    // Append addtional driver java opts about azkaban context\n     String[] requiredJavaOpts = { WORKFLOW_LINK, JOB_LINK, EXECUTION_LINK, ATTEMPT_LINK };\n     for (int i = 0; i < requiredJavaOpts.length; i++) {\n-    \t  driverJavaOptions.append(\" \").append(HadoopJobUtils.javaOptStringFromHadoopConfiguration(conf,\n-    \t            requiredJavaOpts[i]));\n+        driverJavaOptions.append(\" \").append(HadoopJobUtils.javaOptStringFromHadoopConfiguration(conf,\n+                  requiredJavaOpts[i]));\n     }\n-    newArgs[1] = driverJavaOptions.toString();\n-    logger.info(\"newArgs2: \" + Arrays.toString(newArgs));\n+    // Update driver java opts\n+    argArray[1] = driverJavaOptions.toString();\n+  }\n \n-    org.apache.spark.deploy.SparkSubmit$.MODULE$.main(newArgs);\n+  private static void handleDynamicResourceAllocation(String[] argArray) {\n+    // HadoopSparkJob will set env var on this process if we enable dynamic allocation policy for spark jobtype.\n+    // This policy can be enabled in spark jobtype plugin's conf property.\n+    // This policy is designed to enforce dynamic allocation.\n+    String dynamicAllocProp = System.getenv(HadoopSparkJob.SPARK_DYNAMIC_RES_ENV_VAR);\n+    boolean dynamicAllocEnabled = dynamicAllocProp != null && dynamicAllocProp.equals(Boolean.TRUE.toString());\n+    if (dynamicAllocEnabled) {\n+      for (int i = 0; i < argArray.length; i++) {\n+        if (argArray[i] == null) continue;\n+\n+        // If user specifies num of executors, or if user tries to disable dynamic allocation for his application\n+        // by setting some conf params to falase, we need to ignore these settings to enforce the application \n+        // uses dynamic allocation.\n+        if (argArray[i].equals(SparkJobArg.NUM_EXECUTORS.sparkParamName) // --num-executors\n+          || (\n+            argArray[i].equals(SparkJobArg.SPARK_CONF_PREFIX.sparkParamName) // --conf\n+            && (argArray[i+1].startsWith(SPARK_CONF_SHUFFLE_SERVICE_ENABLED) // spark.shuffle.service.enabled=\n+              || argArray[i+1].startsWith(SPARK_CONF_DYNAMIC_ALLOC_ENABLED)) // spark.dynamicAllocation.enabled","path":"plugins/jobtype/src/azkaban/jobtype/HadoopSecureSparkWrapper.java","position":169,"original_position":169,"commit_id":"2c700266a1ea2db43e69eb5007746757a21f8f1a","original_commit_id":"2c700266a1ea2db43e69eb5007746757a21f8f1a","user":{"login":"fangshil","id":2416746,"avatar_url":"https://avatars.githubusercontent.com/u/2416746?v=3","gravatar_id":"","url":"https://api.github.com/users/fangshil","html_url":"https://github.com/fangshil","followers_url":"https://api.github.com/users/fangshil/followers","following_url":"https://api.github.com/users/fangshil/following{/other_user}","gists_url":"https://api.github.com/users/fangshil/gists{/gist_id}","starred_url":"https://api.github.com/users/fangshil/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fangshil/subscriptions","organizations_url":"https://api.github.com/users/fangshil/orgs","repos_url":"https://api.github.com/users/fangshil/repos","events_url":"https://api.github.com/users/fangshil/events{/privacy}","received_events_url":"https://api.github.com/users/fangshil/received_events","type":"User","site_admin":false},"body":"Added","created_at":"2016-08-02T17:13:54Z","updated_at":"2016-08-02T17:13:54Z","html_url":"https://github.com/azkaban/azkaban-plugins/pull/231#discussion_r73197157","pull_request_url":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/231","_links":{"self":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/comments/73197157"},"html":{"href":"https://github.com/azkaban/azkaban-plugins/pull/231#discussion_r73197157"},"pull_request":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/231"}}},"pull_request":{"url":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/231","id":79245456,"html_url":"https://github.com/azkaban/azkaban-plugins/pull/231","diff_url":"https://github.com/azkaban/azkaban-plugins/pull/231.diff","patch_url":"https://github.com/azkaban/azkaban-plugins/pull/231.patch","issue_url":"https://api.github.com/repos/azkaban/azkaban-plugins/issues/231","number":231,"state":"open","locked":false,"title":"update spark jobtype plugin to support multiple spark version","user":{"login":"fangshil","id":2416746,"avatar_url":"https://avatars.githubusercontent.com/u/2416746?v=3","gravatar_id":"","url":"https://api.github.com/users/fangshil","html_url":"https://github.com/fangshil","followers_url":"https://api.github.com/users/fangshil/followers","following_url":"https://api.github.com/users/fangshil/following{/other_user}","gists_url":"https://api.github.com/users/fangshil/gists{/gist_id}","starred_url":"https://api.github.com/users/fangshil/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fangshil/subscriptions","organizations_url":"https://api.github.com/users/fangshil/orgs","repos_url":"https://api.github.com/users/fangshil/repos","events_url":"https://api.github.com/users/fangshil/events{/privacy}","received_events_url":"https://api.github.com/users/fangshil/received_events","type":"User","site_admin":false},"body":"","created_at":"2016-07-28T18:32:25Z","updated_at":"2016-08-02T17:13:54Z","closed_at":null,"merged_at":null,"merge_commit_sha":"11d2a9ef1a99f7c210e83460415abc812c910d87","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/231/commits","review_comments_url":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/231/comments","review_comment_url":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/comments{/number}","comments_url":"https://api.github.com/repos/azkaban/azkaban-plugins/issues/231/comments","statuses_url":"https://api.github.com/repos/azkaban/azkaban-plugins/statuses/2c700266a1ea2db43e69eb5007746757a21f8f1a","head":{"label":"fangshil:master","ref":"master","sha":"2c700266a1ea2db43e69eb5007746757a21f8f1a","user":{"login":"fangshil","id":2416746,"avatar_url":"https://avatars.githubusercontent.com/u/2416746?v=3","gravatar_id":"","url":"https://api.github.com/users/fangshil","html_url":"https://github.com/fangshil","followers_url":"https://api.github.com/users/fangshil/followers","following_url":"https://api.github.com/users/fangshil/following{/other_user}","gists_url":"https://api.github.com/users/fangshil/gists{/gist_id}","starred_url":"https://api.github.com/users/fangshil/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fangshil/subscriptions","organizations_url":"https://api.github.com/users/fangshil/orgs","repos_url":"https://api.github.com/users/fangshil/repos","events_url":"https://api.github.com/users/fangshil/events{/privacy}","received_events_url":"https://api.github.com/users/fangshil/received_events","type":"User","site_admin":false},"repo":{"id":64418597,"name":"azkaban-plugins","full_name":"fangshil/azkaban-plugins","owner":{"login":"fangshil","id":2416746,"avatar_url":"https://avatars.githubusercontent.com/u/2416746?v=3","gravatar_id":"","url":"https://api.github.com/users/fangshil","html_url":"https://github.com/fangshil","followers_url":"https://api.github.com/users/fangshil/followers","following_url":"https://api.github.com/users/fangshil/following{/other_user}","gists_url":"https://api.github.com/users/fangshil/gists{/gist_id}","starred_url":"https://api.github.com/users/fangshil/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fangshil/subscriptions","organizations_url":"https://api.github.com/users/fangshil/orgs","repos_url":"https://api.github.com/users/fangshil/repos","events_url":"https://api.github.com/users/fangshil/events{/privacy}","received_events_url":"https://api.github.com/users/fangshil/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/fangshil/azkaban-plugins","description":"Plugins for Azkaban.","fork":true,"url":"https://api.github.com/repos/fangshil/azkaban-plugins","forks_url":"https://api.github.com/repos/fangshil/azkaban-plugins/forks","keys_url":"https://api.github.com/repos/fangshil/azkaban-plugins/keys{/key_id}","collaborators_url":"https://api.github.com/repos/fangshil/azkaban-plugins/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/fangshil/azkaban-plugins/teams","hooks_url":"https://api.github.com/repos/fangshil/azkaban-plugins/hooks","issue_events_url":"https://api.github.com/repos/fangshil/azkaban-plugins/issues/events{/number}","events_url":"https://api.github.com/repos/fangshil/azkaban-plugins/events","assignees_url":"https://api.github.com/repos/fangshil/azkaban-plugins/assignees{/user}","branches_url":"https://api.github.com/repos/fangshil/azkaban-plugins/branches{/branch}","tags_url":"https://api.github.com/repos/fangshil/azkaban-plugins/tags","blobs_url":"https://api.github.com/repos/fangshil/azkaban-plugins/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/fangshil/azkaban-plugins/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/fangshil/azkaban-plugins/git/refs{/sha}","trees_url":"https://api.github.com/repos/fangshil/azkaban-plugins/git/trees{/sha}","statuses_url":"https://api.github.com/repos/fangshil/azkaban-plugins/statuses/{sha}","languages_url":"https://api.github.com/repos/fangshil/azkaban-plugins/languages","stargazers_url":"https://api.github.com/repos/fangshil/azkaban-plugins/stargazers","contributors_url":"https://api.github.com/repos/fangshil/azkaban-plugins/contributors","subscribers_url":"https://api.github.com/repos/fangshil/azkaban-plugins/subscribers","subscription_url":"https://api.github.com/repos/fangshil/azkaban-plugins/subscription","commits_url":"https://api.github.com/repos/fangshil/azkaban-plugins/commits{/sha}","git_commits_url":"https://api.github.com/repos/fangshil/azkaban-plugins/git/commits{/sha}","comments_url":"https://api.github.com/repos/fangshil/azkaban-plugins/comments{/number}","issue_comment_url":"https://api.github.com/repos/fangshil/azkaban-plugins/issues/comments{/number}","contents_url":"https://api.github.com/repos/fangshil/azkaban-plugins/contents/{+path}","compare_url":"https://api.github.com/repos/fangshil/azkaban-plugins/compare/{base}...{head}","merges_url":"https://api.github.com/repos/fangshil/azkaban-plugins/merges","archive_url":"https://api.github.com/repos/fangshil/azkaban-plugins/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/fangshil/azkaban-plugins/downloads","issues_url":"https://api.github.com/repos/fangshil/azkaban-plugins/issues{/number}","pulls_url":"https://api.github.com/repos/fangshil/azkaban-plugins/pulls{/number}","milestones_url":"https://api.github.com/repos/fangshil/azkaban-plugins/milestones{/number}","notifications_url":"https://api.github.com/repos/fangshil/azkaban-plugins/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/fangshil/azkaban-plugins/labels{/name}","releases_url":"https://api.github.com/repos/fangshil/azkaban-plugins/releases{/id}","deployments_url":"https://api.github.com/repos/fangshil/azkaban-plugins/deployments","created_at":"2016-07-28T18:25:27Z","updated_at":"2016-07-28T18:25:31Z","pushed_at":"2016-07-30T10:01:11Z","git_url":"git://github.com/fangshil/azkaban-plugins.git","ssh_url":"git@github.com:fangshil/azkaban-plugins.git","clone_url":"https://github.com/fangshil/azkaban-plugins.git","svn_url":"https://github.com/fangshil/azkaban-plugins","homepage":"https://azkaban.github.io","size":306721,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"azkaban:master","ref":"master","sha":"7dff4babdddf1f99af3ccc6f19bd01d30a2290c4","user":{"login":"azkaban","id":213618,"avatar_url":"https://avatars.githubusercontent.com/u/213618?v=3","gravatar_id":"","url":"https://api.github.com/users/azkaban","html_url":"https://github.com/azkaban","followers_url":"https://api.github.com/users/azkaban/followers","following_url":"https://api.github.com/users/azkaban/following{/other_user}","gists_url":"https://api.github.com/users/azkaban/gists{/gist_id}","starred_url":"https://api.github.com/users/azkaban/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/azkaban/subscriptions","organizations_url":"https://api.github.com/users/azkaban/orgs","repos_url":"https://api.github.com/users/azkaban/repos","events_url":"https://api.github.com/users/azkaban/events{/privacy}","received_events_url":"https://api.github.com/users/azkaban/received_events","type":"Organization","site_admin":false},"repo":{"id":7566859,"name":"azkaban-plugins","full_name":"azkaban/azkaban-plugins","owner":{"login":"azkaban","id":213618,"avatar_url":"https://avatars.githubusercontent.com/u/213618?v=3","gravatar_id":"","url":"https://api.github.com/users/azkaban","html_url":"https://github.com/azkaban","followers_url":"https://api.github.com/users/azkaban/followers","following_url":"https://api.github.com/users/azkaban/following{/other_user}","gists_url":"https://api.github.com/users/azkaban/gists{/gist_id}","starred_url":"https://api.github.com/users/azkaban/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/azkaban/subscriptions","organizations_url":"https://api.github.com/users/azkaban/orgs","repos_url":"https://api.github.com/users/azkaban/repos","events_url":"https://api.github.com/users/azkaban/events{/privacy}","received_events_url":"https://api.github.com/users/azkaban/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/azkaban/azkaban-plugins","description":"Plugins for Azkaban.","fork":false,"url":"https://api.github.com/repos/azkaban/azkaban-plugins","forks_url":"https://api.github.com/repos/azkaban/azkaban-plugins/forks","keys_url":"https://api.github.com/repos/azkaban/azkaban-plugins/keys{/key_id}","collaborators_url":"https://api.github.com/repos/azkaban/azkaban-plugins/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/azkaban/azkaban-plugins/teams","hooks_url":"https://api.github.com/repos/azkaban/azkaban-plugins/hooks","issue_events_url":"https://api.github.com/repos/azkaban/azkaban-plugins/issues/events{/number}","events_url":"https://api.github.com/repos/azkaban/azkaban-plugins/events","assignees_url":"https://api.github.com/repos/azkaban/azkaban-plugins/assignees{/user}","branches_url":"https://api.github.com/repos/azkaban/azkaban-plugins/branches{/branch}","tags_url":"https://api.github.com/repos/azkaban/azkaban-plugins/tags","blobs_url":"https://api.github.com/repos/azkaban/azkaban-plugins/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/azkaban/azkaban-plugins/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/azkaban/azkaban-plugins/git/refs{/sha}","trees_url":"https://api.github.com/repos/azkaban/azkaban-plugins/git/trees{/sha}","statuses_url":"https://api.github.com/repos/azkaban/azkaban-plugins/statuses/{sha}","languages_url":"https://api.github.com/repos/azkaban/azkaban-plugins/languages","stargazers_url":"https://api.github.com/repos/azkaban/azkaban-plugins/stargazers","contributors_url":"https://api.github.com/repos/azkaban/azkaban-plugins/contributors","subscribers_url":"https://api.github.com/repos/azkaban/azkaban-plugins/subscribers","subscription_url":"https://api.github.com/repos/azkaban/azkaban-plugins/subscription","commits_url":"https://api.github.com/repos/azkaban/azkaban-plugins/commits{/sha}","git_commits_url":"https://api.github.com/repos/azkaban/azkaban-plugins/git/commits{/sha}","comments_url":"https://api.github.com/repos/azkaban/azkaban-plugins/comments{/number}","issue_comment_url":"https://api.github.com/repos/azkaban/azkaban-plugins/issues/comments{/number}","contents_url":"https://api.github.com/repos/azkaban/azkaban-plugins/contents/{+path}","compare_url":"https://api.github.com/repos/azkaban/azkaban-plugins/compare/{base}...{head}","merges_url":"https://api.github.com/repos/azkaban/azkaban-plugins/merges","archive_url":"https://api.github.com/repos/azkaban/azkaban-plugins/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/azkaban/azkaban-plugins/downloads","issues_url":"https://api.github.com/repos/azkaban/azkaban-plugins/issues{/number}","pulls_url":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls{/number}","milestones_url":"https://api.github.com/repos/azkaban/azkaban-plugins/milestones{/number}","notifications_url":"https://api.github.com/repos/azkaban/azkaban-plugins/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/azkaban/azkaban-plugins/labels{/name}","releases_url":"https://api.github.com/repos/azkaban/azkaban-plugins/releases{/id}","deployments_url":"https://api.github.com/repos/azkaban/azkaban-plugins/deployments","created_at":"2013-01-11T20:53:36Z","updated_at":"2016-06-15T12:57:47Z","pushed_at":"2016-07-30T10:01:11Z","git_url":"git://github.com/azkaban/azkaban-plugins.git","ssh_url":"git@github.com:azkaban/azkaban-plugins.git","clone_url":"https://github.com/azkaban/azkaban-plugins.git","svn_url":"https://github.com/azkaban/azkaban-plugins","homepage":"https://azkaban.github.io","size":366788,"stargazers_count":47,"watchers_count":47,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":91,"mirror_url":null,"open_issues_count":28,"forks":91,"open_issues":28,"watchers":47,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/231"},"html":{"href":"https://github.com/azkaban/azkaban-plugins/pull/231"},"issue":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/issues/231"},"comments":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/issues/231/comments"},"review_comments":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/231/comments"},"review_comment":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/pulls/231/commits"},"statuses":{"href":"https://api.github.com/repos/azkaban/azkaban-plugins/statuses/2c700266a1ea2db43e69eb5007746757a21f8f1a"}}}}
{"actor":{"display_login":"fangshil"}}