PLUGIN-1936: Initial Commit

Author: sgarg-CS

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

@@ -25,8 +25,11 @@
 import com.google.common.base.Strings;
 import com.google.gson.Gson;
 import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.google.gson.reflect.TypeToken;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonToken;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.plugin.servicenow.connector.ServiceNowConnectorConfig;
@@ -55,8 +58,12 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -154,7 +161,9 @@ public List<Map<String, String>> fetchTableRecords(
     String accessToken = getAccessToken();
     requestBuilder.setAuthHeader(accessToken);
     RestAPIResponse apiResponse = executeGetWithRetries(requestBuilder.build());
-    return parseResponseToResultListOfMap(apiResponse.getResponseBody());
+    //return parseResponseToResultListOfMap(apiResponse.getResponseBody());
+    return parseResponseStreamToResultListOfMap(apiResponse.getInputStream());
+
   }
 
   private void applyDateRangeToRequest(ServiceNowTableAPIRequestBuilder requestBuilder, String startDate,
@@ -197,6 +206,43 @@ public List<Map<String, String>> parseResponseToResultListOfMap(String responseB
     return GSON.fromJson(ja, type);
   }
 
+  public List<Map<String, String>> parseResponseStreamToResultListOfMap(InputStream in) throws ServiceNowAPIException {
+    List<Map<String, String>> records = new ArrayList<>();
+    // InputStream in = httpResponse.getEntity().getContent();
+    try (InputStreamReader reader = new InputStreamReader(in, StandardCharsets.UTF_8);
+         JsonReader jsonReader = new JsonReader(reader)) {
+      jsonReader.setLenient(true);
+      jsonReader.beginObject();
+      while (jsonReader.hasNext()) {
+        String name = jsonReader.nextName();
+        if (ServiceNowConstants.RESULT.equals(name) && jsonReader.peek() == JsonToken.BEGIN_ARRAY) {
+          jsonReader.beginArray();
+          while (jsonReader.hasNext()) {
+            jsonReader.beginObject();
+            Map<String, String> record = new HashMap<>();
+            while (jsonReader.hasNext()) {
+              String field = jsonReader.nextName();
+              JsonToken token = jsonReader.peek();
+              // JsonElement resultElement = GSON.fromJson(jsonReader, JsonElement.class);
+              // responseBody = resultElement.toString();
+              record.put(field, token == JsonToken.NULL ? null : jsonReader.nextString());
+            }
+            jsonReader.endObject();
+            records.add(record);
+          }
+          jsonReader.endArray();
+        } else {
+          // skip other fields (e.g., metadata like result_count)
+          jsonReader.skipValue();
+        }
+      }
+      jsonReader.endObject();
+      return records;
+    } catch (IOException e) {
+      throw new ServiceNowAPIException(e, null);
+    }
+  }
+
   private String getErrorMessage(String responseBody) {
     try {
       JsonObject jo = GSON.fromJson(responseBody, JsonObject.class);

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

@@ -43,6 +43,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.SocketException;
 import java.util.Collections;
 import java.util.concurrent.Callable;
@@ -82,7 +83,7 @@ public RestAPIResponse executeGet(RestAPIRequest request) throws IOException {
       }
     } catch (ConnectTimeoutException | SocketException e) {
       ServiceNowAPIException exception = new ServiceNowAPIException(e, null);
-      return new RestAPIResponse(Collections.emptyMap(), null, exception);
+      return new RestAPIResponse(Collections.emptyMap(), (InputStream) null, exception);
     }
   }
 

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

@@ -17,21 +17,28 @@
 package io.cdap.plugin.servicenow.restapi;
 
 import com.google.gson.Gson;
+import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
+import com.google.gson.stream.JsonReader;
 import io.cdap.plugin.servicenow.apiclient.ServiceNowAPIException;
 import io.cdap.plugin.servicenow.util.ServiceNowConstants;
 import org.apache.http.Header;
+import org.apache.http.HttpEntity;
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
 import org.apache.http.util.EntityUtils;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
@@ -46,9 +53,14 @@ public class RestAPIResponse {
   private static final Set<Integer> SUCCESS_CODES = new HashSet<>(Arrays.asList(HttpStatus.SC_CREATED,
                                                                                 HttpStatus.SC_OK));
   private final Map<String, String> headers;
-  private final String responseBody;
+  // Deprecated: storing full body as String can cause OOM
+  @Deprecated
+  private String responseBody;
   @Nullable private final ServiceNowAPIException exception;
 
+  // New: store InputStream for streaming consumption
+  private InputStream inputStream;
+
   public RestAPIResponse(
       Map<String, String> headers,
       @Nullable String responseBody,
@@ -58,6 +70,15 @@ public RestAPIResponse(
     this.exception = exception;
   }
 
+  public RestAPIResponse(
+    Map<String, String> headers,
+    InputStream inputStream,
+    @Nullable ServiceNowAPIException exception) {
+    this.headers = headers;
+    this.inputStream = inputStream;
+    this.exception = exception;
+  }
+
   /**
    * Parses HttpResponse into RestAPIResponse object when no errors occur.
    * Throws a {@link ServiceNowAPIException}.
@@ -80,17 +101,27 @@ public static RestAPIResponse parse(HttpResponse httpResponse, String... headerN
 
     ServiceNowAPIException serviceNowAPIException = validateHttpResponse(httpResponse);
     if (serviceNowAPIException != null) {
-      return new RestAPIResponse(headers, null, serviceNowAPIException);
+      return new RestAPIResponse(headers, (InputStream) null, serviceNowAPIException);
     }
 
     String responseBody = null;
     try {
       responseBody = EntityUtils.toString(httpResponse.getEntity());
     } catch (IOException e) {
-      return new RestAPIResponse(headers, null, new ServiceNowAPIException(e, httpResponse));
+      return new RestAPIResponse(headers, (String) null, new ServiceNowAPIException(e, httpResponse));
+    }
+    // Instead of reading the entire entity, store the stream
+    HttpEntity httpEntity = httpResponse.getEntity();
+    InputStream responseStream;
+    try {
+      responseStream = (httpEntity != null) ? httpEntity.getContent() : null;
+    } catch (IOException e) {
+      return new RestAPIResponse(headers, (InputStream) null, new ServiceNowAPIException(e, httpResponse));
     }
     serviceNowAPIException = validateRestApiResponse(httpResponse, responseBody);
-    return new RestAPIResponse(headers, responseBody, serviceNowAPIException);
+    // return new RestAPIResponse(headers, responseBody, serviceNowAPIException);
+    return new RestAPIResponse(headers, responseStream, serviceNowAPIException);
+
   }
 
   public static RestAPIResponse parse(HttpResponse httpResponse) throws IOException {
@@ -131,6 +162,10 @@ public String getResponseBody() {
     return responseBody;
   }
 
+  public InputStream getInputStream() {
+    return inputStream;
+  }
+
   @Nullable
   public ServiceNowAPIException getException() {
     return exception;