[PLUGIN-1706] Implement Retry logic in Salesforce batch source and Mutli Source Plugin for connection timeout issues #225

sgarg-CS · 2023-11-17T15:11:24Z

PLUGIN-1706: Implement Retry logic in Salesforce batch source and Multi-Source Plugin for connection timeout issues

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceSourceConstants.java

itsankit-google

Please add unit tests.

Shubhangi-cs · 2023-11-30T08:03:44Z

Please add unit tests.

Added

itsankit-google · 2023-11-30T18:01:28Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

@@ -50,7 +55,11 @@
 public class SalesforceBulkRecordReader extends RecordReader<Schema, Map<String, ?>> {

  private static final Logger LOG = LoggerFactory.getLogger(SalesforceBulkRecordReader.class);
-
+  public static final Set<String> RETRY_ON_REASON = ImmutableSet.of("Unknown", "InternalServerError",
+                                                                    "ClientInputError", "Timeout");


When is ClientInputError returned? Sounds like a bad request 400 which should not be retried.

ClientInputError is coming when bulk api library not able to parse it properly and this is coming intermittently, we have added fix for this only.

...va/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceQueryExecutionException.java

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceSplitUtil.java

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java

itsankit-google · 2023-12-04T06:16:48Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+        Failsafe.with(SalesforceSplitUtil.getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount))
+          .run(() -> batchInfoList[0] = getQueryResultList(bulkConnection));
+        return batchInfoList[0];
+        //return bulkConnection.getQueryResultList(jobId, batchId).getResult();


nit: remove comment not required.

itsankit-google · 2023-12-04T06:31:49Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+        Failsafe.with(SalesforceSplitUtil.getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount))
+          .run(() -> batchInfoList[0] = getQueryResultList(bulkConnection));
+        return batchInfoList[0];


what is the use of storing batchInfoList and then returning?

return Failsafe.with(SalesforceSplitUtil.getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount)) .run(() -> getQueryResultList(bulkConnection));

As it is used inside the Failsafe block there r chances of exception while retrying. Just used this variable first to capture the successful result before returning.

This does not make sense, how does storing it in the variable help here?

Can you share an example explaining this?

Done the change

itsankit-google · 2023-12-04T11:57:32Z

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java

+      Mockito.when(mockConnection.getQueryResultStream(jobId, batchId, resultId))
+        .thenThrow(new SalesforceQueryExecutionException("Simulated error"))
+        .thenThrow(new SalesforceQueryExecutionException("Simulated error"))
+        .thenReturn(new ByteArrayInputStream(csvStrings[1].getBytes(StandardCharsets.UTF_8)));


nit: identation

itsankit-google · 2023-12-04T12:22:32Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java


+    final InputStream[] queryResponseStream = new InputStream[1];


why this an array of InputStream?

As lambda function is used with Failsafe library, we need to initialize the array of inputstream to use final keyword.

What error did you get with this code?

final InputStream queryResponseStream = Failsafe.with(SalesforceSplitUtil.getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount)) .run(() -> getQueryResultStream(bulkConnection));

itsankit-google · 2023-12-04T12:23:01Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

