PLUGIN-1137 : Make Page Size configurable

docs/ServiceNow-batchsource.md

@@ -59,6 +59,8 @@ ignored if the Mode is set to `Reporting`.
 
 `Display` - will fetch the display values from the ServiceNow tables.
 
+**Page Size**: The default limit for the page size is 5000. Page Size can have 500, 1000, 2000, 3000, 4000 & 5000 value.
+
 Data Types Mapping
 ----------
 

docs/ServiceNowMultiSource-batchsource.md

@@ -35,6 +35,8 @@ Properties
 **Table Name Field**: The name of the field that holds the table name. Must not be the name of any table column that
 will be read. Defaults to `tablename`. Note, the Table name field value will be ignored if the Mode is set to `Table`.
 
+**Page Size**: The default limit for the page size is 5000. Page Size can have 500, 1000, 2000, 3000, 4000 & 5000 value.
+
 Data Types Mapping
 ----------
 

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowBaseSourceConfig.java

@@ -36,7 +36,6 @@
 
 import java.time.LocalDate;
 import java.time.temporal.ChronoUnit;
-
 import javax.annotation.Nullable;
 
 /**
@@ -92,6 +91,11 @@ public class ServiceNowBaseSourceConfig extends PluginConfig {
   @Description("The End date to be used to filter the data. The format must be 'yyyy-MM-dd'.")
   private String endDate;
 
+  @Name(ServiceNowConstants.PROPERTY_PAGE_SIZE)
+  @Macro
+  @Description("The default limit for the page size. ")
+  private Integer pageSize;
+
   @Name(ServiceNowConstants.PROPERTY_TABLE_NAME_FIELD)
   @Macro
   @Nullable
@@ -103,7 +107,8 @@ public class ServiceNowBaseSourceConfig extends PluginConfig {
   public ServiceNowBaseSourceConfig(String referenceName, String tableNameField, String clientId,
                                     String clientSecret, String restApiEndpoint,
                                     String user, String password,
-                                    String valueType, @Nullable String startDate, @Nullable String endDate) {
+                                    String valueType, @Nullable String startDate, @Nullable String endDate,
+                                    Integer pageSize) {
 
     this.referenceName = referenceName;
     this.tableNameField = tableNameField;
@@ -115,6 +120,7 @@ public ServiceNowBaseSourceConfig(String referenceName, String tableNameField, S
     this.valueType = valueType;
     this.startDate = startDate;
     this.endDate = endDate;
+    this.pageSize = pageSize;
   }
 
   public String getReferenceName() {
@@ -165,6 +171,7 @@ public void validate(FailureCollector collector) {
     validateCredentials(collector);
     validateValueType(collector);
     validateDateRange(collector);
+    validatePageSize(collector);
   }
 
   public void validateCredentials(FailureCollector collector) {
@@ -364,4 +371,35 @@ void validateTable(String tableName, FailureCollector collector) {
     }
   }
 
+  /**
+   * Returns the value type chosen.
+   *
+   * @return An instance of Page Size
+   */
+  public Integer getPageSize() {
+    return pageSize;
+  }
+
+  /**
+   * Returns the page Size chosen.
+   */
+  public Integer getPageSize(FailureCollector collector) {
+    if (getPageSize() != null) {
+      return pageSize;
+    }
+    collector.addFailure("Invalid page size: " + pageSize,
+                         String.format("Valid page size are: %s", "500, 1000, 2000, 3000, 4000 and 5000"))
+      .withConfigProperty(ServiceNowConstants.PROPERTY_PAGE_SIZE);
+    collector.getOrThrowException();
+    return null;
+
+  }
+
+  public void validatePageSize(FailureCollector collector) {
+    if (containsMacro(ServiceNowConstants.PROPERTY_PAGE_SIZE)) {
+      return;
+    }
+    getPageSize(collector);
+  }
+
 }

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowInputFormat.java

@@ -22,7 +22,6 @@
 import io.cdap.plugin.servicenow.source.apiclient.ServiceNowTableDataResponse;
 import io.cdap.plugin.servicenow.source.util.SchemaBuilder;
 import io.cdap.plugin.servicenow.source.util.ServiceNowColumn;
