Fix Incorrect Cast in Bulk Handler Failure Handling (#69773) (#69783)

We need to cast the cause and not the original exception here.

Author: original-brownbear

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkProcessorClusterSettingsIT.java

@@ -8,13 +8,17 @@
 
 package org.elasticsearch.action.bulk;
 
+import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
 
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
 
 @ClusterScope(scope = Scope.TEST, numDataNodes = 0)
 public class BulkProcessorClusterSettingsIT extends ESIntegTestCase {
@@ -41,4 +45,12 @@ public void testBulkProcessorAutoCreateRestrictions() throws Exception {
             equalTo("[wontwork] IndexNotFoundException[no such index [wontwork] and [action.auto_create_index] is [false]]"));
         assertFalse("Operation on existing index should succeed", responses[2].isFailed());
     }
+
+    public void testIndexWithDisabledAutoCreateIndex() {
+        assertAcked(client().admin().cluster().prepareUpdateSettings().setTransientSettings(Settings.builder()
+                .put(AutoCreateIndex.AUTO_CREATE_INDEX_SETTING.getKey(), randomFrom("-*", "+.*")).build()).get());
+        final BulkItemResponse itemResponse =
+                client().prepareBulk().add(client().prepareIndex("test-index", "type1").setSource("foo", "bar")).get().getItems()[0];
+        assertThat(itemResponse.getFailure().getCause(), instanceOf(IndexNotFoundException.class));
+    }
 }

server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

@@ -260,7 +260,7 @@ protected void doRun() {
                     public void onFailure(Exception e) {
                         final Throwable cause = ExceptionsHelper.unwrapCause(e);
                         if (cause instanceof IndexNotFoundException) {
-                            indicesThatCannotBeCreated.put(index, (IndexNotFoundException) e);
+                            indicesThatCannotBeCreated.put(index, (IndexNotFoundException) cause);
                         }
                         else if ((cause instanceof ResourceAlreadyExistsException) == false) {
                             // fail all requests involving this index, if create didn't work