AMBARI-12532. Blueprint deployment results in some configurations not being resolved properly. (rnettleton)

ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java

@@ -1360,9 +1360,10 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
             !all.containsKey(cr.getVersionTag()) ||     // tag not set
             cr.getProperties().size() > 0) {            // properties to set
 
-          LOG.info(MessageFormat.format("Applying configuration with tag ''{0}'' to cluster ''{1}''",
+          LOG.info(MessageFormat.format("Applying configuration with tag ''{0}'' to cluster ''{1}''  for configuration type {2}",
               cr.getVersionTag(),
-              request.getClusterName()));
+              request.getClusterName(),
+              cr.getType()));
 
           cr.setClusterName(cluster.getClusterName());
           configurationResponses.add(createConfiguration(cr));

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/BlueprintConfigurationProcessor.java

@@ -197,14 +197,17 @@ public class BlueprintConfigurationProcessor {
   /**
    * Update properties for cluster creation.  This involves updating topology related properties with
    * concrete topology information.
+   *
+   * @return Set of config type names that were updated by this update call
    */
-  public void doUpdateForClusterCreate() throws ConfigurationTopologyException {
+  public Set<String> doUpdateForClusterCreate() throws ConfigurationTopologyException {
+    Set<String> configTypesUpdated = new HashSet<String>();
     Configuration clusterConfig = clusterTopology.getConfiguration();
     Map<String, HostGroupInfo> groupInfoMap = clusterTopology.getHostGroupInfo();
 
     // filter out any properties that should not be included, based on the dependencies
     // specified in the stacks, and the filters defined in this class
-    doFilterPriorToClusterUpdate(clusterConfig);
+    doFilterPriorToClusterUpdate(clusterConfig, configTypesUpdated);
 
     // this needs to be called after doFilterPriorToClusterUpdate() to ensure that the returned
     // set of properties (copy) doesn't include the removed properties.  If an updater
@@ -221,8 +224,15 @@ public class BlueprintConfigurationProcessor {
           // topo cluster scoped configuration which also includes all default and BP properties
           Map<String, String> typeMap = clusterProps.get(type);
           if (typeMap != null && typeMap.containsKey(propertyName)) {
-            clusterConfig.setProperty(type, propertyName, updater.updateForClusterCreate(
-                propertyName, typeMap.get(propertyName), clusterProps, clusterTopology));
+            final String originalValue = typeMap.get(propertyName);
+            final String updatedValue =
+              updater.updateForClusterCreate(propertyName, originalValue, clusterProps, clusterTopology);
+
+            if (!updatedValue.equals(originalValue)) {
+              configTypesUpdated.add(type);
+            }
+
+            clusterConfig.setProperty(type, propertyName, updatedValue);
           }
 
           // host group configs
@@ -231,8 +241,15 @@ public class BlueprintConfigurationProcessor {
             Map<String, Map<String, String>> hgConfigProps = hgConfig.getFullProperties(1);
             Map<String, String> hgTypeMap = hgConfigProps.get(type);
             if (hgTypeMap != null && hgTypeMap.containsKey(propertyName)) {
-              hgConfig.setProperty(type, propertyName, updater.updateForClusterCreate(
-                  propertyName, hgTypeMap.get(propertyName), hgConfigProps, clusterTopology));
+              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);
             }
           }
         }
@@ -254,9 +271,14 @@ public class BlueprintConfigurationProcessor {
         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);
+
+    setMissingConfigurations(clusterConfig, configTypesUpdated);
+
+    return configTypesUpdated;
   }
 
   /**
@@ -340,7 +362,7 @@ public class BlueprintConfigurationProcessor {
     }
   }
 
-  private void doFilterPriorToClusterUpdate(Configuration configuration) {
+  private void doFilterPriorToClusterUpdate(Configuration configuration, Set<String> configTypesUpdated) {
     // getFullProperties returns a copy so changes to it are not reflected in config properties
     Map<String, Map<String, String>> properties = configuration.getFullProperties();
     for (Map.Entry<String, Map<String, String>> configEntry : properties.entrySet()) {
@@ -351,6 +373,7 @@ public class BlueprintConfigurationProcessor {
         String propName = propertyEntry.getKey();
         if (shouldPropertyBeExcludedForClusterUpdate(propName, propertyEntry.getValue(), configType, clusterTopology)) {
           configuration.removeProperty(configType, propName);
+          configTypesUpdated.add(configType);
         }
       }
     }
@@ -2038,11 +2061,11 @@ public class BlueprintConfigurationProcessor {
    *
    * @param configuration  configuration where properties are to be added
    */
-  void setMissingConfigurations(Configuration configuration) {
+  void setMissingConfigurations(Configuration configuration, Set<String> configTypesUpdated) {
     // AMBARI-5206
     final Map<String , String> userProps = new HashMap<String , String>();
 
-    setRetryConfiguration(configuration);
+    setRetryConfiguration(configuration, configTypesUpdated);
 
     Collection<String> services = clusterTopology.getBlueprint().getServices();
     // only add user properties to the map for
@@ -2074,8 +2097,8 @@ public class BlueprintConfigurationProcessor {
       if (configs != null) {
         String user = configs.get(property);
         if (user != null && !user.isEmpty()) {
-          ensureProperty(configuration, "core-site", String.format(proxyUserHosts, user), "*");
-          ensureProperty(configuration, "core-site", String.format(proxyUserGroups, user), "users");
+          ensureProperty(configuration, "core-site", String.format(proxyUserHosts, user), "*", configTypesUpdated);
+          ensureProperty(configuration, "core-site", String.format(proxyUserGroups, user), "users", configTypesUpdated);
         }
       } else {
         LOG.debug("setMissingConfigurations: no user configuration found for type = " + configType +
@@ -2096,18 +2119,26 @@ public class BlueprintConfigurationProcessor {
    *
    * @param configuration cluster configuration
    */
-  private static void setRetryConfiguration(Configuration configuration) {
+  private static void setRetryConfiguration(Configuration configuration, Set<String> configTypesUpdated) {
+    boolean wasUpdated = false;
 
     if (configuration.getPropertyValue(CLUSTER_ENV_CONFIG_TYPE_NAME, COMMAND_RETRY_ENABLED_PROPERTY_NAME) == null) {
       configuration.setProperty(CLUSTER_ENV_CONFIG_TYPE_NAME, COMMAND_RETRY_ENABLED_PROPERTY_NAME, COMMAND_RETRY_ENABLED_DEFAULT);
+      wasUpdated = true;
     }
 
     if (configuration.getPropertyValue(CLUSTER_ENV_CONFIG_TYPE_NAME, COMMANDS_TO_RETRY_PROPERTY_NAME) == null) {
       configuration.setProperty(CLUSTER_ENV_CONFIG_TYPE_NAME, COMMANDS_TO_RETRY_PROPERTY_NAME, COMMANDS_TO_RETRY_DEFAULT);
+      wasUpdated = true;
     }
 
     if (configuration.getPropertyValue(CLUSTER_ENV_CONFIG_TYPE_NAME, COMMAND_RETRY_MAX_TIME_IN_SEC_PROPERTY_NAME) == null) {
       configuration.setProperty(CLUSTER_ENV_CONFIG_TYPE_NAME, COMMAND_RETRY_MAX_TIME_IN_SEC_PROPERTY_NAME, COMMAND_RETRY_MAX_TIME_IN_SEC_DEFAULT);
+      wasUpdated = true;
+    }
+
+    if (wasUpdated) {
+      configTypesUpdated.add(CLUSTER_ENV_CONFIG_TYPE_NAME);
     }
   }
 
@@ -2121,9 +2152,10 @@ public class BlueprintConfigurationProcessor {
    * @param property       property name
    * @param defaultValue   default value
    */
-  private void ensureProperty(Configuration configuration, String type, String property, String defaultValue) {
+  private void ensureProperty(Configuration configuration, String type, String property, String defaultValue, Set<String> configTypesUpdated) {
     if (configuration.getPropertyValue(type, property) == null) {
       configuration.setProperty(type, property, defaultValue);
+      configTypesUpdated.add(type);
     }
   }
 

ambari-server/src/main/java/org/apache/ambari/server/topology/AmbariContext.java

@@ -342,6 +342,51 @@ public class AmbariContext {
     }
   }
 
+  /**
+   * Verifies that all desired configurations have reached the resolved state
+   *   before proceeding with the install
+   *
+   * @param clusterName name of the cluster
+   * @param updatedConfigTypes set of config types that are required to be in the TOPOLOGY_RESOLVED state
+   *
+   * @throws AmbariException upon any system-level error that occurs
+   */
+  public void waitForConfigurationResolution(String clusterName, Set<String> updatedConfigTypes) throws AmbariException {
+    Cluster cluster = getController().getClusters().getCluster(clusterName);
+    boolean shouldWaitForResolution = true;
+    while (shouldWaitForResolution) {
+      int numOfRequestsStillRequiringResolution = 0;
+
+      // for all config types specified
+      for (String actualConfigType : updatedConfigTypes) {
+        // get the actual cluster config for comparison
+        DesiredConfig actualConfig = cluster.getDesiredConfigs().get(actualConfigType);
+        if (!actualConfig.getTag().equals(TopologyManager.TOPOLOGY_RESOLVED_TAG)) {
+          // if any expected config is not resolved, deployment must wait
+          LOG.info("Config type " + actualConfigType + " not resolved yet, Blueprint deployment will wait until configuration update is completed");
+          numOfRequestsStillRequiringResolution++;
+        } else {
+          LOG.info("Config type " + actualConfigType + " is resolved in the cluster config.");
+        }
+      }
+
+      if (numOfRequestsStillRequiringResolution == 0) {
+        // all configs are resolved, deployment can continue
+        LOG.info("All required configuration types are in the " + TopologyManager.TOPOLOGY_RESOLVED_TAG + " state.  Blueprint deployment can now continue.");
+        shouldWaitForResolution = false;
+      } else {
+        LOG.info("Waiting for " + numOfRequestsStillRequiringResolution + " configuration types to be resolved before Blueprint deployment can continue");
+
+        try {
+          // sleep before checking the config again
+          Thread.sleep(100);
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+        }
+      }
+    }
+  }
+
   public boolean doesConfigurationWithTagExist(String clusterName, String tag) {
     boolean isTopologyResolved = false;
     try {

ambari-server/src/main/java/org/apache/ambari/server/topology/ClusterConfigurationRequest.java

@@ -57,7 +57,7 @@ public class ClusterConfigurationRequest {
     // set initial configuration (not topology resolved)
     this.configurationProcessor = new BlueprintConfigurationProcessor(clusterTopology);
     if (setInitial) {
-      setConfigurationsOnCluster(clusterTopology, TopologyManager.INITIAL_CONFIG_TAG);
+      setConfigurationsOnCluster(clusterTopology, TopologyManager.INITIAL_CONFIG_TAG, Collections.<String>emptySet());
     }
   }
 
@@ -68,13 +68,16 @@ public class ClusterConfigurationRequest {
 
   public void process() throws AmbariException, ConfigurationTopologyException {
     // this will update the topo cluster config and all host group configs in the cluster topology
+    Set<String> updatedConfigTypes = Collections.emptySet();
     try {
-      configurationProcessor.doUpdateForClusterCreate();
+      updatedConfigTypes =
+        configurationProcessor.doUpdateForClusterCreate();
     } catch (ConfigurationTopologyException e) {
       //log and continue to set configs on cluster to make progress
       LOG.error("An exception occurred while doing configuration topology update: " + e, e);
     }
-    setConfigurationsOnCluster(clusterTopology, TopologyManager.TOPOLOGY_RESOLVED_TAG);
+
+    setConfigurationsOnCluster(clusterTopology, TopologyManager.TOPOLOGY_RESOLVED_TAG, updatedConfigTypes);
   }
 
   /**
@@ -82,7 +85,7 @@ public class ClusterConfigurationRequest {
    * @param clusterTopology  cluster topology
    * @param tag              config tag
    */
-  public void setConfigurationsOnCluster(ClusterTopology clusterTopology, String tag)  {
+  public void setConfigurationsOnCluster(ClusterTopology clusterTopology, String tag, Set<String> updatedConfigTypes)  {
     //todo: also handle setting of host group scoped configuration which is updated by config processor
     List<BlueprintServiceConfigRequest> configurationRequests = new LinkedList<BlueprintServiceConfigRequest>();
 
@@ -120,7 +123,7 @@ public class ClusterConfigurationRequest {
     globalConfigRequest.addConfigElement("cluster-env", clusterEnvProps,clusterEnvAttributes);
     configurationRequests.add(globalConfigRequest);
 
-    setConfigurationsOnCluster(configurationRequests, tag);
+    setConfigurationsOnCluster(configurationRequests, tag, updatedConfigTypes);
   }
 
   /**
@@ -134,7 +137,7 @@ public class ClusterConfigurationRequest {
    * @param configurationRequests a list of requests to send to the AmbariManagementController.
    */
   private void setConfigurationsOnCluster(List<BlueprintServiceConfigRequest> configurationRequests,
-                                          String tag)  {
+                                          String tag, Set<String> updatedConfigTypes)  {
     // iterate over services to deploy
     for (BlueprintServiceConfigRequest blueprintConfigRequest : configurationRequests) {
       ClusterRequest clusterRequest = null;
@@ -198,6 +201,16 @@ public class ClusterConfigurationRequest {
         LOG.error("ClusterRequest should not be null for service = " + blueprintConfigRequest.getServiceName());
       }
     }
+
+    if (tag.equals(TopologyManager.TOPOLOGY_RESOLVED_TAG)) {
+      // if this is a request to resolve config, then wait until resolution is completed
+      try {
+        // wait until the cluster topology configuration is set/resolved
+        ambariContext.waitForConfigurationResolution(clusterTopology.getClusterName(), updatedConfigTypes);
+      } catch (AmbariException e) {
+        LOG.error("Error while attempting to wait for the cluster configuration to reach TOPOLOGY_RESOLVED state.", e);
+      }
+    }
   }
 
   /**

ambari-server/src/main/java/org/apache/ambari/server/topology/HostRequest.java

@@ -87,7 +87,7 @@ public class HostRequest implements Comparable<HostRequest> {
     this.topology = topology;
 
     createTasks();
-    System.out.println("HostRequest: Created request for host: " +
+    LOG.info("HostRequest: Created request for host: " +
         (hostname == null ? "Host Assignment Pending" : hostname));
   }
 
@@ -120,7 +120,7 @@ public class HostRequest implements Comparable<HostRequest> {
     isOutstanding = hostname == null || !topology.getAmbariContext().
         isHostRegisteredWithCluster(cluster, hostname);
 
-    System.out.println("HostRequest: Successfully recovered host request for host: " +
+    LOG.info("HostRequest: Successfully recovered host request for host: " +
         (hostname == null ? "Host Assignment Pending" : hostname));
   }
 
@@ -187,7 +187,7 @@ public class HostRequest implements Comparable<HostRequest> {
     HostGroup hostGroup = getHostGroup();
     for (String component : hostGroup.getComponents()) {
       if (component == null || component.equals("AMBARI_SERVER")) {
-        System.out.printf("Skipping component %s when creating request\n", component);
+        LOG.info("Skipping component %s when creating request\n", component);
         continue;
       }
 
@@ -469,7 +469,7 @@ public class HostRequest implements Comparable<HostRequest> {
 
     @Override
     public void run() {
-      System.out.println("HostRequest.InstallHostTask: Executing INSTALL task for host: " + hostname);
+      LOG.info("HostRequest.InstallHostTask: Executing INSTALL task for host: " + hostname);
       RequestStatusResponse response = clusterTopology.installHost(hostname);
       // map logical install tasks to physical install tasks
       List<ShortTaskStatus> underlyingTasks = response.getTasks();
@@ -489,6 +489,8 @@ public class HostRequest implements Comparable<HostRequest> {
           }
         }
       }
+
+      LOG.info("HostRequest.InstallHostTask: Exiting INSTALL task for host: " + hostname);
     }
   }
 
@@ -508,7 +510,7 @@ public class HostRequest implements Comparable<HostRequest> {
 
     @Override
     public void run() {
-      System.out.println("HostRequest.StartHostTask: Executing START task for host: " + hostname);
+      LOG.info("HostRequest.StartHostTask: Executing START task for host: " + hostname);
       RequestStatusResponse response = clusterTopology.startHost(hostname);
       // map logical install tasks to physical install tasks
       List<ShortTaskStatus> underlyingTasks = response.getTasks();
@@ -526,6 +528,8 @@ public class HostRequest implements Comparable<HostRequest> {
           }
         }
       }
+
+      LOG.info("HostRequest.StartHostTask: Exiting START task for host: " + hostname);
     }
   }
 

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/BlueprintConfigurationProcessorTest.java

@@ -1764,13 +1764,24 @@ public class BlueprintConfigurationProcessorTest {
     ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
     BlueprintConfigurationProcessor updater = new BlueprintConfigurationProcessor(topology);
 
-    updater.doUpdateForClusterCreate();
+    Set<String> configTypesUpdated =
+      updater.doUpdateForClusterCreate();
+
     String updatedVal = properties.get("yarn-site").get("yarn.resourcemanager.hostname");
     assertEquals("testhost", updatedVal);
     String updatedVal1 = properties.get("oozie-env").get("oozie_heapsize");
     assertEquals("1024m", updatedVal1);
     String updatedVal2 = properties.get("oozie-env").get("oozie_permsize");
     assertEquals("128m", updatedVal2);
+
+    assertEquals("Incorrect number of config types updated",
+      3, configTypesUpdated.size());
+    assertTrue("Expected config type not updated",
+      configTypesUpdated.contains("oozie-env"));
+    assertTrue("Expected config type not updated",
+      configTypesUpdated.contains("yarn-site"));
+    assertTrue("Expected config type not updated",
+      configTypesUpdated.contains("cluster-env"));
   }
 
   @Test
@@ -2190,7 +2201,8 @@ public class BlueprintConfigurationProcessorTest {
 
     BlueprintConfigurationProcessor updater = new BlueprintConfigurationProcessor(topology);
 
-    updater.doUpdateForClusterCreate();
+    Set<String> updatedConfigTypes =
+      updater.doUpdateForClusterCreate();
 
     // after update, verify that the retry properties for commands and installs are set as expected
     assertEquals("Incorrect number of properties added to cluster-env for retry",
@@ -2201,6 +2213,12 @@ public class BlueprintConfigurationProcessorTest {
       "INSTALL,START", clusterEnvProperties.get("commands_to_retry"));
     assertEquals("command_retry_max_time_in_sec was not set to the expected default",
       "600", clusterEnvProperties.get("command_retry_max_time_in_sec"));
+
+    assertEquals("Incorrect number of config types updated by this operation",
+      1, updatedConfigTypes.size());
+
+    assertTrue("Expected type not included in the updated set",
+      updatedConfigTypes.contains("cluster-env"));
   }
 
   @Test
@@ -2223,7 +2241,8 @@ public class BlueprintConfigurationProcessorTest {
 
     BlueprintConfigurationProcessor updater = new BlueprintConfigurationProcessor(topology);
 
-    updater.doUpdateForClusterCreate();
+    Set<String> updatedConfigTypes =
+      updater.doUpdateForClusterCreate();
 
     // after update, verify that the retry properties for commands and installs are set as expected
     // in this case, the customer-provided overrides should be honored, rather than the retry defaults
@@ -2235,6 +2254,9 @@ public class BlueprintConfigurationProcessorTest {
       "TEST", clusterEnvProperties.get("commands_to_retry"));
     assertEquals("command_retry_max_time_in_sec was not set to the expected default",
       "1", clusterEnvProperties.get("command_retry_max_time_in_sec"));
+
+    assertEquals("Incorrect number of config types updated",
+      0, updatedConfigTypes.size());
   }
 
   @Test
@@ -4405,7 +4427,8 @@ public class BlueprintConfigurationProcessorTest {
     ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
     BlueprintConfigurationProcessor updater = new BlueprintConfigurationProcessor(topology);
 
-    updater.doUpdateForClusterCreate();
+    Set<String> updatedConfigTypes =
+      updater.doUpdateForClusterCreate();
 
     // verify that the expected hostname was substituted for the host group name in the config
     assertEquals("HTTPS address HA property not properly exported",
@@ -4456,13 +4479,22 @@ public class BlueprintConfigurationProcessorTest {
     assertFalse("dfs.namenode.rpc-address should have been filtered out of this HA configuration",
       hdfsSiteProperties.containsKey("dfs.namenode.rpc-address"));
 
+
+    // verify that correct configuration types were listed as updated in the returned set
+    assertEquals("Incorrect number of updated config types returned, set = " + updatedConfigTypes,
+      3, updatedConfigTypes.size());
+    assertTrue("Expected config type not found in updated set",
+      updatedConfigTypes.contains("cluster-env"));
+    assertTrue("Expected config type not found in updated set",
+      updatedConfigTypes.contains("hdfs-site"));
+    assertTrue("Expected config type not found in updated set",
+      updatedConfigTypes.contains("hadoop-env"));
   }
 
   @Test
   public void testDoUpdateForClusterWithNameNodeHANotEnabled() throws Exception {
     final String expectedHostName = "c6401.apache.ambari.org";
     final String expectedHostNameTwo = "serverTwo";
-    final String expectedPortNum = "808080";
     final String expectedHostGroupName = "host_group_1";
 
     Map<String, Map<String, String>> properties = new HashMap<String, Map<String, String>>();
@@ -4511,7 +4543,8 @@ public class BlueprintConfigurationProcessorTest {
     ClusterTopology topology = createClusterTopology(bp, clusterConfig, hostGroups);
     BlueprintConfigurationProcessor updater = new BlueprintConfigurationProcessor(topology);
 
-    updater.doUpdateForClusterCreate();
+    Set<String> updatedConfigTypes =
+      updater.doUpdateForClusterCreate();
 
     // verify that the non-HA properties are not filtered out in a non-HA cluster
     assertTrue("dfs.namenode.http-address should have been included in this HA configuration",
@@ -4521,6 +4554,13 @@ public class BlueprintConfigurationProcessorTest {
     assertTrue("dfs.namenode.rpc-address should have been included in this HA configuration",
       hdfsSiteProperties.containsKey("dfs.namenode.rpc-address"));
 
+    // verify that correct configuration types were listed as updated in the returned set
+    assertEquals("Incorrect number of updated config types returned, set = " + updatedConfigTypes,
+      2, updatedConfigTypes.size());
+    assertTrue("Expected config type 'cluster-env' not found in updated set",
+      updatedConfigTypes.contains("cluster-env"));
+    assertTrue("Expected config type 'hdfs-site' not found in updated set",
+      updatedConfigTypes.contains("hdfs-site"));
   }
 
   @Test

ambari-server/src/test/java/org/apache/ambari/server/topology/AmbariContextTest.java

@@ -48,11 +48,13 @@ import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.Config;
 import org.apache.ambari.server.state.ConfigHelper;
+import org.apache.ambari.server.state.DesiredConfig;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.Service;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.configgroup.ConfigGroup;
 import org.easymock.Capture;
+import org.easymock.EasyMockSupport;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -381,4 +383,48 @@ public class AmbariContextTest {
     // test
     context.registerHostWithConfigGroup(HOST1, topology, HOST_GROUP_1);
   }
+
+  @Test
+  public void testWaitForTopologyResolvedStateWithEmptyUpdatedSet() throws Exception {
+    replayAll();
+
+    // verify that wait returns successfully with empty updated list passed in
+    context.waitForConfigurationResolution(CLUSTER_NAME, Collections.<String>emptySet());
+  }
+
+  @Test
+  public void testWaitForTopologyResolvedStateWithRequiredUpdatedSet() throws Exception {
+    final String topologyResolvedState = "TOPOLOGY_RESOLVED";
+    DesiredConfig testHdfsDesiredConfig =
+      new DesiredConfig();
+    testHdfsDesiredConfig.setTag(topologyResolvedState);
+    DesiredConfig testCoreSiteDesiredConfig =
+      new DesiredConfig();
+    testCoreSiteDesiredConfig.setTag(topologyResolvedState);
+    DesiredConfig testClusterEnvDesiredConfig =
+      new DesiredConfig();
+    testClusterEnvDesiredConfig.setTag(topologyResolvedState);
+
+    Map<String, DesiredConfig> testDesiredConfigs =
+      new HashMap<String, DesiredConfig>();
+    testDesiredConfigs.put("hdfs-site", testHdfsDesiredConfig);
+    testDesiredConfigs.put("core-site", testCoreSiteDesiredConfig);
+    testDesiredConfigs.put("cluster-env", testClusterEnvDesiredConfig);
+
+    expect(cluster.getDesiredConfigs()).andReturn(testDesiredConfigs).atLeastOnce();
+
+    replayAll();
+
+    Set<String> testUpdatedConfigTypes =
+      new HashSet<String>();
+    testUpdatedConfigTypes.add("hdfs-site");
+    testUpdatedConfigTypes.add("core-site");
+    testUpdatedConfigTypes.add("cluster-env");
+
+    // verify that wait returns successfully with non-empty list
+    // with all configuration types tagged as "TOPOLOGY_RESOLVED"
+    context.waitForConfigurationResolution(CLUSTER_NAME, testUpdatedConfigTypes);
+  }
+
+
 }