Skip to content

PLUGIN-1936: Read data in a streaming manner #131

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sgarg-CS wants to merge 6 commits into
base: develop
Choose a base branch
from
Open

PLUGIN-1936: Read data in a streaming manner #131

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1f2d9d7
PLUGIN-1936: Initial Commit
sgarg-CS Nov 10, 2025
4235153
PLUGIN-1936: Refactor code to implement streaming logic
sgarg-CS Nov 18, 2025
80033a3
PLUGIN-1936: Fix unexpected closure of stream issue
sgarg-CS Nov 24, 2025
5e99944
PLUGIN-1936 : Fix duplicate record issue
sgarg-CS Nov 24, 2025
cf89907
PLUGIN-1936 : Remove commented code and unused imports
sgarg-CS Nov 25, 2025
2274e97
PLUGIN-1936: Rework based on new changes suggested in the design
sgarg-CS Dec 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,11 @@
<version>${cdap.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.5</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
Expand Down
33 changes: 32 additions & 1 deletion src/main/java/io/cdap/plugin/servicenow/ServiceNowBaseConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,14 @@
package io.cdap.plugin.servicenow;

import com.google.common.annotations.VisibleForTesting;
import com.google.gson.stream.JsonReader;
import io.cdap.cdap.api.annotation.Description;
import io.cdap.cdap.api.annotation.Macro;
import io.cdap.cdap.api.annotation.Name;
import io.cdap.cdap.api.plugin.PluginConfig;
import io.cdap.cdap.etl.api.FailureCollector;
import io.cdap.plugin.common.ConfigUtil;
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.connector.ServiceNowConnectorConfig;
Expand All @@ -31,14 +33,21 @@
import io.cdap.plugin.servicenow.util.SchemaType;
import io.cdap.plugin.servicenow.util.ServiceNowConstants;
import io.cdap.plugin.servicenow.util.SourceValueType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import javax.annotation.Nullable;

/**
* ServiceNow Base Config. Contains connection properties and methods.
*/
public class ServiceNowBaseConfig extends PluginConfig {

private static final Logger log = LoggerFactory.getLogger(ServiceNowBaseConfig.class);
@Name(ConfigUtil.NAME_USE_CONNECTION)
@Nullable
@Description("Whether to use an existing connection.")
Expand Down Expand Up @@ -140,7 +149,7 @@ public void validateTable(String tableName, SourceValueType valueType, FailureCo
requestBuilder.setResponseHeaders(ServiceNowConstants.HEADER_NAME_TOTAL_COUNT);

apiResponse = serviceNowTableAPIClient.executeGetWithRetries(requestBuilder.build());
if (serviceNowTableAPIClient.parseResponseToResultListOfMap(apiResponse.getResponseBody()).isEmpty()) {
if (isResultEmpty(apiResponse)) {
// Removed config property as in case of MultiSource, only first table error was populating.
collector.addFailure("Table: " + tableName + " is empty.", "");
}
Expand All @@ -152,4 +161,26 @@ public void validateTable(String tableName, SourceValueType valueType, FailureCo
}
}

/**
* Check whether the result is empty or not.
Copy link

@Sunish-Dahiya Sunish-Dahiya Dec 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also explain how are we checking is result is empty

* @param restAPIResponse
* @return true if result is empty
* @throws IOException
*/
public boolean isResultEmpty(RestAPIResponse restAPIResponse) throws IOException {
JsonReader reader = new JsonReader(new InputStreamReader(restAPIResponse.getBodyAsStream(),
StandardCharsets.UTF_8));
reader.beginObject();
while (reader.hasNext()) {
String name = reader.nextName();
if (ServiceNowConstants.RESULT.equals(name)) {
reader.beginArray();
return !reader.hasNext();
} else {
reader.skipValue();
}
}
return true;
}

}
80 changes: 64 additions & 16 deletions src/main/java/io/cdap/plugin/servicenow/apiclient/ServiceNowTableAPIClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
import com.google.gson.JsonArray;
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;
Expand Down Expand Up @@ -55,8 +57,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;
Expand Down Expand Up @@ -131,7 +137,7 @@ 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(
public RestAPIResponse fetchTableRecords(
String tableName,
SourceValueType valueType,
String startDate,
Expand All @@ -154,7 +160,7 @@ public List<Map<String, String>> fetchTableRecords(
String accessToken = getAccessToken();
requestBuilder.setAuthHeader(accessToken);
RestAPIResponse apiResponse = executeGetWithRetries(requestBuilder.build());
return parseResponseToResultListOfMap(apiResponse.getResponseBody());
return apiResponse;
}

private void applyDateRangeToRequest(ServiceNowTableAPIRequestBuilder requestBuilder, String startDate,
Expand Down Expand Up @@ -197,6 +203,40 @@ public List<Map<String, String>> parseResponseToResultListOfMap(String responseB
return GSON.fromJson(ja, type);
}

public List<Map<String, String>> parseResponseToResultListOfMap(InputStream in) throws ServiceNowAPIException {
Copy link

@Sunish-Dahiya Sunish-Dahiya Dec 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we creating a List of map? Instead we can directly read it as a JsonObject, something like:

gson.fromJson(reader, ServiceNowRecordObject.class);

try (InputStreamReader reader = new InputStreamReader(in, StandardCharsets.UTF_8);
JsonReader jsonReader = new JsonReader(reader)) {
jsonReader.setLenient(true);
jsonReader.beginObject();

List<Map<String, String>> records = new ArrayList<>();
while (jsonReader.hasNext()) {
String name = jsonReader.nextName();
if (ServiceNowConstants.RESULT.equals(name) && jsonReader.peek() == JsonToken.BEGIN_ARRAY) {
jsonReader.beginArray();
while (jsonReader.hasNext()) {
jsonReader.beginObject();
while (jsonReader.hasNext()) {
Map<String, String> record = new HashMap<>();
String field = jsonReader.nextName();
JsonToken token = jsonReader.peek();
record.put(field, token == JsonToken.NULL ? null : jsonReader.nextString());
records.add(record);
}
jsonReader.endObject();
}
jsonReader.endArray();
} else {
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);
Expand Down Expand Up @@ -231,12 +271,14 @@ private String getErrorMessage(String responseBody) {
* @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>> fetchTableRecordsRetryableMode(String tableName, SourceValueType valueType,
public RestAPIResponse fetchTableRecordsRetryableMode(String tableName, SourceValueType valueType,
String startDate, String endDate, int offset,
int limit) throws ServiceNowAPIException {
final List<Map<String, String>> results = new ArrayList<>();
//final List<Map<String, String>> results = new ArrayList<>();
Copy link

@Sunish-Dahiya Sunish-Dahiya Dec 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please remove

final RestAPIResponse[] restAPIResponse = new RestAPIResponse[1];
Callable<Boolean> fetchRecords = () -> {
results.addAll(fetchTableRecords(tableName, valueType, startDate, endDate, offset, limit));
// results.addAll(fetchTableRecords(tableName, valueType, startDate, endDate, offset, limit));
Copy link

@Sunish-Dahiya Sunish-Dahiya Dec 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please remove

restAPIResponse[0] = fetchTableRecords(tableName, valueType, startDate, endDate, offset, limit);
Copy link

@Sunish-Dahiya Sunish-Dahiya Dec 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you please explain what are we trying to do here? why restAPIResponse is an array

return true;
};

Expand All @@ -254,7 +296,7 @@ public List<Map<String, String>> fetchTableRecordsRetryableMode(String tableName
e, null, false);
}

return results;
return restAPIResponse[0];
}

/**
Expand All @@ -277,8 +319,8 @@ public Schema fetchTableSchema(String tableName, FailureCollector collector) {
}

@VisibleForTesting
public MetadataAPISchemaResponse parseSchemaResponse(String responseBody) {
return GSON.fromJson(responseBody, MetadataAPISchemaResponse.class);
public MetadataAPISchemaResponse parseSchemaResponse(InputStream responseStream) {
return GSON.fromJson(createJsonReader(responseStream), MetadataAPISchemaResponse.class);
}

/**
Expand Down Expand Up @@ -352,7 +394,7 @@ public Schema fetchTableSchema(String tableName, String accessToken, SourceValue
private Schema prepareSchemaWithSchemaAPI(RestAPIResponse restAPIResponse, List<ServiceNowColumn> columns,
String tableName) throws ServiceNowAPIException {
SchemaAPISchemaResponse schemaAPISchemaResponse =
GSON.fromJson(restAPIResponse.getResponseBody(), SchemaAPISchemaResponse.class);
GSON.fromJson(createJsonReader(restAPIResponse.getBodyAsStream()), SchemaAPISchemaResponse.class);

if (schemaAPISchemaResponse.getResult() == null || schemaAPISchemaResponse.getResult().isEmpty()) {
throw new ServiceNowAPIException(
Expand Down Expand Up @@ -386,8 +428,7 @@ private Schema prepareSchemaWithSchemaAPI(RestAPIResponse restAPIResponse, List<
private Schema prepareSchemaWithMetadataAPI(RestAPIResponse restAPIResponse, List<ServiceNowColumn> columns,
String tableName, SourceValueType valueType) throws
ServiceNowAPIException {
MetadataAPISchemaResponse metadataAPISchemaResponse = parseSchemaResponse(restAPIResponse.getResponseBody());

MetadataAPISchemaResponse metadataAPISchemaResponse = parseSchemaResponse(restAPIResponse.getBodyAsStream());
if (metadataAPISchemaResponse.getResult() == null || metadataAPISchemaResponse.getResult().getColumns() == null ||
metadataAPISchemaResponse.getResult().getColumns().isEmpty()) {
throw new ServiceNowAPIException(
Expand All @@ -413,6 +454,11 @@ private Schema prepareSchemaWithMetadataAPI(RestAPIResponse restAPIResponse, Lis
return SchemaBuilder.constructSchema(tableName, columns);
}

public JsonReader createJsonReader(InputStream inputStream) {
Objects.requireNonNull(inputStream, "InputStream must not be null");
return new JsonReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
}

/**
* Get the total number of records in the table
*
Expand Down Expand Up @@ -503,8 +549,8 @@ public String createRecordInDisplayMode(String tableName, HttpEntity entity) thr
}

private String getSystemId(RestAPIResponse restAPIResponse) {
CreateRecordAPIResponse apiResponse = GSON.fromJson(restAPIResponse.getResponseBody(),
CreateRecordAPIResponse.class);
CreateRecordAPIResponse apiResponse = GSON.fromJson(
new InputStreamReader(restAPIResponse.getBodyAsStream(), StandardCharsets.UTF_8), CreateRecordAPIResponse.class);
return apiResponse.getResult().get(ServiceNowConstants.SYSTEM_ID).toString();
}

Expand All @@ -527,7 +573,8 @@ public Map<String, String> getRecordFromServiceNowTable(String tableName, String
requestBuilder.setAuthHeader(accessToken);
restAPIResponse = executeGetWithRetries(requestBuilder.build());

APIResponse apiResponse = GSON.fromJson(restAPIResponse.getResponseBody(), APIResponse.class);
APIResponse apiResponse = GSON.fromJson(
new InputStreamReader(restAPIResponse.getBodyAsStream(), StandardCharsets.UTF_8), APIResponse.class);
return apiResponse.getResult().get(0);
}

Expand All @@ -545,8 +592,9 @@ public Map<String, String> getRecordFromServiceNowTable(String tableName, String
* @throws RuntimeException if the schema response is null or contains no result.
*/
private Schema prepareStringBasedSchema(RestAPIResponse restAPIResponse, List<ServiceNowColumn> columns,
String tableName) {
List<Map<String, String>> result = parseResponseToResultListOfMap(restAPIResponse.getResponseBody());
String tableName) throws ServiceNowAPIException {
List<Map<String, String>> result = parseResponseToResultListOfMap(restAPIResponse.getBodyAsStream());
Copy link

@Sunish-Dahiya Sunish-Dahiya Dec 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should be List of Records/objects instead of map

// List<Map<String, String>> result = parseResponseToResultListOfMap(restAPIResponse.getResponseBody());
Copy link

@Sunish-Dahiya Sunish-Dahiya Dec 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

remove

if (result != null && !result.isEmpty()) {
Map<String, String> firstRecord = result.get(0);
for (String key : firstRecord.keySet()) {
Expand Down
4 changes: 2 additions & 2 deletions src/main/java/io/cdap/plugin/servicenow/connector/ServiceNowConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ private TableList listTables(String accessToken) throws ServiceNowAPIException {
ServiceNowTableAPIClientImpl serviceNowTableAPIClient = new ServiceNowTableAPIClientImpl(config, true);
RestAPIResponse apiResponse =
serviceNowTableAPIClient.executeGetWithRetries(requestBuilder.build());
return GSON.fromJson(apiResponse.getResponseBody(), TableList.class);
return GSON.fromJson(serviceNowTableAPIClient.createJsonReader(apiResponse.getBodyAsStream()), TableList.class);
}

public ConnectorSpec generateSpec(ConnectorContext connectorContext, ConnectorSpecRequest connectorSpecRequest) {
Expand Down Expand Up @@ -184,7 +184,7 @@ private List<StructuredRecord> getTableData(String tableName, int limit)
requestBuilder.setResponseHeaders(ServiceNowConstants.HEADER_NAME_TOTAL_COUNT);
RestAPIResponse apiResponse = serviceNowTableAPIClient.executeGetWithRetries(requestBuilder.build());
List<Map<String, String>> result = serviceNowTableAPIClient.parseResponseToResultListOfMap
(apiResponse.getResponseBody());
(apiResponse.getBodyAsStream());
List<StructuredRecord> recordList = new ArrayList<>();
Schema schema = getSchema(tableName);
if (schema != null) {
Expand Down
1 change: 1 addition & 0 deletions src/main/java/io/cdap/plugin/servicenow/restapi/RestAPIClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading
Loading