honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-02-23 21:38:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Allow double aborts on bulk item requests

Browse Source 
In some cases a request can already be aborted and retried. This means
the condition that aborting a request should only happen when an item
has not been processed yet is too strict. This commit allows for a
double abort. If we attempt to abort an operation that was previously
processed but not aborted, we treat that as a hard failure.

Relates #26434
This commit is contained in:
Jason Tedor 2017-08-31 14:37:02 -04:00
parent 294d167973
commit 111defdfe1
1 changed files with 13 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
core/src/main/java/org/elasticsearch/action/bulk/BulkItemRequest.java
View File

@ -72,13 +72,20 @@ public class BulkItemRequest implements Streamable {
* @throws IllegalStateException If a response already exists for this request * @throws IllegalStateException If a response already exists for this request
*/ */
public void abort(String index, Exception cause) { public void abort(String index, Exception cause) {
if (primaryResponse != null) { if (primaryResponse == null) {
assert false : "Response already exists " + primaryResponse.status(); final BulkItemResponse.Failure failure = new BulkItemResponse.Failure(index, request.type(), request.id(),
throw new IllegalStateException("Item already has a response (status=" + primaryResponse.status() + ")"); Objects.requireNonNull(cause), true);
setPrimaryResponse(new BulkItemResponse(id, request.opType(), failure));
} else {
assert primaryResponse.isFailed() && primaryResponse.getFailure().isAborted()
: "response [" + primaryResponse + "]; cause [" + cause + "]";
if (primaryResponse.isFailed() && primaryResponse.getFailure().isAborted()) {
primaryResponse.getFailure().getCause().addSuppressed(cause);
} else {
throw new IllegalStateException(
"aborting item that with response [" + primaryResponse + "] that was previously processed", cause);
}
} }
final BulkItemResponse.Failure failure = new BulkItemResponse.Failure(index, request.type(), request.id(),
Objects.requireNonNull(cause), true);
setPrimaryResponse(new BulkItemResponse(id, request.opType(), failure));
} }
public static BulkItemRequest readBulkItem(StreamInput in) throws IOException { public static BulkItemRequest readBulkItem(StreamInput in) throws IOException {