error propagation for API calls

src/main/java/io/cdap/plugin/servicenow/ServiceNowBaseConfig.java

@@ -137,7 +137,7 @@ public void validateTable(String tableName, SourceValueType valueType, FailureCo
       // Get the response JSON and fetch the header X-Total-Count. Set the value to recordCount
       requestBuilder.setResponseHeaders(ServiceNowConstants.HEADER_NAME_TOTAL_COUNT);
 
-      apiResponse = serviceNowTableAPIClient.executeGet(requestBuilder.build());
+      apiResponse = serviceNowTableAPIClient.executeGetWithRetries(requestBuilder.build());
       if (serviceNowTableAPIClient.parseResponseToResultListOfMap(apiResponse.getResponseBody()).isEmpty()) {
         // Removed config property as in case of MultiSource, only first table error was populating.
         collector.addFailure("Table: " + tableName + " is empty.", "");

src/main/java/io/cdap/plugin/servicenow/apiclient/ServiceNowAPIException.java

@@ -0,0 +1,77 @@
+package io.cdap.plugin.servicenow.apiclient;
+
+import io.cdap.plugin.servicenow.util.ServiceNowConstants;
+
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.oltu.oauth2.common.exception.OAuthSystemException;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import javax.annotation.Nullable;
+
+/**
+ * Custom Exception class for propagating API errors/exceptions back to caller.
+ */
+public class ServiceNowAPIException extends Exception {
+
+  @Nullable private final HttpResponse httpResponse;
+  @Nullable private final boolean manualRetry;
+
+  private static final Set<Integer> RETRYABLE_CODES = new HashSet<>(Arrays.asList(429,
+      HttpStatus.SC_BAD_GATEWAY,
+      HttpStatus.SC_SERVICE_UNAVAILABLE,
+      HttpStatus.SC_REQUEST_TIMEOUT,
+      HttpStatus.SC_GATEWAY_TIMEOUT));
+
+  public ServiceNowAPIException(String message, @Nullable HttpResponse httpResponse) {
+    super(message);
+    this.httpResponse = httpResponse;
+    this.manualRetry = false;
+  }
+
+  public ServiceNowAPIException(Throwable t, @Nullable HttpResponse httpResponse) {
+    super(t);
+    this.httpResponse = httpResponse;
+    this.manualRetry = false;
+
+  }
+
+  public ServiceNowAPIException(String message, Throwable t,
+      @Nullable HttpResponse httpResponse, boolean manualRetry) {
+    super(message, t);
+    this.httpResponse = httpResponse;
+    this.manualRetry = manualRetry;
+  }
+
+  public String getUnderlyingMessage() {
+    if (this.getCause() != null) {
+      return this.getCause().getMessage();
+    }
+    return null;
+  }
+
+  @Nullable
+  public HttpResponse getHttpResponse() {
+    return httpResponse;
+  }
+
+  public int getStatusCode() {
+    if (httpResponse != null && httpResponse.getStatusLine() != null) {
+      return httpResponse.getStatusLine().getStatusCode();
+    }
+    return 0;
+  }
+
+  public boolean isErrorRetryable() {
+    if (manualRetry) {
+      return true;
+    }
+    Throwable t = this.getCause();
+    return t instanceof OAuthSystemException
+        || (this.getMessage() != null
+        && this.getMessage().contains(ServiceNowConstants.MAXIMUM_EXECUTION_TIME_EXCEEDED))
+        || RETRYABLE_CODES.contains(getStatusCode());
+  }
+}

src/main/java/io/cdap/plugin/servicenow/apiclient/ServiceNowTableAPIClientImpl.java

@@ -56,6 +56,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 import javax.annotation.Nullable;
 
 /**
@@ -76,10 +77,18 @@ public ServiceNowTableAPIClientImpl(ServiceNowConnectorConfig conf) {
     this.conf = conf;
   }
 
-  public String getAccessToken() throws OAuthSystemException, OAuthProblemException {
-    return generateAccessToken(String.format(OAUTH_URL_TEMPLATE, conf.getRestApiEndpoint()),
-                               conf.getClientId(),
-                               conf.getClientSecret(), conf.getUser(), conf.getPassword());
+  public String getAccessToken() throws ServiceNowAPIException {
+    try {
+      return generateAccessToken(String.format(OAUTH_URL_TEMPLATE, conf.getRestApiEndpoint()),
+          conf.getClientId(),
+          conf.getClientSecret(), conf.getUser(), conf.getPassword());
+    } catch (OAuthProblemException | OAuthSystemException e) {
+      throw new ServiceNowAPIException("An error occurred while authenticating.", e, null, false);
+    }
+  }
+
+  private boolean isExceptionRetryable(Throwable t) {
+    return t instanceof ServiceNowAPIException && ((ServiceNowAPIException) t).isErrorRetryable();
   }
 
   /**
@@ -90,6 +99,7 @@ public String getAccessTokenRetryableMode() throws ExecutionException, RetryExce
     Callable fetchToken = this::getAccessToken;
 
     Retryer<String> retryer = RetryerBuilder.<String>newBuilder()
+      .retryIfException(this::isExceptionRetryable)
       .retryIfExceptionOfType(OAuthSystemException.class)
       .withWaitStrategy(WaitStrategies.fixedWait(ServiceNowConstants.BASE_DELAY, TimeUnit.MILLISECONDS))
       .withStopStrategy(StopStrategies.stopAfterAttempt(ServiceNowConstants.MAX_NUMBER_OF_RETRY_ATTEMPTS))
@@ -109,8 +119,14 @@ public String getAccessTokenRetryableMode() throws ExecutionException, RetryExce
    * @param limit     The number of records to be fetched
    * @return The list of Map; each Map representing a table row
    */
-  public List<Map<String, String>> fetchTableRecords(String tableName, SourceValueType valueType, String startDate,
-                                                     String endDate, int offset, int limit) throws IOException {
+  public List<Map<String, String>> fetchTableRecords(
+      String tableName,
+      SourceValueType valueType,
+      String startDate,
+      String endDate,
+      int offset,
+      int limit)
+      throws ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
       this.conf.getRestApiEndpoint(), tableName, false)
       .setExcludeReferenceLink(true)
@@ -123,16 +139,10 @@ public List<Map<String, String>> fetchTableRecords(String tableName, SourceValue
 
     applyDateRangeToRequest(requestBuilder, startDate, endDate);
 
-    try {
-      String accessToken = getAccessToken();
-      requestBuilder.setAuthHeader(accessToken);
-      RestAPIResponse apiResponse = executeGet(requestBuilder.build());
-      return parseResponseToResultListOfMap(apiResponse.getResponseBody());
-    } catch (OAuthSystemException e) {
-      throw new RetryableException("Authentication error occurred", e);
-    } catch (OAuthProblemException e) {
-      throw new IOException("Problem occurred while authenticating", e);
-    }
+    String accessToken = getAccessToken();
+    requestBuilder.setAuthHeader(accessToken);
+    RestAPIResponse apiResponse = executeGetWithRetries(requestBuilder.build());
+    return parseResponseToResultListOfMap(apiResponse.getResponseBody());
   }
 
   private void applyDateRangeToRequest(ServiceNowTableAPIRequestBuilder requestBuilder, String startDate,
@@ -212,23 +222,25 @@ private String getErrorMessage(String responseBody) {
    */
   public List<Map<String, String>> fetchTableRecordsRetryableMode(String tableName, SourceValueType valueType,
                                                                   String startDate, String endDate, int offset,
-                                                                  int limit) throws IOException {
+                                                                  int limit) throws ServiceNowAPIException {
     final List<Map<String, String>> results = new ArrayList<>();
     Callable<Boolean> fetchRecords = () -> {
       results.addAll(fetchTableRecords(tableName, valueType, startDate, endDate, offset, limit));
       return true;
     };
 
     Retryer<Boolean> retryer = RetryerBuilder.<Boolean>newBuilder()
-      .retryIfExceptionOfType(RetryableException.class)
+      .retryIfException(this::isExceptionRetryable)
       .withWaitStrategy(WaitStrategies.exponentialWait(ServiceNowConstants.WAIT_TIME, TimeUnit.MILLISECONDS))
       .withStopStrategy(StopStrategies.stopAfterAttempt(ServiceNowConstants.MAX_NUMBER_OF_RETRY_ATTEMPTS))
       .build();
 
     try {
       retryer.call(fetchRecords);
     } catch (RetryException | ExecutionException e) {
-      throw new IOException(String.format("Data Recovery failed for batch %s to %s.", offset, (offset + limit)), e);
+      throw new ServiceNowAPIException(
+          String.format("Data Recovery failed for batch %s to %s.", offset, (offset + limit)),
+          e, null, false);
     }
 
     return results;
@@ -262,12 +274,11 @@ public SchemaResponse parseSchemaResponse(String responseBody) {
    *
    * @param tableName ServiceNow table name for which schema is getting fetched
    * @return schema for given ServiceNow table
-   * @throws OAuthProblemException
-   * @throws OAuthSystemException
+   * @throws ServiceNowAPIException
    */
   public Schema fetchTableSchema(String tableName)
-      throws OAuthProblemException, OAuthSystemException, IOException {
-    return fetchTableSchema(tableName, getAccessToken());
+      throws ServiceNowAPIException {
+      return fetchTableSchema(tableName, getAccessToken());
   }
 
   /**
@@ -277,14 +288,15 @@ public Schema fetchTableSchema(String tableName)
    * @param accessToken Access Token to use
    * @return schema for given ServiceNow table
    */
-  public Schema fetchTableSchema(String tableName, String accessToken) throws IOException {
+  public Schema fetchTableSchema(String tableName, String accessToken)
+      throws ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
       this.conf.getRestApiEndpoint(), tableName, true)
       .setExcludeReferenceLink(true);
 
     RestAPIResponse apiResponse;
     requestBuilder.setAuthHeader(accessToken);
-    apiResponse = executeGet(requestBuilder.build());
+    apiResponse = executeGetWithRetries(requestBuilder.build());
     SchemaResponse response = parseSchemaResponse(apiResponse.getResponseBody());
     List<ServiceNowColumn> columns = new ArrayList<>();
 
@@ -302,11 +314,10 @@ public Schema fetchTableSchema(String tableName, String accessToken) throws IOEx
    *
    * @param tableName ServiceNow table name for which record count is fetched.
    * @return the table record count
-   * @throws OAuthProblemException
-   * @throws OAuthSystemException
+   * @throws ServiceNowAPIException
    */
   public int getTableRecordCount(String tableName)
-      throws OAuthProblemException, OAuthSystemException, IOException {
+      throws ServiceNowAPIException {
     return getTableRecordCount(tableName, getAccessToken());
   }
 
@@ -316,9 +327,9 @@ public int getTableRecordCount(String tableName)
    * @param tableName ServiceNow table name for which record count is fetched.
    * @param accessToken Access Token for the call
    * @return the table record count
-   * @throws IOException
+   * @throws ServiceNowAPIException
    */
-  public int getTableRecordCount(String tableName, String accessToken) throws IOException {
+  public int getTableRecordCount(String tableName, String accessToken) throws ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
       this.conf.getRestApiEndpoint(), tableName, false)
       .setExcludeReferenceLink(true)
@@ -327,7 +338,7 @@ public int getTableRecordCount(String tableName, String accessToken) throws IOEx
     RestAPIResponse apiResponse = null;
     requestBuilder.setResponseHeaders(ServiceNowConstants.HEADER_NAME_TOTAL_COUNT);
     requestBuilder.setAuthHeader(accessToken);
-    apiResponse = executeGet(requestBuilder.build());
+    apiResponse = executeGetWithRetries(requestBuilder.build());
     return getRecordCountFromHeader(apiResponse);
   }
 
@@ -338,7 +349,7 @@ public int getTableRecordCount(String tableName, String accessToken) throws IOEx
    * @param entity    Details of the Record to be created
    * @description This function is being used in end-to-end (e2e) tests to fetch a record from the ServiceNow Table.
    */
-  public String createRecord(String tableName, HttpEntity entity) throws IOException {
+  public String createRecord(String tableName, HttpEntity entity) throws IOException, ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
       this.conf.getRestApiEndpoint(), tableName, false);
     String systemID;
@@ -352,8 +363,8 @@ public String createRecord(String tableName, HttpEntity entity) throws IOExcepti
       apiResponse = executePost(requestBuilder.build());
 
       systemID = String.valueOf(getSystemId(apiResponse));
-    } catch (OAuthSystemException | OAuthProblemException | UnsupportedEncodingException e) {
-      throw new IOException("Error in creating a new record", e);
+    } catch (IOException e) {
+      throw new ServiceNowAPIException("Error in creating a new record", e, null, false);
     }
     return systemID;
   }
@@ -372,7 +383,7 @@ private String getSystemId(RestAPIResponse restAPIResponse) {
    * @param query The query
    */
   public Map<String, String> getRecordFromServiceNowTable(String tableName, String query)
-      throws OAuthProblemException, OAuthSystemException, IOException {
+      throws ServiceNowAPIException {
 
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
       this.conf.getRestApiEndpoint(), tableName, false)
@@ -381,7 +392,7 @@ public Map<String, String> getRecordFromServiceNowTable(String tableName, String
     RestAPIResponse restAPIResponse;
     String accessToken = getAccessToken();
     requestBuilder.setAuthHeader(accessToken);
-    restAPIResponse = executeGet(requestBuilder.build());
+    restAPIResponse = executeGetWithRetries(requestBuilder.build());
 
     APIResponse apiResponse = GSON.fromJson(restAPIResponse.getResponseBody(), APIResponse.class);
     return apiResponse.getResult().get(0);