AMBARI-8163 - Provide stage resource information via REST API (tbeerbower)

ambari-server/src/main/java/org/apache/ambari/server/api/query/QueryImpl.java

@@ -37,6 +37,7 @@ import org.apache.ambari.server.api.services.Result;
 import org.apache.ambari.server.api.services.ResultImpl;
 import org.apache.ambari.server.api.util.TreeNode;
 import org.apache.ambari.server.api.util.TreeNodeImpl;
+import org.apache.ambari.server.controller.internal.QueryResponseImpl;
 import org.apache.ambari.server.controller.predicate.AndPredicate;
 import org.apache.ambari.server.controller.predicate.EqualsPredicate;
 import org.apache.ambari.server.controller.spi.ClusterController;
@@ -46,9 +47,8 @@ import org.apache.ambari.server.controller.spi.PageRequest;
 import org.apache.ambari.server.controller.spi.PageRequest.StartingPoint;
 import org.apache.ambari.server.controller.spi.PageResponse;
 import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.QueryResponse;
 import org.apache.ambari.server.controller.spi.Request;
-import org.apache.ambari.server.controller.spi.Request.PageInfo;
-import org.apache.ambari.server.controller.spi.Request.SortInfo;
 import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.controller.spi.ResourceProvider;
 import org.apache.ambari.server.controller.spi.Schema;
@@ -378,13 +378,14 @@ public class QueryImpl implements Query, ResourceInstance {
     Set<Resource> resourceSet = new LinkedHashSet<Resource>();
     Set<Resource> providerResourceSet = new LinkedHashSet<Resource>();
 
-    Set<Resource> queryResources = doQuery(resourceType, request, queryPredicate);
+    QueryResponse queryResponse = doQuery(resourceType, request, queryPredicate);
+
     // If there is a page request and the predicate does not contain properties
     // that need to be set
     if ((pageRequest != null || sortRequest != null ) &&
       !populateResourceRequired(resourceType)) {
       PageResponse pageResponse = clusterController.getPage(resourceType,
-        queryResources, request, queryPredicate, pageRequest, sortRequest);
+          queryResponse, request, queryPredicate, pageRequest, sortRequest);
 
       // build a new set
       for (Resource r : pageResponse.getIterable()) {
@@ -392,15 +393,15 @@ public class QueryImpl implements Query, ResourceInstance {
         providerResourceSet.add(r);
       }
     } else {
-      resourceSet.addAll(queryResources);
-      providerResourceSet.addAll(queryResources);
+      resourceSet.addAll(queryResponse.getResources());
+      providerResourceSet.addAll(queryResponse.getResources());
     }
 
     populatedQueryResults.put(null, new QueryResult(
-      request, queryPredicate, userPredicate, getKeyValueMap(), resourceSet));
+      request, queryPredicate, userPredicate, getKeyValueMap(), new QueryResponseImpl(resourceSet)));
 
     queryResults.put(null, new QueryResult(
-      request, queryPredicate, userPredicate, getKeyValueMap(), queryResources));
+      request, queryPredicate, userPredicate, getKeyValueMap(), queryResponse));
 
     clusterController.populateResources(resourceType, providerResourceSet, request, queryPredicate);
     queryForSubResources();
@@ -423,23 +424,23 @@ public class QueryImpl implements Query, ResourceInstance {
       Set<Resource> providerResourceSet = new HashSet<Resource>();
 
       for (QueryResult queryResult : populatedQueryResults.values()) {
-        for (Resource resource : queryResult.getProviderResourceSet()) {
+        for (Resource resource : queryResult.getQueryResponse().getResources()) {
           Map<Resource.Type, String> map = getKeyValueMap(resource, queryResult.getKeyValueMap());
 
           Predicate     queryPredicate = subResource.createPredicate(map, subResource.processedPredicate);
           Set<Resource> resourceSet    = new LinkedHashSet<Resource>();
 
           try {
-            Set<Resource> queryResources = subResource.doQuery(resourceType, request, queryPredicate);
+            Set<Resource> queryResources = subResource.doQuery(resourceType, request, queryPredicate).getResources();
             providerResourceSet.addAll(queryResources);
             resourceSet.addAll(queryResources);
           } catch (NoSuchResourceException e) {
             // do nothing ...
           }
           subResource.queryResults.put(resource,
-              new QueryResult(request, queryPredicate, subResourcePredicate, map, resourceSet));
+              new QueryResult(request, queryPredicate, subResourcePredicate, map, new QueryResponseImpl(resourceSet)));
           subResource.populatedQueryResults.put(resource,
-            new QueryResult(request, queryPredicate, subResourcePredicate, map, resourceSet));
+            new QueryResult(request, queryPredicate, subResourcePredicate, map, new QueryResponseImpl(resourceSet)));
         }
       }
       clusterController.populateResources(resourceType, providerResourceSet, request, null);
