omniture-lib/src/main/java/com/streamsets/pipeline/stage/origin/omniture/OmniturePollingConsumer.java
public void getReport(int reportId)
      throws InterruptedException, ExecutionException, TimeoutException, IOException, StageException {
    int waitTime = 1000;
    Response response = null;
    while (!stop) {
      final AsyncInvoker asyncInvoker = getResource.request()
          .header(WSSE_HEADER, OmnitureAuthUtil.getHeader(username.get(), sharedSecret.get()))
          .async();

      LOG.debug("Getting report using URL {} with report ID {}", getResource.getUri().toURL().toString(), reportId);
      final Future<Response> responseFuture = asyncInvoker.post(Entity.json("{ \"reportID\": " + reportId + " }"));
      response = responseFuture.get(responseTimeoutMillis, TimeUnit.MILLISECONDS);
      String input = response.readEntity(String.class);

      ObjectMapper mapper = new ObjectMapper();
      JsonNode root = mapper.readTree(input);

      // If the report has an error field, it means the report has not finished generating
      if (!root.has("error")) {
        boolean accepted = entityQueue.offer(input, responseTimeoutMillis, TimeUnit.MILLISECONDS);
        if (!accepted) {
          LOG.warn("Response buffer full, dropped record.");
        }
        break;
      } else {
        // Exponential backoff while making subsequent Report.Get requests
        if (root.get("error").textValue().equals("report_not_ready")) {
          waitTime *= 2;
          LOG.info("Report not available. Sleeping for {} seconds", waitTime / 1000);
          Thread.sleep(waitTime);
        } else {
          throw new StageException(Errors.OMNITURE_02,
              root.get("error").get("error_description").asText());
        }
      }
    }
    response.close();
  }
public void getReport(int reportId)
      throws InterruptedException, ExecutionException, TimeoutException, IOException, StageException {
    int waitTime = 1000;
    Response response = null;
    while (!stop) {
      final AsyncInvoker asyncInvoker = getResource.request()
          .header(WSSE_HEADER, OmnitureAuthUtil.getHeader(username.get(), sharedSecret.get()))
          .async();

      LOG.debug("Getting report using URL {} with report ID {}", getResource.getUri().toURL().toString(), reportId);
      final Future<Response> responseFuture = asyncInvoker.post(Entity.json("{ \"reportID\": " + reportId + " }"));
      response = responseFuture.get(responseTimeoutMillis, TimeUnit.MILLISECONDS);
      String input = response.readEntity(String.class);

      ObjectMapper mapper = new ObjectMapper();
      JsonNode root = mapper.readTree(input);

      // If the report has an error field, it means the report has not finished generating
      if (!root.has("error")) {
        boolean accepted = entityQueue.offer(input, responseTimeoutMillis, TimeUnit.MILLISECONDS);
        if (!accepted) {
          LOG.warn("Response buffer full, dropped record.");
        }
        break;
      } else {
        // Exponential backoff while making subsequent Report.Get requests
        if (root.get("error").textValue().equals("report_not_ready")) {
          waitTime *= 2;
          LOG.info("Report not available. Sleeping for {} seconds", waitTime / 1000);
          Thread.sleep(waitTime);
        } else {
          throw new StageException(Errors.OMNITURE_02,
              root.get("error").get("error_description").asText());
        }
      }
    }
    response.close();
  }
Posts a request to the Omniture API to get a report back. Reports may take a while to generate,
so this will loop on the Report.Get request, and ignore any errors indicating that the report
is not yet ready.
@param reportId ID of report to get
@throws InterruptedException
@throws ExecutionException
@throws TimeoutException
@throws IOException