Skip to content

Aggregate bulk operation logs for 429 errors #7227

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
graytaylor0 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Aggregate bulk operation logs for 429 errors #7227

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
22 changes: 19 additions & 3 deletions ...h/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -307,12 +307,20 @@ private BulkOperationRequestResponse handleRetriesAndFailures(final Accumulating
if (attemptNumber % 5 == 1) {
LOG.warn("Bulk Operation Failed. Number of retries {}. Retrying... ", attemptNumber - 1, exceptionFromRequest);
if (exceptionFromRequest == null) {
int throttledCount = 0;
for (final BulkResponseItem bulkItemResponse : bulkResponse.items()) {
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()) {

@srikanthpadakanti srikanthpadakanti Oct 30, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

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?

throttledCount++;
} else {
final ErrorCause error = bulkItemResponse.error();
LOG.warn("index = {} operation = {}, error = {}", bulkItemResponse.index(), bulkItemResponse.operationType(), error != null ? error.reason() : "");
}
}
}
if (throttledCount > 0) {
LOG.warn("Bulk operation throttled for {} documents", throttledCount);

@dlvenable dlvenable Oct 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

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?

}
}
}
bulkRequestNumberOfRetries.increment();
Expand All @@ -325,16 +333,24 @@ private BulkOperationRequestResponse handleRetriesAndFailures(final Accumulating

private void handleFailures(final AccumulatingBulkRequest<BulkOperationWrapper, BulkRequest> bulkRequest, final BulkResponse bulkResponse, final Throwable failure) {
if (failure == null) {
int throttledCount = 0;
for (final BulkResponseItem bulkItemResponse : bulkResponse.items()) {
if(isItemInError(bulkItemResponse)) {
// Skip logging the error for version conflicts when using external versioning
final ErrorCause error = bulkItemResponse.error();
if (isExternalVersioning && error != null && VERSION_CONFLICT_EXCEPTION_TYPE.equals(error.type())) {
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()) {

@srikanthpadakanti srikanthpadakanti Oct 30, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

@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?

throttledCount++;
} else {
LOG.warn("index = {}, operation = {}, status = {}, error = {}", bulkItemResponse.index(), bulkItemResponse.operationType(), bulkItemResponse.status(), error != null ? error.reason() : "");
}
}
}
if (throttledCount > 0) {
LOG.warn("Bulk operation throttled for {} documents", throttledCount);

@dlvenable dlvenable Oct 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Same comment as above.

}
handleFailures(bulkRequest, bulkResponse.items());
} else {
LOG.warn("Bulk Operation Failed.", failure);
Expand Down
Loading