Repository navigation
Aggregate bulk operation logs for 429 errors - #7227
graytaylor0 wants to merge 1 commit into
Conversation
| static final long INITIAL_DELAY_MS = 50; | ||
| static final long MAXIMUM_DELAY_MS = Duration.ofMinutes(10).toMillis(); | ||
| static final String VERSION_CONFLICT_EXCEPTION_TYPE = "version_conflict_engine_exception"; | ||
| private static final int TOO_MANY_REQUESTS_STATUS = RestStatus.TOO_MANY_REQUESTS.getStatus(); |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
PR Reviewer Guide 🔍(Review updated until commit 88e4d01)Here are some key observations to aid the review process:
|
Signed-off-by: Taylor Gray <tylgry@amazon.com>
101dcdc to
88e4d01
Compare
|
Persistent review updated to latest commit 88e4d01 |
| } | ||
| } | ||
| if (throttledCount > 0) { | ||
| LOG.warn("Bulk operation throttled for {} documents", throttledCount); |
There was a problem hiding this comment.
This should match the log line above where this used to go.
It should include the index, operation, and the error reason. On the reason, might OpenSearch give some additional details for the throttling that can be useful?
| } | ||
| } | ||
| if (throttledCount > 0) { | ||
| LOG.warn("Bulk operation throttled for {} documents", throttledCount); |
| if(isItemInError(bulkItemResponse)) { | ||
| final ErrorCause error = bulkItemResponse.error(); | ||
| LOG.warn("index = {} operation = {}, error = {}", bulkItemResponse.index(), bulkItemResponse.operationType(), error != null ? error.reason() : ""); | ||
| if (bulkItemResponse.status() == RestStatus.TOO_MANY_REQUESTS.getStatus()) { |
There was a problem hiding this comment.
The checklist says tests were added, but BulkRetryStrategyTests is unchanged. Could you add a test with several 429 items and one other error, checking that the 429s give one summary line and the other error is still logged on its own?
| continue; | ||
| } | ||
| LOG.warn("index = {}, operation = {}, status = {}, error = {}", bulkItemResponse.index(), bulkItemResponse.operationType(), bulkItemResponse.status(), error != null ? error.reason() : ""); | ||
| if (bulkItemResponse.status() == RestStatus.TOO_MANY_REQUESTS.getStatus()) { |
There was a problem hiding this comment.
@graytaylor0, I think this branch can never run. handleFailures with a response is only called when canRetry(bulkResponse) is false, and any 429 item makes canRetry true, because 429 is not in NON_RETRY_STATUS. When retries run out, execute calls handleFailures with a failure instead, which skips this loop. I checked with a probe over 4,660 runs of mixed item statuses: this method ran 648 times and never saw a 429. Could you remove this hunk and keep only the change in the retry log?
Description
When throttled in the opensearch sink, the pipeline iterates over every bulk operation and logs the same throttled error over and over which is unnecessary. This change aggregates the throttling errors per bulk request into a single log.
Issues Resolved
Resolves #[Issue number to be closed when this PR is merged]
Check List
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.