@Test public void testGetExecutionCommand() throws JSONException, AmbariException { Map<String, Map<String, String>> confs = new HashMap<String, Map<String, String>>(); Map<String, String> configurationsGlobal = new HashMap<String, String>(); configurationsGlobal.put(GLOBAL_NAME1, GLOBAL_VAL1); confs.put(GLOBAL_CONFIG, configurationsGlobal); Map<String, Map<String, String>> confTags = new HashMap<String, Map<String, String>>(); Map<String, String> confTagServiceSite = new HashMap<String, String>(); confTagServiceSite.put("tag", CLUSTER_VERSION_TAG); confTagServiceSite.put("service_override_tag", SERVICE_VERSION_TAG); confTagServiceSite.put("host_override_tag", HOST_VERSION_TAG); confTags.put(SERVICE_SITE_CONFIG, confTagServiceSite); Map<String, String> confTagGlobal = Collections.singletonMap("tag", CLUSTER_VERSION_TAG); confTags.put(GLOBAL_CONFIG, confTagGlobal); ExecutionCommand executionCommand = new ExecutionCommand(); executionCommand.setClusterName(CLUSTER1); executionCommand.setTaskId(1); executionCommand.setRequestAndStage(1, 1); executionCommand.setHostname(HOST1); executionCommand.setRole("NAMENODE"); executionCommand.setRoleParams(Collections.<String, String>emptyMap()); executionCommand.setRoleCommand(RoleCommand.START); executionCommand.setConfigurations(confs); executionCommand.setConfigurationTags(confTags); executionCommand.setServiceName("HDFS"); executionCommand.setCommandType(AgentCommandType.EXECUTION_COMMAND); executionCommand.setCommandParams(Collections.<String, String>emptyMap()); String json = StageUtils.getGson().toJson(executionCommand, ExecutionCommand.class); ExecutionCommandWrapper execCommWrap = new ExecutionCommandWrapper(json); injector.injectMembers(execCommWrap); ExecutionCommand processedExecutionCommand = execCommWrap.getExecutionCommand(); Map<String, String> serviceSiteConfig = processedExecutionCommand.getConfigurations().get(SERVICE_SITE_CONFIG); Assert.assertEquals(SERVICE_SITE_VAL1_S, serviceSiteConfig.get(SERVICE_SITE_NAME1)); Assert.assertEquals(SERVICE_SITE_VAL2_H, serviceSiteConfig.get(SERVICE_SITE_NAME2)); Assert.assertEquals(SERVICE_SITE_VAL3, serviceSiteConfig.get(SERVICE_SITE_NAME3)); Assert.assertEquals(SERVICE_SITE_VAL4, serviceSiteConfig.get(SERVICE_SITE_NAME4)); Assert.assertEquals(SERVICE_SITE_VAL5_S, serviceSiteConfig.get(SERVICE_SITE_NAME5)); Assert.assertEquals(SERVICE_SITE_VAL6_H, serviceSiteConfig.get(SERVICE_SITE_NAME6)); Map<String, String> globalConfig = processedExecutionCommand.getConfigurations().get(GLOBAL_CONFIG); Assert.assertEquals(GLOBAL_VAL1, globalConfig.get(GLOBAL_NAME1)); Assert.assertEquals(GLOBAL_CLUSTER_VAL2, globalConfig.get(GLOBAL_NAME2)); Set<String> serviceSiteKeys = new HashSet<String>(); serviceSiteKeys.addAll(SERVICE_SITE_CLUSTER.keySet()); serviceSiteKeys.addAll(SERVICE_SITE_SERVICE.keySet()); serviceSiteKeys.addAll(SERVICE_SITE_HOST.keySet()); Assert.assertEquals(serviceSiteKeys.size(), serviceSiteConfig.size()); }
public ExecutionCommand getExecutionCommand() { if (executionCommand != null) { return executionCommand; } if( null == jsonExecutionCommand ){ throw new RuntimeException( "Invalid ExecutionCommandWrapper, both object and string representations are null"); } try { executionCommand = gson.fromJson(jsonExecutionCommand, ExecutionCommand.class); if (null == executionCommand.getConfigurations()) { executionCommand.setConfigurations(new TreeMap<String, Map<String, String>>()); } Map<String, Map<String, String>> configurations = executionCommand.getConfigurations(); Long clusterId = hostRoleCommandDAO.findByPK( executionCommand.getTaskId()).getStage().getClusterId(); Cluster cluster = clusters.getClusterById(clusterId); boolean refreshConfigTagsBeforeExecution = executionCommand.getForceRefreshConfigTagsBeforeExecution(); if (refreshConfigTagsBeforeExecution) { Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs(); Map<String, Map<String, String>> configurationTags = configHelper.getEffectiveDesiredTags( cluster, executionCommand.getHostname(), desiredConfigs); LOG.debug( "While scheduling task {} on cluster {}, configurations are being refreshed using desired configurations of {}", executionCommand.getTaskId(), cluster.getClusterName(), desiredConfigs); configurations.clear(); executionCommand.setConfigurationTags(configurationTags); } Map<String, Map<String, String>> configurationTags = executionCommand.getConfigurationTags(); if (null != configurationTags && !configurationTags.isEmpty()) { Map<String, Map<String, String>> configProperties = configHelper .getEffectiveConfigProperties(cluster, configurationTags); for (Map.Entry<String, Map<String, String>> entry : configProperties.entrySet()) { String type = entry.getKey(); Map<String, String> allLevelMergedConfig = entry.getValue(); if (configurations.containsKey(type)) { Map<String, String> mergedConfig = configHelper.getMergedConfig(allLevelMergedConfig, configurations.get(type)); configurations.get(type).clear(); configurations.get(type).putAll(mergedConfig); } else { configurations.put(type, new HashMap<String, String>()); configurations.get(type).putAll(allLevelMergedConfig); } } Map<String, Map<String, Map<String, String>>> configAttributes = configHelper.getEffectiveConfigAttributes( cluster, executionCommand.getConfigurationTags()); for (Map.Entry<String, Map<String, Map<String, String>>> attributesOccurrence : configAttributes.entrySet()) { String type = attributesOccurrence.getKey(); Map<String, Map<String, String>> attributes = attributesOccurrence.getValue(); if (executionCommand.getConfigurationAttributes() != null) { if (!executionCommand.getConfigurationAttributes().containsKey(type)) { executionCommand.getConfigurationAttributes().put(type, new TreeMap<String, Map<String, String>>()); } configHelper.cloneAttributesMap(attributes, executionCommand.getConfigurationAttributes().get(type)); } } } } catch (ClusterNotFoundException cnfe) { LOG.warn( "Unable to lookup the cluster by ID; assuming that there is no cluster and therefore no configs for this execution command: {}", cnfe.getMessage()); return executionCommand; } catch (AmbariException e) { throw new RuntimeException(e); } return executionCommand; }
ExecutionCommandWrapper { public ExecutionCommand getExecutionCommand() { if (executionCommand != null) { return executionCommand; } if( null == jsonExecutionCommand ){ throw new RuntimeException( "Invalid ExecutionCommandWrapper, both object and string representations are null"); } try { executionCommand = gson.fromJson(jsonExecutionCommand, ExecutionCommand.class); if (null == executionCommand.getConfigurations()) { executionCommand.setConfigurations(new TreeMap<String, Map<String, String>>()); } Map<String, Map<String, String>> configurations = executionCommand.getConfigurations(); Long clusterId = hostRoleCommandDAO.findByPK( executionCommand.getTaskId()).getStage().getClusterId(); Cluster cluster = clusters.getClusterById(clusterId); boolean refreshConfigTagsBeforeExecution = executionCommand.getForceRefreshConfigTagsBeforeExecution(); if (refreshConfigTagsBeforeExecution) { Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs(); Map<String, Map<String, String>> configurationTags = configHelper.getEffectiveDesiredTags( cluster, executionCommand.getHostname(), desiredConfigs); LOG.debug( "While scheduling task {} on cluster {}, configurations are being refreshed using desired configurations of {}", executionCommand.getTaskId(), cluster.getClusterName(), desiredConfigs); configurations.clear(); executionCommand.setConfigurationTags(configurationTags); } Map<String, Map<String, String>> configurationTags = executionCommand.getConfigurationTags(); if (null != configurationTags && !configurationTags.isEmpty()) { Map<String, Map<String, String>> configProperties = configHelper .getEffectiveConfigProperties(cluster, configurationTags); for (Map.Entry<String, Map<String, String>> entry : configProperties.entrySet()) { String type = entry.getKey(); Map<String, String> allLevelMergedConfig = entry.getValue(); if (configurations.containsKey(type)) { Map<String, String> mergedConfig = configHelper.getMergedConfig(allLevelMergedConfig, configurations.get(type)); configurations.get(type).clear(); configurations.get(type).putAll(mergedConfig); } else { configurations.put(type, new HashMap<String, String>()); configurations.get(type).putAll(allLevelMergedConfig); } } Map<String, Map<String, Map<String, String>>> configAttributes = configHelper.getEffectiveConfigAttributes( cluster, executionCommand.getConfigurationTags()); for (Map.Entry<String, Map<String, Map<String, String>>> attributesOccurrence : configAttributes.entrySet()) { String type = attributesOccurrence.getKey(); Map<String, Map<String, String>> attributes = attributesOccurrence.getValue(); if (executionCommand.getConfigurationAttributes() != null) { if (!executionCommand.getConfigurationAttributes().containsKey(type)) { executionCommand.getConfigurationAttributes().put(type, new TreeMap<String, Map<String, String>>()); } configHelper.cloneAttributesMap(attributes, executionCommand.getConfigurationAttributes().get(type)); } } } } catch (ClusterNotFoundException cnfe) { LOG.warn( "Unable to lookup the cluster by ID; assuming that there is no cluster and therefore no configs for this execution command: {}", cnfe.getMessage()); return executionCommand; } catch (AmbariException e) { throw new RuntimeException(e); } return executionCommand; } }
ExecutionCommandWrapper { public ExecutionCommand getExecutionCommand() { if (executionCommand != null) { return executionCommand; } if( null == jsonExecutionCommand ){ throw new RuntimeException( "Invalid ExecutionCommandWrapper, both object and string representations are null"); } try { executionCommand = gson.fromJson(jsonExecutionCommand, ExecutionCommand.class); if (null == executionCommand.getConfigurations()) { executionCommand.setConfigurations(new TreeMap<String, Map<String, String>>()); } Map<String, Map<String, String>> configurations = executionCommand.getConfigurations(); Long clusterId = hostRoleCommandDAO.findByPK( executionCommand.getTaskId()).getStage().getClusterId(); Cluster cluster = clusters.getClusterById(clusterId); boolean refreshConfigTagsBeforeExecution = executionCommand.getForceRefreshConfigTagsBeforeExecution(); if (refreshConfigTagsBeforeExecution) { Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs(); Map<String, Map<String, String>> configurationTags = configHelper.getEffectiveDesiredTags( cluster, executionCommand.getHostname(), desiredConfigs); LOG.debug( "While scheduling task {} on cluster {}, configurations are being refreshed using desired configurations of {}", executionCommand.getTaskId(), cluster.getClusterName(), desiredConfigs); configurations.clear(); executionCommand.setConfigurationTags(configurationTags); } Map<String, Map<String, String>> configurationTags = executionCommand.getConfigurationTags(); if (null != configurationTags && !configurationTags.isEmpty()) { Map<String, Map<String, String>> configProperties = configHelper .getEffectiveConfigProperties(cluster, configurationTags); for (Map.Entry<String, Map<String, String>> entry : configProperties.entrySet()) { String type = entry.getKey(); Map<String, String> allLevelMergedConfig = entry.getValue(); if (configurations.containsKey(type)) { Map<String, String> mergedConfig = configHelper.getMergedConfig(allLevelMergedConfig, configurations.get(type)); configurations.get(type).clear(); configurations.get(type).putAll(mergedConfig); } else { configurations.put(type, new HashMap<String, String>()); configurations.get(type).putAll(allLevelMergedConfig); } } Map<String, Map<String, Map<String, String>>> configAttributes = configHelper.getEffectiveConfigAttributes( cluster, executionCommand.getConfigurationTags()); for (Map.Entry<String, Map<String, Map<String, String>>> attributesOccurrence : configAttributes.entrySet()) { String type = attributesOccurrence.getKey(); Map<String, Map<String, String>> attributes = attributesOccurrence.getValue(); if (executionCommand.getConfigurationAttributes() != null) { if (!executionCommand.getConfigurationAttributes().containsKey(type)) { executionCommand.getConfigurationAttributes().put(type, new TreeMap<String, Map<String, String>>()); } configHelper.cloneAttributesMap(attributes, executionCommand.getConfigurationAttributes().get(type)); } } } } catch (ClusterNotFoundException cnfe) { LOG.warn( "Unable to lookup the cluster by ID; assuming that there is no cluster and therefore no configs for this execution command: {}", cnfe.getMessage()); return executionCommand; } catch (AmbariException e) { throw new RuntimeException(e); } return executionCommand; } @AssistedInject  ExecutionCommandWrapper(@Assisted String jsonExecutionCommand); @AssistedInject  ExecutionCommandWrapper(@Assisted ExecutionCommand executionCommand); }
ExecutionCommandWrapper { public ExecutionCommand getExecutionCommand() { if (executionCommand != null) { return executionCommand; } if( null == jsonExecutionCommand ){ throw new RuntimeException( "Invalid ExecutionCommandWrapper, both object and string representations are null"); } try { executionCommand = gson.fromJson(jsonExecutionCommand, ExecutionCommand.class); if (null == executionCommand.getConfigurations()) { executionCommand.setConfigurations(new TreeMap<String, Map<String, String>>()); } Map<String, Map<String, String>> configurations = executionCommand.getConfigurations(); Long clusterId = hostRoleCommandDAO.findByPK( executionCommand.getTaskId()).getStage().getClusterId(); Cluster cluster = clusters.getClusterById(clusterId); boolean refreshConfigTagsBeforeExecution = executionCommand.getForceRefreshConfigTagsBeforeExecution(); if (refreshConfigTagsBeforeExecution) { Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs(); Map<String, Map<String, String>> configurationTags = configHelper.getEffectiveDesiredTags( cluster, executionCommand.getHostname(), desiredConfigs); LOG.debug( "While scheduling task {} on cluster {}, configurations are being refreshed using desired configurations of {}", executionCommand.getTaskId(), cluster.getClusterName(), desiredConfigs); configurations.clear(); executionCommand.setConfigurationTags(configurationTags); } Map<String, Map<String, String>> configurationTags = executionCommand.getConfigurationTags(); if (null != configurationTags && !configurationTags.isEmpty()) { Map<String, Map<String, String>> configProperties = configHelper .getEffectiveConfigProperties(cluster, configurationTags); for (Map.Entry<String, Map<String, String>> entry : configProperties.entrySet()) { String type = entry.getKey(); Map<String, String> allLevelMergedConfig = entry.getValue(); if (configurations.containsKey(type)) { Map<String, String> mergedConfig = configHelper.getMergedConfig(allLevelMergedConfig, configurations.get(type)); configurations.get(type).clear(); configurations.get(type).putAll(mergedConfig); } else { configurations.put(type, new HashMap<String, String>()); configurations.get(type).putAll(allLevelMergedConfig); } } Map<String, Map<String, Map<String, String>>> configAttributes = configHelper.getEffectiveConfigAttributes( cluster, executionCommand.getConfigurationTags()); for (Map.Entry<String, Map<String, Map<String, String>>> attributesOccurrence : configAttributes.entrySet()) { String type = attributesOccurrence.getKey(); Map<String, Map<String, String>> attributes = attributesOccurrence.getValue(); if (executionCommand.getConfigurationAttributes() != null) { if (!executionCommand.getConfigurationAttributes().containsKey(type)) { executionCommand.getConfigurationAttributes().put(type, new TreeMap<String, Map<String, String>>()); } configHelper.cloneAttributesMap(attributes, executionCommand.getConfigurationAttributes().get(type)); } } } } catch (ClusterNotFoundException cnfe) { LOG.warn( "Unable to lookup the cluster by ID; assuming that there is no cluster and therefore no configs for this execution command: {}", cnfe.getMessage()); return executionCommand; } catch (AmbariException e) { throw new RuntimeException(e); } return executionCommand; } @AssistedInject  ExecutionCommandWrapper(@Assisted String jsonExecutionCommand); @AssistedInject  ExecutionCommandWrapper(@Assisted ExecutionCommand executionCommand); ExecutionCommand getExecutionCommand(); AgentCommandType getCommandType(); String getJson(); @Override boolean equals(Object o); @Override int hashCode(); }
ExecutionCommandWrapper { public ExecutionCommand getExecutionCommand() { if (executionCommand != null) { return executionCommand; } if( null == jsonExecutionCommand ){ throw new RuntimeException( "Invalid ExecutionCommandWrapper, both object and string representations are null"); } try { executionCommand = gson.fromJson(jsonExecutionCommand, ExecutionCommand.class); if (null == executionCommand.getConfigurations()) { executionCommand.setConfigurations(new TreeMap<String, Map<String, String>>()); } Map<String, Map<String, String>> configurations = executionCommand.getConfigurations(); Long clusterId = hostRoleCommandDAO.findByPK( executionCommand.getTaskId()).getStage().getClusterId(); Cluster cluster = clusters.getClusterById(clusterId); boolean refreshConfigTagsBeforeExecution = executionCommand.getForceRefreshConfigTagsBeforeExecution(); if (refreshConfigTagsBeforeExecution) { Map<String, DesiredConfig> desiredConfigs = cluster.getDesiredConfigs(); Map<String, Map<String, String>> configurationTags = configHelper.getEffectiveDesiredTags( cluster, executionCommand.getHostname(), desiredConfigs); LOG.debug( "While scheduling task {} on cluster {}, configurations are being refreshed using desired configurations of {}", executionCommand.getTaskId(), cluster.getClusterName(), desiredConfigs); configurations.clear(); executionCommand.setConfigurationTags(configurationTags); } Map<String, Map<String, String>> configurationTags = executionCommand.getConfigurationTags(); if (null != configurationTags && !configurationTags.isEmpty()) { Map<String, Map<String, String>> configProperties = configHelper .getEffectiveConfigProperties(cluster, configurationTags); for (Map.Entry<String, Map<String, String>> entry : configProperties.entrySet()) { String type = entry.getKey(); Map<String, String> allLevelMergedConfig = entry.getValue(); if (configurations.containsKey(type)) { Map<String, String> mergedConfig = configHelper.getMergedConfig(allLevelMergedConfig, configurations.get(type)); configurations.get(type).clear(); configurations.get(type).putAll(mergedConfig); } else { configurations.put(type, new HashMap<String, String>()); configurations.get(type).putAll(allLevelMergedConfig); } } Map<String, Map<String, Map<String, String>>> configAttributes = configHelper.getEffectiveConfigAttributes( cluster, executionCommand.getConfigurationTags()); for (Map.Entry<String, Map<String, Map<String, String>>> attributesOccurrence : configAttributes.entrySet()) { String type = attributesOccurrence.getKey(); Map<String, Map<String, String>> attributes = attributesOccurrence.getValue(); if (executionCommand.getConfigurationAttributes() != null) { if (!executionCommand.getConfigurationAttributes().containsKey(type)) { executionCommand.getConfigurationAttributes().put(type, new TreeMap<String, Map<String, String>>()); } configHelper.cloneAttributesMap(attributes, executionCommand.getConfigurationAttributes().get(type)); } } } } catch (ClusterNotFoundException cnfe) { LOG.warn( "Unable to lookup the cluster by ID; assuming that there is no cluster and therefore no configs for this execution command: {}", cnfe.getMessage()); return executionCommand; } catch (AmbariException e) { throw new RuntimeException(e); } return executionCommand; } @AssistedInject  ExecutionCommandWrapper(@Assisted String jsonExecutionCommand); @AssistedInject  ExecutionCommandWrapper(@Assisted ExecutionCommand executionCommand); ExecutionCommand getExecutionCommand(); AgentCommandType getCommandType(); String getJson(); @Override boolean equals(Object o); @Override int hashCode();  }