data-integrations · AnkitCLI · May 9, 2025 · May 7, 2025
diff --git a/src/e2e-test/features/salesforcemultiobjectsbatchsource/RunTime.feature b/src/e2e-test/features/salesforcemultiobjectsbatchsource/RunTime.feature
@@ -15,11 +15,12 @@
 @SalesforceSalesCloud
 @SFMultiObjectsBatchSource
 @Smoke
+# Skipping these tests due to permission issues for creating custom fields in sObjects. Will Run it once it get resolves.
 @Regression
 
 Feature: Salesforce Multi Objects Batch Source - Run time Scenarios
 
-  @MULTIBATCH-TS-SF-RNTM-01 @BQ_SINK_MULTI_TEST @CREATE_TEST_DATA @CREATE_TEST_DATA2 @DELETE_TEST_DATA @DELETE_TEST_DATA2
+  @MULTIBATCH-TS-SF-RNTM-01 @BQ_SINK_MULTI_TEST @CREATE_TEST_DATA @CREATE_TEST_DATA2 @DELETE_TEST_DATA @DELETE_TEST_DATA2 @Plugin-1890
   Scenario: Verify user should be able to preview, deploy and run pipeline for valid White List
     When Open Datafusion Project to configure pipeline
     And Select data pipeline type as: "Batch"
@@ -58,7 +59,7 @@ Feature: Salesforce Multi Objects Batch Source - Run time Scenarios
     Then Validate the values of records transferred to target Big Query table is equal to the values from multi object source table
 
 
-  @MULTIBATCH-TS-SF-RNTM-02 @BQ_SINK_MULTI_TEST @CREATE_TEST_DATA @CREATE_TEST_DATA2 @DELETE_TEST_DATA @DELETE_TEST_DATA2
+  @MULTIBATCH-TS-SF-RNTM-02 @BQ_SINK_MULTI_TEST @CREATE_TEST_DATA @CREATE_TEST_DATA2 @DELETE_TEST_DATA @DELETE_TEST_DATA2 @Plugin-1890
   Scenario: Verify user should be able to preview, deploy and run pipeline for valid Black List
     When Open Datafusion Project to configure pipeline
     And Select data pipeline type as: "Batch"
@@ -99,7 +100,7 @@ Feature: Salesforce Multi Objects Batch Source - Run time Scenarios
     Then Validate the values of records transferred to target Big Query table is equal to the values from multi object source table
 
 
-  @MULTIBATCH-TS-SF-RNTM-03 @CONNECTION @BQ_SINK_MULTI_TEST @CREATE_TEST_DATA @CREATE_TEST_DATA2 @DELETE_TEST_DATA @DELETE_TEST_DATA2
+  @MULTIBATCH-TS-SF-RNTM-03 @CONNECTION @BQ_SINK_MULTI_TEST @CREATE_TEST_DATA @CREATE_TEST_DATA2 @DELETE_TEST_DATA @DELETE_TEST_DATA2 @Plugin-1890
   Scenario: Verify user should be able to deploy and run the pipeline using connection manager functionality
     When Open Datafusion Project to configure pipeline
     And Select plugin: "Salesforce Multi Objects" from the plugins list as: "Source"

diff --git a/src/e2e-test/features/salesforcemultiobjectsbatchsource/RunTimeWithMacros.feature b/src/e2e-test/features/salesforcemultiobjectsbatchsource/RunTimeWithMacros.feature
@@ -15,11 +15,12 @@
 @SalesforceSalesCloud
 @SFMultiObjectsBatchSource
 @Smoke
+# Skipping these tests due to permission issues for creating custom fields in sObjects. Will Run it once it get resolves.
 @Regression
 
 Feature: Salesforce Multi Objects Batch Source - Run time Scenarios with Macro
 
-  @MULTIBATCH-TS-SF-RNTM-MACRO-01 @BQ_SINK_MULTI_TEST @CREATE_TEST_DATA @CREATE_TEST_DATA2 @DELETE_TEST_DATA @DELETE_TEST_DATA2
+  @MULTIBATCH-TS-SF-RNTM-MACRO-01 @BQ_SINK_MULTI_TEST @CREATE_TEST_DATA @CREATE_TEST_DATA2 @DELETE_TEST_DATA @DELETE_TEST_DATA2 @Plugin-1890
   Scenario: Verify user should be able to preview, deploy a pipeline when plugin is configured with macros for WhiteList
     When Open Datafusion Project to configure pipeline
     And Select data pipeline type as: "Batch"

