AMBARI-4182. Reduce number of transactions in HostResourceProvider, HostComponentResourceProvider. (odiachenko)

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

@@ -40,6 +40,7 @@ import org.apache.ambari.server.api.services.PersistKeyValueService;
 import org.apache.ambari.server.bootstrap.BootStrapImpl;
 import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
 import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.controller.internal.AbstractControllerResourceProvider;
 import org.apache.ambari.server.controller.internal.ClusterControllerImpl;
 import org.apache.ambari.server.controller.internal.StackDefinedPropertyProvider;
 import org.apache.ambari.server.orm.GuiceJpaInitializer;
@@ -447,6 +448,7 @@ public class AmbariServer {
         injector.getInstance(Configuration.class));
     SecurityFilter.init(injector.getInstance(Configuration.class));
     StackDefinedPropertyProvider.init(injector);
+    AbstractControllerResourceProvider.init(injector.getInstance(ResourceProviderFactory.class));
   }
 
   public static void main(String[] args) throws Exception {

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

@@ -31,6 +31,9 @@ import org.apache.ambari.server.actionmanager.HostRoleCommandFactory;
 import org.apache.ambari.server.actionmanager.HostRoleCommandFactoryImpl;
 import org.apache.ambari.server.actionmanager.StageFactory;
 import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.controller.internal.HostComponentResourceProvider;
+import org.apache.ambari.server.controller.internal.HostResourceProvider;
+import org.apache.ambari.server.controller.spi.ResourceProvider;
 import org.apache.ambari.server.orm.PersistenceType;
 import org.apache.ambari.server.scheduler.ExecutionScheduler;
 import org.apache.ambari.server.scheduler.ExecutionSchedulerImpl;
@@ -185,6 +188,14 @@ public class ControllerModule extends AbstractModule {
         Host.class, HostImpl.class).build(HostFactory.class));
     install(new FactoryModuleBuilder().implement(
         Service.class, ServiceImpl.class).build(ServiceFactory.class));
+   
+    
+    install(new FactoryModuleBuilder()
+        .implement(ResourceProvider.class, Names.named("host"), HostResourceProvider.class)
+        .implement(ResourceProvider.class, Names.named("hostComponent"), HostComponentResourceProvider.class)
+        .build(ResourceProviderFactory.class)); 
+
+    
     install(new FactoryModuleBuilder().implement(
         ServiceComponent.class, ServiceComponentImpl.class).build(
         ServiceComponentFactory.class));

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

@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.ambari.server.controller;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.ambari.server.controller.spi.Resource.Type;
+import org.apache.ambari.server.controller.spi.ResourceProvider;
+
+import com.google.inject.name.Named;
+
+public interface ResourceProviderFactory {
+  @Named("host")
+  ResourceProvider getHostResourceProvider(Set<String> propertyIds,
+      Map<Type, String> keyPropertyIds,
+      AmbariManagementController managementController);
+
+  @Named("hostComponent")
+  ResourceProvider getHostComponentResourceProvider(Set<String> propertyIds,
+      Map<Type, String> keyPropertyIds,
+      AmbariManagementController managementController);
+
+}

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

@@ -19,6 +19,7 @@
 package org.apache.ambari.server.controller.internal;
 
 import org.apache.ambari.server.controller.AmbariManagementController;
+import org.apache.ambari.server.controller.ResourceProviderFactory;
 import org.apache.ambari.server.controller.predicate.ArrayPredicate;
 import org.apache.ambari.server.controller.predicate.EqualsPredicate;
 import org.apache.ambari.server.controller.spi.Predicate;
@@ -33,6 +34,7 @@ import java.util.Set;
  */
 public abstract class AbstractControllerResourceProvider extends AbstractResourceProvider {
 
+  private static ResourceProviderFactory resourceProviderFactory;
   /**
    * The management controller to delegate to.
    */
@@ -54,6 +56,10 @@ public abstract class AbstractControllerResourceProvider extends AbstractResourc
     super(propertyIds, keyPropertyIds);
     this.managementController = managementController;
   }
+  
+  public static void init(ResourceProviderFactory factory) {
+    resourceProviderFactory = factory;
+  }
 
 
   // ----- accessors ---------------------------------------------------------
