@Test public void testAction() throws Exception { ExecutionCommand executionCommand = new ExecutionCommand(); Map<String, String> commandParams = new HashMap<>(); HostRoleCommand hrc = EasyMock.createMock(HostRoleCommand.class); SparkShufflePropertyConfig action = new SparkShufflePropertyConfig(); commandParams.put("clusterName", "c1"); executionCommand.setCommandParams(commandParams); executionCommand.setClusterName("c1"); expect(hrc.getRequestId()).andReturn(1L).anyTimes(); expect(hrc.getStageId()).andReturn(2L).anyTimes(); expect(hrc.getExecutionCommandWrapper()).andReturn(new ExecutionCommandWrapper(executionCommand)).anyTimes(); replay(cluster, hrc); clusterField.set(action, m_clusters); action.setExecutionCommand(executionCommand); action.setHostRoleCommand(hrc); CommandReport report = action.execute(null); assertNotNull(report); Cluster c = m_clusters.getCluster("c1"); Config config = c.getDesiredConfigByType("yarn-site"); Map<String, String> map = config.getProperties(); assertTrue(map.containsKey("yarn.nodemanager.aux-services")); assertTrue(map.containsKey("yarn.nodemanager.aux-services.spark_shuffle.class")); assertEquals("some_service,spark_shuffle", map.get("yarn.nodemanager.aux-services")); assertEquals("org.apache.spark.network.yarn.YarnShuffleService", map.get("yarn.nodemanager.aux-services.spark_shuffle.class")); }
@Override public CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext) throws AmbariException, InterruptedException { String clusterName = getExecutionCommand().getClusterName(); Cluster cluster = clusters.getCluster(clusterName); Config yarnSiteConfig = cluster.getDesiredConfigByType(YARN_SITE_CONFIG_TYPE); if (yarnSiteConfig == null) { return createCommandReport(0, HostRoleStatus.FAILED,"{}", String.format("Source type %s not found", YARN_SITE_CONFIG_TYPE), ""); } Map<String, String> yarnSiteProperties = yarnSiteConfig.getProperties(); final List<String> auxSevices; final String oldAuxServices = yarnSiteProperties.get(YARN_NODEMANAGER_AUX_SERVICES); final String newAuxServices; if (yarnSiteProperties.containsKey(YARN_NODEMANAGER_AUX_SERVICES)) { auxSevices = new ArrayList<>(Arrays.asList(oldAuxServices.split(",", -1))); } else { auxSevices = new ArrayList<>(); } if (!auxSevices.contains(SPARK_SHUFFLE_AUX_STR)) { auxSevices.add(SPARK_SHUFFLE_AUX_STR); } newAuxServices = StringUtils.join(auxSevices, ","); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES, newAuxServices); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE); yarnSiteConfig.setProperties(yarnSiteProperties); yarnSiteConfig.save(); return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", String.format("%s was set from %s to %s. %s was set to %s", YARN_NODEMANAGER_AUX_SERVICES, oldAuxServices, newAuxServices, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE), ""); }
SparkShufflePropertyConfig extends AbstractServerAction { @Override public CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext) throws AmbariException, InterruptedException { String clusterName = getExecutionCommand().getClusterName(); Cluster cluster = clusters.getCluster(clusterName); Config yarnSiteConfig = cluster.getDesiredConfigByType(YARN_SITE_CONFIG_TYPE); if (yarnSiteConfig == null) { return createCommandReport(0, HostRoleStatus.FAILED,"{}", String.format("Source type %s not found", YARN_SITE_CONFIG_TYPE), ""); } Map<String, String> yarnSiteProperties = yarnSiteConfig.getProperties(); final List<String> auxSevices; final String oldAuxServices = yarnSiteProperties.get(YARN_NODEMANAGER_AUX_SERVICES); final String newAuxServices; if (yarnSiteProperties.containsKey(YARN_NODEMANAGER_AUX_SERVICES)) { auxSevices = new ArrayList<>(Arrays.asList(oldAuxServices.split(",", -1))); } else { auxSevices = new ArrayList<>(); } if (!auxSevices.contains(SPARK_SHUFFLE_AUX_STR)) { auxSevices.add(SPARK_SHUFFLE_AUX_STR); } newAuxServices = StringUtils.join(auxSevices, ","); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES, newAuxServices); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE); yarnSiteConfig.setProperties(yarnSiteProperties); yarnSiteConfig.save(); return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", String.format("%s was set from %s to %s. %s was set to %s", YARN_NODEMANAGER_AUX_SERVICES, oldAuxServices, newAuxServices, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE), ""); } }
SparkShufflePropertyConfig extends AbstractServerAction { @Override public CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext) throws AmbariException, InterruptedException { String clusterName = getExecutionCommand().getClusterName(); Cluster cluster = clusters.getCluster(clusterName); Config yarnSiteConfig = cluster.getDesiredConfigByType(YARN_SITE_CONFIG_TYPE); if (yarnSiteConfig == null) { return createCommandReport(0, HostRoleStatus.FAILED,"{}", String.format("Source type %s not found", YARN_SITE_CONFIG_TYPE), ""); } Map<String, String> yarnSiteProperties = yarnSiteConfig.getProperties(); final List<String> auxSevices; final String oldAuxServices = yarnSiteProperties.get(YARN_NODEMANAGER_AUX_SERVICES); final String newAuxServices; if (yarnSiteProperties.containsKey(YARN_NODEMANAGER_AUX_SERVICES)) { auxSevices = new ArrayList<>(Arrays.asList(oldAuxServices.split(",", -1))); } else { auxSevices = new ArrayList<>(); } if (!auxSevices.contains(SPARK_SHUFFLE_AUX_STR)) { auxSevices.add(SPARK_SHUFFLE_AUX_STR); } newAuxServices = StringUtils.join(auxSevices, ","); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES, newAuxServices); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE); yarnSiteConfig.setProperties(yarnSiteProperties); yarnSiteConfig.save(); return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", String.format("%s was set from %s to %s. %s was set to %s", YARN_NODEMANAGER_AUX_SERVICES, oldAuxServices, newAuxServices, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE), ""); }  }
SparkShufflePropertyConfig extends AbstractServerAction { @Override public CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext) throws AmbariException, InterruptedException { String clusterName = getExecutionCommand().getClusterName(); Cluster cluster = clusters.getCluster(clusterName); Config yarnSiteConfig = cluster.getDesiredConfigByType(YARN_SITE_CONFIG_TYPE); if (yarnSiteConfig == null) { return createCommandReport(0, HostRoleStatus.FAILED,"{}", String.format("Source type %s not found", YARN_SITE_CONFIG_TYPE), ""); } Map<String, String> yarnSiteProperties = yarnSiteConfig.getProperties(); final List<String> auxSevices; final String oldAuxServices = yarnSiteProperties.get(YARN_NODEMANAGER_AUX_SERVICES); final String newAuxServices; if (yarnSiteProperties.containsKey(YARN_NODEMANAGER_AUX_SERVICES)) { auxSevices = new ArrayList<>(Arrays.asList(oldAuxServices.split(",", -1))); } else { auxSevices = new ArrayList<>(); } if (!auxSevices.contains(SPARK_SHUFFLE_AUX_STR)) { auxSevices.add(SPARK_SHUFFLE_AUX_STR); } newAuxServices = StringUtils.join(auxSevices, ","); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES, newAuxServices); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE); yarnSiteConfig.setProperties(yarnSiteProperties); yarnSiteConfig.save(); return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", String.format("%s was set from %s to %s. %s was set to %s", YARN_NODEMANAGER_AUX_SERVICES, oldAuxServices, newAuxServices, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE), ""); }  @Override CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext); }
SparkShufflePropertyConfig extends AbstractServerAction { @Override public CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext) throws AmbariException, InterruptedException { String clusterName = getExecutionCommand().getClusterName(); Cluster cluster = clusters.getCluster(clusterName); Config yarnSiteConfig = cluster.getDesiredConfigByType(YARN_SITE_CONFIG_TYPE); if (yarnSiteConfig == null) { return createCommandReport(0, HostRoleStatus.FAILED,"{}", String.format("Source type %s not found", YARN_SITE_CONFIG_TYPE), ""); } Map<String, String> yarnSiteProperties = yarnSiteConfig.getProperties(); final List<String> auxSevices; final String oldAuxServices = yarnSiteProperties.get(YARN_NODEMANAGER_AUX_SERVICES); final String newAuxServices; if (yarnSiteProperties.containsKey(YARN_NODEMANAGER_AUX_SERVICES)) { auxSevices = new ArrayList<>(Arrays.asList(oldAuxServices.split(",", -1))); } else { auxSevices = new ArrayList<>(); } if (!auxSevices.contains(SPARK_SHUFFLE_AUX_STR)) { auxSevices.add(SPARK_SHUFFLE_AUX_STR); } newAuxServices = StringUtils.join(auxSevices, ","); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES, newAuxServices); yarnSiteProperties.put(YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE); yarnSiteConfig.setProperties(yarnSiteProperties); yarnSiteConfig.save(); return createCommandReport(0, HostRoleStatus.COMPLETED, "{}", String.format("%s was set from %s to %s. %s was set to %s", YARN_NODEMANAGER_AUX_SERVICES, oldAuxServices, newAuxServices, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE, YARN_NODEMANAGER_AUX_SERVICES_SPARK_SHUFFLE_CLASS_VALUE), ""); }  @Override CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext);  }