@@ -450,7 +451,7 @@ public class QueryImpl implements Query, ResourceInstance {
   /**
    * Query the cluster controller for the resources.
    */
-  private Set<Resource> doQuery(Resource.Type type, Request request, Predicate predicate)
+  private QueryResponse doQuery(Resource.Type type, Request request, Predicate predicate)
       throws UnsupportedPropertyException,
       SystemException,
       NoSuchResourceException,
@@ -547,7 +548,7 @@ public class QueryImpl implements Query, ResourceInstance {
       if (queryParentResource == parentResource) {
 
         Iterable<Resource> iterResource = clusterController.getIterable(
-            resourceDefinition.getType(), queryResult.getProviderResourceSet(),
+            resourceDefinition.getType(), queryResult.getQueryResponse(),
             queryResult.getRequest(), queryResult.getPredicate(), null, null);
 
         for (Resource resource : iterResource) {
@@ -719,7 +720,7 @@ public class QueryImpl implements Query, ResourceInstance {
       Predicate queryUserPredicate = queryResult.getUserPredicate();
       Request   queryRequest       = queryResult.getRequest();
 
-      Set<Resource> providerResourceSet = queryResult.getProviderResourceSet();
+      QueryResponse queryResponse = queryResult.getQueryResponse();
 
       if (hasSubResourcePredicate() && queryUserPredicate != null) {
         queryPredicate = getExtendedPredicate(parentResource, queryUserPredicate);
@@ -729,12 +730,12 @@ public class QueryImpl implements Query, ResourceInstance {
 
       if (pageRequest == null) {
         iterResource = clusterController.getIterable(
-          resourceType, providerResourceSet, queryRequest, queryPredicate,
+          resourceType, queryResponse, queryRequest, queryPredicate,
           null, sortRequest
         );
       } else {
         PageResponse pageResponse = clusterController.getPage(
-          resourceType, providerResourceSet, queryRequest, queryPredicate,
+          resourceType, queryResponse, queryRequest, queryPredicate,
           pageRequest, sortRequest
         );
         iterResource = pageResponse.getIterable();
@@ -899,16 +900,6 @@ public class QueryImpl implements Query, ResourceInstance {
   private Request createRequest() {
     Map<String, String> requestInfoProperties = new HashMap<String, String>();
 
-    PageInfo pageInfo = null;
-    if (null != pageRequest) {
-      pageInfo = new PageInfo(pageRequest);
-    }
-
-    SortInfo sortInfo = null;
-    if (null != sortRequest) {
-      sortInfo = new SortInfo(sortRequest);
-    }
-
     if (pageRequest != null) {
       requestInfoProperties.put(BaseRequest.PAGE_SIZE_PROPERTY_KEY,
           Integer.toString(pageRequest.getPageSize() + pageRequest.getOffset()));
@@ -921,7 +912,7 @@ public class QueryImpl implements Query, ResourceInstance {
 
     if (allProperties) {
       return PropertyHelper.getReadRequest(Collections.<String> emptySet(),
-          requestInfoProperties, null, pageInfo, sortInfo);
+          requestInfoProperties, null, pageRequest, sortRequest);
     }
 
     Map<String, TemporalInfo> mapTemporalInfo    = new HashMap<String, TemporalInfo>();
@@ -939,7 +930,7 @@ public class QueryImpl implements Query, ResourceInstance {
     }
 
     return PropertyHelper.getReadRequest(setProperties, requestInfoProperties,
-        mapTemporalInfo, pageInfo, sortInfo);
+        mapTemporalInfo, pageRequest, sortRequest);
   }
 
 
@@ -1009,17 +1000,18 @@ public class QueryImpl implements Query, ResourceInstance {
     private final Predicate predicate;
     private final Predicate userPredicate;
     private final Map<Resource.Type, String> keyValueMap;
-    private final Set<Resource> providerResourceSet;
+    private final QueryResponse queryResponse;
 
     // ----- Constructor -----------------------------------------------------
 
     private  QueryResult(Request request, Predicate predicate, Predicate userPredicate,
-                         Map<Resource.Type, String> keyValueMap, Set<Resource> providerResourceSet) {
-      this.request             = request;
-      this.predicate           = predicate;
-      this.userPredicate       = userPredicate;
-      this.keyValueMap         = keyValueMap;
-      this.providerResourceSet = providerResourceSet;
+                         Map<Resource.Type, String> keyValueMap,
+                         QueryResponse queryResponse) {
+      this.request        = request;
+      this.predicate      = predicate;
+      this.userPredicate  = userPredicate;
+      this.keyValueMap    = keyValueMap;
+      this.queryResponse  = queryResponse;
     }
 
     // ----- accessors -------------------------------------------------------
@@ -1040,8 +1032,8 @@ public class QueryImpl implements Query, ResourceInstance {
       return keyValueMap;
     }
 
-    public Set<Resource> getProviderResourceSet() {
-      return providerResourceSet;
+    public QueryResponse getQueryResponse() {
+      return queryResponse;
     }
   }
 }

ambari-server/src/main/java/org/apache/ambari/server/api/resources/BaseResourceDefinition.java

@@ -46,6 +46,10 @@ public abstract class BaseResourceDefinition implements ResourceDefinition {
    */
   private Resource.Type m_type;
 
+  /**
+   * The sub-resource type definitions.
+   */
+  private final Set<SubResourceDefinition> subResourceDefinitions;
 
   /**
    * Constructor.
@@ -54,6 +58,22 @@ public abstract class BaseResourceDefinition implements ResourceDefinition {
    */
   public BaseResourceDefinition(Resource.Type resourceType) {
     m_type = resourceType;
+    subResourceDefinitions = Collections.emptySet();
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param resourceType  the resource type
+   * @param subTypes      the sub-resource types
+   */
+  public BaseResourceDefinition(Resource.Type resourceType, Resource.Type ... subTypes) {
+    m_type = resourceType;
+    subResourceDefinitions =  new HashSet<SubResourceDefinition>();
+
+    for (Resource.Type subType : subTypes) {
+      subResourceDefinitions.add(new SubResourceDefinition(subType));
+    }
   }
 
   @Override
@@ -63,7 +83,7 @@ public abstract class BaseResourceDefinition implements ResourceDefinition {
 
   @Override
   public Set<SubResourceDefinition> getSubResourceDefinitions() {
-    return Collections.emptySet();
+    return subResourceDefinitions;
   }
 
   @Override

ambari-server/src/main/java/org/apache/ambari/server/api/resources/RequestResourceDefinition.java

@@ -19,16 +19,14 @@
 package org.apache.ambari.server.api.resources;
 
 
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Set;
-
 import org.apache.ambari.server.api.services.Request;
 import org.apache.ambari.server.api.util.TreeNode;
 import org.apache.ambari.server.controller.internal.RequestResourceProvider;
 import org.apache.ambari.server.controller.spi.Resource;
 
+import java.util.Arrays;
+import java.util.List;
+
 
 /**
  * Request resource definition.
@@ -39,7 +37,7 @@ public class RequestResourceDefinition extends BaseResourceDefinition {
    * Constructor.
    */
   public RequestResourceDefinition() {
-    super(Resource.Type.Request);
+    super(Resource.Type.Request, Resource.Type.Stage, Resource.Type.Task);
   }
 
   @Override
@@ -52,11 +50,6 @@ public class RequestResourceDefinition extends BaseResourceDefinition {
     return "request";
   }
 
-  @Override
-  public Set<SubResourceDefinition> getSubResourceDefinitions() {
-      return Collections.singleton(new SubResourceDefinition(Resource.Type.Task));
-  }
-
   @Override
   public List<PostProcessor> getPostProcessors() {
     return Arrays.asList(new RequestHrefPostProcessor(), new RequestSourceScheduleHrefPostProcessor());

ambari-server/src/main/java/org/apache/ambari/server/api/resources/ResourceInstanceFactoryImpl.java

@@ -325,6 +325,10 @@ public class ResourceInstanceFactoryImpl implements ResourceInstanceFactory {
         resourceDefinition = new UpgradeItemResourceDefinition();
         break;
 
+      case Stage:
+        resourceDefinition = new SimpleResourceDefinition(Resource.Type.Stage, "stage", "stages", Resource.Type.Task);
+        break;
+
       default:
         throw new IllegalArgumentException("Unsupported resource type: " + type);
     }

ambari-server/src/main/java/org/apache/ambari/server/api/resources/SimpleResourceDefinition.java

@@ -0,0 +1,69 @@
+/**
+ * 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.api.resources;
+
+import org.apache.ambari.server.controller.spi.Resource;
+
+/**
+ * Simple concrete resource definition.
+ */
+public class SimpleResourceDefinition extends BaseResourceDefinition {
+
+  /**
+   * The resource singular name.
+   */
+  private final String singularName;
+
+  /**
+   * The resource plural name.
+   */
+  private final String pluralName;
+
+
+  // ----- Constructors ------------------------------------------------------
+
+  /**
+   * Constructor.
+   * @param resourceType  the resource type
+   * @param singularName  the resource singular name
+   * @param pluralName    the resource plural name
+   * @param subTypes      the sub-resource types
+   *
+   */
+  public SimpleResourceDefinition(Resource.Type resourceType, String singularName, String pluralName,
+                                  Resource.Type... subTypes) {
+    super(resourceType, subTypes);
+
+    this.singularName = singularName;
+    this.pluralName   = pluralName;
+  }
+
+
+  // ----- ResourceDefinition ------------------------------------------------
+
+  @Override
+  public String getPluralName() {
+    return pluralName;
+  }
+
+  @Override
+  public String getSingularName() {
+    return singularName;
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/api/services/RequestService.java

@@ -96,12 +96,20 @@ public class RequestService extends BaseService {
         createRequestResource(m_clusterName, null));
   }
 
+  /**
+   * Gets the stage sub-resource.
+   */
+  @Path("{requestId}/stages")
+  public StageService getStageHandler(@PathParam("requestId") String requestId) {
+    return new StageService(m_clusterName, requestId);
+  }
+
   /**
    * Gets the tasks sub-resource.
    */
   @Path("{requestId}/tasks")
   public TaskService getTaskHandler(@PathParam("requestId") String requestId) {
-    return new TaskService(m_clusterName, requestId);
+    return new TaskService(m_clusterName, requestId, null);
   }
 
   /**

ambari-server/src/main/java/org/apache/ambari/server/api/services/StageService.java

@@ -0,0 +1,168 @@
+/**
+ * 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.api.services;
+
+
+import org.apache.ambari.server.api.resources.ResourceInstance;
+import org.apache.ambari.server.controller.spi.Resource;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.UriInfo;
+import java.util.HashMap;
+import java.util.Map;
+
+
+/**
+ * Service responsible for stage resource requests.
+ */
+public class StageService extends BaseService {
+  /**
+   * Parent cluster name.
+   */
+  private String m_clusterName;
+
+  /**
+   * Parent request id.
+   */
+  private String m_requestId;
+
+
+  // ----- Constructors ------------------------------------------------------
+
+  /**
+   * Constructor.
+   *
+   * @param clusterName  cluster id
+   * @param requestId    request id
+   */
+  public StageService(String clusterName, String requestId) {
+    m_clusterName = clusterName;
+    m_requestId = requestId;
+  }
+
+
+  // ----- StageService ------------------------------------------------------
+
+  /**
+   * Handles URL: /clusters/{clusterID}/requests/{requestID}/stages/{stageID} or
+   * /requests/{requestId}/stages/{stageID}
+   * Get a specific stage.
+   *
+   * @param headers  http headers
+   * @param ui       uri info
+   * @param stageId  stage id
+   *
+   * @return stage resource representation
+   */
+  @GET
+  @Path("{stageId}")
+  @Produces("text/plain")
+  public Response getStage(String body, @Context HttpHeaders headers, @Context UriInfo ui,
+                           @PathParam("stageId") String stageId) {
+
+    return handleRequest(headers, body, ui, Request.Type.GET,
+        createStageResource(m_clusterName, m_requestId, stageId));
+  }
+
+  /**
+   * Handles URL: /clusters/{clusterId}/requests/{requestID}/stages or /requests/{requestID}/stages
+   * Get all stages for a request.
+   *
+   * @param headers  http headers
+   * @param ui       uri info
+   *
+   * @return stage collection resource representation
+   */
+  @GET
+  @Produces("text/plain")
+  public Response getStages(String body, @Context HttpHeaders headers, @Context UriInfo ui) {
+    return handleRequest(headers, body, ui, Request.Type.GET,
+        createStageResource(m_clusterName, m_requestId, null));
+  }
+
+  /**
+   * Gets the tasks sub-resource.
+   */
+  @Path("{stageId}/tasks")
+  public TaskService getTaskHandler(@PathParam("stageId") String stageId) {
+    return new TaskService(m_clusterName, m_requestId, stageId);
+  }
+
+  /**
+   * Handles: PUT /clusters/{clusterId}/requests/{requestId}/stages/{stageId} or /requests/{requestId}/stages/{stageId}
+   * Change state of existing stages.
+   *
+   * @param body        http body
+   * @param headers     http headers
+   * @param ui          uri info
+   * @return information regarding the created services
+   */
+  @PUT
+  @Path("{stageId}")
+  @Produces("text/plain")
+  public Response updateStages(String body, @Context HttpHeaders headers, @Context UriInfo ui,
+                               @PathParam("stageId") String stageId) {
+    return handleRequest(headers, body, ui, Request.Type.PUT, createStageResource(m_clusterName, m_requestId, stageId));
+  }
+
+  /**
+   * Handles: POST /clusters/{clusterId}/requests/{requestId}/stages or /requests/{requestId}/stages
+   * Create multiple services.
+   *
+   * @param body        http body
+   * @param headers     http headers
+   * @param ui          uri info
+   * @return information regarding the created services
+   */
+  @POST
+  @Produces("text/plain")
+  public Response createStages(String body, @Context HttpHeaders headers, @Context UriInfo ui) {
+
+    return handleRequest(headers, body, ui, Request.Type.POST, createStageResource(m_clusterName, m_requestId, null));
+  }
+
+  /**
+   * Create a stage resource instance.
+   *
+   * @param clusterName  cluster name
+   * @param requestId    request id
+   * @param stageId      stage id
+   *
+   * @return a stage resource instance
+   */
+  ResourceInstance createStageResource(String clusterName, String requestId, String stageId) {
+    Map<Resource.Type,String> mapIds = new HashMap<Resource.Type, String>();
+
+    if (clusterName != null) {
+      mapIds.put(Resource.Type.Cluster, clusterName);
+    }
+    mapIds.put(Resource.Type.Request, requestId);
+    mapIds.put(Resource.Type.Stage, stageId);
+
+    return createResource(Resource.Type.Stage, mapIds);
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/api/services/TaskService.java

@@ -46,15 +46,21 @@ public class TaskService extends BaseService {
    */
   private String m_requestId;
 
+  /**
+   * Parent stage id.
+   */
+  private String m_stageId;
+
   /**
    * Constructor.
-   *
    * @param clusterName  cluster id
    * @param requestId    request id
+   * @param stageId      stage id
    */
-  public TaskService(String clusterName, String requestId) {
+  public TaskService(String clusterName, String requestId, String stageId) {
     m_clusterName = clusterName;
     m_requestId = requestId;
+    m_stageId = stageId;
   }
 
   /**
@@ -74,7 +80,7 @@ public class TaskService extends BaseService {
                           @PathParam("taskId") String taskId) {
 
     return handleRequest(headers, body, ui, Request.Type.GET,
-        createTaskResource(m_clusterName, m_requestId, taskId));
+        createTaskResource(m_clusterName, m_requestId, m_stageId, taskId));
   }
 
   /**
@@ -90,7 +96,7 @@ public class TaskService extends BaseService {
   @Produces("text/plain")
   public Response getComponents(String body, @Context HttpHeaders headers, @Context UriInfo ui) {
     return handleRequest(headers, body, ui, Request.Type.GET,
-        createTaskResource(m_clusterName, m_requestId, null));
+        createTaskResource(m_clusterName, m_requestId, m_stageId, null));
   }
 
   /**
@@ -98,14 +104,16 @@ public class TaskService extends BaseService {
    *
    * @param clusterName  cluster name
    * @param requestId    request id
+   * @param stageId      stage id
    * @param taskId       task id
    *
    * @return a task resource instance
    */
-  ResourceInstance createTaskResource(String clusterName, String requestId, String taskId) {
+  ResourceInstance createTaskResource(String clusterName, String requestId, String stageId, String taskId) {
     Map<Resource.Type,String> mapIds = new HashMap<Resource.Type, String>();
     mapIds.put(Resource.Type.Cluster, clusterName);
     mapIds.put(Resource.Type.Request, requestId);
+    mapIds.put(Resource.Type.Stage, stageId);
     mapIds.put(Resource.Type.Task, taskId);
 
     return createResource(Resource.Type.Task, mapIds);

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

@@ -27,9 +27,11 @@ import java.util.Set;
 
 import org.apache.ambari.server.StaticallyInject;
 import org.apache.ambari.server.controller.AlertHistoryRequest;
+import org.apache.ambari.server.controller.spi.ExtendedResourceProvider;
 import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
 import org.apache.ambari.server.controller.spi.NoSuchResourceException;
 import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.QueryResponse;
 import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.spi.RequestStatus;
 import org.apache.ambari.server.controller.spi.Resource;
@@ -47,9 +49,8 @@ import com.google.inject.Inject;
  * ResourceProvider for Alert History
  */
 @StaticallyInject
-public class AlertHistoryResourceProvider extends AbstractResourceProvider {
+public class AlertHistoryResourceProvider extends AbstractResourceProvider implements ExtendedResourceProvider{
 
-  public static final String ALERT_HISTORY = "AlertHistory";
   public static final String ALERT_HISTORY_DEFINITION_ID = "AlertHistory/definition_id";
   public static final String ALERT_HISTORY_DEFINITION_NAME = "AlertHistory/definition_name";
   public static final String ALERT_HISTORY_ID = "AlertHistory/id";
@@ -107,8 +108,6 @@ public class AlertHistoryResourceProvider extends AbstractResourceProvider {
 
   /**
    * Constructor.
-   *
-   * @param controller
    */
   AlertHistoryResourceProvider() {
     super(PROPERTY_IDS, KEY_PROPERTY_IDS);
@@ -163,23 +162,10 @@ public class AlertHistoryResourceProvider extends AbstractResourceProvider {
     Set<Resource> results = new LinkedHashSet<Resource>();
     Set<String> requestPropertyIds = getRequestPropertyIds(request, predicate);
 
-    Request.PageInfo pageInfo = request.getPageInfo();
-    Request.SortInfo sortInfo = request.getSortInfo();
-
     AlertHistoryRequest historyRequest = new AlertHistoryRequest();
-    historyRequest.Predicate = predicate;
-
-    if (null != pageInfo) {
-      historyRequest.Pagination = pageInfo.getPageRequest();
-
-      pageInfo.setResponsePaged(true);
-      pageInfo.setTotalCount(s_dao.getCount(predicate));
-    }
-
-    if (null != sortInfo) {
-      sortInfo.setResponseSorted(true);
-      historyRequest.Sort = sortInfo.getSortRequest();
-    }
+    historyRequest.Predicate  = predicate;
+    historyRequest.Pagination = request.getPageRequest();
+    historyRequest.Sort       = request.getSortRequest();
 
     List<AlertHistoryEntity> entities = s_dao.findAll(historyRequest);
     for (AlertHistoryEntity entity : entities) {
@@ -189,6 +175,18 @@ public class AlertHistoryResourceProvider extends AbstractResourceProvider {
     return results;
   }
 
+  @Override
+  public QueryResponse queryForResources(Request request, Predicate predicate)
+      throws SystemException, UnsupportedPropertyException,
+      NoSuchResourceException, NoSuchParentResourceException {
+
+    return new QueryResponseImpl(
+        getResources(request, predicate),
+        request.getSortRequest() != null,
+        request.getPageRequest() != null,
+        s_dao.getCount(predicate));
+  }
+
   /**
    * Converts the {@link AlertHistoryEntity} to a {@link Resource}.
    *
@@ -196,7 +194,7 @@ public class AlertHistoryResourceProvider extends AbstractResourceProvider {
    *          the entity to convert (not {@code null}).
    * @param requestedIds
    *          the properties requested (not {@code null}).
-   * @return
+   * @return the new {@link Resource}
    */
   private Resource toResource(AlertHistoryEntity entity,
       Set<String> requestedIds) {

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

@@ -27,9 +27,11 @@ import java.util.Set;
 
 import org.apache.ambari.server.StaticallyInject;
 import org.apache.ambari.server.controller.AlertNoticeRequest;
+import org.apache.ambari.server.controller.spi.ExtendedResourceProvider;
 import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
 import org.apache.ambari.server.controller.spi.NoSuchResourceException;
 import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.QueryResponse;
 import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.spi.RequestStatus;
 import org.apache.ambari.server.controller.spi.Resource;
@@ -49,9 +51,8 @@ import com.google.inject.Inject;
  * ResourceProvider for Alert History
  */
 @StaticallyInject
-public class AlertNoticeResourceProvider extends AbstractResourceProvider {
+public class AlertNoticeResourceProvider extends AbstractResourceProvider implements ExtendedResourceProvider {
 
-  public static final String ALERT_HISTORY = "AlertNotice";
   public static final String ALERT_NOTICE_ID = "AlertNotice/id";
   public static final String ALERT_NOTICE_STATE = "AlertNotice/notification_state";
   public static final String ALERT_NOTICE_UUID = "AlertNotice/uuid";
@@ -99,8 +100,6 @@ public class AlertNoticeResourceProvider extends AbstractResourceProvider {
 
   /**
    * Constructor.
-   *
-   * @param controller
    */
   AlertNoticeResourceProvider() {
     super(PROPERTY_IDS, KEY_PROPERTY_IDS);
@@ -155,23 +154,10 @@ public class AlertNoticeResourceProvider extends AbstractResourceProvider {
     Set<String> requestPropertyIds = getRequestPropertyIds(request, predicate);
     Set<Resource> results = new LinkedHashSet<Resource>();
 
-    Request.PageInfo pageInfo = request.getPageInfo();
-    Request.SortInfo sortInfo = request.getSortInfo();
-
     AlertNoticeRequest noticeRequest = new AlertNoticeRequest();
-    noticeRequest.Predicate = predicate;
-
-    if (null != pageInfo) {
-      noticeRequest.Pagination = pageInfo.getPageRequest();
-
-      pageInfo.setResponsePaged(true);
-      pageInfo.setTotalCount(s_dao.getNoticesCount(predicate));
-    }
-
-    if (null != sortInfo) {
-      sortInfo.setResponseSorted(true);
-      noticeRequest.Sort = sortInfo.getSortRequest();
-    }
+    noticeRequest.Predicate  = predicate;
+    noticeRequest.Pagination = request.getPageRequest();
+    noticeRequest.Sort       = request.getSortRequest();
 
     List<AlertNoticeEntity> entities = s_dao.findAllNotices(noticeRequest);
     for (AlertNoticeEntity entity : entities) {
@@ -181,16 +167,26 @@ public class AlertNoticeResourceProvider extends AbstractResourceProvider {
     return results;
   }
 
+  @Override
+  public QueryResponse queryForResources(Request request, Predicate predicate)
+      throws SystemException, UnsupportedPropertyException,
+      NoSuchResourceException, NoSuchParentResourceException {
+
+    return new QueryResponseImpl(
+        getResources(request, predicate),
+        request.getSortRequest() != null,
+        request.getPageRequest() != null,
+        s_dao.getNoticesCount(predicate));
+  }
+
   /**
    * Converts the {@link AlertNoticeEntity} to a {@link Resource}.
    *
-   * @param clusterName
-   *          the name of the cluster (not {@code null}).
    * @param entity
    *          the entity to convert (not {@code null}).
    * @param requestedIds
    *          the properties requested (not {@code null}).
-   * @return
+   * @return the new {@link Resource}
    */
   private Resource toResource(AlertNoticeEntity entity,
       Set<String> requestedIds) {

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

@@ -32,6 +32,7 @@ import java.util.Set;
 import java.util.TreeSet;
 
 import org.apache.ambari.server.controller.spi.ClusterController;
+import org.apache.ambari.server.controller.spi.ExtendedResourceProvider;
 import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
 import org.apache.ambari.server.controller.spi.NoSuchResourceException;
 import org.apache.ambari.server.controller.spi.PageRequest;
@@ -39,6 +40,7 @@ import org.apache.ambari.server.controller.spi.PageResponse;
 import org.apache.ambari.server.controller.spi.Predicate;
 import org.apache.ambari.server.controller.spi.PropertyProvider;
 import org.apache.ambari.server.controller.spi.ProviderModule;
+import org.apache.ambari.server.controller.spi.QueryResponse;
 import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.spi.RequestStatus;
 import org.apache.ambari.server.controller.spi.Resource;
@@ -72,8 +74,8 @@ public class ClusterControllerImpl implements ClusterController {
   /**
    * Map of resource providers keyed by resource type.
    */
-  private final Map<Resource.Type, ResourceProvider> resourceProviders =
-      new HashMap<Resource.Type, ResourceProvider>();
+  private final Map<Resource.Type, ExtendedResourceProviderWrapper> resourceProviders =
+      new HashMap<Resource.Type, ExtendedResourceProviderWrapper>();
 
   /**
    * Map of property provider lists keyed by resource type.
@@ -109,18 +111,15 @@ public class ClusterControllerImpl implements ClusterController {
   // ----- ClusterController -------------------------------------------------
 
   @Override
-  public Set<Resource> getResources(Type type, Request request, Predicate predicate)
+  public QueryResponse getResources(Type type, Request request, Predicate predicate)
       throws UnsupportedPropertyException, NoSuchResourceException,
              NoSuchParentResourceException, SystemException {
-    Set<Resource> resources;
+    QueryResponse queryResponse = null;
 
-    ResourceProvider provider = ensureResourceProvider(type);
+    ExtendedResourceProviderWrapper provider = ensureResourceProviderWrapper(type);
     ensurePropertyProviders(type);
 
-    if (provider == null) {
-      resources = Collections.emptySet();
-    } else {
-
+    if (provider != null) {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Using resource provider "
             + provider.getClass().getName()
@@ -130,15 +129,16 @@ public class ClusterControllerImpl implements ClusterController {
       checkProperties(type, request, predicate);
 
       // get the resources
-      resources = provider.getResources(request, predicate);
+      queryResponse = provider.queryForResources(request, predicate);
 
       // if a specific resource was asked for and not found then throw exception
-      if (predicate != null && (resources == null || resources.isEmpty())) {
+      if (predicate != null &&
+          (queryResponse == null || queryResponse.getResources() == null || queryResponse.getResources().isEmpty())) {
         throw new NoSuchResourceException(
             "The requested resource doesn't exist: " + type.toString() + " not found, " + predicate);
       }
     }
-    return resources;
+    return queryResponse == null ? new QueryResponseImpl(Collections.<Resource>emptySet()) : queryResponse;
   }
 
   @Override
@@ -157,16 +157,16 @@ public class ClusterControllerImpl implements ClusterController {
   }
 
   @Override
-  public Iterable<Resource> getIterable(Type type, Set<Resource> providerResources,
+  public Iterable<Resource> getIterable(Type type, QueryResponse queryResponse,
                                         Request request, Predicate predicate,
                                         PageRequest pageRequest,
                                         SortRequest sortRequest)
       throws NoSuchParentResourceException, UnsupportedPropertyException, NoSuchResourceException, SystemException {
-    return getPage(type, providerResources, request, predicate, pageRequest, sortRequest).getIterable();
+    return getPage(type, queryResponse, request, predicate, pageRequest, sortRequest).getIterable();
   }
 
   @Override
-  public PageResponse getPage(Type type, Set<Resource> providerResources,
+  public PageResponse getPage(Type type, QueryResponse queryResponse,
                               Request request, Predicate predicate,
                               PageRequest pageRequest, SortRequest sortRequest)
       throws UnsupportedPropertyException,
@@ -174,21 +174,17 @@ public class ClusterControllerImpl implements ClusterController {
       NoSuchResourceException,
       NoSuchParentResourceException {
 
-    ResourceProvider provider = ensureResourceProvider(type);
+    Set<Resource> providerResources = queryResponse.getResources();
 
-    ResourcePredicateEvaluator evaluator = provider instanceof ResourcePredicateEvaluator ?
-        (ResourcePredicateEvaluator) provider : DEFAULT_RESOURCE_PREDICATE_EVALUATOR;
+    ExtendedResourceProviderWrapper provider  = ensureResourceProviderWrapper(type);
 
     int totalCount = 0;
     Set<Resource> resources = providerResources;
 
     if (!providerResources.isEmpty()) {
-      Request.PageInfo pageInfo = request.getPageInfo();
-      Request.SortInfo sortInfo = request.getSortInfo();
-
       // determine if the provider has already paged & sorted the results
-      boolean providerAlreadyPaged = (null != pageInfo && pageInfo.isResponsePaged());
-      boolean providerAlreadySorted = (null != sortInfo && sortInfo.isResponseSorted());
+      boolean providerAlreadyPaged  = queryResponse.isPagedResponse();
+      boolean providerAlreadySorted = queryResponse.isSortedResponse();
 
       // conditionally create a comparator if there is a sort
       Comparator<Resource> resourceComparator = comparator;
@@ -216,16 +212,16 @@ public class ClusterControllerImpl implements ClusterController {
         switch (pageRequest.getStartingPoint()) {
           case Beginning:
             return getPageFromOffset(pageRequest.getPageSize(), 0, resources,
-                predicate, evaluator);
+                predicate, provider);
           case End:
             return getPageToOffset(pageRequest.getPageSize(), -1, resources,
-                predicate, evaluator);
+                predicate, provider);
           case OffsetStart:
             return getPageFromOffset(pageRequest.getPageSize(),
-                pageRequest.getOffset(), resources, predicate, evaluator);
+                pageRequest.getOffset(), resources, predicate, provider);
           case OffsetEnd:
             return getPageToOffset(pageRequest.getPageSize(),
-                pageRequest.getOffset(), resources, predicate, evaluator);
+                pageRequest.getOffset(), resources, predicate, provider);
           case PredicateStart:
           case PredicateEnd:
             // TODO : need to support the following cases for pagination
@@ -234,12 +230,12 @@ public class ClusterControllerImpl implements ClusterController {
             break;
         }
       } else if (providerAlreadyPaged) {
-        totalCount = pageInfo.getTotalCount();
+        totalCount = queryResponse.getTotalResourceCount();
       }
     }
 
     return new PageResponseImpl(new ResourceIterable(resources, predicate,
-      evaluator), 0, null, null, totalCount);
+        provider), 0, null, null, totalCount);
   }
 
   /**
@@ -345,17 +341,28 @@ public class ClusterControllerImpl implements ClusterController {
 
   @Override
   public ResourceProvider ensureResourceProvider(Type type) {
+    return ensureResourceProviderWrapper(type);
+  }
+
+
+  // ----- helper methods ----------------------------------------------------
+
+  /**
+   * Get the extended resource provider for the given type, creating it if required.
+   *
+   * @param type  the resource type
+   *
+   * @return the resource provider
+   */
+  private ExtendedResourceProviderWrapper ensureResourceProviderWrapper(Type type) {
     synchronized (resourceProviders) {
       if (!resourceProviders.containsKey(type)) {
-        resourceProviders.put(type, providerModule.getResourceProvider(type));
+        resourceProviders.put(type, new ExtendedResourceProviderWrapper(providerModule.getResourceProvider(type)));
       }
     }
     return resourceProviders.get(type);
   }
 
-
-  // ----- helper methods ----------------------------------------------------
-
   /**
    * Get an iterable set of resources filtered by the given request and
    * predicate objects.
@@ -405,9 +412,9 @@ public class ClusterControllerImpl implements ClusterController {
       NoSuchResourceException,
       NoSuchParentResourceException {
 
-    Set<Resource> providerResources = getResources(type, request, predicate);
-    populateResources(type, providerResources, request, predicate);
-    return getPage(type, providerResources, request, predicate, pageRequest, sortRequest);
+    QueryResponse queryResponse = getResources(type, request, predicate);
+    populateResources(type, queryResponse.getResources(), request, predicate);
+    return getPage(type, queryResponse, request, predicate, pageRequest, sortRequest);
   }
 
   /**
@@ -881,4 +888,105 @@ public class ClusterControllerImpl implements ClusterController {
       }
     }
   }
+
+
+  // ----- inner class : ExtendedResourceProviderWrapper ---------------------
+
+  /**
+   * Wrapper class that allows the cluster controller to treat all resource providers the same.
+   */
+  private static class ExtendedResourceProviderWrapper implements ExtendedResourceProvider, ResourcePredicateEvaluator {
+
+    /**
+     * The delegate resource provider.
+     */
+    private final ResourceProvider resourceProvider;
+
+    /**
+     * The delegate predicate evaluator. {@code null} if the given delegate resource provider is not an
+     * instance of {@link ResourcePredicateEvaluator}
+     */
+    private final ResourcePredicateEvaluator evaluator;
+
+    /**
+     * The delegate extended resource provider.  {@code null} if the given delegate resource provider is not an
+     * instance of {@link ExtendedResourceProvider}
+     */
+    private final ExtendedResourceProvider extendedResourceProvider;
+
+
+    // ----- Constructors ----------------------------------------------------
+
+    /**
+     * Constructor.
+     *
+     * @param resourceProvider  the delegate resource provider
+     */
+    public ExtendedResourceProviderWrapper(ResourceProvider resourceProvider) {
+      this.resourceProvider = resourceProvider;
+
+      extendedResourceProvider = resourceProvider instanceof ExtendedResourceProvider ?
+          (ExtendedResourceProvider) resourceProvider : null;
+
+      evaluator = resourceProvider instanceof ResourcePredicateEvaluator ?
+          (ResourcePredicateEvaluator) resourceProvider : DEFAULT_RESOURCE_PREDICATE_EVALUATOR;
+    }
+
+
+    // ----- ExtendedResourceProvider ----------------------------------------
+
+    @Override
+    public QueryResponse queryForResources(Request request, Predicate predicate)
+        throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+      return extendedResourceProvider == null ?
+          new QueryResponseImpl(resourceProvider.getResources(request, predicate)) :
+          extendedResourceProvider.queryForResources(request, predicate);
+    }
+
+
+    // ----- ResourceProvider ------------------------------------------------
+
+    @Override
+    public RequestStatus createResources(Request request)
+        throws SystemException, UnsupportedPropertyException,
+               ResourceAlreadyExistsException, NoSuchParentResourceException {
+      return resourceProvider.createResources(request);
+    }
+
+    @Override
+    public Set<Resource> getResources(Request request, Predicate predicate)
+        throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+      return resourceProvider.getResources(request, predicate);
+    }
+
+    @Override
+    public RequestStatus updateResources(Request request, Predicate predicate)
+        throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+      return resourceProvider.updateResources(request, predicate);
+    }
+
+    @Override
+    public RequestStatus deleteResources(Predicate predicate)
+        throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
+      return resourceProvider.deleteResources(predicate);
+    }
+
+    @Override
+    public Map<Type, String> getKeyPropertyIds() {
+      return resourceProvider.getKeyPropertyIds();
+    }
+
+    @Override
+    public Set<String> checkPropertyIds(Set<String> propertyIds) {
+      return resourceProvider.checkPropertyIds(propertyIds);
+    }
+
+
+    // ----- ResourcePredicateEvaluator --------------------------------------
+
+    @Override
+    public boolean evaluate(Predicate predicate, Resource resource) {
+      return evaluator.evaluate(predicate, resource);
+    }
+  }
 }

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

@@ -104,6 +104,8 @@ public class DefaultProviderModule extends AbstractProviderModule {
         return new ClusterStackVersionResourceProvider();
       case HostStackVersion:
         return new HostStackVersionResourceProvider();
+      case Stage:
+        return new StageResourceProvider();
 
       default:
         return AbstractControllerResourceProvider.getResourceProvider(type, propertyIds,

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

@@ -0,0 +1,103 @@
+/*
+ * 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.internal;
+
+import org.apache.ambari.server.controller.spi.QueryResponse;
+import org.apache.ambari.server.controller.spi.Resource;
+
+import java.util.Set;
+
+/**
+ * Simple {@link org.apache.ambari.server.controller.spi.QueryResponse} implementation.
+ */
+public class QueryResponseImpl implements QueryResponse {
+
+  /**
+   * The set of resources returned by the query.
+   */
+  private final Set<Resource> resources;
+
+  /**
+   * {@code true} if the response is sorted.
+   */
+  private final boolean sortedResponse;
+
+  /**
+   * {@code true} if the response is paginated.
+   */
+  private final boolean pagedResponse;
+
+  /**
+   * The total number of resources returned by the query.
+   */
+  private final int totalResourceCount;
+
+
+  // ----- Constructors ------------------------------------------------------
+
+  /**
+   * Constructor.
+   *
+   * @param resources the set of resources returned by the query.
+   */
+  public QueryResponseImpl(Set<Resource> resources) {
+    this.resources          = resources;
+    this.sortedResponse     = false;
+    this.pagedResponse      = false;
+    this.totalResourceCount = 0;
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param resources           the set of resources returned by the query.
+   * @param sortedResponse      indicates whether or not the response is sorted
+   * @param pagedResponse       indicates whether or not the response is paged
+   * @param totalResourceCount  the total number of resources returned by the query
+   */
+  public QueryResponseImpl(Set<Resource> resources, boolean sortedResponse,
+                           boolean pagedResponse, int totalResourceCount) {
+    this.resources          = resources;
+    this.sortedResponse     = sortedResponse;
+    this.pagedResponse      = pagedResponse;
+    this.totalResourceCount = totalResourceCount;
+  }
+
+
+  // ----- QueryResponse -----------------------------------------------------
+
+  @Override
+  public Set<Resource> getResources() {
+    return resources;
+  }
+
+  @Override
+  public boolean isSortedResponse() {
+    return sortedResponse;
+  }
+
+  @Override
+  public boolean isPagedResponse() {
+    return pagedResponse;
+  }
+
+  @Override
+  public int getTotalResourceCount() {
+    return totalResourceCount;
+  }
+}

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

@@ -25,8 +25,10 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
 
+import org.apache.ambari.server.controller.spi.PageRequest;
 import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.spi.ResourceProvider;
+import org.apache.ambari.server.controller.spi.SortRequest;
 import org.apache.ambari.server.controller.spi.TemporalInfo;
 
 /**
@@ -61,13 +63,14 @@ public class RequestImpl implements Request {
    * An optional page request which a concrete {@link ResourceProvider} can use
    * to return a slice of results.
    */
-  private PageInfo m_pageInfo = null;
+  private PageRequest m_pageRequest = null;
 
   /**
    * An optional sort request which a concrete {@link ResourceProvider} can use
    * to return sorted results.
    */
-  private SortInfo m_sortInfo = null;
+  private SortRequest m_sortRequest = null;
+
 
   // ----- Constructors ------------------------------------------------------
 
@@ -80,8 +83,23 @@ public class RequestImpl implements Request {
    * @param mapTemporalInfo        temporal info
    */
   public RequestImpl(Set<String> propertyIds, Set<Map<String, Object>> properties,
-                     Map<String, String> requestInfoProperties, Map<String,
-                     TemporalInfo> mapTemporalInfo) {
+                     Map<String, String> requestInfoProperties, Map<String,TemporalInfo> mapTemporalInfo) {
+    this(propertyIds, properties, requestInfoProperties, mapTemporalInfo, null, null);
+  }
+
+  /**
+   * Create a request.
+   *
+   * @param propertyIds            property ids associated with the request; may be null
+   * @param properties             resource properties associated with the request; may be null
+   * @param requestInfoProperties  request properties; may be null
+   * @param mapTemporalInfo        temporal info
+   * @param sortRequest            the sort request information; may be null
+   * @param pageRequest            the page request information; may be null
+   */
+  public RequestImpl(Set<String> propertyIds, Set<Map<String, Object>> properties,
+                     Map<String, String> requestInfoProperties, Map<String, TemporalInfo> mapTemporalInfo,
+                     SortRequest sortRequest, PageRequest pageRequest) {
     this.propertyIds = propertyIds == null ?
         Collections.unmodifiableSet(new HashSet<String>()) :
         Collections.unmodifiableSet(propertyIds);
@@ -95,8 +113,10 @@ public class RequestImpl implements Request {
         Collections.unmodifiableMap(requestInfoProperties);
 
     setTemporalInfo(mapTemporalInfo);
-  }
 
+    m_sortRequest = sortRequest;
+    m_pageRequest = pageRequest;
+  }
 
   // ----- Request -----------------------------------------------------------
 
@@ -124,50 +144,14 @@ public class RequestImpl implements Request {
     m_mapTemporalInfo = mapTemporalInfo;
   }
 
-  /**
-   * {@inheritDoc}
-   */
   @Override
-  public PageInfo getPageInfo() {
-    return m_pageInfo;
+  public PageRequest getPageRequest() {
+    return m_pageRequest;
   }
 
-  /**
-   * {@inheritDoc}
-   */
   @Override
-  public SortInfo getSortInfo() {
-    return m_sortInfo;
-  }
-
-  /**
-   * Sets the pagination request that a {@link ResourceProvider} can optionally
-   * use when creating its result set.
-   * <p/>
-   * If the result set is being paginated by the {@link ResourceProvider}, then
-   * {@link PageInfo#isResponsePaged()} must be invoked with {@code true} by the
-   * provider.
-   *
-   * @param pageInfo
-   *          the page request, or {@code null} if none.
-   */
-  public void setPageInfo(PageInfo pageInfo) {
-    m_pageInfo = pageInfo;
-  }
-
-  /**
-   * Sets the sorting request that a {@link ResourceProvider} can optionally use
-   * when creating its result set.
-   * <p/>
-   * If the result set is being paginated by the {@link ResourceProvider}, then
-   * {@link SortInfo#isResponseSorted()} must be invoked with {@code true} by
-   * the provider.
-   *
-   * @param sortInfo
-   *          the sort request, or {@code null} if none.
-   */
-  public void setSortInfo(SortInfo sortInfo) {
-    m_sortInfo = sortInfo;
+  public SortRequest getSortRequest() {
+    return m_sortRequest;
   }
 
   @Override

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

@@ -0,0 +1,325 @@
+/**
+ * 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.internal;
+
+import org.apache.ambari.server.StaticallyInject;
+import org.apache.ambari.server.actionmanager.HostRoleStatus;
+import org.apache.ambari.server.controller.spi.ExtendedResourceProvider;
+import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
+import org.apache.ambari.server.controller.spi.NoSuchResourceException;
+import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.QueryResponse;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.RequestStatus;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
+import org.apache.ambari.server.controller.spi.SystemException;
+import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
+import org.apache.ambari.server.orm.dao.StageDAO;
+import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
+import org.apache.ambari.server.orm.entities.StageEntity;
+import org.apache.ambari.server.state.Cluster;
+import org.apache.ambari.server.state.Clusters;
+
+import javax.inject.Inject;
+import javax.inject.Provider;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * ResourceProvider for Stage
+ */
+@StaticallyInject
+public class StageResourceProvider extends AbstractResourceProvider implements ExtendedResourceProvider {
+
+  /**
+   * Used for querying stage resources.
+   */
+  @Inject
+  private static StageDAO dao = null;
+
+  /**
+   * Used to get cluster information.
+   */
+  private static Clusters clusters = null;
+
+  @Inject
+  private static Provider<Clusters> clustersProvider = null;
+
+  /**
+   * Stage property constants.
+   */
+  public static final String STAGE_STAGE_ID = "Stage/stage_id";
+  public static final String STAGE_CLUSTER_NAME = "Stage/cluster_name";
+  public static final String STAGE_REQUEST_ID = "Stage/request_id";
+  public static final String STAGE_LOG_INFO = "Stage/log_info";
+  public static final String STAGE_CONTEXT = "Stage/context";
+  public static final String STAGE_CLUSTER_HOST_INFO = "Stage/cluster_host_info";
+  public static final String STAGE_COMMAND_PARAMS = "Stage/command_params";
+  public static final String STAGE_HOST_PARAMS = "Stage/host_params";
+  public static final String STAGE_PROGRESS_PERCENT = "Stage/progress_percent";
+  public static final String STAGE_STATUS = "Stage/status";
+  public static final String STAGE_START_TIME = "Stage/start_time";
+  public static final String STAGE_END_TIME = "Stage/end_time";
+
+  /**
+   * The property ids for a stage resource.
+   */
+  private static final Set<String> PROPERTY_IDS = new HashSet<String>();
+
+  /**
+   * The key property ids for a stage resource.
+   */
+  private static final Map<Resource.Type, String> KEY_PROPERTY_IDS =
+      new HashMap<Resource.Type, String>();
+
+  static {
+    // properties
+    PROPERTY_IDS.add(STAGE_STAGE_ID);
+    PROPERTY_IDS.add(STAGE_CLUSTER_NAME);
+    PROPERTY_IDS.add(STAGE_REQUEST_ID);
+    PROPERTY_IDS.add(STAGE_LOG_INFO);
+    PROPERTY_IDS.add(STAGE_CONTEXT);
+    PROPERTY_IDS.add(STAGE_CLUSTER_HOST_INFO);
+    PROPERTY_IDS.add(STAGE_COMMAND_PARAMS);
+    PROPERTY_IDS.add(STAGE_HOST_PARAMS);
+    PROPERTY_IDS.add(STAGE_PROGRESS_PERCENT);
+    PROPERTY_IDS.add(STAGE_STATUS);
+    PROPERTY_IDS.add(STAGE_START_TIME);
+    PROPERTY_IDS.add(STAGE_END_TIME);
+
+    // keys
+    KEY_PROPERTY_IDS.put(Resource.Type.Stage, STAGE_STAGE_ID);
+    KEY_PROPERTY_IDS.put(Resource.Type.Cluster, STAGE_CLUSTER_NAME);
+    KEY_PROPERTY_IDS.put(Resource.Type.Request, STAGE_REQUEST_ID);
+  }
+
+
+  // ----- Constructors ------------------------------------------------------
+
+  /**
+   * Constructor.
+   */
+  StageResourceProvider() {
+    super(PROPERTY_IDS, KEY_PROPERTY_IDS);
+  }
+
+
+  // ----- AbstractResourceProvider ------------------------------------------
+
+  @Override
+  protected Set<String> getPKPropertyIds() {
+    return new HashSet<String>(KEY_PROPERTY_IDS.values());
+  }
+
+
+  // ----- ResourceProvider --------------------------------------------------
+
+  @Override
+  public RequestStatus createResources(Request request) throws SystemException,
+      UnsupportedPropertyException, ResourceAlreadyExistsException,
+      NoSuchParentResourceException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public RequestStatus updateResources(Request request, Predicate predicate)
+      throws SystemException, UnsupportedPropertyException,
+      NoSuchResourceException, NoSuchParentResourceException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public RequestStatus deleteResources(Predicate predicate)
+      throws SystemException, UnsupportedPropertyException,
+      NoSuchResourceException, NoSuchParentResourceException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public Set<Resource> getResources(Request request, Predicate predicate)
+      throws SystemException, UnsupportedPropertyException,
+      NoSuchResourceException, NoSuchParentResourceException {
+
+    ensureClusters();
+
+    Set<Resource> results     = new LinkedHashSet<Resource>();
+    Set<String>   propertyIds = getRequestPropertyIds(request, predicate);
+
+    List<StageEntity> entities = dao.findAll(request, predicate);
+    for (StageEntity entity : entities) {
+      results.add(toResource(entity, propertyIds));
+    }
+    return results;
+  }
+
+
+  // ----- ExtendedResourceProvider ------------------------------------------
+
+  @Override
+  public QueryResponse queryForResources(Request request, Predicate predicate)
+      throws SystemException, UnsupportedPropertyException,
+      NoSuchResourceException, NoSuchParentResourceException {
+
+    Set<Resource> results = getResources(request, predicate);
+
+    return new QueryResponseImpl(results, request.getSortRequest() != null, false, results.size());
+  }
+
+
+  // ----- helper methods ----------------------------------------------------
+
+  /**
+   * Converts the {@link StageEntity} to a {@link Resource}.
+   *
+   * @param entity        the entity to convert (not {@code null})
+   * @param requestedIds  the properties requested (not {@code null})
+   *
+   * @return the new resource
+   */
+  private Resource toResource(StageEntity entity,
+                              Set<String> requestedIds) {
+
+    Resource resource = new ResourceImpl(Resource.Type.Stage);
+
+    Long clusterId = entity.getClusterId();
+    if (clusterId != null) {
+      try {
+        Cluster cluster = clusters.getClusterById(clusterId);
+
+        setResourceProperty(resource, STAGE_CLUSTER_NAME, cluster.getClusterName(), requestedIds);
+      } catch (Exception e) {
+        LOG.error("Can not get information for cluster " + clusterId + ".", e );
+      }
+    }
+
+    setResourceProperty(resource, STAGE_STAGE_ID, entity.getStageId(), requestedIds);
+    setResourceProperty(resource, STAGE_REQUEST_ID, entity.getRequestId(), requestedIds);
+    setResourceProperty(resource, STAGE_CONTEXT, entity.getRequestContext(), requestedIds);
+    setResourceProperty(resource, STAGE_CLUSTER_HOST_INFO, entity.getClusterHostInfo(), requestedIds);
+    setResourceProperty(resource, STAGE_COMMAND_PARAMS, entity.getCommandParamsStage(), requestedIds);
+    setResourceProperty(resource, STAGE_HOST_PARAMS, entity.getHostParamsStage(), requestedIds);
+
+    Collection<HostRoleCommandEntity> tasks = entity.getHostRoleCommands();
+
+    Long startTime = tasks.isEmpty() ? 0L : Long.MAX_VALUE;
+    Long endTime   = 0L;
+
+    for (HostRoleCommandEntity task : tasks) {
+      startTime = Math.min(task.getStartTime(), startTime);
+      endTime   = Math.max(task.getEndTime(), endTime);
+    }
+
+    setResourceProperty(resource, STAGE_START_TIME, startTime, requestedIds);
+    setResourceProperty(resource, STAGE_END_TIME, endTime, requestedIds);
+
+    int taskCount = tasks.size();
+
+    Map<HostRoleStatus, Integer> taskStatusCounts = calculateTaskStatusCounts(tasks);
+
+    setResourceProperty(resource, STAGE_PROGRESS_PERCENT, calculateProgressPercent(taskStatusCounts, taskCount),
+        requestedIds);
+    setResourceProperty(resource, STAGE_STATUS, calculateSummaryStatus(taskStatusCounts, taskCount).toString(),
+        requestedIds);
+
+    return resource;
+  }
+
+  /**
+   * Calculate the percent complete for the stage based on its tasks.
+   *
+   * @param counters    counts of tasks that are in various states
+   * @param totalTasks  total number of tasks in request
+   *
+   * @return the percent complete for the stage
+   */
+  private double calculateProgressPercent(Map<HostRoleStatus, Integer> counters, double totalTasks) {
+    return ((counters.get(HostRoleStatus.QUEUED) * 0.09 +
+        counters.get(HostRoleStatus.IN_PROGRESS) * 0.35 +
+        counters.get(HostRoleStatus.COMPLETED)) / totalTasks) * 100.0;
+  }
+
+  /**
+   * Calculate the stage status based on the status of its tasks.
+   *
+   * @param counters    counts of tasks that are in various states
+   * @param totalTasks  total number of tasks in request
+   *
+   * @return summary request status based on statuses of tasks in different states.
+   */
+  private HostRoleStatus calculateSummaryStatus(Map<HostRoleStatus, Integer> counters, int totalTasks) {
+    return counters.get(HostRoleStatus.FAILED) > 0 ? HostRoleStatus.FAILED :
+        counters.get(HostRoleStatus.ABORTED) > 0 ? HostRoleStatus.ABORTED :
+        counters.get(HostRoleStatus.TIMEDOUT) > 0 ? HostRoleStatus.TIMEDOUT :
+        counters.get(HostRoleStatus.IN_PROGRESS) > 0 ? HostRoleStatus.IN_PROGRESS :
+        counters.get(HostRoleStatus.COMPLETED) == totalTasks ? HostRoleStatus.COMPLETED : HostRoleStatus.PENDING;
+  }
+
+  /**
+   * Returns counts of tasks that are in various states.
+   *
+   * @param tasks  the collection of tasks
+   *
+   * @return a map of counts of tasks keyed by the task status
+   */
+  private Map<HostRoleStatus, Integer> calculateTaskStatusCounts(Collection<HostRoleCommandEntity> tasks) {
+    Map<HostRoleStatus, Integer> counters = new HashMap<HostRoleStatus, Integer>();
+    // initialize
+    for (HostRoleStatus hostRoleStatus : HostRoleStatus.values()) {
+      counters.put(hostRoleStatus, 0);
+    }
+    // calculate counts
+    for (HostRoleCommandEntity hostRoleCommand : tasks) {
+      HostRoleStatus status = hostRoleCommand.getStatus();
+      // count tasks where isCompletedState() == true as COMPLETED
+      // but don't count tasks with COMPLETED status twice
+      if (status.isCompletedState() && status != HostRoleStatus.COMPLETED) {
+        // Increase total number of completed tasks;
+        counters.put(HostRoleStatus.COMPLETED, counters.get(HostRoleStatus.COMPLETED) + 1);
+      }
+      // Increment counter for particular status
+      counters.put(status, counters.get(status) + 1);
+    }
+
+    // We overwrite the value to have the sum converged
+    counters.put(HostRoleStatus.IN_PROGRESS,
+        tasks.size() -
+        counters.get(HostRoleStatus.COMPLETED) -
+        counters.get(HostRoleStatus.QUEUED) -
+        counters.get(HostRoleStatus.PENDING));
+
+    return counters;
+  }
+
+  /**
+   * Ensure that cluster information is available.
+   *
+   * @return the clusters information
+   */
+  private synchronized Clusters ensureClusters() {
+    if (clusters == null) {
+      clusters = clustersProvider.get();
+    }
+    return clusters;
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/controller/spi/ClusterController.java

@@ -37,7 +37,7 @@ public interface ClusterController extends SchemaFactory {
    * @param request     the request object which defines the desired set of properties
    * @param predicate   the predicate object which filters which resources are returned
    *
-   * @return an iterable object of the requested resources
+   * @return the query response
    *
    * @throws UnsupportedPropertyException thrown if the request or predicate contain
    *                                      unsupported property ids
@@ -45,7 +45,7 @@ public interface ClusterController extends SchemaFactory {
    * @throws NoSuchResourceException no matching resource(s) found
    * @throws NoSuchParentResourceException a specified parent resource doesn't exist
    */
-  Set<Resource> getResources(Resource.Type type, Request request, Predicate predicate)
+  QueryResponse getResources(Resource.Type type, Request request, Predicate predicate)
       throws UnsupportedPropertyException,
       NoSuchResourceException,
       NoSuchParentResourceException,
@@ -74,12 +74,12 @@ public interface ClusterController extends SchemaFactory {
    * Get an iterable set of resources from the given set of resources filtered by the
    * given request and predicate objects.
    *
-   * @param type               type of resources
-   * @param providerResources  set of populated Resources
-   * @param request            the request
-   * @param predicate          the predicate object which filters which resources are returned
-   * @param pageRequest        the page request for a paginated response
-   * @param sortRequest        the sortRequest object which defines if the resources need to be sorted
+   * @param type           type of resources
+   * @param queryResponse  the response from the resource query
+   * @param request        the request
+   * @param predicate      the predicate object which filters which resources are returned
+   * @param pageRequest    the page request for a paginated response
+   * @param sortRequest    the sortRequest object which defines if the resources need to be sorted
    *
    * @return a page response representing the requested page of resources
    *
@@ -89,7 +89,7 @@ public interface ClusterController extends SchemaFactory {
    * @throws NoSuchResourceException no matching resource(s) found
    * @throws NoSuchParentResourceException a specified parent resource doesn't exist
    */
-  Iterable<Resource> getIterable(Resource.Type type, Set<Resource> providerResources,
+  Iterable<Resource> getIterable(Resource.Type type, QueryResponse queryResponse,
                                  Request request, Predicate predicate,
                                  PageRequest pageRequest,
                                  SortRequest sortRequest)
@@ -102,11 +102,11 @@ public interface ClusterController extends SchemaFactory {
    * Get a page of resources from the given set filtered by the given request,
    * predicate objects and page request.
    *
-   * @param type               type of resources
-   * @param providerResources  set of populated Resources
-   * @param request            the request
-   * @param predicate          the predicate object which filters which resources are returned
-   * @param pageRequest        the page request for a paginated response
+   * @param type           type of resources
+   * @param queryResponse  the response from the resource query
+   * @param request        the request
+   * @param predicate      the predicate object which filters which resources are returned
+   * @param pageRequest    the page request for a paginated response
    *
    * @return a page response representing the requested page of resources
    *
@@ -116,7 +116,7 @@ public interface ClusterController extends SchemaFactory {
    * @throws NoSuchResourceException no matching resource(s) found
    * @throws NoSuchParentResourceException a specified parent resource doesn't exist
    */
-  PageResponse getPage(Resource.Type type, Set<Resource> providerResources,
+  PageResponse getPage(Resource.Type type, QueryResponse queryResponse,
                        Request request, Predicate predicate,
                        PageRequest pageRequest, SortRequest sortRequest)
       throws UnsupportedPropertyException,

ambari-server/src/main/java/org/apache/ambari/server/controller/spi/ExtendedResourceProvider.java

@@ -0,0 +1,49 @@
+/*
+ * 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.spi;
+
+/**
+ * Extended resource provider interface that adds the ability to return a
+ * {@link org.apache.ambari.server.controller.spi.QueryResponse} from a
+ * resource query.
+ *
+ * If a resource provider supports paging and/or sorting of the resources
+ * acquired through a query then it should implement
+ * {@link org.apache.ambari.server.controller.spi.ExtendedResourceProvider}.
+ */
+public interface ExtendedResourceProvider extends ResourceProvider {
+  /**
+   * Query for resources that match the given predicate.
+   *
+   * @param request    the request object which defines the desired set of properties
+   * @param predicate  the predicate object which can be used to filter which
+   *                   resources are returned
+   *
+   * @return the response from the resource query
+   *
+   * @throws SystemException an internal system exception occurred
+   * @throws UnsupportedPropertyException the request contains unsupported property ids
+   * @throws NoSuchResourceException the requested resource instance doesn't exist
+   * @throws NoSuchParentResourceException a parent resource of the requested resource doesn't exist
+   */
+  public QueryResponse queryForResources(Request request, Predicate predicate)
+      throws SystemException,
+      UnsupportedPropertyException,
+      NoSuchResourceException,
+      NoSuchParentResourceException;
+}

ambari-server/src/main/java/org/apache/ambari/server/controller/spi/QueryResponse.java

@@ -0,0 +1,58 @@
+/*
+ * 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.spi;
+
+import java.util.Set;
+
+/**
+ * Resource query response.  Used for resource queries
+ * that may include the paging and/or sorting of the
+ * returned resources.
+ */
+public interface QueryResponse {
+
+  /**
+   * Get the set of resources returned in the response.
+   *
+   * @return the set of resources
+   */
+  public Set<Resource> getResources();
+
+  /**
+   * Determine whether or not the response is sorted.
+   *
+   * @return {@code true} if the response is sorted
+   */
+  public boolean isSortedResponse();
+
+  /**
+   * Determine whether or not the response is paginated.
+   *
+   * @return {@code true} if the response is paginated
+   */
+  public boolean isPagedResponse();
+
+  /**
+   * Get the the total number of resources for the query result.
+   * May be different than the size of the resource set for a
+   * paged response.
+   *
+   * @return total the total number of resources in the query result
+   */
+  public int getTotalResourceCount();
+}

ambari-server/src/main/java/org/apache/ambari/server/controller/spi/Request.java

@@ -69,7 +69,7 @@ public interface Request {
    *
    * @return the page request information.
    */
-  public PageInfo getPageInfo();
+  public PageRequest getPageRequest();
 
   /**
    * Obtain information to order the results by. This structure can be used for
@@ -78,146 +78,5 @@ public interface Request {
    *
    * @return the sort request information.
    */
-  public SortInfo getSortInfo();
-
-  /**
-   * The {@link PageInfo} class encapsulates the page request and optional
-   * provider page response data. It is used so that a {@link ResourceProvider}
-   * can indicate whether the provided {@link PageRequest} was already applied
-   * to the result set.
-   */
-  public final class PageInfo {
-    /**
-     * {@code true} if the reponse is already paginated.
-     */
-    private boolean m_pagedResponse = false;
-
-    /**
-     * If {@link #m_pagedResponse} is {@code true} then this value will
-     * represent the total items that the page slice originated from.
-     */
-    private int m_totalCount = 0;
-
-    /**
-     * The initial page request, not {@code null}.
-     */
-    private final PageRequest m_pageRequest;
-
-    /**
-     * Constructor.
-     *
-     * @param pageRequest
-     */
-    public PageInfo(PageRequest pageRequest) {
-      m_pageRequest = pageRequest;
-    }
-
-    /**
-     * Gets the page request.
-     *
-     * @return the page request (never {@code null}).
-     */
-    public PageRequest getPageRequest() {
-      return m_pageRequest;
-    }
-
-    /**
-     * Gets whether the response has already been paginated from a larger result
-     * set.
-     *
-     * @return {@code true} if the response is already paged, {@code false}
-     *         otherwise.
-     */
-    public boolean isResponsePaged() {
-      return m_pagedResponse;
-    }
-
-    /**
-     * Gets whether the response has already been paginated from a larger result
-     * set.
-     *
-     * @param responsePaged
-     *          {@code true} if the response is already paged, {@code false}
-     *          otherwise.
-     */
-    public void setResponsePaged(boolean responsePaged) {
-      m_pagedResponse = responsePaged;
-    }
-
-    /**
-     * Gets the total count of items from which the paged result set was taken.
-     *
-     * @return the totalCount the total count of items in the un-paginated set.
-     */
-    public int getTotalCount() {
-      return m_totalCount;
-    }
-
-    /**
-     * Sets the total count of items from which the paged result set was taken.
-     *
-     * @param totalCount
-     *          the totalCount the total count of items in the un-paginated set.
-     */
-    public void setTotalCount(int totalCount) {
-      m_totalCount = totalCount;
-    }
-  }
-
-  /**
-   * The {@link SortInfo} class encapsulates the sort request and optional
-   * provider sort response data. It is used so that a {@link ResourceProvider}
-   * can indicate whether the provided {@link SortRequest} was already applied
-   * to the result set.
-   */
-  public final class SortInfo {
-    /**
-     * {@code true} if the response is already sorted.
-     */
-    private boolean m_sortedResponse = false;
-
-    /**
-     * The initial sort request, not {@code null}.
-     */
-    private final SortRequest m_sortRequest;
-
-    /**
-     * Constructor.
-     *
-     * @param sortRequest
-     */
-    public SortInfo(SortRequest sortRequest) {
-      m_sortRequest = sortRequest;
-    }
-
-    /**
-     * Gets the sort request.
-     *
-     * @return the sort request (never {@code null}).
-     */
-    public SortRequest getSortRequest() {
-      return m_sortRequest;
-    }
-
-    /**
-     * Sets whether the response is already sorted.
-     *
-     * @param responseSorted
-     *          {@code true} if the response is already sorted, {@code false}
-     *          otherwise.
-     */
-    public void setResponseSorted(boolean responseSorted) {
-      m_sortedResponse = responseSorted;
-    }
-
-    /**
-     * Gets whether the response is already sorted.
-     *
-     * @return {@code true} if the response is already sorted, {@code false}
-     *         otherwise.
-     */
-    public boolean isResponseSorted() {
-      return m_sortedResponse;
-    }
-  }
+  public SortRequest getSortRequest();
 }

ambari-server/src/main/java/org/apache/ambari/server/controller/spi/Resource.java

@@ -132,7 +132,8 @@ public interface Resource {
     ClusterStackVersion,
     HostStackVersion,
     Upgrade,
-    UpgradeItem;
+    UpgradeItem,
+    Stage;
 
     /**
      * Get the {@link Type} that corresponds to this InternalType.
@@ -226,6 +227,7 @@ public interface Resource {
     public static final Type HostStackVersion = InternalType.HostStackVersion.getType();
     public static final Type Upgrade = InternalType.Upgrade.getType();
     public static final Type UpgradeItem = InternalType.UpgradeItem.getType();
+    public static final Type Stage = InternalType.Stage.getType();
 
     /**
      * The type name.

ambari-server/src/main/java/org/apache/ambari/server/controller/utilities/PropertyHelper.java

@@ -29,10 +29,10 @@ import java.util.regex.Pattern;
 
 import org.apache.ambari.server.controller.internal.PropertyInfo;
 import org.apache.ambari.server.controller.internal.RequestImpl;
+import org.apache.ambari.server.controller.spi.PageRequest;
 import org.apache.ambari.server.controller.spi.Request;
-import org.apache.ambari.server.controller.spi.Request.PageInfo;
-import org.apache.ambari.server.controller.spi.Request.SortInfo;
 import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.controller.spi.SortRequest;
 import org.apache.ambari.server.controller.spi.TemporalInfo;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.type.TypeReference;
@@ -373,23 +373,18 @@ public class PropertyHelper {
    *          request info properties
    * @param mapTemporalInfo
    *          the temporal info
-   * @param pageInfo
+   * @param pageRequest
    *          an optional page request, or {@code null} for none.
-   * @param sortInfo
+   * @param sortRequest
    *          an optional sort request, or {@code null} for none.
    */
   public static Request getReadRequest(Set<String> propertyIds,
       Map<String, String> requestInfoProperties,
-      Map<String, TemporalInfo> mapTemporalInfo, PageInfo pageInfo,
-      SortInfo sortInfo) {
+      Map<String, TemporalInfo> mapTemporalInfo, PageRequest pageRequest,
+      SortRequest sortRequest) {
 
-    RequestImpl request = new RequestImpl(propertyIds, null,
-        requestInfoProperties,
-        mapTemporalInfo);
-
-    request.setPageInfo(pageInfo);
-    request.setSortInfo(sortInfo);
-    return request;
+    return  new RequestImpl(propertyIds, null, requestInfoProperties,
+        mapTemporalInfo, sortRequest, pageRequest);
   }
 
   /**

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/StageDAO.java

@@ -23,12 +23,24 @@ import com.google.inject.Provider;
 import com.google.inject.Singleton;
 import com.google.inject.persist.Transactional;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
+import org.apache.ambari.server.api.query.JpaPredicateVisitor;
+import org.apache.ambari.server.api.query.JpaSortBuilder;
+import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.utilities.PredicateHelper;
 import org.apache.ambari.server.orm.RequiresSession;
 import org.apache.ambari.server.orm.entities.StageEntity;
 import org.apache.ambari.server.orm.entities.StageEntityPK;
+import org.apache.ambari.server.orm.entities.StageEntity_;
 import org.apache.ambari.server.utils.StageUtils;
+import org.eclipse.persistence.config.HintValues;
+import org.eclipse.persistence.config.QueryHints;
+
 import javax.persistence.EntityManager;
 import javax.persistence.TypedQuery;
+import javax.persistence.criteria.CriteriaQuery;
+import javax.persistence.criteria.Order;
+import javax.persistence.metamodel.SingularAttribute;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Collection;
@@ -139,4 +151,75 @@ public class StageDAO {
   public void removeByPK(StageEntityPK stageEntityPK) {
     remove(findByPK(stageEntityPK));
   }
+
+  /**
+   * Finds all {@link org.apache.ambari.server.orm.entities.StageEntity} that match the provided
+   * {@link org.apache.ambari.server.controller.spi.Predicate}. This method will make JPA do the heavy lifting
+   * of providing a slice of the result set.
+   *
+   * @param request
+   * @return
+   */
+  @Transactional
+  public List<StageEntity> findAll(Request request, Predicate predicate) {
+    EntityManager entityManager = entityManagerProvider.get();
+
+    // convert the Ambari predicate into a JPA predicate
+    StagePredicateVisitor visitor = new StagePredicateVisitor();
+    PredicateHelper.visit(predicate, visitor);
+
+    CriteriaQuery<StageEntity> query = visitor.getCriteriaQuery();
+    javax.persistence.criteria.Predicate jpaPredicate = visitor.getJpaPredicate();
+
+    if (jpaPredicate != null) {
+      query.where(jpaPredicate);
+    }
+
+    // sorting
+    JpaSortBuilder<StageEntity> sortBuilder = new JpaSortBuilder<StageEntity>();
+    List<Order> sortOrders = sortBuilder.buildSortOrders(request.getSortRequest(), visitor);
+    query.orderBy(sortOrders);
+
+    TypedQuery<StageEntity> typedQuery = entityManager.createQuery(query);
+
+    // !!! https://bugs.eclipse.org/bugs/show_bug.cgi?id=398067
+    // ensure that an associated entity with a JOIN is not stale; this causes
+    // the associated StageEntity to be stale
+    typedQuery.setHint(QueryHints.REFRESH, HintValues.TRUE);
+
+    return daoUtils.selectList(typedQuery);
+  }
+
+  /**
+   * The {@link org.apache.ambari.server.orm.dao.StageDAO.StagePredicateVisitor} is used to convert an Ambari
+   * {@link org.apache.ambari.server.controller.spi.Predicate} into a JPA {@link javax.persistence.criteria.Predicate}.
+   */
+  private final class StagePredicateVisitor extends
+      JpaPredicateVisitor<StageEntity> {
+
+    /**
+     * Constructor.
+     *
+     */
+    public StagePredicateVisitor() {
+      super(entityManagerProvider.get(), StageEntity.class);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Class<StageEntity> getEntityClass() {
+      return StageEntity.class;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public List<? extends SingularAttribute<?, ?>> getPredicateMapping(
+        String propertyId) {
+      return StageEntity_.getPredicateMapping().get(propertyId);
+    }
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/orm/entities/StageEntity_.java

@@ -0,0 +1,87 @@
+/*
+ * 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.orm.entities;
+
+import org.apache.ambari.server.controller.internal.StageResourceProvider;
+
+import javax.persistence.metamodel.SingularAttribute;
+import javax.persistence.metamodel.StaticMetamodel;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The {@link StageEntity_} is a strongly typed metamodel for creating
+ * {@link javax.persistence.criteria.CriteriaQuery} for {@link StageEntity}.
+ */
+@StaticMetamodel(StageEntity.class)
+public class StageEntity_ {
+  public static volatile SingularAttribute<StageEntity, Long> clusterId;
+  public static volatile SingularAttribute<StageEntity, Long> requestId;
+  public static volatile SingularAttribute<StageEntity, Long> stageId;
+
+  public static volatile SingularAttribute<StageEntity, String> logInfo;
+  public static volatile SingularAttribute<StageEntity, String> requestContext;
+
+  public static volatile SingularAttribute<StageEntity, byte[]> clusterHostInfo;
+  public static volatile SingularAttribute<StageEntity, byte[]> commandParamsStage;
+  public static volatile SingularAttribute<StageEntity, byte[]> hostParamsStage;
+
+  public static volatile SingularAttribute<StageEntity, RequestEntity> request;
+
+  /**
+   * Gets a mapping of between a resource provider property.
+   * <p/>
+   * This is used when converting an Ambari {@link org.apache.ambari.server.controller.spi.Predicate} into a JPA
+   * {@link javax.persistence.criteria.Predicate} and we need a type-safe
+   * conversion between "category/property" and JPA field names.
+   * <p/>
+   * Multiple {@link SingularAttribute} instances can be chained together in
+   * order to provide an {@code entity.subEntity.field} reference.
+   *
+   * @return a mapping of between a resource provider property
+   */
+  public static Map<String, List<? extends SingularAttribute<StageEntity, ?>>> getPredicateMapping() {
+    Map<String, List<? extends SingularAttribute<StageEntity, ?>>> mapping = new HashMap<String, List<? extends SingularAttribute<StageEntity, ?>>>();
+
+    mapping.put(StageResourceProvider.STAGE_REQUEST_ID,
+        Collections.singletonList(requestId));
+
+    mapping.put(StageResourceProvider.STAGE_STAGE_ID,
+        Collections.singletonList(stageId));
+
+    mapping.put(StageResourceProvider.STAGE_LOG_INFO,
+        Collections.singletonList(logInfo));
+
+    mapping.put(StageResourceProvider.STAGE_CONTEXT,
+        Collections.singletonList(requestContext));
+
+    mapping.put(StageResourceProvider.STAGE_CLUSTER_HOST_INFO,
+        Collections.singletonList(clusterHostInfo));
+
+    mapping.put(StageResourceProvider.STAGE_COMMAND_PARAMS,
+        Collections.singletonList(commandParamsStage));
+
+    mapping.put(StageResourceProvider.STAGE_HOST_PARAMS,
+        Collections.singletonList(hostParamsStage));
+
+    return mapping;
+  }
+}

ambari-server/src/test/java/org/apache/ambari/server/api/resources/SimpleResourceDefinitionTest.java

@@ -0,0 +1,45 @@
+/**
+ * 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.api.resources;
+
+import org.apache.ambari.server.controller.spi.Resource;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * SimpleResourceDefinition tests.
+ */
+public class SimpleResourceDefinitionTest {
+
+  @Test
+  public void testGetPluralName() throws Exception {
+    ResourceDefinition resourceDefinition =
+        new SimpleResourceDefinition(Resource.Type.Stage, "stage", "stages", Resource.Type.Task);
+
+    assertEquals("stages", resourceDefinition.getPluralName());
+  }
+
+  @Test
+  public void testGetSingularName() throws Exception {
+    ResourceDefinition resourceDefinition =
+        new SimpleResourceDefinition(Resource.Type.Stage, "stage", "stages", Resource.Type.Task);
+
+    assertEquals("stage", resourceDefinition.getSingularName());
+  }
+}

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

@@ -48,8 +48,6 @@ import org.apache.ambari.server.controller.spi.Predicate;
 import org.apache.ambari.server.controller.spi.PropertyProvider;
 import org.apache.ambari.server.controller.spi.ProviderModule;
 import org.apache.ambari.server.controller.spi.Request;
-import org.apache.ambari.server.controller.spi.Request.PageInfo;
-import org.apache.ambari.server.controller.spi.Request.SortInfo;
 import org.apache.ambari.server.controller.spi.RequestStatus;
 import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
@@ -907,7 +905,7 @@ public class ClusterControllerImplTest {
   }
 
   /**
-   * Tests that when a {@link ResourceProviderPageResponse} is present on the
+   * Tests that when a {@link PageResponse} is present on the
    * {@link Request}, in-memory paging is not performed.
    *
    * @throws Exception
@@ -951,21 +949,15 @@ public class ClusterControllerImplTest {
     Set<Resource> providerResources = new LinkedHashSet<Resource>();
     providerResources.add(new ResourceImpl(Resource.Type.AlertHistory));
 
-    PageInfo pageInfo = new PageInfo(pageRequest);
-    pageInfo.setResponsePaged(true);
-
-    SortInfo sortInfo = new SortInfo(sortRequest);
-    sortInfo.setResponseSorted(true);
-
     Request request = PropertyHelper.getReadRequest(propertyIds, null, null,
-        pageInfo, sortInfo);
+        pageRequest, sortRequest);
 
     Predicate predicate = new PredicateBuilder().property(
         AlertHistoryResourceProvider.ALERT_HISTORY_HOSTNAME).equals(
         "c6401.ambari.apache.org").toPredicate();
 
     PageResponse pageResponse = controller.getPage(Resource.Type.AlertHistory,
-        providerResources, request, predicate, pageRequest, sortRequest);
+        new QueryResponseImpl(providerResources, true, true, 0), request, predicate, pageRequest, sortRequest);
 
     verify(providerModule, resourceProvider, pageRequest, sortRequest);
   }

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

@@ -0,0 +1,83 @@
+/*
+ * 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.internal;
+
+import junit.framework.Assert;
+import org.apache.ambari.server.controller.spi.QueryResponse;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.junit.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * QueryResponseImpl tests
+ */
+public class QueryResponseImplTest {
+
+  @Test
+  public void testGetResources() throws Exception {
+    Set<Resource> resources = new HashSet<Resource>();
+    resources.add(new ResourceImpl(Resource.Type.Stage));
+
+    QueryResponse queryResponse = new QueryResponseImpl(resources);
+    Assert.assertEquals(resources, queryResponse.getResources());
+  }
+
+  @Test
+  public void testIsSortedResponse() throws Exception {
+    Set<Resource> resources = new HashSet<Resource>();
+    resources.add(new ResourceImpl(Resource.Type.Stage));
+
+    QueryResponse queryResponse = new QueryResponseImpl(resources);
+    Assert.assertFalse(queryResponse.isSortedResponse());
+
+    queryResponse = new QueryResponseImpl(resources, false, true, 0);
+    Assert.assertFalse(queryResponse.isSortedResponse());
+
+    queryResponse = new QueryResponseImpl(resources, true, true, 0);
+    Assert.assertTrue(queryResponse.isSortedResponse());
+  }
+
+  @Test
+  public void testIsPagedResponse() throws Exception {
+    Set<Resource> resources = new HashSet<Resource>();
+    resources.add(new ResourceImpl(Resource.Type.Stage));
+
+    QueryResponse queryResponse = new QueryResponseImpl(resources);
+    Assert.assertFalse(queryResponse.isPagedResponse());
+
+    queryResponse = new QueryResponseImpl(resources, true, false, 0);
+    Assert.assertFalse(queryResponse.isPagedResponse());
+
+    queryResponse = new QueryResponseImpl(resources, true, true, 0);
+    Assert.assertTrue(queryResponse.isPagedResponse());
+  }
+
+  @Test
+  public void testGetTotalResourceCount() throws Exception {
+    Set<Resource> resources = new HashSet<Resource>();
+    resources.add(new ResourceImpl(Resource.Type.Stage));
+
+    QueryResponse queryResponse = new QueryResponseImpl(resources);
+    Assert.assertEquals(0, queryResponse.getTotalResourceCount());
+
+    queryResponse = new QueryResponseImpl(resources, true, false, 99);
+    Assert.assertEquals(99, queryResponse.getTotalResourceCount());
+  }
+}

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

@@ -0,0 +1,202 @@
+/**
+ * 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.internal;
+
+import com.google.inject.Binder;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Module;
+import com.google.inject.util.Modules;
+import org.apache.ambari.server.actionmanager.HostRoleStatus;
+import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.QueryResponse;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.Resource;
+import org.apache.ambari.server.metadata.ActionMetadata;
+import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
+import org.apache.ambari.server.orm.dao.StageDAO;
+import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
+import org.apache.ambari.server.orm.entities.StageEntity;
+import org.apache.ambari.server.state.Cluster;
+import org.apache.ambari.server.state.Clusters;
+import org.easymock.EasyMock;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+import static org.easymock.EasyMock.createNiceMock;
+import static org.easymock.EasyMock.createStrictMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.junit.Assert.*;
+
+public class StageResourceProviderTest {
+
+  private StageDAO dao = null;
+  private Injector injector;
+
+  @Before
+  public void before() {
+    dao = createStrictMock(StageDAO.class);
+
+    // create an injector which will inject the mocks
+    injector = Guice.createInjector(Modules.override(
+        new InMemoryDefaultTestModule()).with(new MockModule()));
+
+    Assert.assertNotNull(injector);
+  }
+
+
+  @Test
+  public void testCreateResources() throws Exception {
+    StageResourceProvider provider = new StageResourceProvider();
+
+    Request request = createNiceMock(Request.class);
+    try {
+      provider.createResources(request);
+      fail("Expected UnsupportedOperationException");
+    } catch (UnsupportedOperationException e) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testUpdateResources() throws Exception {
+    StageResourceProvider provider = new StageResourceProvider();
+
+    Request request = createNiceMock(Request.class);
+    Predicate predicate = createNiceMock(Predicate.class);
+    try {
+      provider.updateResources(request, predicate);
+      fail("Expected UnsupportedOperationException");
+    } catch (UnsupportedOperationException e) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testDeleteResources() throws Exception {
+    StageResourceProvider provider = new StageResourceProvider();
+
+    Predicate predicate = createNiceMock(Predicate.class);
+    try {
+      provider.deleteResources(predicate);
+      fail("Expected UnsupportedOperationException");
+    } catch (UnsupportedOperationException e) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testGetResources() throws Exception {
+    StageResourceProvider provider = new StageResourceProvider();
+
+    Request request = createNiceMock(Request.class);
+    Predicate predicate = createNiceMock(Predicate.class);
+
+    List<StageEntity> entities = getStageEntities();
+
+    expect(dao.findAll(request, predicate)).andReturn(entities);
+
+    replay(dao, request, predicate);
+
+    Set<Resource> resources = provider.getResources(request, predicate);
+
+    Assert.assertEquals(1, resources.size());
+
+    Resource resource = resources.iterator().next();
+
+    Assert.assertEquals(100.0, resource.getPropertyValue(StageResourceProvider.STAGE_PROGRESS_PERCENT));
+    Assert.assertEquals("COMPLETED", resource.getPropertyValue(StageResourceProvider.STAGE_STATUS));
+    Assert.assertEquals(1000L, resource.getPropertyValue(StageResourceProvider.STAGE_START_TIME));
+    Assert.assertEquals(2500L, resource.getPropertyValue(StageResourceProvider.STAGE_END_TIME));
+
+    verify(dao);
+
+  }
+
+  @Test
+  public void testQueryForResources() throws Exception {
+    StageResourceProvider provider = new StageResourceProvider();
+
+    Request request = createNiceMock(Request.class);
+    Predicate predicate = createNiceMock(Predicate.class);
+
+    List<StageEntity> entities = getStageEntities();
+
+    expect(dao.findAll(request, predicate)).andReturn(entities);
+
+    replay(dao, request, predicate);
+
+    QueryResponse response =  provider.queryForResources(request, predicate);
+
+    Set<Resource> resources = response.getResources();
+
+    Assert.assertEquals(1, resources.size());
+
+    Assert.assertFalse(response.isSortedResponse());
+    Assert.assertFalse(response.isPagedResponse());
+    Assert.assertEquals(1, response.getTotalResourceCount());
+
+    verify(dao);
+  }
+
+  private List<StageEntity> getStageEntities() {
+    StageEntity stage = new StageEntity();
+
+    HostRoleCommandEntity task1 = new HostRoleCommandEntity();
+    task1.setStatus(HostRoleStatus.COMPLETED);
+    task1.setStartTime(1000L);
+    task1.setEndTime(2000L);
+
+    HostRoleCommandEntity task2 = new HostRoleCommandEntity();
+    task2.setStatus(HostRoleStatus.COMPLETED);
+    task2.setStartTime(1500L);
+    task2.setEndTime(2500L);
+
+
+    Collection<HostRoleCommandEntity> tasks = new HashSet<HostRoleCommandEntity>();
+    tasks.add(task1);
+    tasks.add(task2);
+
+    stage.setHostRoleCommands(tasks);
+
+    List<StageEntity> entities = new LinkedList<StageEntity>();
+    entities.add(stage);
+    return entities;
+  }
+
+  private class MockModule implements Module {
+    @Override
+    public void configure(Binder binder) {
+      binder.bind(StageDAO.class).toInstance(dao);
+      binder.bind(Clusters.class).toInstance(
+          EasyMock.createNiceMock(Clusters.class));
+      binder.bind(Cluster.class).toInstance(
+          EasyMock.createNiceMock(Cluster.class));
+      binder.bind(ActionMetadata.class);
+    }
+  }
+}

ambari-server/src/test/java/org/apache/ambari/server/orm/dao/StageDAOTest.java

@@ -0,0 +1,176 @@
+/**
+ * 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.orm.dao;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.persist.PersistService;
+import org.apache.ambari.server.controller.internal.StageResourceProvider;
+import org.apache.ambari.server.controller.internal.SortRequestImpl;
+import org.apache.ambari.server.controller.spi.Predicate;
+import org.apache.ambari.server.controller.spi.Request;
+import org.apache.ambari.server.controller.spi.SortRequest;
+import org.apache.ambari.server.controller.spi.SortRequestProperty;
+import org.apache.ambari.server.controller.utilities.PredicateBuilder;
+import org.apache.ambari.server.controller.utilities.PropertyHelper;
+import org.apache.ambari.server.orm.GuiceJpaInitializer;
+import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
+import org.apache.ambari.server.orm.OrmTestHelper;
+import org.apache.ambari.server.orm.entities.RequestEntity;
+import org.apache.ambari.server.orm.entities.StageEntity;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * StageDAO tests.
+ */
+public class StageDAOTest {
+
+  private Injector injector;
+  private StageDAO stageDao;
+
+  @Before
+  public void setup() throws Exception {
+    injector = Guice.createInjector(new InMemoryDefaultTestModule());
+    injector.getInstance(GuiceJpaInitializer.class);
+
+    stageDao = injector.getInstance(StageDAO.class);
+
+    OrmTestHelper helper     = injector.getInstance(OrmTestHelper.class);
+    Long          clusterId  = helper.createCluster();
+    RequestDAO    requestDao = injector.getInstance(RequestDAO.class);
+
+    RequestEntity requestEntity = new RequestEntity();
+    requestEntity.setClusterId(clusterId);
+    requestEntity.setStartTime(1000L);
+    requestEntity.setEndTime(1200L);
+    requestEntity.setRequestId(99L);
+    requestDao.create(requestEntity);
+
+    for (int i = 0; i < 5; i++) {
+      StageEntity definition = new StageEntity();
+      definition.setClusterId(clusterId);
+      definition.setRequestId(99L);
+      definition.setStageId((long) (100 + i));
+      definition.setLogInfo("log info for " + i);
+      definition.setRequestContext("request context for " + i);
+      definition.setRequest(requestEntity);
+      stageDao.create(definition);
+    }
+
+    List<StageEntity> definitions = stageDao.findAll();
+    assertNotNull(definitions);
+    assertEquals(5, definitions.size());
+  }
+
+  @After
+  public void teardown() {
+    injector.getInstance(PersistService.class).stop();
+    injector = null;
+  }
+
+  /**
+   * Tests that the Ambari {@link org.apache.ambari.server.controller.spi.Predicate}
+   * can be converted and submitted to JPA correctly to return a restricted result set.
+   */
+  @Test
+  public void testStagePredicate() throws Exception {
+
+    Predicate predicate = new PredicateBuilder().property(
+        StageResourceProvider.STAGE_CLUSTER_NAME).equals("c1").toPredicate();
+
+    List<StageEntity> entities = stageDao.findAll(PropertyHelper.getReadRequest(), predicate);
+    assertEquals(5, entities.size());
+
+    predicate = new PredicateBuilder().
+        property(StageResourceProvider.STAGE_CONTEXT).equals("request context for 3").or().
+        property(StageResourceProvider.STAGE_CONTEXT).equals("request context for 4").
+        toPredicate();
+
+    entities = stageDao.findAll(PropertyHelper.getReadRequest(), predicate);
+    assertEquals(2, entities.size());
+  }
+
+  /**
+   * Tests that JPA does the sorting work for us.
+   */
+  @Test
+  public void testStageSorting() throws Exception {
+    List<SortRequestProperty> sortProperties = new ArrayList<SortRequestProperty>();
+    SortRequest sortRequest = new SortRequestImpl(sortProperties);
+
+    Predicate predicate = new PredicateBuilder().property(
+        StageResourceProvider.STAGE_CLUSTER_NAME).equals("c1").toPredicate();
+
+
+    sortProperties.add(new SortRequestProperty(
+        StageResourceProvider.STAGE_LOG_INFO, SortRequest.Order.ASC));
+
+    Request request = PropertyHelper.getReadRequest(new HashSet<String>(Arrays.<String>asList()),
+        null, null, null, sortRequest);
+
+    // get back all 5
+    List<StageEntity> entities = stageDao.findAll(request, predicate);
+    assertEquals(5, entities.size());
+
+    // assert sorting ASC
+    String lastInfo = null;
+    for (StageEntity entity : entities) {
+      if (lastInfo == null) {
+        lastInfo = entity.getLogInfo();
+        continue;
+      }
+
+      String currentInfo = entity.getLogInfo();
+      assertTrue(lastInfo.compareTo(currentInfo) <= 0);
+      lastInfo = currentInfo;
+    }
+
+    // clear and do DESC
+    sortProperties.clear();
+    sortProperties.add(new SortRequestProperty(
+        StageResourceProvider.STAGE_LOG_INFO, SortRequest.Order.DESC));
+
+    // get back all 5
+    entities = stageDao.findAll(request, predicate);
+    assertEquals(5, entities.size());
+
+    // assert sorting DESC
+    lastInfo = null;
+    for (StageEntity entity : entities) {
+      if (null == lastInfo) {
+        lastInfo = entity.getLogInfo();
+        continue;
+      }
+
+      String currentInfo = entity.getLogInfo();
+      assertTrue(lastInfo.compareTo(currentInfo) >= 0);
+      lastInfo = currentInfo;
+    }
+  }
+}