diff --git a/src/e2e-test/features/salesforcestreamingsource/DesignTime.feature b/src/e2e-test/features/salesforcestreamingsource/DesignTime.feature
@@ -27,6 +27,7 @@ Feature: Salesforce Streaming Source - Design time scenarios
     And Navigate to the properties page of plugin: "Salesforce"
     And fill Authentication properties for Salesforce Admin user
     And Enter input plugin property: "pushTopicName" with value: "topic.name"
+    And Enter input plugin property: "sObjectName" with value: "sobject.Automation_custom_c"
     And Select dropdown plugin property: "pushTopicNotifyCreate" with option value: "Enabled"
     And Select dropdown plugin property: "pushTopicNotifyUpdate" with option value: "Enabled"
     And Select dropdown plugin property: "pushTopicNotifyDelete" with option value: "Enabled"

diff --git a/src/e2e-test/java/io/cdap/plugin/BQValidation.java b/src/e2e-test/java/io/cdap/plugin/BQValidation.java
@@ -51,7 +51,7 @@ public class BQValidation {
 
   public static boolean validateSalesforceAndBQRecordValues(String objectName, String bqTable) throws
     IOException, InterruptedException {
-    String uniqueRecordId = SalesforceClient.queryObjectId(objectName);
+    List<String> uniqueRecordIds = SalesforceClient.queryObjectId(objectName);
 
     List<JsonObject> bigQueryResponse = new ArrayList<>();
     List<Object> bigQueryRows = new ArrayList<>();
@@ -60,8 +60,11 @@ public static boolean validateSalesforceAndBQRecordValues(String objectName, Str
       JsonObject jsonData = gson.fromJson(String.valueOf(rows), JsonObject.class);
       bigQueryResponse.add(jsonData);
     }
-    List<JsonObject> sObjectResponse;
-    sObjectResponse = SalesforceClient.queryObject(uniqueRecordId, objectName);
+    List<JsonObject> sObjectResponse = new ArrayList<>();
+    for (String recordId : uniqueRecordIds) {
+      JsonObject record = SalesforceClient.queryObject(recordId, objectName);
+      sObjectResponse.add(record);
+    }
     return compareSalesforceAndJsonData(sObjectResponse, bigQueryResponse, bqTable);
   }
 
@@ -82,9 +85,12 @@ public static boolean validateSalesforceMultiObjectToBQRecordValues() throws IOE
         JsonObject jsonData = gson.fromJson(String.valueOf(row), JsonObject.class);
         bigQueryResponse.add(jsonData);
       }
-      String uniqueRecordId = SalesforceClient.queryObjectId(currentObject);
-      List<JsonObject> sObjectResponse;
-      sObjectResponse = SalesforceClient.queryObject(uniqueRecordId, currentObject);
+      List<String> uniqueRecordIds = SalesforceClient.queryObjectId(currentObject);
+      List<JsonObject> sObjectResponse = new ArrayList<>();
+      for (String recordId : uniqueRecordIds) {
+        JsonObject record = SalesforceClient.queryObject(recordId, currentObject);
+        sObjectResponse.add(record);
+      }
       boolean isValid = compareSalesforceAndJsonData(
         sObjectResponse, bigQueryResponse, currentTargetTable);
 
@@ -141,14 +147,9 @@ public static boolean compareSalesforceAndJsonData(List<JsonObject> salesforceDa
       Assert.fail("bigQueryData is null");
       return result;
     }
