fix loop when updating reposity and creating snapshot

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

Author: kkewwei

CHANGELOG.md

@@ -31,6 +31,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ### Fixed
 - Fix bytes parameter on `_cat/recovery` ([#17598](https://github.com/opensearch-project/OpenSearch/pull/17598))
+- 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

@@ -45,8 +45,12 @@
 import java.util.Collection;
 import java.util.Collections;
 
+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.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.hasToString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.sameInstance;
@@ -122,4 +126,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/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.closed) {
+            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);