[Docs] Docs tests should wait for async execution to complete (#28481)

This commit splits the async execution documentation into 2 parts, one
for the async method itself and one for the action listener. This allows
to add more doc and to use CountDownLatches in doc tests to wait for
asynchronous operations to be completed before moving to the next test.

It also renames few files.

Related to #28457

Author: tlrx

client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/CRUDDocumentationIT.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.LatchedActionListener;
 import org.elasticsearch.action.bulk.BackoffPolicy;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkProcessor;
@@ -60,11 +61,11 @@
 import org.elasticsearch.script.ScriptType;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 
-import java.io.IOException;
 import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import static java.util.Collections.emptyMap;
@@ -227,9 +228,8 @@ public void testIndex() throws Exception {
         }
         {
             IndexRequest request = new IndexRequest("posts", "doc", "async").source("field", "value");
-
-            // tag::index-execute-async
-            client.indexAsync(request, new ActionListener<IndexResponse>() {
+            // tag::index-execute-listener
+            ActionListener<IndexResponse> listener = new ActionListener<IndexResponse>() {
                 @Override
                 public void onResponse(IndexResponse indexResponse) {
                     // <1>
@@ -239,10 +239,18 @@ public void onResponse(IndexResponse indexResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            // end::index-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::index-execute-async
+            client.indexAsync(request, listener); // <1>
             // end::index-execute-async
 
-            assertBusy(() -> assertTrue(client.exists(new GetRequest("posts", "doc", "async"))));
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
     }
 
@@ -490,8 +498,8 @@ public void testUpdate() throws Exception {
         {
             UpdateRequest request = new UpdateRequest("posts", "doc", "async").doc("reason", "async update").docAsUpsert(true);
 
-            // tag::update-execute-async
-            client.updateAsync(request, new ActionListener<UpdateResponse>() {
+            // tag::update-execute-listener
+            ActionListener<UpdateResponse> listener = new ActionListener<UpdateResponse>() {
                 @Override
                 public void onResponse(UpdateResponse updateResponse) {
                     // <1>
@@ -501,10 +509,18 @@ public void onResponse(UpdateResponse updateResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            // end::update-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::update-execute-async
+            client.updateAsync(request, listener); // <1>
             // end::update-execute-async
 
-            assertBusy(() -> assertTrue(client.exists(new GetRequest("posts", "doc", "async"))));
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
     }
 
@@ -602,8 +618,8 @@ public void testDelete() throws Exception {
 
             DeleteRequest request = new DeleteRequest("posts", "doc", "async");
 
-            // tag::delete-execute-async
-            client.deleteAsync(request, new ActionListener<DeleteResponse>() {
+            // tag::delete-execute-listener
+            ActionListener<DeleteResponse> listener = new ActionListener<DeleteResponse>() {
                 @Override
                 public void onResponse(DeleteResponse deleteResponse) {
                     // <1>
@@ -613,14 +629,22 @@ public void onResponse(DeleteResponse deleteResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            // end::delete-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::delete-execute-async
+            client.deleteAsync(request, listener); // <1>
             // end::delete-execute-async
 
-            assertBusy(() -> assertFalse(client.exists(new GetRequest("posts", "doc", "async"))));
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
     }
 
-    public void testBulk() throws IOException {
+    public void testBulk() throws Exception {
         RestHighLevelClient client = highLevelClient();
         {
             // tag::bulk-request
@@ -696,8 +720,8 @@ public void testBulk() throws IOException {
             request.waitForActiveShards(ActiveShardCount.ALL); // <2>
             // end::bulk-request-active-shards
 
-            // tag::bulk-execute-async
-            client.bulkAsync(request, new ActionListener<BulkResponse>() {
+            // tag::bulk-execute-listener
+            ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
                 @Override
                 public void onResponse(BulkResponse bulkResponse) {
                     // <1>
@@ -707,12 +731,22 @@ public void onResponse(BulkResponse bulkResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            // end::bulk-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::bulk-execute-async
+            client.bulkAsync(request, listener); // <1>
             // end::bulk-execute-async
+
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
     }
 
-    public void testGet() throws IOException {
+    public void testGet() throws Exception {
         RestHighLevelClient client = highLevelClient();
         {
             String mappings = "{\n" +
@@ -839,8 +873,9 @@ public void testGet() throws IOException {
         }
         {
             GetRequest request = new GetRequest("posts", "doc", "1");
-            //tag::get-execute-async
-            client.getAsync(request, new ActionListener<GetResponse>() {
+
+            //tag::get-execute-listener
+            ActionListener<GetResponse> listener = new ActionListener<GetResponse>() {
                 @Override
                 public void onResponse(GetResponse getResponse) {
                     // <1>
@@ -850,8 +885,18 @@ public void onResponse(GetResponse getResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            //end::get-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            //tag::get-execute-async
+            client.getAsync(request, listener); // <1>
             //end::get-execute-async
+
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
         {
             //tag::get-indexnotfound
@@ -879,7 +924,7 @@ public void onFailure(Exception e) {
         }
     }
 
-    public void testBulkProcessor() throws InterruptedException, IOException {
+    public void testBulkProcessor() throws InterruptedException {
         RestHighLevelClient client = highLevelClient();
         {
             // tag::bulk-processor-init

client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/IndicesClientDocumentationIT.java

@@ -21,6 +21,7 @@
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.LatchedActionListener;
 import org.elasticsearch.action.admin.indices.alias.Alias;
 import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
 import org.elasticsearch.action.admin.indices.close.CloseIndexResponse;
@@ -33,14 +34,16 @@
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.ESRestHighLevelClientTestCase;
-import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.RestStatus;
 
 import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 /**
  * This class is used to generate the Java Indices API documentation.
@@ -118,8 +121,8 @@ public void testDeleteIndexAsync() throws Exception {
         {
             DeleteIndexRequest request = new DeleteIndexRequest("posts");
 
-            // tag::delete-index-execute-async
-            client.indices().deleteAsync(request, new ActionListener<DeleteIndexResponse>() {
+            // tag::delete-index-execute-listener
+            ActionListener<DeleteIndexResponse> listener = new ActionListener<DeleteIndexResponse>() {
                 @Override
                 public void onResponse(DeleteIndexResponse deleteIndexResponse) {
                     // <1>
@@ -129,14 +132,18 @@ public void onResponse(DeleteIndexResponse deleteIndexResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            // end::delete-index-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::delete-index-execute-async
+            client.indices().deleteAsync(request, listener); // <1>
             // end::delete-index-execute-async
 
-            assertBusy(() -> {
-                // TODO Use Indices Exist API instead once it exists
-                Response response = client.getLowLevelClient().performRequest("HEAD", "posts");
-                assertTrue(RestStatus.NOT_FOUND.getStatus() == response.getStatusLine().getStatusCode());
-            });
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
     }
 
@@ -206,8 +213,9 @@ public void testCreateIndexAsync() throws Exception {
 
         {
             CreateIndexRequest request = new CreateIndexRequest("twitter");
-            // tag::create-index-execute-async
-            client.indices().createAsync(request, new ActionListener<CreateIndexResponse>() {
+
+            // tag::create-index-execute-listener
+            ActionListener<CreateIndexResponse> listener = new ActionListener<CreateIndexResponse>() {
                 @Override
                 public void onResponse(CreateIndexResponse createIndexResponse) {
                     // <1>
@@ -217,14 +225,18 @@ public void onResponse(CreateIndexResponse createIndexResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            // end::create-index-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::create-index-execute-async
+            client.indices().createAsync(request, listener); // <1>
             // end::create-index-execute-async
 
-            assertBusy(() -> {
-                // TODO Use Indices Exist API instead once it exists
-                Response response = client.getLowLevelClient().performRequest("HEAD", "twitter");
-                assertTrue(RestStatus.OK.getStatus() == response.getStatusLine().getStatusCode());
-            });
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
     }
 
@@ -269,8 +281,8 @@ public void testOpenIndex() throws Exception {
             assertTrue(acknowledged);
             assertTrue(shardsAcked);
 
-            // tag::open-index-execute-async
-            client.indices().openAsync(request, new ActionListener<OpenIndexResponse>() {
+            // tag::open-index-execute-listener
+            ActionListener<OpenIndexResponse> listener = new ActionListener<OpenIndexResponse>() {
                 @Override
                 public void onResponse(OpenIndexResponse openIndexResponse) {
                     // <1>
@@ -280,14 +292,18 @@ public void onResponse(OpenIndexResponse openIndexResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            // end::open-index-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::open-index-execute-async
+            client.indices().openAsync(request, listener); // <1>
             // end::open-index-execute-async
 
-            assertBusy(() -> {
-                // TODO Use Indices Exist API instead once it exists
-                Response response = client.getLowLevelClient().performRequest("HEAD", "index");
-                assertTrue(RestStatus.OK.getStatus() == response.getStatusLine().getStatusCode());
-            });
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
 
         {
@@ -304,7 +320,7 @@ public void onFailure(Exception e) {
         }
     }
 
-    public void testCloseIndex() throws IOException {
+    public void testCloseIndex() throws Exception {
         RestHighLevelClient client = highLevelClient();
 
         {
@@ -339,8 +355,8 @@ public void testCloseIndex() throws IOException {
             // end::close-index-response
             assertTrue(acknowledged);
 
-            // tag::close-index-execute-async
-            client.indices().closeAsync(request, new ActionListener<CloseIndexResponse>() {
+            // tag::close-index-execute-listener
+            ActionListener<CloseIndexResponse> listener = new ActionListener<CloseIndexResponse>() {
                 @Override
                 public void onResponse(CloseIndexResponse closeIndexResponse) {
                     // <1>
@@ -350,8 +366,18 @@ public void onResponse(CloseIndexResponse closeIndexResponse) {
                 public void onFailure(Exception e) {
                     // <2>
                 }
-            });
+            };
+            // end::close-index-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::close-index-execute-async
+            client.indices().closeAsync(request, listener); // <1>
             // end::close-index-execute-async
+
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
 
         {