-    int jsonObjectIdx = 0;
-    if (salesforceData.size() > 0) {
-      salesforceData.get(jsonObjectIdx).entrySet().size();
-    }
-    // Get the column count of the first JsonObject in bigQueryData
-    int columnCountSource = 0;
-    if (bigQueryData.size() > 0) {
-      columnCountSource = bigQueryData.get(jsonObjectIdx).entrySet().size();
+    if (salesforceData.isEmpty() || bigQueryData.isEmpty()) {
+      Assert.fail("One or both datasets are empty");
+      return result;
     }
 
     BigQuery bigQuery = BigQueryOptions.getDefaultInstance().getService();
@@ -158,39 +159,39 @@ public static boolean compareSalesforceAndJsonData(List<JsonObject> salesforceDa
     TableId tableRef = TableId.of(projectId, dataset, tableName);
     // Get the table schema
     Schema schema = bigQuery.getTable(tableRef).getDefinition().getSchema();
-    // Iterate over the fields
-    int currentColumnCount = 1;
-    while (currentColumnCount <= columnCountSource) {
+
+    for (int rowIndex = 0; rowIndex < salesforceData.size(); rowIndex++) {
+      JsonObject salesforceRow = salesforceData.get(rowIndex);
+      JsonObject bigQueryRow = bigQueryData.get(rowIndex);
+
       for (Field field : schema.getFields()) {
         String columnName = field.getName();
         String columnType = field.getType().toString();
 
         switch (columnType) {
 
           case "BOOLEAN":
-            boolean sourceAsBoolean = salesforceData.get(jsonObjectIdx).get(columnName).getAsBoolean();
-            boolean targetAsBoolean = bigQueryData.get(jsonObjectIdx).get(columnName).getAsBoolean();
+            boolean sourceAsBoolean = salesforceRow.get(columnName).getAsBoolean();
+            boolean targetAsBoolean = bigQueryRow.get(columnName).getAsBoolean();
             Assert.assertEquals("Different values found for column : %s", sourceAsBoolean, targetAsBoolean);
             break;
 
           case "FLOAT":
-            double sourceVal = salesforceData.get(jsonObjectIdx).get(columnName).getAsDouble();
-            double targetVal = bigQueryData.get(jsonObjectIdx).get(columnName).getAsDouble();
+            double sourceVal = salesforceRow.get(columnName).getAsDouble();
+            double targetVal = bigQueryRow.get(columnName).getAsDouble();
             Assert.assertEquals(String.format("Different values found for column: %s", columnName), 0,
                                 Double.compare(sourceVal, targetVal));
             break;
 
           case "TIMESTAMP":
             OffsetDateTime sourceTimestamp = OffsetDateTime.parse(
-              salesforceData.get(jsonObjectIdx)
-                .get(columnName)
+              salesforceRow.get(columnName)
                 .getAsString(),
               DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ")
             );
 
             OffsetDateTime targetTimestamp = OffsetDateTime.parse(
-              bigQueryData.get(jsonObjectIdx)
-                .get(columnName)
+              bigQueryRow.get(columnName)
                 .getAsString()
             );
             Assert.assertEquals("Different values found for column : %s", sourceTimestamp, targetTimestamp);
@@ -200,23 +201,21 @@ public static boolean compareSalesforceAndJsonData(List<JsonObject> salesforceDa
             DateTimeFormatter formatterSource = DateTimeFormatter.ofPattern("HH:mm:ss.SSSX");
             DateTimeFormatter formatterTarget = DateTimeFormatter.ofPattern("HH:mm:ss");
             LocalTime sourceTime = LocalTime.parse(
-              salesforceData.get(jsonObjectIdx)
-                .get(columnName)
+              salesforceRow.get(columnName)
                 .getAsString(), formatterSource
             );
             LocalTime targetTime = LocalTime.parse(
-              bigQueryData.get(jsonObjectIdx)
-                .get(columnName)
+              bigQueryRow.get(columnName)
                 .getAsString(), formatterTarget
             );
             Assert.assertEquals("Different values found for column : %s", sourceTime, targetTime);
             break;
 
           case "DATE":
-            JsonElement jsonElementSource = salesforceData.get(jsonObjectIdx).get(columnName);
+            JsonElement jsonElementSource = salesforceRow.get(columnName);
             Date sourceDate = (jsonElementSource != null && !jsonElementSource.isJsonNull()) ? Date.valueOf(
               jsonElementSource.getAsString()) : null;
-            JsonElement jsonElementTarget = bigQueryData.get(jsonObjectIdx).get(columnName);
+            JsonElement jsonElementTarget = bigQueryRow.get(columnName);
             Date targetDate = (jsonElementTarget != null && !jsonElementTarget.isJsonNull()) ? Date.valueOf(
               jsonElementTarget.getAsString()) : null;
             Assert.assertEquals("Different values found for column : %s", sourceDate, targetDate);
@@ -231,21 +230,19 @@ public static boolean compareSalesforceAndJsonData(List<JsonObject> salesforceDa
             if (columnName.equals("Col_GeoLocation__c")) {
               break;
             } else {
-              JsonElement sourceElement = salesforceData.get(jsonObjectIdx).get(columnName);
+              JsonElement sourceElement = salesforceRow.get(columnName);
               String sourceString = (sourceElement != null && !sourceElement.isJsonNull())
                 ? sourceElement.getAsString() : null;
-              JsonElement targetElement = bigQueryData.get(jsonObjectIdx).get(columnName);
+              JsonElement targetElement = bigQueryRow.get(columnName);
               String targetString = (targetElement != null && !targetElement.isJsonNull())
                 ? targetElement.getAsString() : null;
               Assert.assertEquals(String.format("Different  values found for column : %s", columnName),
                                   String.valueOf(sourceString), String.valueOf(targetString));
-              break;
+            }
+            break;
             }
         }
-        currentColumnCount++;
       }
-      jsonObjectIdx++;
-    }
     Assert.assertFalse("Number of rows in Source table is greater than the number of rows in Target table",
                        salesforceData.size() > bigQueryData.size());
     return true;

diff --git a/src/e2e-test/java/io/cdap/plugin/salesforcestreamingsource/stepsdesign/DesignTimeSteps.java b/src/e2e-test/java/io/cdap/plugin/salesforcestreamingsource/stepsdesign/DesignTimeSteps.java
@@ -35,6 +35,7 @@
 
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 /**
  * Design-time steps of Salesforce Streaming plugins.
@@ -70,8 +71,10 @@ public void fillUniqueTopicNameInRuntimeArguments(String runtimeArgumentKey) {
 
   @Then("Update existing salesforce records")
   public void updateExistingSalesforceRecords() {
-    String uniqueRecordId = SalesforceClient.queryObjectId(customObject);
-    SalesforceClient.updateObject(uniqueRecordId, customObject);
+    List<String> uniqueRecordIds = SalesforceClient.queryObjectId(customObject);
+    for (String recordId : uniqueRecordIds) {
+      SalesforceClient.updateObject(recordId, customObject);
+    }
   }
 
 

diff --git a/src/e2e-test/java/io/cdap/plugin/tests/hooks/TestSetupHooks.java b/src/e2e-test/java/io/cdap/plugin/tests/hooks/TestSetupHooks.java
@@ -37,6 +37,7 @@
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.text.ParseException;
+import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.UUID;
 
@@ -65,16 +66,20 @@ public static void createTestObject2() throws UnsupportedEncodingException, Pars
 
   @After(order = 2, value = "@DELETE_TEST_DATA")
   public static void deleteTestObject() {
-    String uniqueRecordId = SalesforceClient.queryObjectId(SObjects.AUTOMATION_CUSTOM__C.value);
-    SalesforceClient.deleteId(uniqueRecordId, SObjects.AUTOMATION_CUSTOM__C.value);
-    BeforeActions.scenario.write("Record - " + uniqueRecordId + " deleted successfully");
+    List<String> uniqueRecordIds = SalesforceClient.queryObjectId(SObjects.AUTOMATION_CUSTOM__C.value);
+    for (String recordId : uniqueRecordIds) {
+    SalesforceClient.deleteId(recordId, SObjects.AUTOMATION_CUSTOM__C.value);
+    BeforeActions.scenario.write("Record - " + recordId + " deleted successfully");
+    }
   }
 
   @After(order = 2, value = "@DELETE_TEST_DATA2")
   public static void deleteTestObject2() {
-    String uniqueRecordId = SalesforceClient.queryObjectId(SObjects.AUTOMATION_CUSTOM2__C.value);
-    SalesforceClient.deleteId(uniqueRecordId, SObjects.AUTOMATION_CUSTOM2__C.value);
-    BeforeActions.scenario.write("Record - " + uniqueRecordId + " deleted successfully");
+    List<String> uniqueRecordIds = SalesforceClient.queryObjectId(SObjects.AUTOMATION_CUSTOM2__C.value);
+    for (String recordId : uniqueRecordIds) {
+      SalesforceClient.deleteId(recordId, SObjects.AUTOMATION_CUSTOM__C.value);
+      BeforeActions.scenario.write("Record - " + recordId + " deleted successfully");
+    }
   }
 
 

diff --git a/src/e2e-test/java/io/cdap/plugin/tests/runner/TestRunner.java b/src/e2e-test/java/io/cdap/plugin/tests/runner/TestRunner.java
@@ -32,7 +32,7 @@
     "io.cdap.plugin.salesforcemultiobjectsbatchsource.stepsdesign",
     "io.cdap.plugin.bigquery.stepsdesign",
     "io.cdap.plugin.tests.hooks", "stepsdesign"},
-  tags = {"@Regression"},
+  tags = {"@Regression and not @Plugin-1890"},
   monochrome = true,
   plugin = {"pretty", "html:target/cucumber-html-report", "json:target/cucumber-reports/cucumber.json",
     "junit:target/cucumber-reports/cucumber.xml"}

diff --git a/src/e2e-test/java/io/cdap/plugin/utils/SalesforceClient.java b/src/e2e-test/java/io/cdap/plugin/utils/SalesforceClient.java
@@ -149,7 +149,7 @@ public static String createObject(JSONObject objectJson, String objectName) thro
     return uniqueRecordId;
   }
 
-  public static List<JsonObject> queryObject(String id, String objectName) {
+  public static JsonObject queryObject(String id, String objectName) {
     getAccessToken();
     HttpClient httpClient = HttpClientBuilder.create().build();
     String baseUri = loginInstanceUrl + REST_ENDPOINT + API_VERSION;
@@ -167,12 +167,12 @@ public static List<JsonObject> queryObject(String id, String objectName) {
         String responseString = EntityUtils.toString(response.getEntity());
         Gson gson = new Gson();
         JsonObject objectResponseInJson = gson.fromJson(responseString, JsonObject.class);
-        sfobjectResponse.add(objectResponseInJson);
+        return objectResponseInJson;
       }
     } catch (IOException ioException) {
       logger.info("Error in establishing connection to Salesforce: " + ioException);
     }
-    return sfobjectResponse;
+    return null;
   }
 
   public static void deletePushTopic(String pushTopicName) {
@@ -233,10 +233,11 @@ public static void deleteId(String id, String objectName) {
     }
   }
 
-  public static String queryObjectId(String objectName) {
+  public static List<String> queryObjectId(String objectName) {
     getAccessToken();
     HttpClient httpClient = HttpClientBuilder.create().build();
     String baseUri = loginInstanceUrl + REST_ENDPOINT + API_VERSION;
+    List<String> idList = new ArrayList<>();
 
     try {
       String query = "SELECT Id FROM " + objectName;
@@ -259,14 +260,15 @@ public static String queryObjectId(String objectName) {
         JsonArray records = queryResponse.getAsJsonArray("records");
         for (JsonElement record : records) {
           JsonObject recordObject = record.getAsJsonObject();
-          uniqueRecordId = recordObject.get("Id").getAsString();
-          logger.info("Queried Object id from response: " + uniqueRecordId);
+          String id = recordObject.get("Id").getAsString();
+          idList.add(id);
+          logger.info("Queried Object id from response: " + id);
         }
       }
     } catch (IOException ioException) {
       logger.info("Error in establishing connection to Salesforce: " + ioException);
     }
-    return uniqueRecordId;
+    return idList;
   }
 
     public static void updateObject(String id, String objectName) {

diff --git a/src/e2e-test/resources/BigQuery/BigQueryCreateTableQuery.txt b/src/e2e-test/resources/BigQuery/BigQueryCreateTableQuery.txt
@@ -1,3 +1,2 @@
 create table `DATASET.TABLE_NAME` (Name STRING, Col_Timestamp__c TIMESTAMP, Col_Date__c DATE, Col_Currency__c FLOAT64,
- Col_Email__c STRING, Col_Number__c FLOAT64, Col_GeoLocation__Latitude__s FLOAT64,
-Col_GeoLocation__Longitude__s FLOAT64, Col__c STRING, Col_Url__c STRING, Col_Time__c TIME, Col_Text__c STRING)
+ Col_Email__c STRING, Col_Number__c FLOAT64, Col__c STRING, Col_Url__c STRING, Col_Time__c TIME, Col_Text__c STRING)
diff --git a/src/e2e-test/resources/BigQuery/BigQueryInsertDataQuery.txt b/src/e2e-test/resources/BigQuery/BigQueryInsertDataQuery.txt
@@ -1,5 +1,5 @@
 insert into `DATASET.TABLE_NAME` (Name, Col_Timestamp__c, Col_Date__c, Col_Currency__c, Col_Email__c, Col_Number__c,
-Col_GeoLocation__Latitude__s, Col_GeoLocation__Longitude__s, Col__c, Col_Url__c, Col_Time__c, Col_Text__c) values
-('adam','2019-03-10 04:50:01 UTC','2021-01-28',61.823765812,'[email protected]',898365444,37.794116,-122.3432,
+Col__c, Col_Url__c, Col_Time__c, Col_Text__c) values
+('adam','2019-03-10 04:50:01 UTC','2021-01-28',61.823765812,'[email protected]',898365444,
 '984746334','abc/123','20:26:34','find');