@@ -83,6 +89,7 @@ public abstract class AbstractControllerResourceProvider extends AbstractResourc
                                                      Set<String> propertyIds,
                                                      Map<Resource.Type, String> keyPropertyIds,
                                                      AmbariManagementController managementController) {
+
     switch (type) {
       case Cluster:
         return new ClusterResourceProvider(propertyIds, keyPropertyIds, managementController);
@@ -91,9 +98,9 @@ public abstract class AbstractControllerResourceProvider extends AbstractResourc
       case Component:
         return new ComponentResourceProvider(propertyIds, keyPropertyIds, managementController);
       case Host:
-        return new HostResourceProvider(propertyIds, keyPropertyIds, managementController);
+        return resourceProviderFactory.getHostResourceProvider(propertyIds, keyPropertyIds, managementController);
       case HostComponent:
-        return new HostComponentResourceProvider(propertyIds, keyPropertyIds, managementController);
+        return resourceProviderFactory.getHostComponentResourceProvider(propertyIds, keyPropertyIds, managementController);
       case Configuration:
         return new ConfigurationResourceProvider(propertyIds, keyPropertyIds, managementController);
       case Action:

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

@@ -42,10 +42,15 @@ import org.apache.ambari.server.controller.spi.SystemException;
 import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
 
+import com.google.inject.Injector;
+import com.google.inject.assistedinject.Assisted;
+import com.google.inject.assistedinject.AssistedInject;
+import com.google.inject.persist.Transactional;
+
 /**
  * Resource provider for host component resources.
  */
-class HostComponentResourceProvider extends AbstractControllerResourceProvider {
+public class HostComponentResourceProvider extends AbstractControllerResourceProvider {
 
   // ----- Property ID constants ---------------------------------------------
 
@@ -113,9 +118,10 @@ class HostComponentResourceProvider extends AbstractControllerResourceProvider {
    * @param keyPropertyIds        the key property ids
    * @param managementController  the management controller
    */
-  HostComponentResourceProvider(Set<String> propertyIds,
-                                Map<Resource.Type, String> keyPropertyIds,
-                                AmbariManagementController managementController) {
+  @AssistedInject
+  public HostComponentResourceProvider(@Assisted Set<String> propertyIds,
+                                @Assisted Map<Resource.Type, String> keyPropertyIds,
+                                @Assisted AmbariManagementController managementController) {
     super(propertyIds, keyPropertyIds, managementController);
   }
 
@@ -147,6 +153,7 @@ class HostComponentResourceProvider extends AbstractControllerResourceProvider {
   }
 
   @Override
+  @Transactional
   public Set<Resource> getResources(Request request, Predicate predicate)
       throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
 

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

@@ -63,10 +63,16 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 
+import com.google.inject.Injector;
+import com.google.inject.assistedinject.Assisted;
+import com.google.inject.assistedinject.AssistedInject;
+import com.google.inject.persist.Transactional;
+
+
 /**
  * Resource provider for host resources.
  */
-class HostResourceProvider extends AbstractControllerResourceProvider {
+public class HostResourceProvider extends AbstractControllerResourceProvider {
 
   // ----- Property ID constants ---------------------------------------------
 
@@ -97,8 +103,12 @@ class HostResourceProvider extends AbstractControllerResourceProvider {
       PropertyHelper.getPropertyId("Hosts", "last_registration_time");
   protected static final String HOST_DISK_INFO_PROPERTY_ID =
       PropertyHelper.getPropertyId("Hosts", "disk_info");
+  
+  
   protected static final String HOST_HOST_STATUS_PROPERTY_ID =
       PropertyHelper.getPropertyId("Hosts", "host_status");
+  
+  
   protected static final String HOST_HOST_HEALTH_REPORT_PROPERTY_ID =
       PropertyHelper.getPropertyId("Hosts", "host_health_report");
   protected static final String HOST_STATE_PROPERTY_ID =
@@ -121,9 +131,10 @@ class HostResourceProvider extends AbstractControllerResourceProvider {
    * @param keyPropertyIds        the key property ids
    * @param managementController  the management controller
    */
-  HostResourceProvider(Set<String> propertyIds,
-                       Map<Resource.Type, String> keyPropertyIds,
-                       AmbariManagementController managementController) {
+  @AssistedInject
+  HostResourceProvider(@Assisted Set<String> propertyIds,
+                       @Assisted Map<Resource.Type, String> keyPropertyIds,
+                       @Assisted AmbariManagementController managementController) {
     super(propertyIds, keyPropertyIds, managementController);
   }
 
@@ -154,6 +165,7 @@ class HostResourceProvider extends AbstractControllerResourceProvider {
   }
 
   @Override
+  @Transactional
   public Set<Resource> getResources(Request request, Predicate predicate)
       throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
 

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

@@ -20,6 +20,7 @@ package org.apache.ambari.server.controller.internal;
 
 import org.apache.ambari.server.controller.AmbariManagementController;
 import org.apache.ambari.server.controller.RequestStatusResponse;
+import org.apache.ambari.server.controller.ResourceProviderFactory;
 import org.apache.ambari.server.controller.ServiceComponentHostRequest;
 import org.apache.ambari.server.controller.ServiceComponentHostResponse;
 import org.apache.ambari.server.controller.spi.Predicate;
@@ -42,6 +43,7 @@ import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Set;
 
+import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.createNiceMock;
 import static org.easymock.EasyMock.eq;
@@ -59,13 +61,26 @@ public class HostComponentResourceProviderTest {
 
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    HostComponentResourceProvider hostComponentResourceProvider = 
+        new HostComponentResourceProvider(PropertyHelper.getPropertyIds(type),
+        PropertyHelper.getKeyPropertyIds(type),
+        managementController);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
 
     managementController.createHostComponents(
         AbstractResourceProviderTest.Matcher.getHostComponentRequestSet(
             "Cluster100", "Service100", "Component100", "Host100", null, null));
+    
+    expect(resourceProviderFactory.getHostComponentResourceProvider(anyObject(Set.class),
+        anyObject(Map.class),
+        eq(managementController))).
+        andReturn(hostComponentResourceProvider).anyTimes();
+    
 
     // replay
-    replay(managementController, response);
+    replay(managementController, response, resourceProviderFactory);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -93,7 +108,7 @@ public class HostComponentResourceProviderTest {
     provider.createResources(request);
 
     // verify
-    verify(managementController, response);
+    verify(managementController, response, resourceProviderFactory);
   }
 
   @Test
@@ -101,6 +116,10 @@ public class HostComponentResourceProviderTest {
     Resource.Type type = Resource.Type.HostComponent;
 
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostComponentResourceProvider = createNiceMock(HostComponentResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
 
     Set<ServiceComponentHostResponse> allResponse = new HashSet<ServiceComponentHostResponse>();
     StackId stackId = new StackId("HDP-0.1");
@@ -127,19 +146,11 @@ public class HostComponentResourceProviderTest {
         HostComponentResourceProvider.HOST_COMPONENT_DESIRED_STACK_ID_PROPERTY_ID, stackId2.getStackId());
 
     // set expectations
-    expect(managementController.getHostComponents(
-        AbstractResourceProviderTest.Matcher.getHostComponentRequestSet(
-            "Cluster100", null, null, null, null, null))).andReturn(allResponse).once();
-
-    // replay
-    replay(managementController);
-
-    ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
-        type,
-        PropertyHelper.getPropertyIds(type),
-        PropertyHelper.getKeyPropertyIds(type),
-        managementController);
-
+    expect(resourceProviderFactory.getHostComponentResourceProvider(anyObject(Set.class),
+        anyObject(Map.class),
+        eq(managementController))).
+        andReturn(hostComponentResourceProvider).anyTimes();
+    
     Set<String> propertyIds = new HashSet<String>();
 
     propertyIds.add(HostComponentResourceProvider.HOST_COMPONENT_CLUSTER_NAME_PROPERTY_ID);
@@ -152,6 +163,52 @@ public class HostComponentResourceProviderTest {
     Predicate predicate = new PredicateBuilder().property(
         HostComponentResourceProvider.HOST_COMPONENT_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").toPredicate();
     Request request = PropertyHelper.getReadRequest(propertyIds);
+    
+    Set<Resource> hostsComponentResources = new HashSet<Resource>();
+    
+    Resource hostsComponentResource1 = new ResourceImpl(Resource.Type.HostComponent);
+    hostsComponentResource1.setProperty(HostComponentResourceProvider.HOST_COMPONENT_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
+    hostsComponentResource1.setProperty(HostComponentResourceProvider.HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "Host100");
+    hostsComponentResource1.setProperty(HostComponentResourceProvider.HOST_COMPONENT_SERVICE_NAME_PROPERTY_ID, "Service100");
+    hostsComponentResource1.setProperty(HostComponentResourceProvider.HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "Component100");
+    hostsComponentResource1.setProperty(HostComponentResourceProvider.HOST_COMPONENT_STATE_PROPERTY_ID, State.INSTALLED.name());
+    hostsComponentResource1.setProperty(HostComponentResourceProvider.HOST_COMPONENT_DESIRED_STATE_PROPERTY_ID, State.STARTED.name());
+    hostsComponentResource1.setProperty(HostComponentResourceProvider.HOST_COMPONENT_STACK_ID_PROPERTY_ID, stackId.getStackId());
+    hostsComponentResource1.setProperty(HostComponentResourceProvider.HOST_COMPONENT_DESIRED_STACK_ID_PROPERTY_ID, stackId2.getStackId());
+    Resource hostsComponentResource2 = new ResourceImpl(Resource.Type.HostComponent);
+    hostsComponentResource2.setProperty(HostComponentResourceProvider.HOST_COMPONENT_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
+    hostsComponentResource2.setProperty(HostComponentResourceProvider.HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "Host100");
+    hostsComponentResource2.setProperty(HostComponentResourceProvider.HOST_COMPONENT_SERVICE_NAME_PROPERTY_ID, "Service100");
+    hostsComponentResource2.setProperty(HostComponentResourceProvider.HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "Component101");
+    hostsComponentResource2.setProperty(HostComponentResourceProvider.HOST_COMPONENT_STATE_PROPERTY_ID, State.INSTALLED.name());
+    hostsComponentResource2.setProperty(HostComponentResourceProvider.HOST_COMPONENT_DESIRED_STATE_PROPERTY_ID, State.STARTED.name());
+    hostsComponentResource2.setProperty(HostComponentResourceProvider.HOST_COMPONENT_STACK_ID_PROPERTY_ID, stackId.getStackId());
+    hostsComponentResource2.setProperty(HostComponentResourceProvider.HOST_COMPONENT_DESIRED_STACK_ID_PROPERTY_ID, stackId2.getStackId());
+    Resource hostsComponentResource3 = new ResourceImpl(Resource.Type.HostComponent);
+    hostsComponentResource3.setProperty(HostComponentResourceProvider.HOST_COMPONENT_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
+    hostsComponentResource3.setProperty(HostComponentResourceProvider.HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "Host100");
+    hostsComponentResource3.setProperty(HostComponentResourceProvider.HOST_COMPONENT_SERVICE_NAME_PROPERTY_ID, "Service100");
+    hostsComponentResource3.setProperty(HostComponentResourceProvider.HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "Component102");
+    hostsComponentResource3.setProperty(HostComponentResourceProvider.HOST_COMPONENT_STATE_PROPERTY_ID, State.INSTALLED.name());
+    hostsComponentResource3.setProperty(HostComponentResourceProvider.HOST_COMPONENT_DESIRED_STATE_PROPERTY_ID, State.STARTED.name());
+    hostsComponentResource3.setProperty(HostComponentResourceProvider.HOST_COMPONENT_STACK_ID_PROPERTY_ID, stackId.getStackId());
+    hostsComponentResource3.setProperty(HostComponentResourceProvider.HOST_COMPONENT_DESIRED_STACK_ID_PROPERTY_ID, stackId2.getStackId());
+    hostsComponentResources.add(hostsComponentResource1);
+    hostsComponentResources.add(hostsComponentResource2);
+    hostsComponentResources.add(hostsComponentResource3);
+    
+    expect(hostComponentResourceProvider.getResources(eq(request), eq(predicate))).andReturn(hostsComponentResources).anyTimes();
+
+    // replay
+    replay(managementController, resourceProviderFactory, hostComponentResourceProvider);
+
+    ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
+        type,
+        PropertyHelper.getPropertyIds(type),
+        PropertyHelper.getKeyPropertyIds(type),
+        managementController);
+
+
     Set<Resource> resources = provider.getResources(request, predicate);
 
     Assert.assertEquals(3, resources.size());
@@ -169,7 +226,7 @@ public class HostComponentResourceProviderTest {
     }
 
     // verify
-    verify(managementController);
+    verify(managementController, resourceProviderFactory, hostComponentResourceProvider);
   }
 
   @Test
@@ -178,6 +235,7 @@ public class HostComponentResourceProviderTest {
 
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
 
     Map<String, String> mapRequestProps = new HashMap<String, String>();
     mapRequestProps.put("context", "Called from a test");
@@ -185,6 +243,11 @@ public class HostComponentResourceProviderTest {
     Set<ServiceComponentHostResponse> nameResponse = new HashSet<ServiceComponentHostResponse>();
     nameResponse.add(new ServiceComponentHostResponse(
         "Cluster102", "Service100", "Component100", "Host100", "STARTED", "", "", ""));
+    
+    HostComponentResourceProvider provider = 
+        new HostComponentResourceProvider(PropertyHelper.getPropertyIds(type),
+        PropertyHelper.getKeyPropertyIds(type),
+        managementController);
 
     // set expectations
     expect(managementController.getHostComponents(
@@ -193,15 +256,14 @@ public class HostComponentResourceProviderTest {
         AbstractResourceProviderTest.Matcher.getHostComponentRequestSet(
             "Cluster102", null, "Component100", "Host100", null, "STARTED"),
             eq(mapRequestProps), eq(false))).andReturn(response).once();
+    
+    expect(resourceProviderFactory.getHostComponentResourceProvider(anyObject(Set.class),
+        anyObject(Map.class),
+        eq(managementController))).
+        andReturn(provider).anyTimes();
 
     // replay
-    replay(managementController, response);
-
-    ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
-        type,
-        PropertyHelper.getPropertyIds(type),
-        PropertyHelper.getKeyPropertyIds(type),
-        managementController);
+    replay(managementController, response, resourceProviderFactory);
 
     Map<String, Object> properties = new LinkedHashMap<String, Object>();
 
@@ -216,7 +278,7 @@ public class HostComponentResourceProviderTest {
     provider.updateResources(request, predicate);
 
     // verify
-    verify(managementController, response);
+    verify(managementController, response, resourceProviderFactory);
   }
 
   @Test
@@ -225,6 +287,11 @@ public class HostComponentResourceProviderTest {
 
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
+    
+    HostComponentResourceProvider provider = 
+        new HostComponentResourceProvider(PropertyHelper.getPropertyIds(type),
+        PropertyHelper.getKeyPropertyIds(type),
+        managementController);
 
     // set expectations
     expect(managementController.deleteHostComponents(
@@ -234,12 +301,6 @@ public class HostComponentResourceProviderTest {
     // replay
     replay(managementController, response);
 
-    ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
-        type,
-        PropertyHelper.getPropertyIds(type),
-        PropertyHelper.getKeyPropertyIds(type),
-        managementController);
-
     AbstractResourceProviderTest.TestObserver observer = new AbstractResourceProviderTest.TestObserver();
 
     ((ObservableResourceProvider)provider).addObserver(observer);
@@ -276,12 +337,10 @@ public class HostComponentResourceProviderTest {
 
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
 
-    AbstractResourceProvider provider =
-        (AbstractResourceProvider) AbstractControllerResourceProvider.getResourceProvider(
-            Resource.Type.HostComponent,
-            propertyIds,
-            keyPropertyIds,
-            managementController);
+    HostComponentResourceProvider provider = 
+        new HostComponentResourceProvider(propertyIds,
+        keyPropertyIds,
+        managementController);
 
     Set<String> unsupported = provider.checkPropertyIds(Collections.singleton("foo"));
     Assert.assertTrue(unsupported.isEmpty());

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

@@ -28,6 +28,7 @@ import org.apache.ambari.server.controller.AmbariManagementControllerImpl;
 import org.apache.ambari.server.controller.HostResponse;
 import org.apache.ambari.server.controller.RequestStatusResponse;
 import org.apache.ambari.server.controller.HostRequest;
+import org.apache.ambari.server.controller.ResourceProviderFactory;
 import org.apache.ambari.server.controller.ServiceComponentHostRequest;
 import org.apache.ambari.server.controller.ServiceComponentHostResponse;
 import org.apache.ambari.server.controller.spi.Predicate;
@@ -57,6 +58,7 @@ import java.util.Map;
 import java.util.Set;
 
 import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.eq;
 import static org.easymock.EasyMock.capture;
 import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.createNiceMock;
@@ -80,11 +82,15 @@ public class HostResourceProviderTest {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     Clusters clusters = createNiceMock(Clusters.class);
     RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
-
-    expect(managementController.getClusters()).andReturn(clusters);
-
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
+    
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class), anyObject(Map.class),
+        eq(managementController))).andReturn(hostResourceProvider).anyTimes();
     // replay
-    replay(managementController, response, clusters);
+    replay(managementController, response, clusters, resourceProviderFactory, hostResourceProvider);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -124,10 +130,10 @@ public class HostResourceProviderTest {
     Host host3 = createNiceMock(Host.class);
     HostHealthStatus healthStatus = createNiceMock(HostHealthStatus.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
-
-    HostResponse hostResponse1 = createNiceMock(HostResponse.class);
-    HostResponse hostResponse2 = createNiceMock(HostResponse.class);
-    HostResponse hostResponse3 = createNiceMock(HostResponse.class);
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
 
     List<Host> hosts = new LinkedList<Host>();
     hosts.add(host1);
@@ -141,6 +147,10 @@ public class HostResourceProviderTest {
     expect(managementController.getClusters()).andReturn(clusters).anyTimes();
     expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
     expect(managementController.getHostComponents((Set<ServiceComponentHostRequest>) anyObject())).andReturn(Collections.<ServiceComponentHostResponse>emptySet()).anyTimes();
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class),
+                                                           anyObject(Map.class),
+                                                           eq(managementController))).
+                                                           andReturn(hostResourceProvider).anyTimes();
 
     expect(clusters.getHosts()).andReturn(hosts).anyTimes();
 
@@ -154,28 +164,41 @@ public class HostResourceProviderTest {
     expect(host2.getHostName()).andReturn("Host101").anyTimes();
     expect(host3.getHostName()).andReturn("Host102").anyTimes();
 
-    expect(host1.convertToResponse()).andReturn(hostResponse1);
-    expect(host2.convertToResponse()).andReturn(hostResponse2);
-    expect(host3.convertToResponse()).andReturn(hostResponse3);
-
-    expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
-    expect(hostResponse2.getClusterName()).andReturn("Cluster100").anyTimes();
-    expect(hostResponse3.getClusterName()).andReturn("Cluster100").anyTimes();
-    expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
-    expect(hostResponse2.getHostname()).andReturn("Host101").anyTimes();
-    expect(hostResponse3.getHostname()).andReturn("Host102").anyTimes();
-    expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
-    expect(hostResponse2.getHealthStatus()).andReturn(healthStatus).anyTimes();
-    expect(hostResponse3.getHealthStatus()).andReturn(healthStatus).anyTimes();
-
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();
+    
+    Set<String> propertyIds = new HashSet<String>();
+
+    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
+
+    Predicate predicate =
+        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
+            toPredicate();
+    Request request = PropertyHelper.getReadRequest(propertyIds);
+    
+    Set<Resource> hostsResources = new HashSet<Resource>();
+    
+    Resource hostResource1 = new ResourceImpl(Resource.Type.Host);
+    hostResource1.setProperty(PropertyHelper.getPropertyId("Hosts", "cluster_name"), "Cluster100");
+    hostResource1.setProperty(PropertyHelper.getPropertyId("Hosts", "host_name"), "Host100");
+    Resource hostResource2 = new ResourceImpl(Resource.Type.Host);
+    hostResource2.setProperty(PropertyHelper.getPropertyId("Hosts", "cluster_name"), "Cluster100");
+    hostResource2.setProperty(PropertyHelper.getPropertyId("Hosts", "host_name"), "Host101");
+    Resource hostResource3 = new ResourceImpl(Resource.Type.Host);
+    hostResource3.setProperty(PropertyHelper.getPropertyId("Hosts", "cluster_name"), "Cluster100");
+    hostResource3.setProperty(PropertyHelper.getPropertyId("Hosts", "host_name"), "Host102");
+    hostsResources.add(hostResource1);
+    hostsResources.add(hostResource2);
+    hostsResources.add(hostResource3);
+    
+    expect(hostResourceProvider.getResources(eq(request), eq(predicate))).andReturn(hostsResources).anyTimes();
+    
 
     // replay
     replay(managementController, clusters, cluster,
         host1, host2, host3,
-        hostResponse1, hostResponse2, hostResponse3,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -183,15 +206,7 @@ public class HostResourceProviderTest {
         PropertyHelper.getKeyPropertyIds(type),
         managementController);
 
-    Set<String> propertyIds = new HashSet<String>();
-
-    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
-
-    Predicate predicate =
-        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
-            toPredicate();
-    Request request = PropertyHelper.getReadRequest(propertyIds);
+    
     Set<Resource> resources = provider.getResources(request, predicate);
 
     Assert.assertEquals(3, resources.size());
@@ -203,7 +218,6 @@ public class HostResourceProviderTest {
     // verify
     verify(managementController, clusters, cluster,
         host1, host2, host3,
-        hostResponse1, hostResponse2, hostResponse3,
         healthStatus, ambariMetaInfo);
   }
 
@@ -214,15 +228,16 @@ public class HostResourceProviderTest {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     Clusters clusters = createNiceMock(Clusters.class);
     Cluster cluster = createNiceMock(Cluster.class);
-    Host host1 = createNiceMock(Host.class);
     HostHealthStatus healthStatus = createNiceMock(HostHealthStatus.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
     ComponentInfo componentInfo = createNiceMock(ComponentInfo.class);
 
     HostResponse hostResponse1 = createNiceMock(HostResponse.class);
-
-    List<Host> hosts = new LinkedList<Host>();
-    hosts.add(host1);
+    
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
 
     Set<Cluster> clusterSet = new HashSet<Cluster>();
     clusterSet.add(cluster);
@@ -241,15 +256,10 @@ public class HostResourceProviderTest {
     expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
     expect(managementController.getHostComponents((Set<ServiceComponentHostRequest>) anyObject())).andReturn(responses).anyTimes();
 
-    expect(clusters.getHosts()).andReturn(hosts).anyTimes();
-
     expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
 
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
 
-    expect(host1.getHostName()).andReturn("Host100").anyTimes();
-
-    expect(host1.convertToResponse()).andReturn(hostResponse1);
 
     expect(hostResponse1.getClusterName()).andReturn("").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
@@ -264,13 +274,36 @@ public class HostResourceProviderTest {
         (String) anyObject(), (String) anyObject())).andReturn(componentInfo).anyTimes();
 
     expect(componentInfo.getCategory()).andReturn("MASTER").anyTimes();
+    
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class),
+                                                               anyObject(Map.class),
+                                                               eq(managementController))).
+                                                               andReturn(hostResourceProvider).anyTimes();
+    
+    Set<String> propertyIds = new HashSet<String>();
+
+    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
+
+    Predicate predicate =
+        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
+            toPredicate();
+    Request request = PropertyHelper.getReadRequest(propertyIds);
+    
+    Set<Resource> hostsResources = new HashSet<Resource>();
+    
+    Resource hostResource1 = new ResourceImpl(Resource.Type.Host);
+    hostResource1.setProperty(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID, HealthStatus.HEALTHY.name());
+    hostsResources.add(hostResource1);
+    
+    expect(hostResourceProvider.getResources(eq(request), eq(predicate))).andReturn(hostsResources).anyTimes();
 
 
     // replay
     replay(managementController, clusters, cluster,
-        host1,
         hostResponse1, componentInfo,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -278,16 +311,6 @@ public class HostResourceProviderTest {
         PropertyHelper.getKeyPropertyIds(type),
         managementController);
 
-    Set<String> propertyIds = new HashSet<String>();
-
-    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
-
-    Predicate predicate =
-        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
-            toPredicate();
-    Request request = PropertyHelper.getReadRequest(propertyIds);
     Set<Resource> resources = provider.getResources(request, predicate);
 
     Assert.assertEquals(1, resources.size());
@@ -300,7 +323,6 @@ public class HostResourceProviderTest {
 
     // verify
     verify(managementController, clusters, cluster,
-        host1,
         hostResponse1, componentInfo,
         healthStatus, ambariMetaInfo);
   }
@@ -312,16 +334,17 @@ public class HostResourceProviderTest {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     Clusters clusters = createNiceMock(Clusters.class);
     Cluster cluster = createNiceMock(Cluster.class);
-    Host host1 = createNiceMock(Host.class);
     HostHealthStatus healthStatus = createNiceMock(HostHealthStatus.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
     StackId stackId = createNiceMock(StackId.class);
     ComponentInfo componentInfo = createNiceMock(ComponentInfo.class);
 
     HostResponse hostResponse1 = createNiceMock(HostResponse.class);
-
-    List<Host> hosts = new LinkedList<Host>();
-    hosts.add(host1);
+    
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
 
     Set<Cluster> clusterSet = new HashSet<Cluster>();
     clusterSet.add(cluster);
@@ -340,16 +363,10 @@ public class HostResourceProviderTest {
     expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
     expect(managementController.getHostComponents((Set<ServiceComponentHostRequest>) anyObject())).andReturn(responses).anyTimes();
 
-    expect(clusters.getHosts()).andReturn(hosts).anyTimes();
-
     expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
 
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
 
-    expect(host1.getHostName()).andReturn("Host100").anyTimes();
-
-    expect(host1.convertToResponse()).andReturn(hostResponse1);
-
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
@@ -363,13 +380,36 @@ public class HostResourceProviderTest {
         (String) anyObject(), (String) anyObject())).andReturn(componentInfo).anyTimes();
 
     expect(componentInfo.getCategory()).andReturn("MASTER").anyTimes();
+    
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class), anyObject(Map.class),
+        eq(managementController))).andReturn(hostResourceProvider).anyTimes();
+    
+    Set<String> propertyIds = new HashSet<String>();
+
+    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
+
+    Predicate predicate =
+        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
+            toPredicate();
+    Request request = PropertyHelper.getReadRequest(propertyIds);
+    
+    
+    Set<Resource> hostsResources = new HashSet<Resource>();
+       
+    Resource hostResource1 = new ResourceImpl(Resource.Type.Host);
+    hostResource1.setProperty(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
+    hostResource1.setProperty(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID, HealthStatus.HEALTHY.name());
+    hostsResources.add(hostResource1);
+    
+    expect(hostResourceProvider.getResources(eq(request), eq(predicate))).andReturn(hostsResources).anyTimes();
 
 
     // replay
     replay(managementController, clusters, cluster,
-        host1,
         hostResponse1, stackId, componentInfo,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -377,16 +417,7 @@ public class HostResourceProviderTest {
         PropertyHelper.getKeyPropertyIds(type),
         managementController);
 
-    Set<String> propertyIds = new HashSet<String>();
 
-    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
-
-    Predicate predicate =
-        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
-            toPredicate();
-    Request request = PropertyHelper.getReadRequest(propertyIds);
     Set<Resource> resources = provider.getResources(request, predicate);
 
     Assert.assertEquals(1, resources.size());
@@ -399,9 +430,8 @@ public class HostResourceProviderTest {
 
     // verify
     verify(managementController, clusters, cluster,
-        host1,
         hostResponse1, stackId, componentInfo,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
   }
 
   @Test
@@ -411,17 +441,16 @@ public class HostResourceProviderTest {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     Clusters clusters = createNiceMock(Clusters.class);
     Cluster cluster = createNiceMock(Cluster.class);
-    Host host1 = createNiceMock(Host.class);
     HostHealthStatus healthStatus = createNiceMock(HostHealthStatus.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
     StackId stackId = createNiceMock(StackId.class);
     ComponentInfo componentInfo = createNiceMock(ComponentInfo.class);
-
     HostResponse hostResponse1 = createNiceMock(HostResponse.class);
-
-    List<Host> hosts = new LinkedList<Host>();
-    hosts.add(host1);
-
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
+    
     Set<Cluster> clusterSet = new HashSet<Cluster>();
     clusterSet.add(cluster);
 
@@ -439,16 +468,10 @@ public class HostResourceProviderTest {
     expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
     expect(managementController.getHostComponents((Set<ServiceComponentHostRequest>) anyObject())).andReturn(responses).anyTimes();
 
-    expect(clusters.getHosts()).andReturn(hosts).anyTimes();
-
     expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
 
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
 
-    expect(host1.getHostName()).andReturn("Host100").anyTimes();
-
-    expect(host1.convertToResponse()).andReturn(hostResponse1);
-
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
@@ -462,13 +485,36 @@ public class HostResourceProviderTest {
         (String) anyObject(), (String) anyObject())).andReturn(componentInfo).anyTimes();
 
     expect(componentInfo.getCategory()).andReturn("MASTER").anyTimes();
+    
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class), anyObject(Map.class),
+        eq(managementController))).andReturn(hostResourceProvider).anyTimes();
+    
+    Set<String> propertyIds = new HashSet<String>();
+
+    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
+
+    Predicate predicate =
+        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
+            toPredicate();
+    Request request = PropertyHelper.getReadRequest(propertyIds);
+    
+    
+    Set<Resource> hostsResources = new HashSet<Resource>();
+    
+    Resource hostResource1 = new ResourceImpl(Resource.Type.Host);
+    hostResource1.setProperty(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
+    hostResource1.setProperty(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID, HealthStatus.UNHEALTHY.name());
+    hostsResources.add(hostResource1);
+    
+    expect(hostResourceProvider.getResources(eq(request), eq(predicate))).andReturn(hostsResources).anyTimes();
 
 
     // replay
     replay(managementController, clusters, cluster,
-        host1,
         hostResponse1, stackId, componentInfo,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -476,16 +522,7 @@ public class HostResourceProviderTest {
         PropertyHelper.getKeyPropertyIds(type),
         managementController);
 
-    Set<String> propertyIds = new HashSet<String>();
-
-    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
 
-    Predicate predicate =
-        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
-            toPredicate();
-    Request request = PropertyHelper.getReadRequest(propertyIds);
     Set<Resource> resources = provider.getResources(request, predicate);
 
     Assert.assertEquals(1, resources.size());
@@ -498,9 +535,8 @@ public class HostResourceProviderTest {
 
     // verify
     verify(managementController, clusters, cluster,
-        host1,
         hostResponse1, stackId, componentInfo,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
   }
 
   @Test
@@ -510,15 +546,14 @@ public class HostResourceProviderTest {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     Clusters clusters = createNiceMock(Clusters.class);
     Cluster cluster = createNiceMock(Cluster.class);
-    Host host1 = createNiceMock(Host.class);
     HostHealthStatus healthStatus = createNiceMock(HostHealthStatus.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
-
     HostResponse hostResponse1 = createNiceMock(HostResponse.class);
-
-    List<Host> hosts = new LinkedList<Host>();
-    hosts.add(host1);
-
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
+    
     Set<Cluster> clusterSet = new HashSet<Cluster>();
     clusterSet.add(cluster);
 
@@ -526,16 +561,10 @@ public class HostResourceProviderTest {
     expect(managementController.getClusters()).andReturn(clusters).anyTimes();
     expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
 
-    expect(clusters.getHosts()).andReturn(hosts).anyTimes();
-
     expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
 
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
-
-    expect(host1.getHostName()).andReturn("Host100").anyTimes();
-
-    expect(host1.convertToResponse()).andReturn(hostResponse1);
-
+    
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
@@ -543,13 +572,33 @@ public class HostResourceProviderTest {
 
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.UNKNOWN).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("UNKNOWN").anyTimes();
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class), anyObject(Map.class),
+        eq(managementController))).andReturn(hostResourceProvider).anyTimes();
+    
+    Set<Resource> hostsResources = new HashSet<Resource>();
+    
+    Resource hostResource1 = new ResourceImpl(Resource.Type.Host);
+    hostResource1.setProperty(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
+    hostResource1.setProperty(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID, HealthStatus.UNKNOWN.name());
+    hostsResources.add(hostResource1);
+    
+    Set<String> propertyIds = new HashSet<String>();
 
+    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
+
+    Predicate predicate =
+        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
+            toPredicate();
+    Request request = PropertyHelper.getReadRequest(propertyIds);
+    
+    expect(hostResourceProvider.getResources(eq(request), eq(predicate))).andReturn(hostsResources).anyTimes();
 
     // replay
     replay(managementController, clusters, cluster,
-        host1,
         hostResponse1,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -557,16 +606,6 @@ public class HostResourceProviderTest {
         PropertyHelper.getKeyPropertyIds(type),
         managementController);
 
-    Set<String> propertyIds = new HashSet<String>();
-
-    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
-
-    Predicate predicate =
-        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
-            toPredicate();
-    Request request = PropertyHelper.getReadRequest(propertyIds);
     Set<Resource> resources = provider.getResources(request, predicate);
 
     Assert.assertEquals(1, resources.size());
@@ -579,9 +618,8 @@ public class HostResourceProviderTest {
 
     // verify
     verify(managementController, clusters, cluster,
-        host1,
         hostResponse1,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
   }
 
   @Test
@@ -591,16 +629,15 @@ public class HostResourceProviderTest {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     Clusters clusters = createNiceMock(Clusters.class);
     Cluster cluster = createNiceMock(Cluster.class);
-    Host host1 = createNiceMock(Host.class);
     HostHealthStatus healthStatus = createNiceMock(HostHealthStatus.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
     StackId stackId = createNiceMock(StackId.class);
     ComponentInfo componentInfo = createNiceMock(ComponentInfo.class);
-
     HostResponse hostResponse1 = createNiceMock(HostResponse.class);
-
-    List<Host> hosts = new LinkedList<Host>();
-    hosts.add(host1);
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
 
     Set<Cluster> clusterSet = new HashSet<Cluster>();
     clusterSet.add(cluster);
@@ -618,37 +655,46 @@ public class HostResourceProviderTest {
     expect(managementController.getClusters()).andReturn(clusters).anyTimes();
     expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
     expect(managementController.getHostComponents((Set<ServiceComponentHostRequest>) anyObject())).andReturn(responses).anyTimes();
-
-    expect(clusters.getHosts()).andReturn(hosts).anyTimes();
-
     expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
-
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
-
-    expect(host1.getHostName()).andReturn("Host100").anyTimes();
-
-    expect(host1.convertToResponse()).andReturn(hostResponse1);
-
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
     expect(hostResponse1.getStatus()).andReturn(HealthStatus.ALERT.name()).anyTimes();
-
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();
-
-
     expect(ambariMetaInfo.getComponentCategory((String) anyObject(), (String) anyObject(),
         (String) anyObject(), (String) anyObject())).andReturn(componentInfo).anyTimes();
-
     expect(componentInfo.getCategory()).andReturn("SLAVE").anyTimes();
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class), anyObject(Map.class),
+        eq(managementController))).andReturn(hostResourceProvider).anyTimes();
+    
+    
+    Set<String> propertyIds = new HashSet<String>();
+
+    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
+    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
+
+    Predicate predicate =
+        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
+            toPredicate();
+    Request request = PropertyHelper.getReadRequest(propertyIds);
+    
+    Set<Resource> hostsResources = new HashSet<Resource>();
+    
+    Resource hostResource1 = new ResourceImpl(Resource.Type.Host);
+    hostResource1.setProperty(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
+    hostResource1.setProperty(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID, HealthStatus.ALERT.name());
+    hostsResources.add(hostResource1);
+    
+    expect(hostResourceProvider.getResources(eq(request), eq(predicate))).andReturn(hostsResources).anyTimes();
 
 
     // replay
     replay(managementController, clusters, cluster,
-        host1,
         hostResponse1, stackId, componentInfo,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -656,16 +702,7 @@ public class HostResourceProviderTest {
         PropertyHelper.getKeyPropertyIds(type),
         managementController);
 
-    Set<String> propertyIds = new HashSet<String>();
-
-    propertyIds.add(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_NAME_PROPERTY_ID);
-    propertyIds.add(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID);
-
-    Predicate predicate =
-        new PredicateBuilder().property(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID).equals("Cluster100").
-            toPredicate();
-    Request request = PropertyHelper.getReadRequest(propertyIds);
+    
     Set<Resource> resources = provider.getResources(request, predicate);
 
     Assert.assertEquals(1, resources.size());
@@ -678,9 +715,8 @@ public class HostResourceProviderTest {
 
     // verify
     verify(managementController, clusters, cluster,
-        host1,
         hostResponse1, stackId, componentInfo,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
   }
 
   @Test
@@ -689,14 +725,13 @@ public class HostResourceProviderTest {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     Clusters clusters = createNiceMock(Clusters.class);
     Cluster cluster = createNiceMock(Cluster.class);
-    Host host1 = createNiceMock(Host.class);
     HostHealthStatus healthStatus = createNiceMock(HostHealthStatus.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
-
     HostResponse hostResponse1 = createNiceMock(HostResponse.class);
-
-    List<Host> hosts = new LinkedList<Host>();
-    hosts.add(host1);
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
 
     Set<Cluster> clusterSet = new HashSet<Cluster>();
     clusterSet.add(cluster);
@@ -705,36 +740,25 @@ public class HostResourceProviderTest {
     expect(managementController.getClusters()).andReturn(clusters).anyTimes();
     expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
     expect(managementController.getHostComponents((Set<ServiceComponentHostRequest>) anyObject())).andReturn(Collections.<ServiceComponentHostResponse>emptySet()).anyTimes();
-
-    expect(clusters.getHosts()).andReturn(hosts).anyTimes();
-    expect(clusters.getHost("Host100")).andReturn(host1).anyTimes();
-
     expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
-
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
-
-    expect(host1.getHostName()).andReturn("Host100").anyTimes();
-
-    expect(host1.convertToResponse()).andReturn(hostResponse1);
-
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
-
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class), anyObject(Map.class),
+        eq(managementController))).andReturn(hostResourceProvider).anyTimes();
 
     // replay
     replay(managementController, clusters, cluster,
-        host1,
         hostResponse1,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
 
     Map<String, Object> properties = new LinkedHashMap<String, Object>();
 
     properties.put(HostResourceProvider.HOST_CLUSTER_NAME_PROPERTY_ID, "Cluster100");
     properties.put(HostResourceProvider.HOST_NAME_PROPERTY_ID, "Host100");
-    
     properties.put(PropertyHelper.getPropertyId("Hosts.desired_config", "type"), "global");
     properties.put(PropertyHelper.getPropertyId("Hosts.desired_config", "tag"), "version1");
     properties.put(PropertyHelper.getPropertyId("Hosts.desired_config.properties", "a"), "b");
@@ -757,9 +781,8 @@ public class HostResourceProviderTest {
     
     // verify
     verify(managementController, clusters, cluster,
-        host1,
         hostResponse1,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
   }
 
   @Test
@@ -769,14 +792,13 @@ public class HostResourceProviderTest {
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
     Clusters clusters = createNiceMock(Clusters.class);
     Cluster cluster = createNiceMock(Cluster.class);
-    Host host1 = createNiceMock(Host.class);
     HostHealthStatus healthStatus = createNiceMock(HostHealthStatus.class);
     AmbariMetaInfo ambariMetaInfo = createNiceMock(AmbariMetaInfo.class);
-
     HostResponse hostResponse1 = createNiceMock(HostResponse.class);
-
-    List<Host> hosts = new LinkedList<Host>();
-    hosts.add(host1);
+    ResourceProviderFactory resourceProviderFactory = createNiceMock(ResourceProviderFactory.class);
+    ResourceProvider hostResourceProvider = createNiceMock(HostResourceProvider.class);
+    
+    AbstractControllerResourceProvider.init(resourceProviderFactory);
 
     Set<Cluster> clusterSet = new HashSet<Cluster>();
     clusterSet.add(cluster);
@@ -785,30 +807,20 @@ public class HostResourceProviderTest {
     expect(managementController.getClusters()).andReturn(clusters).anyTimes();
     expect(managementController.getAmbariMetaInfo()).andReturn(ambariMetaInfo).anyTimes();
     expect(managementController.getHostComponents((Set<ServiceComponentHostRequest>) anyObject())).andReturn(Collections.<ServiceComponentHostResponse>emptySet()).anyTimes();
-
-    expect(clusters.getHosts()).andReturn(hosts).anyTimes();
-    expect(clusters.getHost("Host100")).andReturn(host1).anyTimes();
-
     expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
-
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
-
-    expect(host1.getHostName()).andReturn("Host100").anyTimes();
-
-    expect(host1.convertToResponse()).andReturn(hostResponse1);
-
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
-
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();
+    expect(resourceProviderFactory.getHostResourceProvider(anyObject(Set.class), anyObject(Map.class),
+        eq(managementController))).andReturn(hostResourceProvider).anyTimes();
 
     // replay
     replay(managementController, clusters, cluster,
-        host1,
         hostResponse1,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
 
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
@@ -832,9 +844,8 @@ public class HostResourceProviderTest {
 
     // verify
     verify(managementController, clusters, cluster,
-        host1,
         hostResponse1,
-        healthStatus, ambariMetaInfo);
+        healthStatus, ambariMetaInfo, resourceProviderFactory, hostResourceProvider);
   }
 
   @Test
@@ -856,13 +867,9 @@ public class HostResourceProviderTest {
     expect(managementController.getClusters()).andReturn(clusters).anyTimes();
     expect(clusters.getHosts()).andReturn(hosts).anyTimes();
     expect(clusters.getHost("Host100")).andReturn(host1).anyTimes();
-
     expect(clusters.getCluster("Cluster100")).andReturn(cluster).anyTimes();
-
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
-
     expect(host1.getHostName()).andReturn("Host100").anyTimes();
-
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();
 
@@ -871,11 +878,7 @@ public class HostResourceProviderTest {
         host1,
         healthStatus);
 
-    ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
-        type,
-        PropertyHelper.getPropertyIds(type),
-        PropertyHelper.getKeyPropertyIds(type),
-        managementController);
+    ResourceProvider provider = getHostProvider(managementController);
 
     AbstractResourceProviderTest.TestObserver observer = new AbstractResourceProviderTest.TestObserver();
 
@@ -902,11 +905,9 @@ public class HostResourceProviderTest {
   public static HostResourceProvider getHostProvider(AmbariManagementController managementController) {
     Resource.Type type = Resource.Type.Host;
 
-    return (HostResourceProvider) AbstractControllerResourceProvider.getResourceProvider(
-        type,
-        PropertyHelper.getPropertyIds(type),
-        PropertyHelper.getKeyPropertyIds(type),
-        managementController);
+    return new HostResourceProvider(PropertyHelper.getPropertyIds(type),
+                                    PropertyHelper.getKeyPropertyIds(type),
+                                    managementController);
   }
 
   @Test