Retry on createIndex failure

Author: JVerwolf

modules/reindex/src/internalClusterTest/java/org/elasticsearch/migration/FeatureMigrationIT.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.migration;
 
+import com.carrotsearch.randomizedtesting.annotations.Timeout;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.action.admin.cluster.migration.GetFeatureUpgradeStatusAction;
 import org.elasticsearch.action.admin.cluster.migration.GetFeatureUpgradeStatusRequest;
@@ -277,7 +278,6 @@ public void testMigrateIndexWithWriteBlock() throws Exception {
         });
     }
 
-    @AwaitsFix(bugUrl = "ES-10666") // This test uncovered an existing issue
     public void testIndexBlockIsRemovedWhenAliasRequestFails() throws Exception {
         createSystemIndexForDescriptor(INTERNAL_UNMANAGED);
         ensureGreen();

server/src/main/java/org/elasticsearch/upgrades/SystemIndexMigrator.java

@@ -207,49 +207,11 @@ public void run(SystemIndexMigrationTaskState taskState) {
         }
 
         // Kick off our callback "loop" - finishIndexAndLoop calls back into prepareNextIndex
-        cleanUpPreviousMigration(
-            taskState,
-            clusterState,
-            state -> prepareNextIndex(state, state2 -> migrateSingleIndex(state2, this::finishIndexAndLoop), stateFeatureName)
-        );
-    }
-
-    private void cleanUpPreviousMigration(
-        SystemIndexMigrationTaskState taskState,
-        ClusterState currentState,
-        Consumer<ClusterState> listener
-    ) {
         logger.debug("cleaning up previous migration, task state: [{}]", taskState == null ? "null" : Strings.toString(taskState));
-        if (taskState != null && taskState.getCurrentIndex() != null) {
-            SystemIndexMigrationInfo migrationInfo;
-            try {
-                migrationInfo = SystemIndexMigrationInfo.fromTaskState(
-                    taskState,
-                    systemIndices,
-                    currentState.metadata(),
-                    indexScopedSettings
-                );
-            } catch (Exception e) {
-                markAsFailed(e);
-                return;
-            }
-            final String newIndexName = migrationInfo.getNextIndexName();
-            logger.info("removing index [{}] from previous incomplete migration", newIndexName);
-
-            migrationInfo.createClient(baseClient)
-                .admin()
-                .indices()
-                .prepareDelete(newIndexName)
-                .execute(ActionListener.wrap(ackedResponse -> {
-                    if (ackedResponse.isAcknowledged()) {
-                        logger.debug("successfully removed index [{}]", newIndexName);
-                        clearResults(clusterService, ActionListener.wrap(listener::accept, this::markAsFailed));
-                    }
-                }, this::markAsFailed));
-        } else {
-            logger.debug("no incomplete index to remove");
-            clearResults(clusterService, ActionListener.wrap(listener::accept, this::markAsFailed));
-        }
+        clearResults(clusterService, ActionListener.wrap(state ->
+                prepareNextIndex(state2 ->
+                    migrateSingleIndex(state2, this::finishIndexAndLoop), stateFeatureName),
+            this::markAsFailed));
     }
 
     private void finishIndexAndLoop(BulkByScrollResponse bulkResponse) {
@@ -290,7 +252,6 @@ private void finishIndexAndLoop(BulkByScrollResponse bulkResponse) {
             );
         } else {
             prepareNextIndex(
-                clusterService.state(),
                 state2 -> migrateSingleIndex(state2, this::finishIndexAndLoop),
                 lastMigrationInfo.getFeatureName()
             );
@@ -303,7 +264,6 @@ private void recordIndexMigrationSuccess(SystemIndexMigrationInfo lastMigrationI
             SingleFeatureMigrationResult.success(),
             ActionListener.wrap(state -> {
                 prepareNextIndex(
-                    state,
                     clusterState -> migrateSingleIndex(clusterState, this::finishIndexAndLoop),
                     lastMigrationInfo.getFeatureName()
                 );
@@ -312,7 +272,7 @@ private void recordIndexMigrationSuccess(SystemIndexMigrationInfo lastMigrationI
         updateTask.submit(clusterService);
     }
 
-    private void prepareNextIndex(ClusterState clusterState, Consumer<ClusterState> listener, String lastFeatureName) {
+    private void prepareNextIndex(Consumer<ClusterState> listener, String lastFeatureName) {
         synchronized (migrationQueue) {
             assert migrationQueue != null;
             if (migrationQueue.isEmpty()) {
@@ -424,7 +384,7 @@ private void migrateSingleIndex(ClusterState clusterState, Consumer<BulkByScroll
         logger.info("migrating index [{}] from feature [{}] to new index [{}]", oldIndexName, migrationInfo.getFeatureName(), newIndexName);
         ActionListener<BulkByScrollResponse> innerListener = ActionListener.wrap(listener::accept, this::markAsFailed);
         try {
-            createIndex(migrationInfo, innerListener.delegateFailureAndWrap((delegate, shardsAcknowledgedResponse) -> {
+            createIndexRetryOnFailure(migrationInfo, innerListener.delegateFailureAndWrap((delegate, shardsAcknowledgedResponse) -> {
                 logger.debug(
                     "while migrating [{}] , got create index response: [{}]",
                     oldIndexName,
@@ -508,7 +468,26 @@ private void migrateSingleIndex(ClusterState clusterState, Consumer<BulkByScroll
         }
     }
 
+    //        ActionListener<BulkByScrollResponse> innerListener = ActionListener.wrap(listener::accept, this::markAsFailed);
+    private <T> void retryAfterFailureHandler(
+        ActionListener<T> listener,
+        Consumer<ActionListener<T>> retryableAction,
+        Consumer<Exception> failureHandler
+    ) {
+        retryableAction.accept(ActionListener.wrap(listener::onResponse, e -> {
+            logger.error("error occurred while executing retryable action", e);
+            failureHandler.accept(e);
+            listener.onFailure(e);
+        }));
+    }
+
     private void createIndex(SystemIndexMigrationInfo migrationInfo, ActionListener<ShardsAcknowledgedResponse> listener) {
+        logger.info(
+            "creating new system index [{}] from feature [{}]",
+            migrationInfo.getNextIndexName(),
+            migrationInfo.getFeatureName()
+        );
+
         final CreateIndexClusterStateUpdateRequest createRequest = new CreateIndexClusterStateUpdateRequest(
             "migrate-system-index",
             migrationInfo.getNextIndexName(),
@@ -534,6 +513,40 @@ private void createIndex(SystemIndexMigrationInfo migrationInfo, ActionListener<
         );
     }
 
+
+    private void createIndexRetryOnFailure(SystemIndexMigrationInfo migrationInfo, ActionListener<ShardsAcknowledgedResponse> listener) {
+        createIndex(migrationInfo, ActionListener.wrap(listener::onResponse, e -> {
+            logger.warn(
+                "createIndex failed, retrying after removing index [{}] from previous attempt",
+                migrationInfo.getNextIndexName()
+            );
+            deleteIndex(migrationInfo, ActionListener.wrap(
+                cleanupResponse -> createIndex(migrationInfo, listener),
+                e2 -> {
+                    logger.warn("createIndex failed after retrying, aborting", e2);
+                    listener.onFailure(e2);
+                }
+            ));
+        }));
+    }
+
+    private <T> void deleteIndex(
+        SystemIndexMigrationInfo migrationInfo,
+        ActionListener<AcknowledgedResponse> listener
+    ) {
+        logger.info("removing index [{}] from feature [{}]", migrationInfo.getNextIndexName(), migrationInfo.getFeatureName());
+        String newIndexName = migrationInfo.getNextIndexName();
+        baseClient.admin()
+            .indices()
+            .prepareDelete(newIndexName)
+            .execute(ActionListener.wrap(ackedResponse -> {
+                if (ackedResponse.isAcknowledged()) {
+                    logger.info("successfully removed index [{}]", newIndexName);
+                    listener.onResponse(ackedResponse);
+                }
+            }, listener::onFailure));
+    }
+
     private void setAliasAndRemoveOldIndex(SystemIndexMigrationInfo migrationInfo, ActionListener<IndicesAliasesResponse> listener) {
         final IndicesAliasesRequestBuilder aliasesRequest = migrationInfo.createClient(baseClient).admin().indices().prepareAliases();
         aliasesRequest.removeIndex(migrationInfo.getCurrentIndexName());
@@ -638,7 +651,7 @@ public void markAsFailed(Exception e) {
         String featureName = Optional.ofNullable(migrationInfo).map(SystemIndexMigrationInfo::getFeatureName).orElse("<unknown feature>");
         String indexName = Optional.ofNullable(migrationInfo).map(SystemIndexMigrationInfo::getCurrentIndexName).orElse("<unknown index>");
 
-        MigrationResultsUpdateTask.upsert(
+        MigrationResultsUpdateTask.upsert( // this is for the Custom Metadata, not the Persistent Task
             featureName,
             SingleFeatureMigrationResult.failure(indexName, e),
             ActionListener.wrap(state -> super.markAsFailed(e), exception -> super.markAsFailed(e))