Rethrow Failed Assertion in BlobStoreTestUtils (elastic#53859)

This fixes two issues:

1. Currently, the future here is never resolved on assertion error so a failing test would take a full minute
to complete until the future times out.
2. S3 tests overide this method to busy assert on this method. This only works if an assertion error makes it
to the calling thread.

Closes elastic#53508

Author: original-brownbear

test/framework/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreTestUtil.java

@@ -112,27 +112,35 @@ public static boolean blobExists(BlobContainer container, String blobName) throw
      *                 of this assertion must pass an executor on those when using such an implementation.
      */
     public static void assertConsistency(BlobStoreRepository repository, Executor executor) {
-        final PlainActionFuture<Void> listener = PlainActionFuture.newFuture();
-        executor.execute(ActionRunnable.run(listener, () -> {
-            final BlobContainer blobContainer = repository.blobContainer();
-            final long latestGen;
-            try (DataInputStream inputStream = new DataInputStream(blobContainer.readBlob("index.latest"))) {
-                latestGen = inputStream.readLong();
-            } catch (NoSuchFileException e) {
-                throw new AssertionError("Could not find index.latest blob for repo [" + repository + "]");
-            }
-            assertIndexGenerations(blobContainer, latestGen);
-            final RepositoryData repositoryData;
-            try (InputStream blob = blobContainer.readBlob("index-" + latestGen);
-                 XContentParser parser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
-                     LoggingDeprecationHandler.INSTANCE, blob)) {
-                repositoryData = RepositoryData.snapshotsFromXContent(parser, latestGen);
+        final PlainActionFuture<AssertionError> listener = PlainActionFuture.newFuture();
+        executor.execute(ActionRunnable.supply(listener, () -> {
+            try {
+                final BlobContainer blobContainer = repository.blobContainer();
+                final long latestGen;
+                try (DataInputStream inputStream = new DataInputStream(blobContainer.readBlob("index.latest"))) {
+                    latestGen = inputStream.readLong();
+                } catch (NoSuchFileException e) {
+                    throw new AssertionError("Could not find index.latest blob for repo [" + repository + "]");
+                }
+                assertIndexGenerations(blobContainer, latestGen);
+                final RepositoryData repositoryData;
+                try (InputStream blob = blobContainer.readBlob("index-" + latestGen);
+                     XContentParser parser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
+                         LoggingDeprecationHandler.INSTANCE, blob)) {
+                    repositoryData = RepositoryData.snapshotsFromXContent(parser, latestGen);
+                }
+                assertIndexUUIDs(blobContainer, repositoryData);
+                assertSnapshotUUIDs(repository, repositoryData);
+                assertShardIndexGenerations(blobContainer, repositoryData.shardGenerations());
+                return null;
+            } catch (AssertionError e) {
+                return e;
             }
-            assertIndexUUIDs(blobContainer, repositoryData);
-            assertSnapshotUUIDs(repository, repositoryData);
-            assertShardIndexGenerations(blobContainer, repositoryData.shardGenerations());
         }));
-        listener.actionGet(TimeValue.timeValueMinutes(1L));
+        final AssertionError err = listener.actionGet(TimeValue.timeValueMinutes(1L));
+        if (err != null) {
+            throw new AssertionError(err);
+        }
     }
 
     private static void assertIndexGenerations(BlobContainer repoRoot, long latestGen) throws IOException {