-    throws AsyncApiException, InterruptedException {
-
+    throws AsyncApiException, InterruptedException, SalesforceQueryExecutionException {
+    final String[][] batchInfoList = new String[1][];


Is this used anywhere?

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java

itsankit-google · 2023-12-04T12:34:11Z

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java

+
+  @Test
+  public void testSetupParserWithRetrySuccess() throws Exception {
+    MapToRecordTransformer transformer = new MapToRecordTransformer();


The first usage of transformer seem to be at line 457, as a recommended practice, local variables shouldn't be initialized more than 2-3 lines before they are used.

itsankit-google · 2023-12-05T12:29:16Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceSplitUtil.java

+    final BatchInfo[] batchInfo = new BatchInfo[1];
+    Failsafe.with(getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount))
+      .run(() -> batchInfo[0] = createBatchFromStream(bulkConnection, query, job));


Please fix this as well similarly.

final BatchInfo batchInfo = Failsafe.with(getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount)) .run(() -> createBatchFromStream(bulkConnection, query, job));

itsankit-google · 2023-12-05T12:33:46Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceSplitUtil.java

+      LOG.error("The bulk query job {} failed. Job State: {}", job.getId(), job.getState());
+      if (SalesforceBulkRecordReader.RETRY_ON_REASON.contains(exception.getExceptionCode())) {
+        throw new SalesforceQueryExecutionException(exception.getMessage());
+      }


if the error is not retryable, we are ignoring the error here, is it expected?

error logged

itsankit-google · 2023-12-05T12:34:02Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+      LOG.error("The bulk query job {} failed.", jobId);
+      if (RETRY_ON_REASON.contains(exception.getExceptionCode())) {
+        throw new SalesforceQueryExecutionException(exception.getMessage());
+      }


if the error is not retryable, we are ignoring the error here, is it expected?

error logged

itsankit-google · 2023-12-05T12:34:25Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+      LOG.error("The bulk query job {} failed.", jobId);
+      if (RETRY_ON_REASON.contains(exception.getExceptionCode())) {
+        throw new SalesforceQueryExecutionException(exception.getMessage());
+      }


if the error is not retryable, we are ignoring the error here, is it expected?

error logged

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java

itsankit-google · 2023-12-07T05:55:33Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+      if (RETRY_ON_REASON.contains(exception.getExceptionCode())) {
+        throw new SalesforceQueryExecutionException(exception.getMessage());
+      }
+      throw new AsyncApiException("Failed to parse query result list ", AsyncExceptionCode.ClientInputError, exception);


Why are we overriding the AsyncApiException, can't we just throw exception? Same applies to all places.

itsankit-google · 2023-12-07T05:55:56Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+    } catch (AsyncApiException exception) {
+      LOG.error("The bulk query job {} failed.", jobId);
+      if (RETRY_ON_REASON.contains(exception.getExceptionCode())) {
+        throw new SalesforceQueryExecutionException(exception.getMessage());


This way you are loosing all the additional information from original AsyncApiException stack trace. There are only some case where you would want to hide all details of the underlying exception by using e.getMessage instead of attaching a root cause. I don't see why would we want to have it here.

Same applies to all the places in this PR.

itsankit-google · 2023-12-07T05:57:16Z

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java

+    Long initialRetryDuration = 5L;
+    Long maxRetryDuration = 80L;


You probably want immediate retries, otherwise this test would be waiting unnecessary.

itsankit-google

Users should also be able to disable the retry.

We should add a Retry Configuration section in the UI.

There should be a toggle to disable the behavior and fine tune the retries(which is already added).

itsankit-google · 2023-12-07T06:02:13Z

Please don't remove the build label, it helps ensuring that we are not introducing any compile time errors.

itsankit-google · 2023-12-07T06:17:55Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+    try {
+      return bulkConnection.getQueryResultStream(jobId, batchId, resultIds[resultIdIndex]);
+    } catch (AsyncApiException exception) {
+      LOG.error("The bulk query job {} failed.", jobId);


If you catch it and log it and rethrow, it is considered as an anti-pattern, ref: https://stackoverflow.com/questions/6639963/why-is-log-and-throw-considered-an-anti-pattern

It is okay to log if you think this information will not be logged anywhere above in the stack, but it should not be an error log since we are also retrying the error, it can be a warn.

Same comment applies to all places in PR.

Logging the jobId as warn might still be useful as I cannot see it getting logged in upper stack anywhere.

Logged as a warning

itsankit-google · 2023-12-08T05:39:41Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+        return Failsafe.with(SalesforceSplitUtil.getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount))
+          .get(() -> getQueryResultList(bulkConnection));


I am wondering if we would need to unwrap those unchecked exceptions to prevent unneeded layers / cluttered message, see: https://github.com/data-integrations/bigquery-delta-plugins/pull/40/files#diff-427b4fbd30b2c80fb381e0e8fc3a6e2424ec9f0c94e8ed0881b0bc49402fc385R268-R279 for reference.

Needs to be changed at all the three places where it is used or here only?

The comment applies to all similar places.

itsankit-google · 2023-12-11T19:56:39Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

+          }
+          throw e;
+        }
+


nit: remove extra empty line.

Shubhangi-cs · 2023-12-13T10:39:20Z

Users should also be able to disable the retry.

We should add a Retry Configuration section in the UI.

There should be a toggle to disable the behavior and fine tune the retries(which is already added).

Added

vikasrathee-cs · 2023-12-14T11:15:53Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceSplitUtil.java

