draft for 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,33 @@
+package io.cdap.plugin.servicenow.apiclient;
+
+import org.apache.http.HttpResponse;
+
+import javax.annotation.Nullable;
+
+public class ServiceNowAPIException extends Exception {
+
+  @Nullable private final HttpResponse httpResponse;
+  private final boolean isErrorRetryable;
+
+  public ServiceNowAPIException(
+      Throwable t, @Nullable HttpResponse httpResponse, boolean isErrorRetryable) {
+    super(t);
+    this.httpResponse = httpResponse;
+    this.isErrorRetryable = isErrorRetryable;
+  }
+
+  public ServiceNowAPIException(
+      String message, @Nullable HttpResponse httpResponse, boolean isErrorRetryable) {
+    super(message);
+    this.httpResponse = httpResponse;
+    this.isErrorRetryable = isErrorRetryable;
+  }
+
+  public HttpResponse getHttpResponse() {
+    return httpResponse;
+  }
+
+  public boolean isErrorRetryable() {
+    return isErrorRetryable;
+  }
+}

src/main/java/io/cdap/plugin/servicenow/connector/ServiceNowConnector.java

@@ -40,6 +40,7 @@
 import io.cdap.plugin.common.ConfigUtil;
 import io.cdap.plugin.common.Constants;
 import io.cdap.plugin.common.ReferenceNames;
+import io.cdap.plugin.servicenow.apiclient.ServiceNowAPIException;
 import io.cdap.plugin.servicenow.apiclient.ServiceNowTableAPIClientImpl;
 import io.cdap.plugin.servicenow.apiclient.ServiceNowTableAPIRequestBuilder;
 import io.cdap.plugin.servicenow.restapi.RestAPIResponse;
@@ -61,10 +62,7 @@
 import javax.annotation.Nullable;
 import javax.ws.rs.core.MediaType;
 
-
-/**
- * ServiceNow Connector Plugin
- */
+/** ServiceNow Connector Plugin */
 @Plugin(type = Connector.PLUGIN_TYPE)
 @Name(ServiceNowConstants.PLUGIN_NAME)
 @Description("Connection to access data in Servicenow tables.")
@@ -88,21 +86,21 @@ public void test(ConnectorContext connectorContext) throws ValidationException {
   }
 
   @Override
