OpenSearch Sink should make the number of retries configurable - Issue #2291 #2339

kkondaka · 2023-03-01T06:24:51Z

Description

Opensearch Sink retries the retryable operations forever. This changes adds configuration option to opensearch sink to limit the number of retries.

Issues Resolved

Resolves #2291

Check List

[X ] New functionality includes testing.
New functionality has been documented.
- New functionality has javadoc added
[ X] Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

…opensearch-project#2291 Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

codecov-commenter · 2023-03-01T06:43:23Z

Codecov Report

Merging #2339 (a166221) into main (6d4eb42) will decrease coverage by 0.17%.
The diff coverage is n/a.

📣 This organization is not using Codecov’s GitHub App Integration. We recommend you install it so Codecov can continue to function properly for your repositories. Learn more

@@             Coverage Diff              @@
##               main    #2339      +/-   ##
============================================
- Coverage     93.81%   93.65%   -0.17%     
+ Complexity     1977     1975       -2     
============================================
  Files           229      229              
  Lines          5499     5499              
  Branches        442      442              
============================================
- Hits           5159     5150       -9     
- Misses          230      241      +11     
+ Partials        110      108       -2

see 1 file with indirect coverage changes

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

cmanning09 · 2023-03-03T22:04:48Z

...rch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/RetryConfiguration.java


    public Builder withDlqFile(final String dlqFile) {
      checkNotNull(dlqFile, "dlqFile cannot be null.");
      this.dlqFile = dlqFile;
      return this;
    }

