From 72735be673b5ca0bff72944ccc2a61551cdf6af8 Mon Sep 17 00:00:00 2001
From: Yannick Welsch <yannick@welsch.lu>
Date: Thu, 6 Jun 2019 14:04:45 +0200
Subject: [PATCH] Fix NPE when rejecting bulk updates (#42923)

Single updates use a different internal code path than updates that are wrapped in a bulk request.
While working on a refactoring to bring both closer together I've noticed that bulk updates were
failing some of the tests that single updates passed. In particular, bulk updates cause
NullPointerExceptions to be thrown and listeners not being properly notified when being rejected
from the thread pool.
---
 .../action/bulk/TransportShardBulkAction.java      |  3 +--
 .../java/org/elasticsearch/update/UpdateIT.java    | 14 +++++++++++++-
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java b/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
index 372923c9036..ad803589f9d 100644
--- a/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
+++ b/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
@@ -342,11 +342,10 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
                                                   BulkItemResponse operationResponse,
                                                   final UpdateHelper.Result translate) {
         final BulkItemResponse response;
-        DocWriteResponse.Result translatedResult = translate.getResponseResult();
         if (operationResponse.isFailed()) {
             response = new BulkItemResponse(operationResponse.getItemId(), DocWriteRequest.OpType.UPDATE, operationResponse.getFailure());
         } else {
-
+            final DocWriteResponse.Result translatedResult = translate.getResponseResult();
             final UpdateResponse updateResponse;
             if (translatedResult == DocWriteResponse.Result.CREATED || translatedResult == DocWriteResponse.Result.UPDATED) {
                 final IndexRequest updateIndexRequest = translate.action();
diff --git a/server/src/test/java/org/elasticsearch/update/UpdateIT.java b/server/src/test/java/org/elasticsearch/update/UpdateIT.java
index 7652c503450..1a0df1c4a10 100644
--- a/server/src/test/java/org/elasticsearch/update/UpdateIT.java
+++ b/server/src/test/java/org/elasticsearch/update/UpdateIT.java
@@ -24,6 +24,7 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.admin.indices.alias.Alias;
+import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.delete.DeleteResponse;
 import org.elasticsearch.action.get.GetResponse;
@@ -700,7 +701,18 @@ public class UpdateIT extends ESIntegTestCase {
                                         .setRetryOnConflict(retryOnConflict)
                                         .setUpsert(jsonBuilder().startObject().field("field", 1).endObject())
                                         .request();
-                                client().update(ur, new UpdateListener(j));
+                                if (randomBoolean()) {
+                                    client().update(ur, new UpdateListener(j));
+                                } else {
+                                    client().prepareBulk().add(ur).execute(ActionListener.map(new UpdateListener(j), br -> {
+                                        final BulkItemResponse ir = br.getItems()[0];
+                                        if (ir.isFailed()) {
+                                            throw ir.getFailure().getCause();
+                                        } else {
+                                            return ir.getResponse();
+                                        }
+                                    }));
+                                }
                             } catch (NoNodeAvailableException nne) {
                                 updateRequestsOutstanding.release();
                                 synchronized (failedMap) {