From 201dc87aae5dcc4bbaf092c0033424b8bec15b37 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Christoph=20B=C3=BCscher?= <christoph@elastic.co>
Date: Mon, 6 Jul 2015 20:07:26 +0200
Subject: [PATCH] Fix: Use correct OpType on Failure in BulkItemResponse

When a bulk request fails on a Delete or Update request, the BulkItemResponse
reports incorrect "index" operation in the response. This PR fixes this
for the case of closed indices as reported in #9821 but also for
other failures and adds tests for the two cases covered.

Closes #9821
---
 .../action/bulk/BulkItemResponse.java         |  2 --
 .../action/bulk/TransportBulkAction.java      | 14 ++++++--
 .../org/elasticsearch/document/BulkTests.java | 34 +++++++++++++++++++
 3 files changed, 45 insertions(+), 5 deletions(-)

diff --git a/core/src/main/java/org/elasticsearch/action/bulk/BulkItemResponse.java b/core/src/main/java/org/elasticsearch/action/bulk/BulkItemResponse.java
index 3b9d2e03ba0..80e86eaaf17 100644
--- a/core/src/main/java/org/elasticsearch/action/bulk/BulkItemResponse.java
+++ b/core/src/main/java/org/elasticsearch/action/bulk/BulkItemResponse.java
@@ -19,7 +19,6 @@
 
 package org.elasticsearch.action.bulk;
 
-import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionWriteResponse;
 import org.elasticsearch.action.delete.DeleteResponse;
@@ -28,7 +27,6 @@ import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Streamable;
-import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.RestStatus;
 
 import java.io.IOException;
diff --git a/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java b/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
index e97af0f3da3..783db0c33db 100644
--- a/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
+++ b/core/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
@@ -168,13 +168,13 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
         } else if (request instanceof DeleteRequest) {
             DeleteRequest deleteRequest = (DeleteRequest) request;
             if (index.equals(deleteRequest.index())) {
-                responses.set(idx, new BulkItemResponse(idx, "index", new BulkItemResponse.Failure(deleteRequest.index(), deleteRequest.type(), deleteRequest.id(), e)));
+                responses.set(idx, new BulkItemResponse(idx, "delete", new BulkItemResponse.Failure(deleteRequest.index(), deleteRequest.type(), deleteRequest.id(), e)));
                 return true;
             }
         } else if (request instanceof UpdateRequest) {
             UpdateRequest updateRequest = (UpdateRequest) request;
             if (index.equals(updateRequest.index())) {
-                responses.set(idx, new BulkItemResponse(idx, "index", new BulkItemResponse.Failure(updateRequest.index(), updateRequest.type(), updateRequest.id(), e)));
+                responses.set(idx, new BulkItemResponse(idx, "update", new BulkItemResponse.Failure(updateRequest.index(), updateRequest.type(), updateRequest.id(), e)));
                 return true;
             }
         } else {
@@ -379,7 +379,15 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
         if (unavailableException != null) {
             BulkItemResponse.Failure failure = new BulkItemResponse.Failure(request.index(), request.type(), request.id(),
                     unavailableException);
-            BulkItemResponse bulkItemResponse = new BulkItemResponse(idx, "index", failure);
+            String operationType = "unknown";
+            if (request instanceof IndexRequest) {
+                operationType = "index";
+            } else if (request instanceof DeleteRequest) {
+                operationType = "delete";
+            } else if (request instanceof UpdateRequest) {
+                operationType = "update";
+            }
+            BulkItemResponse bulkItemResponse = new BulkItemResponse(idx, operationType, failure);
             responses.set(idx, bulkItemResponse);
             // make sure the request gets never processed again
             bulkRequest.requests.set(idx, null);
diff --git a/core/src/test/java/org/elasticsearch/document/BulkTests.java b/core/src/test/java/org/elasticsearch/document/BulkTests.java
index 8be32de6feb..e4d3af7cdcc 100644
--- a/core/src/test/java/org/elasticsearch/document/BulkTests.java
+++ b/core/src/test/java/org/elasticsearch/document/BulkTests.java
@@ -725,5 +725,39 @@ public class BulkTests extends ElasticsearchIntegrationTest {
         assertThat(bulkResponse.hasFailures(), is(true));
         assertThat(bulkResponse.getItems().length, is(5));
     }
+
+    @Test // issue 9821
+    public void testFailedRequestsOnClosedIndex() throws Exception {
+        createIndex("bulkindex1");
+        ensureYellow();
+
+        client().prepareIndex("bulkindex1", "index1_type", "1").setSource("text", "test").get();
+        assertAcked(client().admin().indices().prepareClose("bulkindex1"));
+
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(new IndexRequest("bulkindex1", "index1_type", "1").source("text", "hallo1"))
+                .add(new UpdateRequest("bulkindex1", "index1_type", "1").doc("foo", "bar"))
+                .add(new DeleteRequest("bulkindex1", "index1_type", "1")).refresh(true);
+
+        BulkResponse bulkResponse = client().bulk(bulkRequest).get();
+        assertThat(bulkResponse.hasFailures(), is(true));
+        BulkItemResponse[] responseItems = bulkResponse.getItems();
+        assertThat(responseItems.length, is(3));
+        assertThat(responseItems[0].getOpType(), is("index"));
+        assertThat(responseItems[1].getOpType(), is("update"));
+        assertThat(responseItems[2].getOpType(), is("delete"));
+    }
+
+    @Test // issue 9821
+    public void testInvalidIndexNamesCorrectOpType() {
+        BulkResponse bulkResponse = client().prepareBulk()
+                .add(client().prepareIndex().setIndex("INVALID.NAME").setType("type1").setId("1").setSource("field", 1))
+                .add(client().prepareUpdate().setIndex("INVALID.NAME").setType("type1").setId("1").setDoc("field", randomInt()))
+                .add(client().prepareDelete().setIndex("INVALID.NAME").setType("type1").setId("1")).get();
+        assertThat(bulkResponse.getItems().length, is(3));
+        assertThat(bulkResponse.getItems()[0].getOpType(), is("index"));
+        assertThat(bulkResponse.getItems()[1].getOpType(), is("update"));
+        assertThat(bulkResponse.getItems()[2].getOpType(), is("delete"));
+    }
 }