bck

Author: original-brownbear

server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.snapshots;
 
-import com.carrotsearch.randomizedtesting.annotations.Repeat;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Version;
@@ -107,6 +106,7 @@
 import org.elasticsearch.cluster.service.ClusterApplierService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.service.MasterService;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.network.NetworkModule;
 import org.elasticsearch.common.settings.ClusterSettings;
@@ -193,7 +193,6 @@
 import static org.hamcrest.Matchers.hasSize;
 import static org.mockito.Mockito.mock;
 
-@Repeat(iterations = 5000)
 public class SnapshotResiliencyTests extends ESTestCase {
 
     private DeterministicTaskQueue deterministicTaskQueue;
@@ -203,16 +202,15 @@ public class SnapshotResiliencyTests extends ESTestCase {
     private Path tempDir;
 
     /**
-     * Whether to use eventually consistent blob store in tests.
+     * Context shared by all the node's {@link Repository} instances if the eventually consistent blobstore is to be used.
+     * {@code null} if not using the eventually consistent blobstore.
      */
-    private boolean eventuallyConsistent;
+    @Nullable private MockEventuallyConsistentRepository.Context blobStoreContext;
 
-    private MockEventuallyConsistentRepository.Context blobStoreContext;
     @Before
     public void createServices() {
         tempDir = createTempDir();
-        eventuallyConsistent = randomBoolean();
-        if (eventuallyConsistent) {
+        if (randomBoolean()) {
             blobStoreContext = new MockEventuallyConsistentRepository.Context();
         }
         deterministicTaskQueue =
@@ -1086,16 +1084,9 @@ searchTransportService, new SearchPhaseController(searchService::createReduceCon
             client.initialize(actions, () -> clusterService.localNode().getId(), transportService.getRemoteClusterService());
         }
 
-        private Repository.Factory getRepoFactory(final Environment environment) {
+        private Repository.Factory getRepoFactory(Environment environment) {
             // Run half the tests with the eventually consistent repository
-            if (eventuallyConsistent) {
-                return metaData -> {
-                    final Repository repository = new MockEventuallyConsistentRepository(
-                        metaData, environment, xContentRegistry(), deterministicTaskQueue, blobStoreContext);
-                    repository.start();
-                    return repository;
-                };
-            } else {
+            if (blobStoreContext == null) {
                 return metaData -> {
                     final Repository repository = new FsRepository(metaData, environment, xContentRegistry(), threadPool) {
                         @Override
@@ -1106,6 +1097,13 @@ protected void assertSnapshotOrGenericThread() {
                     repository.start();
                     return repository;
                 };
+            } else {
+                return metaData -> {
+                    final Repository repository = new MockEventuallyConsistentRepository(
+                        metaData, environment, xContentRegistry(), deterministicTaskQueue, blobStoreContext);
+                    repository.start();
+                    return repository;
+                };
             }
         }
         public void restart() {

server/src/test/java/org/elasticsearch/snapshots/mockstore/MockEventuallyConsistentRepository.java

@@ -43,7 +43,14 @@
 import java.util.Set;
 
 /**
- * Mock Repository that simulates the mechanics of an eventually consistent blobstore.
+ * Mock Repository that simulates the eventually consistent behaviour of AWS S3 as documented in the
+ * <a href="https://docs.aws.amazon.com/AmazonS3/latest/dev/Introduction.html#ConsistencyModel">AWS S3 docs</a>.
+ * Specifically this implementation simulates:
+ * <ul>
+ *     <li>First read after write is consistent for each blob. (see S3 docs for specifics)</li>
+ *     <li>Deletes and updates to a blob can become visible with a delay.</li>
+ *     <li>Blobs can become visible to list operations with a delay.</li>
+ * </ul>
  */
 public class MockEventuallyConsistentRepository extends FsRepository {
 
@@ -68,14 +75,20 @@ protected BlobStore createBlobStore() throws Exception {
         return new MockBlobStore(super.createBlobStore());
     }
 
+    /**
+     * Context that must be shared between all instances of {@link MockEventuallyConsistentRepository} in a test run.
+     */
     public static final class Context {
 
+        /**
+         * Map of blob path to a tuple of cached non-existent blobs in them and a map of child blob name to {@link Runnable} that when
+         * executed will create the blob.
+         */
         private final Map<BlobPath, Tuple<Set<String>, Map<String, Runnable>>> state = new HashMap<>();
 
         public Tuple<Set<String>, Map<String, Runnable>> getState(BlobPath path) {
             return state.computeIfAbsent(path, p -> new Tuple<>(new HashSet<>(), new HashMap<>()));
         }
-
     }
 
     private class MockBlobStore extends BlobStoreWrapper {
@@ -125,11 +138,12 @@ private void ensureReadAfterWrite(String blobName) {
 
             @Override
             public void deleteBlob(String blobName) {
+                // TODO: simulate longer delays here once the S3 blob store implementation can handle them
                 deterministicTaskQueue.scheduleNow(() -> {
                     try {
                         super.deleteBlob(blobName);
                     } catch (DirectoryNotEmptyException | NoSuchFileException e) {
-                        // ignored
+                        // ignored since neither of these exceptions would occur on S3
                     } catch (IOException e) {
                         throw new AssertionError(e);
                     }
@@ -154,6 +168,7 @@ public void writeBlob(String blobName, InputStream inputStream, long blobSize, b
                         throw new AssertionError(e);
                     }
                 });
+                // TODO: simulate longer delays here once the S3 blob store implementation can handle them
                 deterministicTaskQueue.scheduleNow(() -> {
                     if (pendingWrites.containsKey(blobName)) {
                         pendingWrites.remove(blobName).run();