Wait for concurrent reads in ESIndexInputTestCase (#51425)

Today `ESIndexInputTestCase#randomReadAndSlice` performs some reads on
background threads, but does not necessarily wait for these reads to finish.
It's possible that the underlying `IndexInput` may be closed before these reads
complete, causing them to fail.

Author: DaveCTurner

test/framework/src/main/java/org/elasticsearch/common/lucene/store/ESIndexInputTestCase.java

@@ -109,7 +109,8 @@ protected byte[] randomReadAndSlice(IndexInput indexInput, int length) throws IO
                 case 5:
                     // Read clone or slice concurrently
                     final int cloneCount = between(1, 3);
-                    final CountDownLatch countDownLatch = new CountDownLatch(1 + cloneCount);
+                    final CountDownLatch startLatch = new CountDownLatch(1 + cloneCount);
+                    final CountDownLatch finishLatch = new CountDownLatch(cloneCount);
 
                     final PlainActionFuture<byte[]> mainThreadResultFuture = new PlainActionFuture<>();
                     final int mainThreadReadStart = readPos;
@@ -133,8 +134,8 @@ protected void doRun() throws Exception {
                                     final int sliceEnd = between(readEnd, length);
                                     clone = indexInput.slice("concurrent slice (0, " + sliceEnd + ") of " + indexInput, 0L, sliceEnd);
                                 }
-                                countDownLatch.countDown();
-                                countDownLatch.await();
+                                startLatch.countDown();
+                                startLatch.await();
                                 clone.seek(readStart);
                                 final byte[] cloneResult = randomReadAndSlice(clone, readEnd);
                                 if (randomBoolean()) {
@@ -155,20 +156,26 @@ protected void doRun() throws Exception {
                                 }
                             }
 
+                            @Override
+                            public void onAfter() {
+                                finishLatch.countDown();
+                            }
+
                             @Override
                             public void onRejection(Exception e) {
                                 // all threads are busy, and queueing can lead this test to deadlock, so we need take no action
-                                countDownLatch.countDown();
+                                startLatch.countDown();
                             }
                         });
                     }
 
                     try {
-                        countDownLatch.countDown();
-                        countDownLatch.await();
+                        startLatch.countDown();
+                        startLatch.await();
                         ActionListener.completeWith(mainThreadResultFuture, () -> randomReadAndSlice(indexInput, mainThreadReadEnd));
                         System.arraycopy(mainThreadResultFuture.actionGet(), readPos, output, readPos, mainThreadReadEnd - readPos);
                         readPos = mainThreadReadEnd;
+                        finishLatch.await();
                     } catch (InterruptedException e) {
                         throw new AssertionError(e);
                     }