-  public BrowseDetail browse(ConnectorContext connectorContext, BrowseRequest browseRequest) throws IOException {
-    ServiceNowTableAPIClientImpl serviceNowTableAPIClient = new ServiceNowTableAPIClientImpl(config);
+  public BrowseDetail browse(ConnectorContext connectorContext, BrowseRequest browseRequest)
+      throws IOException {
+    ServiceNowTableAPIClientImpl serviceNowTableAPIClient =
+        new ServiceNowTableAPIClientImpl(config);
     try {
       String accessToken = serviceNowTableAPIClient.getAccessToken();
       return browse(connectorContext, accessToken);
-    } catch (OAuthSystemException | OAuthProblemException e) {
+    } catch (OAuthSystemException | OAuthProblemException | ServiceNowAPIException e) {
       throw new IOException(e);
     }
   }
 
-  /**
-   * Browse Details for the given AccessToken.
-   */
-  public BrowseDetail browse(ConnectorContext connectorContext,
-                             String accessToken) throws IOException {
+  /** Browse Details for the given AccessToken. */
+  public BrowseDetail browse(ConnectorContext connectorContext, String accessToken)
+      throws ServiceNowAPIException {
     int count = 0;
     FailureCollector collector = connectorContext.getFailureCollector();
     config.validateCredentialsFields(collector);
@@ -112,10 +110,12 @@ public BrowseDetail browse(ConnectorContext connectorContext,
     for (int i = 0; i < table.length; i++) {
       String name = table[i].getName();
       String label = table[i].getLabel();
-      BrowseEntity.Builder entity = (BrowseEntity.builder(name, name, ENTITY_TYPE_TABLE).
-        canBrowse(false).canSample(true));
-      entity.addProperty(LABEL_NAME, BrowseEntityPropertyValue.builder(label, BrowseEntityPropertyValue.
-        PropertyType.STRING).build());
+      BrowseEntity.Builder entity =
+          (BrowseEntity.builder(name, name, ENTITY_TYPE_TABLE).canBrowse(false).canSample(true));
+      entity.addProperty(
+          LABEL_NAME,
+          BrowseEntityPropertyValue.builder(label, BrowseEntityPropertyValue.PropertyType.STRING)
+              .build());
       browseDetailBuilder.addEntity(entity.build());
       count++;
     }
@@ -125,64 +125,73 @@ public BrowseDetail browse(ConnectorContext connectorContext,
   /**
    * @return the list of tables.
    */
-  private TableList listTables(String accessToken) throws IOException {
-    ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      config.getRestApiEndpoint(), OBJECT_TABLE_LIST, false);
+  private TableList listTables(String accessToken) throws ServiceNowAPIException {
+    ServiceNowTableAPIRequestBuilder requestBuilder =
+        new ServiceNowTableAPIRequestBuilder(config.getRestApiEndpoint(), OBJECT_TABLE_LIST, false);
     requestBuilder.setAuthHeader(accessToken);
     requestBuilder.setAcceptHeader(MediaType.APPLICATION_JSON);
     requestBuilder.setContentTypeHeader(MediaType.APPLICATION_JSON);
-    ServiceNowTableAPIClientImpl serviceNowTableAPIClient = new ServiceNowTableAPIClientImpl(config);
-    RestAPIResponse apiResponse = serviceNowTableAPIClient.executeGet(requestBuilder.build());
+    ServiceNowTableAPIClientImpl serviceNowTableAPIClient =
+        new ServiceNowTableAPIClientImpl(config);
+    RestAPIResponse apiResponse =
+        serviceNowTableAPIClient.executeGetWithRetries(requestBuilder.build());
     return GSON.fromJson(apiResponse.getResponseBody(), TableList.class);
   }
 
-  public ConnectorSpec generateSpec(ConnectorContext connectorContext, ConnectorSpecRequest connectorSpecRequest) {
+  public ConnectorSpec generateSpec(
+      ConnectorContext connectorContext, ConnectorSpecRequest connectorSpecRequest) {
     ConnectorSpec.Builder specBuilder = ConnectorSpec.builder();
     Map<String, String> properties = new HashMap<>();
     properties.put(io.cdap.plugin.common.ConfigUtil.NAME_USE_CONNECTION, "true");
     properties.put(ConfigUtil.NAME_CONNECTION, connectorSpecRequest.getConnectionWithMacro());
     String tableName = connectorSpecRequest.getPath();
     if (tableName != null) {
       properties.put(ServiceNowConstants.PROPERTY_TABLE_NAME, tableName);
-      properties.put(Constants.Reference.REFERENCE_NAME, ReferenceNames.cleanseReferenceName(tableName));
+      properties.put(
+          Constants.Reference.REFERENCE_NAME, ReferenceNames.cleanseReferenceName(tableName));
     }
     Schema schema = getSchema(tableName);
     if (schema != null) {
       specBuilder.setSchema(schema);
     }
-    return specBuilder.addRelatedPlugin(new PluginSpec(ServiceNowConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE,
-                                                       properties))
-      .addRelatedPlugin(new PluginSpec(ServiceNowConstants.PLUGIN_NAME, BatchSink.PLUGIN_TYPE, properties)).build();
+    return specBuilder
+        .addRelatedPlugin(
+            new PluginSpec(ServiceNowConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE, properties))
+        .addRelatedPlugin(
+            new PluginSpec(ServiceNowConstants.PLUGIN_NAME, BatchSink.PLUGIN_TYPE, properties))
+        .build();
   }
 
   @Override
-  public List<StructuredRecord> sample(ConnectorContext connectorContext, SampleRequest sampleRequest)
-    throws IOException {
+  public List<StructuredRecord> sample(
+      ConnectorContext connectorContext, SampleRequest sampleRequest) throws IOException {
     String table = sampleRequest.getPath();
     if (table == null) {
       throw new IllegalArgumentException("Path should contain table name.");
     }
     try {
       return getTableData(table, sampleRequest.getLimit());
-    } catch (OAuthProblemException | OAuthSystemException e) {
-      throw new IOException("Unable to fetch the data.");
+    } catch (OAuthProblemException | OAuthSystemException | ServiceNowAPIException e) {
+      throw new IOException("Unable to fetch the data.", e);
     }
   }
 
   private List<StructuredRecord> getTableData(String tableName, int limit)
-      throws OAuthProblemException, OAuthSystemException, IOException {
-    ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      config.getRestApiEndpoint(), tableName, false)
-      .setExcludeReferenceLink(true)
-      .setDisplayValue(SourceValueType.SHOW_DISPLAY_VALUE)
-      .setLimit(limit);
-    ServiceNowTableAPIClientImpl serviceNowTableAPIClient = new ServiceNowTableAPIClientImpl(config);
+      throws OAuthProblemException, OAuthSystemException, ServiceNowAPIException {
+    ServiceNowTableAPIRequestBuilder requestBuilder =
+        new ServiceNowTableAPIRequestBuilder(config.getRestApiEndpoint(), tableName, false)
+            .setExcludeReferenceLink(true)
+            .setDisplayValue(SourceValueType.SHOW_DISPLAY_VALUE)
+            .setLimit(limit);
+    ServiceNowTableAPIClientImpl serviceNowTableAPIClient =
+        new ServiceNowTableAPIClientImpl(config);
     String accessToken = serviceNowTableAPIClient.getAccessToken();
     requestBuilder.setAuthHeader(accessToken);
     requestBuilder.setResponseHeaders(ServiceNowConstants.HEADER_NAME_TOTAL_COUNT);
-    RestAPIResponse apiResponse = serviceNowTableAPIClient.executeGet(requestBuilder.build());
-    List<Map<String, String>> result = serviceNowTableAPIClient.parseResponseToResultListOfMap
-      (apiResponse.getResponseBody());
+    RestAPIResponse apiResponse =
+        serviceNowTableAPIClient.executeGetWithRetries(requestBuilder.build());
+    List<Map<String, String>> result =
+        serviceNowTableAPIClient.parseResponseToResultListOfMap(apiResponse.getResponseBody());
     List<StructuredRecord> recordList = new ArrayList<>();
     Schema schema = getSchema(tableName);
     if (schema != null) {
@@ -191,23 +200,25 @@ private List<StructuredRecord> getTableData(String tableName, int limit)
         StructuredRecord.Builder recordBuilder = StructuredRecord.builder(schema);
         for (Schema.Field field : tableFields) {
           String fieldName = field.getName();
-          ServiceNowRecordConverter.convertToValue(fieldName, field.getSchema(), result.get(i), recordBuilder);
+          ServiceNowRecordConverter.convertToValue(
+              fieldName, field.getSchema(), result.get(i), recordBuilder);
         }
         StructuredRecord structuredRecord = recordBuilder.build();
         recordList.add(structuredRecord);
       }
     }
     return recordList;
-
   }
 
   @Nullable
   private Schema getSchema(String tableName) {
     SourceQueryMode mode = SourceQueryMode.TABLE;
-    List<ServiceNowTableInfo> tableInfo = ServiceNowInputFormat.fetchTableInfo(mode, config, tableName,
-                                                                               null);
-    Schema schema = tableInfo.stream().findFirst().isPresent() ? tableInfo.stream().findFirst().get().getSchema() :
-      null;
+    List<ServiceNowTableInfo> tableInfo =
+        ServiceNowInputFormat.fetchTableInfo(mode, config, tableName, null);
+    Schema schema =
+        tableInfo.stream().findFirst().isPresent()
+            ? tableInfo.stream().findFirst().get().getSchema()
+            : null;
     return schema;
   }
 }

src/main/java/io/cdap/plugin/servicenow/restapi/RestAPIClient.java

@@ -16,9 +16,18 @@
 
 package io.cdap.plugin.servicenow.restapi;
 
+import com.github.rholder.retry.Attempt;
+import com.github.rholder.retry.RetryException;
+import com.github.rholder.retry.Retryer;
+import com.github.rholder.retry.RetryerBuilder;
+import com.github.rholder.retry.StopStrategies;
+import com.github.rholder.retry.WaitStrategies;
+import com.google.common.base.Predicate;
 import com.jcraft.jsch.IO;
 import io.cdap.plugin.servicenow.apiclient.NonRetryableException;
 import io.cdap.plugin.servicenow.apiclient.RetryableException;
+import io.cdap.plugin.servicenow.apiclient.ServiceNowAPIException;
+import io.cdap.plugin.servicenow.util.ServiceNowConstants;
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.methods.CloseableHttpResponse;
@@ -34,6 +43,7 @@
 import org.apache.oltu.oauth2.common.exception.OAuthProblemException;
 import org.apache.oltu.oauth2.common.exception.OAuthSystemException;
 import org.apache.oltu.oauth2.common.message.types.GrantType;
+import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,10 +53,11 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 
-/**
- * An abstract class to call Rest API.
- */
+/** An abstract class to call Rest API. */
 public abstract class RestAPIClient {
   private static final Logger LOG = LoggerFactory.getLogger(RestAPIClient.class);
 
@@ -67,6 +78,60 @@ public RestAPIResponse executeGet(RestAPIRequest request) throws IOException {
     }
   }
 
+  /**
+   * Executes the Rest API request and returns the response with retries.
+   *
+   * @param request the Rest API request.
+   * @return an instance of RestAPIResponse object.
+   * @throws ServiceNowAPIException
+   */
+  public RestAPIResponse executeGetWithRetries(RestAPIRequest request)
+      throws ServiceNowAPIException {
+    Callable<RestAPIResponse> callable = () -> executeGet(request);
+    return handleExecution(getRetryer(), callable);
+  }
+
+  private RestAPIResponse handleExecution(
+      Retryer<RestAPIResponse> retryer, Callable<RestAPIResponse> callable)
+      throws ServiceNowAPIException {
+    try {
+      RestAPIResponse response = retryer.call(callable);
+      // Execution is successful
+      if (response.hasException()) {
+        // Execution is successful and returned non retryable error
+        throw response.getException();
+      }
+      return response;
+    } catch (RetryException e) {
+      // Execution successful, returned retryable error and retries exhausted
+      Attempt<?> apiResponseAttempt = e.getLastFailedAttempt();
+      if (apiResponseAttempt.hasException()) {
+        // last attempt has execution failure
+        throw new ServiceNowAPIException(apiResponseAttempt.getExceptionCause(), null, false);
+      } else {
+        // last execution attempt was successful but has an error response
+        // if execution is successful, it's expected to have a exception in response object
+        RestAPIResponse response = (RestAPIResponse) apiResponseAttempt.getResult();
+        throw response.getException();
+      }
+    } catch (ExecutionException e) {
+      // Execution failed with error
+      throw new ServiceNowAPIException(e, null, false);
+    }
+  }
+
+  private Retryer<RestAPIResponse> getRetryer() {
+    return RetryerBuilder.<RestAPIResponse>newBuilder()
+        .retryIfResult(
+            restAPIResponse ->
+                restAPIResponse.hasException() && restAPIResponse.getException().isErrorRetryable())
+        .withWaitStrategy(
+            WaitStrategies.exponentialWait(ServiceNowConstants.WAIT_TIME, TimeUnit.MILLISECONDS))
+        .withStopStrategy(
+            StopStrategies.stopAfterAttempt(ServiceNowConstants.MAX_NUMBER_OF_RETRY_ATTEMPTS))
+        .build();
+  }
+
   /**
    * Executes the Rest API request and returns the response.
    *
@@ -78,7 +143,8 @@ public RestAPIResponse executePost(RestAPIRequest request) throws IOException {
     request.getHeaders().entrySet().forEach(e -> httpPost.addHeader(e.getKey(), e.getValue()));
     httpPost.setEntity(request.getEntity());
 
-    // We're retrying all transport exceptions while executing the HTTP POST method and the generic transport
+    // We're retrying all transport exceptions while executing the HTTP POST method and the generic
+    // transport
     // exceptions in HttpClient are represented by the standard java.io.IOException class
     // https://hc.apache.org/httpclient-legacy/exception-handling.html
     try (CloseableHttpClient httpClient = HttpClientBuilder.create().build()) {
@@ -99,20 +165,25 @@ public RestAPIResponse executePost(RestAPIRequest request) throws IOException {
    * @throws OAuthSystemException
    * @throws OAuthProblemException
    */
-  protected String generateAccessToken(String restApiEndpoint, String clientId, String clientSecret, String user,
-                                       String password) throws OAuthSystemException, OAuthProblemException {
+  protected String generateAccessToken(
+      String restApiEndpoint, String clientId, String clientSecret, String user, String password)
+      throws OAuthSystemException, OAuthProblemException {
     String token = "NO-VALUE";
 
     OAuthClient client = new OAuthClient(new URLConnectionClient());
-    OAuthClientRequest request = OAuthClientRequest.tokenLocation(restApiEndpoint)
-      .setGrantType(GrantType.PASSWORD)
-      .setClientId(clientId)
-      .setClientSecret(clientSecret)
-      .setUsername(user)
-      .setPassword(password)
-      .buildBodyMessage();
+    OAuthClientRequest request =
+        OAuthClientRequest.tokenLocation(restApiEndpoint)
+            .setGrantType(GrantType.PASSWORD)
+            .setClientId(clientId)
+            .setClientSecret(clientSecret)
+            .setUsername(user)
+            .setPassword(password)
+            .buildBodyMessage();
 
-    token = client.accessToken(request, OAuth.HttpMethod.POST, OAuthJSONAccessTokenResponse.class).getAccessToken();
+    token =
+        client
+            .accessToken(request, OAuth.HttpMethod.POST, OAuthJSONAccessTokenResponse.class)
+            .getAccessToken();
     return token;
   }
 }