Fix BulkWithUpdatesIT and CloseIndexIT

As of today the Close Index API does its best to close indices,
but closing an index with ongoing recoveries might or might not
be acknowledged depending of the values of the max seq number
and global checkpoint at the time the
TransportVerifyShardBeforeClose action is executed.

These tests failed because they always expect that the index is
correctly closed on the first try, which is not always the case.
Instead we need to retry the closing until it succeed.

Closes #37571

server/src/test/java/org/elasticsearch/action/bulk/BulkWithUpdatesIT.java

@@ -34,6 +34,7 @@ import org.elasticsearch.client.Requests;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.VersionType;
+import org.elasticsearch.indices.IndexClosedException;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.script.MockScriptPlugin;
 import org.elasticsearch.script.Script;
@@ -57,6 +58,7 @@ import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitC
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 
@@ -569,7 +571,7 @@ public class BulkWithUpdatesIT extends ESIntegTestCase {
         SearchResponse searchResponse = client().prepareSearch("bulkindex*").get();
         assertHitCount(searchResponse, 3);
 
-        assertAcked(client().admin().indices().prepareClose("bulkindex2"));
+        assertBusy(() -> assertAcked(client().admin().indices().prepareClose("bulkindex2")));
 
         BulkResponse bulkResponse = client().bulk(bulkRequest).get();
         assertThat(bulkResponse.hasFailures(), is(true));
@@ -581,7 +583,7 @@ public class BulkWithUpdatesIT extends ESIntegTestCase {
         createIndex("bulkindex1");
 
         client().prepareIndex("bulkindex1", "index1_type", "1").setSource("text", "test").get();
-        assertAcked(client().admin().indices().prepareClose("bulkindex1"));
+        assertBusy(() -> assertAcked(client().admin().indices().prepareClose("bulkindex1")));
 
         BulkRequest bulkRequest = new BulkRequest().setRefreshPolicy(RefreshPolicy.IMMEDIATE);
         bulkRequest.add(new IndexRequest("bulkindex1", "index1_type", "1").source(Requests.INDEX_CONTENT_TYPE, "text", "hallo1"))
@@ -593,8 +595,11 @@ public class BulkWithUpdatesIT extends ESIntegTestCase {
         BulkItemResponse[] responseItems = bulkResponse.getItems();
         assertThat(responseItems.length, is(3));
         assertThat(responseItems[0].getOpType(), is(OpType.INDEX));
+        assertThat(responseItems[0].getFailure().getCause(), instanceOf(IndexClosedException.class));
         assertThat(responseItems[1].getOpType(), is(OpType.UPDATE));
+        assertThat(responseItems[1].getFailure().getCause(), instanceOf(IndexClosedException.class));
         assertThat(responseItems[2].getOpType(), is(OpType.DELETE));
+        assertThat(responseItems[2].getFailure().getCause(), instanceOf(IndexClosedException.class));
     }
 
     // issue 9821

server/src/test/java/org/elasticsearch/indices/state/CloseIndexIT.java

@@ -21,7 +21,6 @@ package org.elasticsearch.indices.state;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.ActiveShardCount;
-import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
@@ -41,6 +40,7 @@ import java.util.stream.IntStream;
 
 import static java.util.Collections.emptySet;
 import static java.util.stream.Collectors.toList;
+import static org.elasticsearch.action.support.IndicesOptions.lenientExpandOpen;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.hamcrest.Matchers.containsString;
@@ -64,9 +64,9 @@ public class CloseIndexIT extends ESIntegTestCase {
         assertThat(e.getMessage(), is("no such index [test2]"));
     }
 
-    public void testCloseOneMissingIndexIgnoreMissing() {
+    public void testCloseOneMissingIndexIgnoreMissing() throws Exception {
         createIndex("test1");
-        assertAcked(client().admin().indices().prepareClose("test1", "test2").setIndicesOptions(IndicesOptions.lenientExpandOpen()));
+        assertBusy(() -> assertAcked(client().admin().indices().prepareClose("test1", "test2").setIndicesOptions(lenientExpandOpen())));
         assertIndexIsClosed("test1");
     }
 
@@ -90,7 +90,7 @@ public class CloseIndexIT extends ESIntegTestCase {
         indexRandom(randomBoolean(), false, randomBoolean(), IntStream.range(0, nbDocs)
             .mapToObj(i -> client().prepareIndex(indexName, "_doc", String.valueOf(i)).setSource("num", i)).collect(toList()));
 
-        assertAcked(client().admin().indices().prepareClose(indexName));
+        assertBusy(() -> assertAcked(client().admin().indices().prepareClose(indexName)));
         assertIndexIsClosed(indexName);
 
         assertAcked(client().admin().indices().prepareOpen(indexName));
@@ -106,15 +106,15 @@ public class CloseIndexIT extends ESIntegTestCase {
                 .mapToObj(i -> client().prepareIndex(indexName, "_doc", String.valueOf(i)).setSource("num", i)).collect(toList()));
         }
         // First close should be acked
-        assertAcked(client().admin().indices().prepareClose(indexName));
+        assertBusy(() -> assertAcked(client().admin().indices().prepareClose(indexName)));
         assertIndexIsClosed(indexName);
 
         // Second close should be acked too
-        assertAcked(client().admin().indices().prepareClose(indexName));
+        assertBusy(() -> assertAcked(client().admin().indices().prepareClose(indexName)));
         assertIndexIsClosed(indexName);
     }
 
-    public void testCloseUnassignedIndex() {
+    public void testCloseUnassignedIndex() throws Exception {
         final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
         assertAcked(prepareCreate(indexName)
             .setWaitForActiveShards(ActiveShardCount.NONE)
@@ -124,7 +124,7 @@ public class CloseIndexIT extends ESIntegTestCase {
         assertThat(clusterState.metaData().indices().get(indexName).getState(), is(IndexMetaData.State.OPEN));
         assertThat(clusterState.routingTable().allShards().stream().allMatch(ShardRouting::unassigned), is(true));
 
-        assertAcked(client().admin().indices().prepareClose(indexName));
+        assertBusy(() -> assertAcked(client().admin().indices().prepareClose(indexName)));
         assertIndexIsClosed(indexName);
     }
 
@@ -172,7 +172,7 @@ public class CloseIndexIT extends ESIntegTestCase {
             indexer.setAssertNoFailuresOnStop(false);
 
             waitForDocs(randomIntBetween(10, 50), indexer);
-            assertAcked(client().admin().indices().prepareClose(indexName));
+            assertBusy(() -> assertAcked(client().admin().indices().prepareClose(indexName)));
             indexer.stop();
             nbDocs += indexer.totalIndexedDocs();