+    public Builder withMaxRetries(final Integer maxRetries) {


Can we provide more min and max validations around the maxRetries. There are some edge cases like negative numbers and Integer.MAX_VALUE that I am not convinced we should support.

Integer.MAX_VALUE should be OK to support indefinite retries.

cmanning09 · 2023-03-03T22:05:34Z

...nsearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

@@ -138,10 +138,12 @@ private void doInitializeInternal() throws IOException {
    OpenSearchTransport transport = new RestClientTransport(restHighLevelClient.getLowLevelClient(), new PreSerializedJsonpMapper());
    openSearchClient = new OpenSearchClient(transport);
    bulkRequestSupplier = () -> new JavaClientAccumulatingBulkRequest(new BulkRequest.Builder());
+    final int maxRetries = openSearchSinkConfig.getRetryConfiguration().getMaxRetries();


Let's update the README.md file with this new configuration option.

cmanning09 · 2023-03-03T22:17:56Z

...src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategyTests.java

@@ -68,7 +68,7 @@ public void testCanRetry() {
        AccumulatingBulkRequest accumulatingBulkRequest = mock(AccumulatingBulkRequest.class);


Which one of these tests are validating we are honoring the configured retry config?

Added new tests to test retry config.

dlvenable · 2023-03-03T22:45:31Z

...in/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/AccumulatingBulkRequest.java

@@ -24,6 +24,8 @@

    int getOperationsCount();

+    int incrementAndGetRetryCount();


I think this is not the right place to have this.

The purpose of this class is to accumulate documents to know when to flush. This is a different concern than handling retries.

Also, each new retry may issue a new AccumulatingBulkRequest which would lose the count.

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dlvenable · 2023-03-08T01:22:19Z

...in/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/AccumulatingBulkRequest.java

@@ -26,6 +26,10 @@

    int incrementAndGetRetryCount();

+    void setRetryCount(int count);


Adding the setter will help the issue of not losing count. But, I still think that there should be a distinct class for handling the retry counts. The AccumulatingBulkRequest can focus on counting accumulations which is its current responsibility.

Another class like RetryableBulkRequest could handle the retry counting.

…pper into os-limit-retries rebasing

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dlvenable · 2023-03-13T20:50:16Z

...g/opensearch/dataprepper/plugins/sink/opensearch/bulk/JavaClientAccumulatingBulkRequest.java

@@ -19,6 +19,7 @@ public class JavaClientAccumulatingBulkRequest implements AccumulatingBulkReques
    private BulkRequest.Builder bulkRequestBuilder;
    private long currentBulkSize = 0L;
    private int operationCount = 0;
+    private int retryCount = 0;


I think this was left behind from the previous revision.

dlvenable · 2023-03-13T20:53:27Z

...src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategyTests.java

+
+        private BulkResponse bulkMaxRetriesResponse(final BulkRequest bulkRequest) {
+            final int requestSize = bulkRequest.operations().size();
+            assert requestSize == 4;


Please use assertThat(requestSize, equalTo(4));

dlvenable · 2023-03-13T21:02:14Z

...arch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategy.java

            } else {
                final int numberOfDocs = bulkRequestForRetry.getOperationsCount();
-                if (firstAttempt) {
+                if (retryCount == 1) {


It might be good to create a variable boolean firstAttempt = retryCount == 1. It is a little unclear in this function that retryCount = 1 actually means that there have been no retries and this is the first attempt.

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

cmanning09 · 2023-03-14T14:53:12Z

...arch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategy.java

+                                          final BackOffUtils backOffUtils,
+                                          final BulkResponse bulkResponse,
+                                          Exception e) throws InterruptedException {
+        boolean doRetry = (Objects.isNull(e)) ? canRetry(bulkResponse) : canRetry(e);


minor: this can be final to avoid unintentional future re-assignment

cmanning09 · 2023-03-14T15:02:05Z

...arch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategy.java

+                                          final BulkResponse bulkResponse,
+                                          Exception e) throws InterruptedException {
+        boolean doRetry = (Objects.isNull(e)) ? canRetry(bulkResponse) : canRetry(e);
+        if (!Objects.isNull(bulkResponse) && doRetry && retryCount == 1) { // first attempt


Why does doRetry need to be true here? If it is false and some documents were successfully sent on the first attempt we should increment the counter for those documents.

@cmanning09 I guess you are right. I just refactored the existing code. It looks like even existing code is incrementing stat only when canRetry is true. Modified the code to increment the stat even when the doRetry is not true.

cmanning09 · 2023-03-14T15:03:14Z

...arch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategy.java

+            bulkRequestNumberOfRetries.increment();
+        } else {
+            if (doRetry && retryCount >= maxRetries) {
+                e = new RuntimeException(String.format("Number of retries reached the limit of max retries(configured value %d)", maxRetries));


We should avoid re-assignment of method parameters. This can be dangerous and leads to confusing code.

In this case, the caller gets the new exception returned too, so that it gets passed back. Basically it is an in & out parameter to the function handleRetriesAndFailures

If we need to return an object we should use the standard mechanism for returning an object and return the object as a return value. This current behavior is a side affect that happens conditionally and is not documented. I am concerned this will be problematic in the future.

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

OpenSearch Sink should make the number of retries configurable - Issue …

dc1dc6f

…opensearch-project#2291 Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

kkondaka requested a review from a team as a code owner March 1, 2023 06:24

cmanning09 reviewed Mar 3, 2023

View reviewed changes

dlvenable requested changes Mar 3, 2023

View reviewed changes

kkondaka added 2 commits March 6, 2023 23:15

Addressed review comments

739c543

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

Fixed build failure and updated the documentation

6b6a0c1

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dlvenable requested changes Mar 8, 2023

View reviewed changes

kkondaka added 2 commits March 8, 2023 04:50

Merge branch 'main' of https://github.com/opensearch-project/data-pre…

c88e995

…pper into os-limit-retries rebasing

Modified to keep track of request retry counts in BulkRetryStrategy

6ec4834

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dlvenable requested changes Mar 13, 2023

View reviewed changes

Addressed review comments

a166221

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

cmanning09 reviewed Mar 14, 2023

View reviewed changes

Addressed review comments

35b5e18

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

cmanning09 approved these changes Mar 16, 2023

View reviewed changes

dlvenable approved these changes Mar 16, 2023

View reviewed changes

dlvenable merged commit 5cf8659 into opensearch-project:main Mar 16, 2023

cmanning09 mentioned this pull request Apr 26, 2023

[BUG] OpenSearch Sink failures are masked by max retry in the logs #2597

Closed

kkondaka deleted the os-limit-retries branch July 13, 2023 04:31

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

OpenSearch Sink should make the number of retries configurable - Issue #2291 #2339

OpenSearch Sink should make the number of retries configurable - Issue #2291 #2339

kkondaka commented Mar 1, 2023

codecov-commenter commented Mar 1, 2023 •

edited

Loading

cmanning09 Mar 3, 2023

kkondaka Mar 6, 2023 •

edited

Loading

cmanning09 Mar 3, 2023

cmanning09 Mar 3, 2023

kkondaka Mar 13, 2023

dlvenable Mar 3, 2023

dlvenable Mar 8, 2023

dlvenable Mar 13, 2023

dlvenable Mar 13, 2023

dlvenable Mar 13, 2023

cmanning09 Mar 14, 2023

cmanning09 Mar 14, 2023

kkondaka Mar 14, 2023

cmanning09 Mar 14, 2023

kkondaka Mar 14, 2023

cmanning09 Mar 16, 2023

		@@ -68,7 +68,7 @@ public void testCanRetry() {
		AccumulatingBulkRequest accumulatingBulkRequest = mock(AccumulatingBulkRequest.class);

		@@ -24,6 +24,8 @@

		int getOperationsCount();

		int incrementAndGetRetryCount();

		@@ -26,6 +26,10 @@

		int incrementAndGetRetryCount();

		void setRetryCount(int count);

OpenSearch Sink should make the number of retries configurable - Issue #2291 #2339

OpenSearch Sink should make the number of retries configurable - Issue #2291 #2339

Conversation

kkondaka commented Mar 1, 2023

Description

Issues Resolved

Check List

codecov-commenter commented Mar 1, 2023 • edited Loading

Codecov Report

Choose a reason for hiding this comment

kkondaka Mar 6, 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

codecov-commenter commented Mar 1, 2023 •

edited

Loading

kkondaka Mar 6, 2023 •

edited

Loading