@Test public void testRecommendConfiguration_EmptyConfiguration_applyStackDefaultsOnly() throws Exception { final String expectedHostGroupName = "host_group_1"; Map<String, Map<String, String>> properties = new HashMap<String, Map<String, String>>(); Collection<String> hgComponents = new HashSet<String>(); hgComponents.add("NAMENODE"); hgComponents.add("SECONDARY_NAMENODE"); hgComponents.add("RESOURCEMANAGER"); TestHostGroup group1 = new TestHostGroup("group1", hgComponents, Collections.singleton("testhost")); Collection<String> hgComponents2 = new HashSet<String>(); hgComponents2.add("DATANODE"); hgComponents2.add("HDFS_CLIENT"); TestHostGroup group2 = new TestHostGroup("group2", hgComponents2, Collections.singleton(expectedHostGroupName)); Collection<TestHostGroup> hostGroups = new HashSet<TestHostGroup>(); hostGroups.add(group1); hostGroups.add(group2); Map<String, Map<String, String>> parentProperties = new HashMap<String, Map<String, String>>(); Configuration parentConfig = new Configuration(parentProperties, Collections.<String, Map<String, Map<String, String>>>emptyMap(), createStackDefaults()); Configuration clusterConfig = new Configuration(properties, Collections.<String, Map<String, Map<String, String>>>emptyMap(), parentConfig); ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups); topology.getAdvisedConfigurations().putAll(createAdvisedConfigMap()); topology.setConfigRecommendationStrategy(ConfigRecommendationStrategy.ONLY_STACK_DEFAULTS_APPLY); BlueprintConfigurationProcessor configProcessor = new BlueprintConfigurationProcessor(topology); reset(stack); expect(stack.getConfiguration(bp.getServices())).andReturn(createStackDefaults()).anyTimes(); Set<String> emptySet = Collections.emptySet(); expect(stack.getExcludedConfigurationTypes(anyObject(String.class))).andReturn(emptySet).anyTimes(); replay(stack); configProcessor.doUpdateForClusterCreate(); assertEquals("stackDefaultUpgraded", clusterConfig.getPropertyValue("core-site", "fs.stackDefault.key1")); assertNull(clusterConfig.getPropertyValue("core-site", "fs.stackDefault.key2")); assertNull(clusterConfig.getPropertyValue("core-site", "fs.notStackDefault")); }
public Set<String> doUpdateForClusterCreate() throws ConfigurationTopologyException { Set<String> configTypesUpdated = new HashSet<String>(); Configuration clusterConfig = clusterTopology.getConfiguration(); Map<String, HostGroupInfo> groupInfoMap = clusterTopology.getHostGroupInfo(); doRecommendConfigurations(clusterConfig, configTypesUpdated); doFilterPriorToClusterUpdate(clusterConfig, configTypesUpdated); Map<String, Map<String, String>> clusterProps = clusterConfig.getFullProperties(); for (Map<String, Map<String, PropertyUpdater>> updaterMap : createCollectionOfUpdaters()) { for (Map.Entry<String, Map<String, PropertyUpdater>> entry : updaterMap.entrySet()) { String type = entry.getKey(); for (Map.Entry<String, PropertyUpdater> updaterEntry : entry.getValue().entrySet()) { String propertyName = updaterEntry.getKey(); PropertyUpdater updater = updaterEntry.getValue(); Map<String, String> typeMap = clusterProps.get(type); if (typeMap != null && typeMap.containsKey(propertyName) && typeMap.get(propertyName) != null) { final String originalValue = typeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, clusterProps, clusterTopology); if(updatedValue == null ) { continue; } if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } clusterConfig.setProperty(type, propertyName, updatedValue); } for (HostGroupInfo groupInfo : groupInfoMap.values()) { Configuration hgConfig = groupInfo.getConfiguration(); Map<String, Map<String, String>> hgConfigProps = hgConfig.getFullProperties(1); Map<String, String> hgTypeMap = hgConfigProps.get(type); if (hgTypeMap != null && hgTypeMap.containsKey(propertyName)) { final String originalValue = hgTypeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, hgConfigProps, clusterTopology); if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } hgConfig.setProperty(type, propertyName, updatedValue); } } } } } if (clusterTopology.isNameNodeHAEnabled()) { Map<String, String> hdfsSiteConfig = clusterConfig.getFullProperties().get("hdfs-site"); String nameservices = hdfsSiteConfig.get("dfs.nameservices"); String int_nameservices = hdfsSiteConfig.get("dfs.internal.nameservices"); if(int_nameservices == null && nameservices != null) { clusterConfig.setProperty("hdfs-site", "dfs.internal.nameservices", nameservices); } if (! isNameNodeHAInitialActiveNodeSet(clusterProps) && ! isNameNodeHAInitialStandbyNodeSet(clusterProps)) { Collection<String> nnHosts = clusterTopology.getHostAssignmentsForComponent("NAMENODE"); if (nnHosts.size() != 2) { throw new ConfigurationTopologyException("NAMENODE HA requires exactly 2 hosts running NAMENODE but there are: " + nnHosts.size() + " Hosts: " + nnHosts); } Iterator<String> nnHostIterator = nnHosts.iterator(); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_active", nnHostIterator.next()); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_standby", nnHostIterator.next()); configTypesUpdated.add("hadoop-env"); } } setMissingConfigurations(clusterConfig, configTypesUpdated); trimProperties(clusterConfig, clusterTopology); return configTypesUpdated; }
BlueprintConfigurationProcessor { public Set<String> doUpdateForClusterCreate() throws ConfigurationTopologyException { Set<String> configTypesUpdated = new HashSet<String>(); Configuration clusterConfig = clusterTopology.getConfiguration(); Map<String, HostGroupInfo> groupInfoMap = clusterTopology.getHostGroupInfo(); doRecommendConfigurations(clusterConfig, configTypesUpdated); doFilterPriorToClusterUpdate(clusterConfig, configTypesUpdated); Map<String, Map<String, String>> clusterProps = clusterConfig.getFullProperties(); for (Map<String, Map<String, PropertyUpdater>> updaterMap : createCollectionOfUpdaters()) { for (Map.Entry<String, Map<String, PropertyUpdater>> entry : updaterMap.entrySet()) { String type = entry.getKey(); for (Map.Entry<String, PropertyUpdater> updaterEntry : entry.getValue().entrySet()) { String propertyName = updaterEntry.getKey(); PropertyUpdater updater = updaterEntry.getValue(); Map<String, String> typeMap = clusterProps.get(type); if (typeMap != null && typeMap.containsKey(propertyName) && typeMap.get(propertyName) != null) { final String originalValue = typeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, clusterProps, clusterTopology); if(updatedValue == null ) { continue; } if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } clusterConfig.setProperty(type, propertyName, updatedValue); } for (HostGroupInfo groupInfo : groupInfoMap.values()) { Configuration hgConfig = groupInfo.getConfiguration(); Map<String, Map<String, String>> hgConfigProps = hgConfig.getFullProperties(1); Map<String, String> hgTypeMap = hgConfigProps.get(type); if (hgTypeMap != null && hgTypeMap.containsKey(propertyName)) { final String originalValue = hgTypeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, hgConfigProps, clusterTopology); if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } hgConfig.setProperty(type, propertyName, updatedValue); } } } } } if (clusterTopology.isNameNodeHAEnabled()) { Map<String, String> hdfsSiteConfig = clusterConfig.getFullProperties().get("hdfs-site"); String nameservices = hdfsSiteConfig.get("dfs.nameservices"); String int_nameservices = hdfsSiteConfig.get("dfs.internal.nameservices"); if(int_nameservices == null && nameservices != null) { clusterConfig.setProperty("hdfs-site", "dfs.internal.nameservices", nameservices); } if (! isNameNodeHAInitialActiveNodeSet(clusterProps) && ! isNameNodeHAInitialStandbyNodeSet(clusterProps)) { Collection<String> nnHosts = clusterTopology.getHostAssignmentsForComponent("NAMENODE"); if (nnHosts.size() != 2) { throw new ConfigurationTopologyException("NAMENODE HA requires exactly 2 hosts running NAMENODE but there are: " + nnHosts.size() + " Hosts: " + nnHosts); } Iterator<String> nnHostIterator = nnHosts.iterator(); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_active", nnHostIterator.next()); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_standby", nnHostIterator.next()); configTypesUpdated.add("hadoop-env"); } } setMissingConfigurations(clusterConfig, configTypesUpdated); trimProperties(clusterConfig, clusterTopology); return configTypesUpdated; } }
BlueprintConfigurationProcessor { public Set<String> doUpdateForClusterCreate() throws ConfigurationTopologyException { Set<String> configTypesUpdated = new HashSet<String>(); Configuration clusterConfig = clusterTopology.getConfiguration(); Map<String, HostGroupInfo> groupInfoMap = clusterTopology.getHostGroupInfo(); doRecommendConfigurations(clusterConfig, configTypesUpdated); doFilterPriorToClusterUpdate(clusterConfig, configTypesUpdated); Map<String, Map<String, String>> clusterProps = clusterConfig.getFullProperties(); for (Map<String, Map<String, PropertyUpdater>> updaterMap : createCollectionOfUpdaters()) { for (Map.Entry<String, Map<String, PropertyUpdater>> entry : updaterMap.entrySet()) { String type = entry.getKey(); for (Map.Entry<String, PropertyUpdater> updaterEntry : entry.getValue().entrySet()) { String propertyName = updaterEntry.getKey(); PropertyUpdater updater = updaterEntry.getValue(); Map<String, String> typeMap = clusterProps.get(type); if (typeMap != null && typeMap.containsKey(propertyName) && typeMap.get(propertyName) != null) { final String originalValue = typeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, clusterProps, clusterTopology); if(updatedValue == null ) { continue; } if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } clusterConfig.setProperty(type, propertyName, updatedValue); } for (HostGroupInfo groupInfo : groupInfoMap.values()) { Configuration hgConfig = groupInfo.getConfiguration(); Map<String, Map<String, String>> hgConfigProps = hgConfig.getFullProperties(1); Map<String, String> hgTypeMap = hgConfigProps.get(type); if (hgTypeMap != null && hgTypeMap.containsKey(propertyName)) { final String originalValue = hgTypeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, hgConfigProps, clusterTopology); if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } hgConfig.setProperty(type, propertyName, updatedValue); } } } } } if (clusterTopology.isNameNodeHAEnabled()) { Map<String, String> hdfsSiteConfig = clusterConfig.getFullProperties().get("hdfs-site"); String nameservices = hdfsSiteConfig.get("dfs.nameservices"); String int_nameservices = hdfsSiteConfig.get("dfs.internal.nameservices"); if(int_nameservices == null && nameservices != null) { clusterConfig.setProperty("hdfs-site", "dfs.internal.nameservices", nameservices); } if (! isNameNodeHAInitialActiveNodeSet(clusterProps) && ! isNameNodeHAInitialStandbyNodeSet(clusterProps)) { Collection<String> nnHosts = clusterTopology.getHostAssignmentsForComponent("NAMENODE"); if (nnHosts.size() != 2) { throw new ConfigurationTopologyException("NAMENODE HA requires exactly 2 hosts running NAMENODE but there are: " + nnHosts.size() + " Hosts: " + nnHosts); } Iterator<String> nnHostIterator = nnHosts.iterator(); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_active", nnHostIterator.next()); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_standby", nnHostIterator.next()); configTypesUpdated.add("hadoop-env"); } } setMissingConfigurations(clusterConfig, configTypesUpdated); trimProperties(clusterConfig, clusterTopology); return configTypesUpdated; } BlueprintConfigurationProcessor(ClusterTopology clusterTopology); }
BlueprintConfigurationProcessor { public Set<String> doUpdateForClusterCreate() throws ConfigurationTopologyException { Set<String> configTypesUpdated = new HashSet<String>(); Configuration clusterConfig = clusterTopology.getConfiguration(); Map<String, HostGroupInfo> groupInfoMap = clusterTopology.getHostGroupInfo(); doRecommendConfigurations(clusterConfig, configTypesUpdated); doFilterPriorToClusterUpdate(clusterConfig, configTypesUpdated); Map<String, Map<String, String>> clusterProps = clusterConfig.getFullProperties(); for (Map<String, Map<String, PropertyUpdater>> updaterMap : createCollectionOfUpdaters()) { for (Map.Entry<String, Map<String, PropertyUpdater>> entry : updaterMap.entrySet()) { String type = entry.getKey(); for (Map.Entry<String, PropertyUpdater> updaterEntry : entry.getValue().entrySet()) { String propertyName = updaterEntry.getKey(); PropertyUpdater updater = updaterEntry.getValue(); Map<String, String> typeMap = clusterProps.get(type); if (typeMap != null && typeMap.containsKey(propertyName) && typeMap.get(propertyName) != null) { final String originalValue = typeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, clusterProps, clusterTopology); if(updatedValue == null ) { continue; } if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } clusterConfig.setProperty(type, propertyName, updatedValue); } for (HostGroupInfo groupInfo : groupInfoMap.values()) { Configuration hgConfig = groupInfo.getConfiguration(); Map<String, Map<String, String>> hgConfigProps = hgConfig.getFullProperties(1); Map<String, String> hgTypeMap = hgConfigProps.get(type); if (hgTypeMap != null && hgTypeMap.containsKey(propertyName)) { final String originalValue = hgTypeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, hgConfigProps, clusterTopology); if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } hgConfig.setProperty(type, propertyName, updatedValue); } } } } } if (clusterTopology.isNameNodeHAEnabled()) { Map<String, String> hdfsSiteConfig = clusterConfig.getFullProperties().get("hdfs-site"); String nameservices = hdfsSiteConfig.get("dfs.nameservices"); String int_nameservices = hdfsSiteConfig.get("dfs.internal.nameservices"); if(int_nameservices == null && nameservices != null) { clusterConfig.setProperty("hdfs-site", "dfs.internal.nameservices", nameservices); } if (! isNameNodeHAInitialActiveNodeSet(clusterProps) && ! isNameNodeHAInitialStandbyNodeSet(clusterProps)) { Collection<String> nnHosts = clusterTopology.getHostAssignmentsForComponent("NAMENODE"); if (nnHosts.size() != 2) { throw new ConfigurationTopologyException("NAMENODE HA requires exactly 2 hosts running NAMENODE but there are: " + nnHosts.size() + " Hosts: " + nnHosts); } Iterator<String> nnHostIterator = nnHosts.iterator(); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_active", nnHostIterator.next()); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_standby", nnHostIterator.next()); configTypesUpdated.add("hadoop-env"); } } setMissingConfigurations(clusterConfig, configTypesUpdated); trimProperties(clusterConfig, clusterTopology); return configTypesUpdated; } BlueprintConfigurationProcessor(ClusterTopology clusterTopology); Map<String, Map<String, PropertyUpdater>> getRemovePropertyUpdaters(); void initRemovePropertyUpdaters(); Collection<String> getRequiredHostGroups(); Set<String> doUpdateForClusterCreate(); void doUpdateForBlueprintExport(); void doNameNodeHAUpdate(); void doYarnResourceManagerHAUpdate(); void doOozieServerHAUpdate(); }
BlueprintConfigurationProcessor { public Set<String> doUpdateForClusterCreate() throws ConfigurationTopologyException { Set<String> configTypesUpdated = new HashSet<String>(); Configuration clusterConfig = clusterTopology.getConfiguration(); Map<String, HostGroupInfo> groupInfoMap = clusterTopology.getHostGroupInfo(); doRecommendConfigurations(clusterConfig, configTypesUpdated); doFilterPriorToClusterUpdate(clusterConfig, configTypesUpdated); Map<String, Map<String, String>> clusterProps = clusterConfig.getFullProperties(); for (Map<String, Map<String, PropertyUpdater>> updaterMap : createCollectionOfUpdaters()) { for (Map.Entry<String, Map<String, PropertyUpdater>> entry : updaterMap.entrySet()) { String type = entry.getKey(); for (Map.Entry<String, PropertyUpdater> updaterEntry : entry.getValue().entrySet()) { String propertyName = updaterEntry.getKey(); PropertyUpdater updater = updaterEntry.getValue(); Map<String, String> typeMap = clusterProps.get(type); if (typeMap != null && typeMap.containsKey(propertyName) && typeMap.get(propertyName) != null) { final String originalValue = typeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, clusterProps, clusterTopology); if(updatedValue == null ) { continue; } if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } clusterConfig.setProperty(type, propertyName, updatedValue); } for (HostGroupInfo groupInfo : groupInfoMap.values()) { Configuration hgConfig = groupInfo.getConfiguration(); Map<String, Map<String, String>> hgConfigProps = hgConfig.getFullProperties(1); Map<String, String> hgTypeMap = hgConfigProps.get(type); if (hgTypeMap != null && hgTypeMap.containsKey(propertyName)) { final String originalValue = hgTypeMap.get(propertyName); final String updatedValue = updater.updateForClusterCreate(propertyName, originalValue, hgConfigProps, clusterTopology); if (!updatedValue.equals(originalValue)) { configTypesUpdated.add(type); } hgConfig.setProperty(type, propertyName, updatedValue); } } } } } if (clusterTopology.isNameNodeHAEnabled()) { Map<String, String> hdfsSiteConfig = clusterConfig.getFullProperties().get("hdfs-site"); String nameservices = hdfsSiteConfig.get("dfs.nameservices"); String int_nameservices = hdfsSiteConfig.get("dfs.internal.nameservices"); if(int_nameservices == null && nameservices != null) { clusterConfig.setProperty("hdfs-site", "dfs.internal.nameservices", nameservices); } if (! isNameNodeHAInitialActiveNodeSet(clusterProps) && ! isNameNodeHAInitialStandbyNodeSet(clusterProps)) { Collection<String> nnHosts = clusterTopology.getHostAssignmentsForComponent("NAMENODE"); if (nnHosts.size() != 2) { throw new ConfigurationTopologyException("NAMENODE HA requires exactly 2 hosts running NAMENODE but there are: " + nnHosts.size() + " Hosts: " + nnHosts); } Iterator<String> nnHostIterator = nnHosts.iterator(); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_active", nnHostIterator.next()); clusterConfig.setProperty("hadoop-env", "dfs_ha_initial_namenode_standby", nnHostIterator.next()); configTypesUpdated.add("hadoop-env"); } } setMissingConfigurations(clusterConfig, configTypesUpdated); trimProperties(clusterConfig, clusterTopology); return configTypesUpdated; } BlueprintConfigurationProcessor(ClusterTopology clusterTopology); Map<String, Map<String, PropertyUpdater>> getRemovePropertyUpdaters(); void initRemovePropertyUpdaters(); Collection<String> getRequiredHostGroups(); Set<String> doUpdateForClusterCreate(); void doUpdateForBlueprintExport(); void doNameNodeHAUpdate(); void doYarnResourceManagerHAUpdate(); void doOozieServerHAUpdate();  }