fix loop when updating reposity and creating snapshot

Signed-off-by: kkewwei <[email protected]>
Signed-off-by: kkewwei <[email protected]>

Author: kkewwei

CHANGELOG-3.0.md

@@ -15,6 +15,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Removed
 
 ### Fixed
+- Fix simultaneously creating a snapshot and updating the repository can potentially trigger an infinite loop ([#17532](https://github.com/opensearch-project/OpenSearch/pull/17532))
 
 ### Security
 

server/src/internalClusterTest/java/org/opensearch/repositories/RepositoriesServiceIT.java

@@ -32,7 +32,9 @@
 
 package org.opensearch.repositories;
 
+import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesRequest;
 import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesResponse;
+import org.opensearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
 import org.opensearch.cluster.metadata.RepositoryMetadata;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.plugins.Plugin;
@@ -42,9 +44,14 @@
 import org.opensearch.test.OpenSearchIntegTestCase;
 import org.opensearch.transport.client.Client;
 
+import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
 
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasToString;
+import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS;
+import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_SHARDS;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
@@ -122,4 +129,66 @@ public void testSystemRepositoryCantBeCreated() {
 
         assertThrows(RepositoryException.class, () -> createRepository(repositoryName, FsRepository.TYPE, repoSettings));
     }
+
+    public void testCreatSnapAndUpdateReposityCauseInfiniteLoop() throws InterruptedException {
+        // create index
+        internalCluster();
+        String indexName = "test-index";
+        createIndex(indexName, Settings.builder().put(SETTING_NUMBER_OF_REPLICAS, 0).put(SETTING_NUMBER_OF_SHARDS, 1).build());
+        index(indexName, "_doc", "1", Collections.singletonMap("user", generateRandomStringArray(1, 10, false, false)));
+        flush(indexName);
+
+        // create repository
+        final String repositoryName = "test-repo";
+        Settings.Builder repoSettings = Settings.builder()
+            .put("location", randomRepoPath())
+            .put("max_snapshot_bytes_per_sec", "10mb")
+            .put("max_restore_bytes_per_sec", "10mb");
+        OpenSearchIntegTestCase.putRepositoryWithNoSettingOverrides(
+            client().admin().cluster(),
+            repositoryName,
+            FsRepository.TYPE,
+            true,
+            repoSettings
+        );
+
+        String snapshotName = "test-snapshot";
+        Runnable createSnapshot = () -> {
+            logger.info("--> begining snapshot");
+            client().admin()
+                .cluster()
+                .prepareCreateSnapshot(repositoryName, snapshotName)
+                .setWaitForCompletion(true)
+                .setIndices(indexName)
+                .get();
+            logger.info("--> finishing snapshot");
+        };
+
+        // snapshot mab be failed when updating repository
+        Thread thread = new Thread(() -> {
+            try {
+            createSnapshot.run();
+            } catch (Exception e) {
+                assertThat(e, instanceOf(RepositoryException.class));
+                assertThat(e, hasToString(containsString(("the repository is closed"))));
+            }
+        });
+        thread.start();
+
+        logger.info("--> begin to reset repository");
+        repoSettings = Settings.builder().put("location", randomRepoPath()).put("max_snapshot_bytes_per_sec", "300mb");
+        OpenSearchIntegTestCase.putRepositoryWithNoSettingOverrides(
+            client().admin().cluster(),
+            repositoryName,
+            FsRepository.TYPE,
+            true,
+            repoSettings
+        );
+        logger.info("--> finish to reset repository");
+
+        // after updating repository, snapshot should be success
+        createSnapshot.run();
+
+        thread.join();
+    }
 }

server/src/main/java/org/opensearch/repositories/FilterRepository.java

@@ -345,4 +345,9 @@ public void stop() {
     public void close() {
         in.close();
     }
+
+    @Override
+    public boolean isOpen() {
+        return in.isOpen();
+    }
 }

server/src/main/java/org/opensearch/repositories/Repository.java

@@ -611,4 +611,7 @@ default void reload(RepositoryMetadata repositoryMetadata) {}
      * Validate the repository metadata
      */
     default void validateMetadata(RepositoryMetadata repositoryMetadata) {}
+
+    boolean isOpen();
+
 }

server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java

@@ -551,6 +551,8 @@ protected static long calculateMaxWithinIntLimit(long defaultThresholdOfHeap, lo
      */
     protected volatile int bufferSize;
 
+    private volatile boolean closed;
+
     /**
      * Constructs new BlobStoreRepository
      * @param repositoryMetadata   The metadata for this repository including name and settings
@@ -630,6 +632,7 @@ protected void doClose() {
         }
         if (store != null) {
             try {
+                closed = true;
                 store.close();
             } catch (Exception t) {
                 logger.warn("cannot close blob store", t);
@@ -643,6 +646,10 @@ public void executeConsistentStateUpdate(
         String source,
         Consumer<Exception> onFailure
     ) {
+        if (this.isOpen() == false) {
+            onFailure.accept(new RepositoryException(metadata.name(), "the repository is closed, maybe updated or deleted, please retry"));
+            return;
+        }
         final RepositoryMetadata repositoryMetadataStart = metadata;
         getRepositoryData(ActionListener.wrap(repositoryData -> {
             final ClusterStateUpdateTask updateTask = createUpdateTask.apply(repositoryData);
@@ -4690,6 +4697,11 @@ private void checkAborted() {
         }
     }
 
+    @Override
+    public boolean isOpen() {
+        return closed == false;
+    }
+
     private static void failStoreIfCorrupted(Store store, Exception e) {
         if (Lucene.isCorruptionException(e)) {
             try {

server/src/test/java/org/opensearch/repositories/RepositoriesServiceTests.java

@@ -841,6 +841,11 @@ public void cloneRemoteStoreIndexShardSnapshot(
 
         }
 
+        @Override
+        public boolean isOpen() {
+            return isClosed == false;
+        }
+
         @Override
         public Lifecycle.State lifecycleState() {
             return null;

test/framework/src/main/java/org/opensearch/index/shard/RestoreOnlyRepository.java

@@ -73,14 +73,18 @@ public RestoreOnlyRepository(String indexName) {
         this.indexName = indexName;
     }
 
+    private volatile boolean closed;
+
     @Override
     protected void doStart() {}
 
     @Override
     protected void doStop() {}
 
     @Override
-    protected void doClose() {}
+    protected void doClose() {
+        closed = true;
+    }
 
     @Override
     public RepositoryMetadata getMetadata() {
@@ -249,4 +253,9 @@ public void cloneRemoteStoreIndexShardSnapshot(
     ) {
         throw new UnsupportedOperationException("Unsupported for restore-only repository");
     }
+
+    @Override
+    public boolean isOpen() {
+        return closed == false;
+    }
 }