@GET @Path("/estimator/{pipelineId}") @Produces(MediaType.APPLICATION_JSON)
  public String getPrediction(
      @PathParam(value = "pipelineId") String pipelineId)
      throws SolverException, SkylineStoreException {
    // first, try to grab the predicted resource allocation from the skyline
    // store
    RLESparseResourceAllocation result = skylineStore.getEstimation(pipelineId);
    // if received resource allocation is null, then run the solver
    if (result == null) {
      RecurrenceId recurrenceId = new RecurrenceId(pipelineId, "*");
      Map<RecurrenceId, List<ResourceSkyline>> jobHistory =
          skylineStore.getHistory(recurrenceId);
      result = solver.solve(jobHistory);
    }
    final String prediction = gson.toJson(result, rleType);
    LOGGER.debug("Predict resource requests for pipelineId: {}." + pipelineId);

    return prediction;
  }
@Test public void testGetPrediction() {
    // first, parse the log
    final String logFile = "resourceEstimatorService.txt";
    WebResource webResource = resource();
    webResource.path(parseLogCommand).type(MediaType.APPLICATION_XML_TYPE)
        .post(logFile);
    webResource = resource().path(getHistorySkylineCommand);
    String response = webResource.get(String.class);
    Map<RecurrenceId, List<ResourceSkyline>> jobHistory =
        gson.fromJson(response,
            new TypeToken<Map<RecurrenceId, List<ResourceSkyline>>>() {
            }.getType());
    checkResult("tpch_q12_0", jobHistory);
    checkResult("tpch_q12_1", jobHistory);
    // then, try to get estimated resource allocation from skyline store
    webResource = resource().path(getEstimatedSkylineCommand);
    response = webResource.get(String.class);
    Assert.assertEquals("null", response);
    // then, we call estimator module to make the prediction
    webResource = resource().path(makeEstimationCommand);
    response = webResource.get(String.class);
    RLESparseResourceAllocation skylineList =
        gson.fromJson(response, new TypeToken<RLESparseResourceAllocation>() {
        }.getType());
    Assert.assertEquals(1,
        skylineList.getCapacityAtTime(0).getMemorySize() / containerMemAlloc);
    Assert.assertEquals(1058,
        skylineList.getCapacityAtTime(10).getMemorySize() / containerMemAlloc);
    Assert.assertEquals(2538,
        skylineList.getCapacityAtTime(15).getMemorySize() / containerMemAlloc);
    Assert.assertEquals(2484,
        skylineList.getCapacityAtTime(20).getMemorySize() / containerMemAlloc);
    // then, we get estimated resource allocation for tpch_q12
    webResource = resource().path(getEstimatedSkylineCommand);
    response = webResource.get(String.class);
    final RLESparseResourceAllocation skylineList2 =
        gson.fromJson(response, new TypeToken<RLESparseResourceAllocation>() {
        }.getType());
    compareRLESparseResourceAllocation(skylineList, skylineList2);
    // then, we call estimator module again to directly get estimated resource
    // allocation from skyline store
    webResource = resource().path(makeEstimationCommand);
    response = webResource.get(String.class);
    final RLESparseResourceAllocation skylineList3 =
        gson.fromJson(response, new TypeToken<RLESparseResourceAllocation>() {
        }.getType());
    compareRLESparseResourceAllocation(skylineList, skylineList3);
    // finally, test delete
    webResource = resource().path(deleteHistoryCommand);
    webResource.delete();
    webResource = resource().path(getHistorySkylineCommand);
    response = webResource.get(String.class);
    jobHistory = gson.fromJson(response,
        new TypeToken<Map<RecurrenceId, List<ResourceSkyline>>>() {
        }.getType());
    // jobHistory should only have info for tpch_q12_0
    Assert.assertEquals(1, jobHistory.size());
    final String pipelineId =
        ((RecurrenceId) jobHistory.keySet().toArray()[0]).getRunId();
    Assert.assertEquals("tpch_q12_0", pipelineId);
  }