-import io.cdap.plugin.servicenow.source.util.ServiceNowConstants;
 import io.cdap.plugin.servicenow.source.util.ServiceNowTableInfo;
 import io.cdap.plugin.servicenow.source.util.SourceQueryMode;
 import org.apache.hadoop.conf.Configuration;
@@ -45,6 +44,7 @@
  */
 public class ServiceNowInputFormat extends InputFormat<NullWritable, StructuredRecord> {
   private static final Logger LOG = LoggerFactory.getLogger(ServiceNowInputFormat.class);
+  private int pageSize;
 
   /**
    * Updates the jobConfig with the ServiceNow table information, which will then be read in getSplit() function.
@@ -114,28 +114,29 @@ private static ServiceNowTableInfo getTableMetaData(String tableName, ServiceNow
   @Override
   public List<InputSplit> getSplits(JobContext jobContext) throws IOException, InterruptedException {
     ServiceNowJobConfiguration jobConfig = new ServiceNowJobConfiguration(jobContext.getConfiguration());
+    pageSize = jobConfig.getPluginConf().getPageSize().intValue();
 
     List<ServiceNowTableInfo> tableInfos = jobConfig.getTableInfos();
     List<InputSplit> resultSplits = new ArrayList<>();
 
     for (ServiceNowTableInfo tableInfo : tableInfos) {
       String tableName = tableInfo.getTableName();
       int totalRecords = tableInfo.getRecordCount();
-      if (totalRecords <= ServiceNowConstants.PAGE_SIZE) {
+      if (totalRecords <= pageSize) {
         // add single split for table and continue
         resultSplits.add(new ServiceNowInputSplit(tableName, 0));
         continue;
       }
 
-      int pages = (tableInfo.getRecordCount() / ServiceNowConstants.PAGE_SIZE);
-      if (tableInfo.getRecordCount() % ServiceNowConstants.PAGE_SIZE > 0) {
+      int pages = (tableInfo.getRecordCount() / pageSize);
+      if (tableInfo.getRecordCount() % pageSize > 0) {
         pages++;
       }
       int offset = 0;
 
       for (int page = 1; page <= pages; page++) {
         resultSplits.add(new ServiceNowInputSplit(tableName, offset));
-        offset += ServiceNowConstants.PAGE_SIZE;
+        offset += pageSize;
       }
     }
 

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowMultiInputFormat.java

@@ -23,9 +23,7 @@
 import io.cdap.plugin.servicenow.source.apiclient.ServiceNowTableDataResponse;
 import io.cdap.plugin.servicenow.source.util.SchemaBuilder;
 import io.cdap.plugin.servicenow.source.util.ServiceNowColumn;
-import io.cdap.plugin.servicenow.source.util.ServiceNowConstants;
 import io.cdap.plugin.servicenow.source.util.ServiceNowTableInfo;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapreduce.InputFormat;
@@ -49,7 +47,8 @@
  * ServiceNow input format.
  */
 public class ServiceNowMultiInputFormat extends InputFormat<NullWritable, StructuredRecord> {
-  private static final Logger LOG = LoggerFactory.getLogger(ServiceNowInputFormat.class);
+  private static final Logger LOG = LoggerFactory.getLogger(ServiceNowMultiInputFormat.class);
+  private int pageSize;
 
   /**
    * Updates the jobConfig with the ServiceNow table information, which will then be read in getSplit() function.
@@ -111,6 +110,7 @@ private static ServiceNowTableInfo getTableMetaData(String tableName, ServiceNow
   @Override
   public List<InputSplit> getSplits(JobContext jobContext) throws IOException, InterruptedException {
     ServiceNowJobConfiguration jobConfig = new ServiceNowJobConfiguration(jobContext.getConfiguration());
+    pageSize = jobConfig.getPluginConf().getPageSize().intValue();
 
     List<ServiceNowTableInfo> tableInfos = jobConfig.getTableInfos();
     List<InputSplit> resultSplits = new ArrayList<>();
@@ -119,15 +119,15 @@ public List<InputSplit> getSplits(JobContext jobContext) throws IOException, Int
       String tableName = tableInfo.getTableName();
       int totalRecords = tableInfo.getRecordCount();
 
-      int pages = (tableInfo.getRecordCount() / ServiceNowConstants.PAGE_SIZE);
-      if (tableInfo.getRecordCount() % ServiceNowConstants.PAGE_SIZE > 0) {
+      int pages = (tableInfo.getRecordCount() / pageSize);
+      if (tableInfo.getRecordCount() % pageSize > 0) {
         pages++;
       }
       int offset = 0;
 
       for (int page = 1; page <= pages; page++) {
         resultSplits.add(new ServiceNowInputSplit(tableName, offset));
-        offset += ServiceNowConstants.PAGE_SIZE;
+        offset += pageSize;
       }
     }
 

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowMultiRecordReader.java

@@ -21,7 +21,6 @@
 import io.cdap.plugin.servicenow.source.apiclient.ServiceNowTableAPIClientImpl;
 import io.cdap.plugin.servicenow.source.apiclient.ServiceNowTableDataResponse;
 import io.cdap.plugin.servicenow.source.util.SchemaBuilder;
-import io.cdap.plugin.servicenow.source.util.ServiceNowConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -85,16 +84,16 @@ public void fetchData() {
     ServiceNowTableAPIClientImpl restApi = new ServiceNowTableAPIClientImpl(multiSourcePluginConf);
 
     // Get the table data
-    results =
-      restApi.fetchTableRecords(tableName, multiSourcePluginConf.getStartDate(), multiSourcePluginConf.getEndDate(),
-                                split.getOffset(), ServiceNowConstants.PAGE_SIZE);
+    results = restApi.fetchTableRecords(tableName, multiSourcePluginConf.getStartDate(),
+      multiSourcePluginConf.getEndDate(), split.getOffset(), multiSourcePluginConf.getPageSize());
 
-    LOG.debug("size={}", results.size());
     if (!results.isEmpty()) {
+      LOG.debug("size={}", results.size());
       fetchSchema(restApi);
     }
 
     iterator = results.iterator();
+
   }
 
   private void fetchSchema(ServiceNowTableAPIClientImpl restApi) {

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowMultiSourceConfig.java

@@ -21,21 +21,10 @@
 import io.cdap.cdap.api.annotation.Macro;
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.cdap.etl.api.FailureCollector;
-import io.cdap.plugin.common.IdUtils;
-import io.cdap.plugin.servicenow.restapi.RestAPIResponse;
-import io.cdap.plugin.servicenow.source.apiclient.ServiceNowTableAPIClientImpl;
-import io.cdap.plugin.servicenow.source.apiclient.ServiceNowTableAPIRequestBuilder;
 import io.cdap.plugin.servicenow.source.util.ServiceNowConstants;
 import io.cdap.plugin.servicenow.source.util.Util;
 
-import org.apache.http.HttpStatus;
-import org.apache.oltu.oauth2.common.exception.OAuthProblemException;
-import org.apache.oltu.oauth2.common.exception.OAuthSystemException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.util.Set;
-
 import javax.annotation.Nullable;
 
 /**
@@ -65,9 +54,9 @@ public class ServiceNowMultiSourceConfig extends ServiceNowBaseSourceConfig {
   public ServiceNowMultiSourceConfig(String referenceName, String tableNameField, String clientId,
                                      String clientSecret, String restApiEndpoint, String user, String password,
                                      String valueType, @Nullable String startDate, @Nullable String endDate,
-                                     String tableNames) {
+                                     Integer pageSize, String tableNames) {
     super(referenceName, tableNameField, clientId, clientSecret, restApiEndpoint, user, password, valueType, startDate,
-      endDate);
+          endDate, pageSize);
     this.tableNames = tableNames;
   }
 

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowRecordReader.java

@@ -21,7 +21,6 @@
 import io.cdap.plugin.servicenow.source.apiclient.ServiceNowTableAPIClientImpl;
 import io.cdap.plugin.servicenow.source.apiclient.ServiceNowTableDataResponse;
 import io.cdap.plugin.servicenow.source.util.SchemaBuilder;
-import io.cdap.plugin.servicenow.source.util.ServiceNowConstants;
 import io.cdap.plugin.servicenow.source.util.SourceQueryMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -37,7 +36,6 @@ public class ServiceNowRecordReader extends ServiceNowBaseRecordReader {
   private static final Logger LOG = LoggerFactory.getLogger(ServiceNowRecordReader.class);
   private final ServiceNowSourceConfig pluginConf;
 
-
   ServiceNowRecordReader(ServiceNowSourceConfig pluginConf) {
     super();
     this.pluginConf = pluginConf;
@@ -93,7 +91,7 @@ private void fetchData() {
 
     // Get the table data
     results = restApi.fetchTableRecords(tableName, pluginConf.getStartDate(), pluginConf.getEndDate(),
-                                        split.getOffset(), ServiceNowConstants.PAGE_SIZE);
+                                        split.getOffset(), pluginConf.getPageSize());
     LOG.debug("size={}", results.size());
     if (!results.isEmpty()) {
       fetchSchema(restApi);

src/main/java/io/cdap/plugin/servicenow/source/ServiceNowSourceConfig.java

@@ -24,11 +24,8 @@
 import io.cdap.plugin.servicenow.source.util.SourceApplication;
 import io.cdap.plugin.servicenow.source.util.SourceQueryMode;
 import io.cdap.plugin.servicenow.source.util.Util;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.Optional;
-
 import javax.annotation.Nullable;
 
 /**
@@ -81,9 +78,10 @@ public class ServiceNowSourceConfig extends ServiceNowBaseSourceConfig {
   public ServiceNowSourceConfig(String referenceName, String queryMode, @Nullable String applicationName,
                                 @Nullable String tableNameField, @Nullable String tableName, String clientId,
                                 String clientSecret, String restApiEndpoint, String user, String password,
-                                String valueType, @Nullable String startDate, @Nullable String endDate) {
+                                String valueType, @Nullable String startDate, @Nullable String endDate,
+                                Integer pageSize) {
     super(referenceName, tableNameField, clientId, clientSecret, restApiEndpoint, user, password, valueType, startDate,
-      endDate);
+      endDate, pageSize);
     this.referenceName = referenceName;
     this.queryMode = queryMode;
     this.applicationName = applicationName;

src/main/java/io/cdap/plugin/servicenow/source/apiclient/RetriableException.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright © 2022 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.servicenow.source.apiclient;
+
+/**
+ * Custom Exception Class for handling retrying API calls
+ */
+public class RetriableException extends RuntimeException {
+
+  private static final long serialVersionUID = 1L;
+
+  public RetriableException() {
+         super();
+  }
+
+}

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

@@ -198,12 +198,16 @@ private int getRecordCountFromHeader(RestAPIResponse apiResponse) {
 
   public List<Map<String, Object>> parseResponseToResultListOfMap(String responseBody) {
     Gson gson = new Gson();
+    long start = System.currentTimeMillis();
+
     JsonObject jo = gson.fromJson(responseBody, JsonObject.class);
     JsonArray ja = jo.getAsJsonArray("result");
 
     Type type = new TypeToken<List<Map<String, Object>>>() {
     }.getType();
 
+    long end = System.currentTimeMillis();
+    LOG.info("parsing response took {}ms ",  (end - start));
     return gson.fromJson(ja, type);
   }
 

src/main/java/io/cdap/plugin/servicenow/source/util/ServiceNowConstants.java

@@ -96,6 +96,11 @@ public interface ServiceNowConstants {
    */
   String PROPERTY_END_DATE = "endDate";
 
+  /**
+   * Configuration property name used to specify page size.
+   */
+  String PROPERTY_PAGE_SIZE = "pageSize";
+
   /**
    * Configuration property name used to specify table name field.
    */
@@ -111,11 +116,6 @@ public interface ServiceNowConstants {
    */
   String DATE_FORMAT = "yyyy-MM-dd";
 
-  /**
-   * The max limit for the page size.
-   */
-  int PAGE_SIZE = 5000;
-
   /**
    * The total count.
    */

src/test/java/io/cdap/plugin/servicenow/source/ServiceNowMultiInputFormatTest.java

@@ -25,7 +25,7 @@ public class ServiceNowMultiInputFormatTest {
   public void testFetchTablesInfoWithEmptyTableNames() {
     ServiceNowMultiSourceConfig config = new ServiceNowMultiSourceConfig("Reference Name",
      "tableName", "client_id", "Client Secret", "http://example.com",
-     "user", "password", "Actual", "2021-12-30", "2021-12-31", "");
+     "user", "password", "Actual", "2021-12-30", "2021-12-31", 5000, "");
     Assert.assertTrue(ServiceNowMultiInputFormat
                         .fetchTablesInfo(config)
                         .isEmpty());