+      null, ConcurrencyMode.Parallel, ContentType.CSV);
+    final BatchInfo batchInfo;
+    try {
+      if (retryOnBackendError) {


instead of having if else for whole code, if !retryEnabled, set maxretryattempts to 0 at all places. Keep failsafe with 0 maxRetryCount.

change this at all places

vikasrathee-cs · 2023-12-14T11:19:40Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

@@ -73,6 +90,10 @@ public SalesforceBulkRecordReader(Schema schema) {
    this.jobId = jobId;
    this.batchId = batchId;
    this.resultIds = resultIds;
+    initialRetryDuration = SalesforceSourceConstants.DEFAULT_INITIAL_RETRY_DURATION_SECONDS;


remove this code, instead use like this in initialize method.
conf.get(SalesforceSourceConstants.CONFIG_INITIAL_RETRY_DURATION, SalesforceSourceConstants.DEFAULT_INITIAL_RETRY_DURATION_SECONDS)

this is used for testing purpose only.

still it is not solving any purpose, if those properties are null in config, these value will be overidden by null

It will require extra mocking for conf and stubs for all the values required for the test cases.

itsankit-google · 2023-12-15T07:24:40Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceSourceConfig.java

+                         @Nullable String consumerKey,
+                         @Nullable String consumerSecret,
+                         @Nullable String username,
+                         @Nullable String password,
+                         @Nullable String loginUrl,
+                         @Nullable Integer connectTimeout,
+                         @Nullable Integer readTimeout,
+                         @Nullable String query,
+                         @Nullable String sObjectName,
+                         @Nullable String datetimeAfter,
+                         @Nullable String datetimeBefore,
+                         @Nullable String duration,
+                         @Nullable String offset,
+                         @Nullable String schema,
+                         @Nullable String securityToken,
+                         @Nullable String operation,
+                         @Nullable Long initialRetryDuration,
+                         @Nullable Long maxRetryDuration,
+                         @Nullable Integer maxRetryCount,
+                         Boolean retryOnBackendError,
+                         @Nullable OAuthInfo oAuthInfo,
+                         @Nullable Boolean enablePKChunk,
+                         @Nullable Integer chunkSize,
+                         @Nullable String parent,
+                         @Nullable String proxyUrl) {


nit: indentation

itsankit-google · 2023-12-15T07:26:17Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReader.java

-    csvParser = CSVParser.parse(queryResponseStream, StandardCharsets.UTF_8, csvFormat);
-    if (csvParser.getHeaderMap().isEmpty()) {
-      throw new IllegalStateException("Empty response was received from Salesforce, but csv header was expected.");
+    int maxRetryCount = isRetryRequired ? maxRetryAttempts : 0;


when maxRetryCount is 0 it should not use FailSafe to call the salesforce query method.

This comment applies to all places in the PR.

done the changes

itsankit-google · 2023-12-18T06:22:27Z

...va/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceQueryExecutionException.java

+ * the License.
+ */
+
+import io.cdap.cdap.api.retry.RetryableException;


remove unused imports.

itsankit-google · 2023-12-18T06:24:42Z

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBaseSourceConfig.java

  }


  public Map<ChronoUnit, Integer> getDuration() {
    return extractRangeValue(SalesforceSourceConstants.PROPERTY_DURATION, duration);
  }

+  public Boolean isRetryRequired() {
+    return retryOnBackendError;


This would be null in case of older pipelines. Did we verify it?

My intent was that we should handle it somewhere at the top level or in this class.

I leave it up to you to decide.

Done the change

My intent was that we should handle it somewhere at the top level or in this class.

I leave it up to you to decide.

ya changed it in config as well

itsankit-google · 2023-12-18T06:29:56Z

...va/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceQueryExecutionException.java

@@ -0,0 +1,39 @@
+package io.cdap.plugin.salesforce.plugin.source.batch.util;


this statement should be below the license.

itsankit-google · 2023-12-18T07:36:24Z

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java

+   * The FailsafeException is thrown with the expected error message.
+   * The retry policy is correctly configured with the specified configurations.
+   */
+  @Test(expected = FailsafeException.class)


why FailsafeException here?

Javadoc comment suggests that The FailsafeException is thrown with the expected error message. Where are we comparing the expected error message?

FailsafeException is expected as retry limit will get exceeded, changed the javadoc accordingly

itsankit-google · 2023-12-18T09:10:26Z

Please squash commits before merge.

sgarg-CS requested a review from vikasrathee-cs November 17, 2023 15:15

sgarg-CS force-pushed the feature/retry-bulk-query branch 2 times, most recently from 9a76af8 to c8b6161 Compare November 17, 2023 15:35

sgarg-CS changed the title ~~Implement Retry logic in Salesforce batch source and Mutli Source Plugin for connection timeout issues~~ [PLUGIN-1706] Implement Retry logic in Salesforce batch source and Mutli Source Plugin for connection timeout issues Nov 24, 2023

vikasrathee-cs requested a review from itsankit-google November 27, 2023 13:32

itsankit-google added the build label Nov 27, 2023

itsankit-google reviewed Nov 28, 2023

View reviewed changes

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/util/SalesforceSourceConstants.java Outdated Show resolved Hide resolved

itsankit-google reviewed Nov 28, 2023

View reviewed changes

Vipinofficial11 force-pushed the feature/retry-bulk-query branch from 2f78859 to 8c56415 Compare November 29, 2023 08:36

itsankit-google reviewed Nov 30, 2023

View reviewed changes

vikasrathee-cs requested review from itsankit-google and removed request for vikasrathee-cs December 1, 2023 05:59

itsankit-google reviewed Dec 4, 2023

View reviewed changes

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java Show resolved Hide resolved

itsankit-google reviewed Dec 4, 2023

View reviewed changes

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java Outdated Show resolved Hide resolved

itsankit-google reviewed Dec 4, 2023

View reviewed changes

itsankit-google reviewed Dec 5, 2023

View reviewed changes

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceBulkRecordReaderTest.java Show resolved Hide resolved

vikasrathee-cs removed the build label Dec 6, 2023

itsankit-google reviewed Dec 7, 2023

View reviewed changes

itsankit-google added the build label Dec 7, 2023

itsankit-google reviewed Dec 7, 2023

View reviewed changes

itsankit-google reviewed Dec 8, 2023

View reviewed changes

sau42shri self-requested a review December 8, 2023 12:04

itsankit-google reviewed Dec 11, 2023

View reviewed changes

Shubhangi-cs force-pushed the feature/retry-bulk-query branch from 0554370 to 7d64c9d Compare December 12, 2023 05:27

vikasrathee-cs reviewed Dec 14, 2023

View reviewed changes

itsankit-google reviewed Dec 15, 2023

View reviewed changes

itsankit-google reviewed Dec 18, 2023

View reviewed changes

itsankit-google approved these changes Dec 18, 2023

View reviewed changes

Shubhangi-cs force-pushed the feature/retry-bulk-query branch 4 times, most recently from 8030eb4 to 5e490ac Compare December 18, 2023 12:05

Retry logic for bulk query jobs

b92576e

Shubhangi-cs force-pushed the feature/retry-bulk-query branch from 5e490ac to b92576e Compare December 18, 2023 12:06

Shubhangi-cs merged commit a84eecd into data-integrations:develop Dec 18, 2023
3 checks passed

This was referenced Dec 18, 2023

Salesforce Release 1.6.2 #231

Closed

Feature/enhancement 1.6.2 Release #232

Merged


		final InputStream[] queryResponseStream = new InputStream[1];

		return Failsafe.with(SalesforceSplitUtil.getRetryPolicy(initialRetryDuration, maxRetryDuration, maxRetryCount))
		.get(() -> getQueryResultList(bulkConnection));

		@@ -0,0 +1,39 @@
		package io.cdap.plugin.salesforce.plugin.source.batch.util;

[PLUGIN-1706] Implement Retry logic in Salesforce batch source and Mutli Source Plugin for connection timeout issues #225

[PLUGIN-1706] Implement Retry logic in Salesforce batch source and Mutli Source Plugin for connection timeout issues #225

Conversation

sgarg-CS commented Nov 17, 2023 • edited by jira bot Loading

itsankit-google left a comment

Choose a reason for hiding this comment

Shubhangi-cs commented Nov 30, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

itsankit-google Dec 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

itsankit-google Dec 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

itsankit-google Dec 5, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

itsankit-google Dec 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

itsankit-google left a comment

Choose a reason for hiding this comment

itsankit-google commented Dec 7, 2023

itsankit-google Dec 7, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Shubhangi-cs commented Dec 13, 2023

vikasrathee-cs Dec 14, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sgarg-CS commented Nov 17, 2023 •

edited by jira bot

Loading

itsankit-google Dec 4, 2023 •

edited

Loading

itsankit-google Dec 4, 2023 •

edited

Loading

itsankit-google Dec 5, 2023 •

edited

Loading

itsankit-google Dec 7, 2023 •

edited

Loading

itsankit-google Dec 7, 2023 •

edited

Loading

vikasrathee-cs Dec 14, 2023 •

edited

Loading