simplifying

Author: masseyke

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamPersistentTaskExecutor.java

@@ -44,6 +44,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.function.BiConsumer;
 
 import static org.elasticsearch.xpack.core.deprecation.DeprecatedIndexPredicate.getReindexRequiredPredicate;
 
@@ -217,41 +218,58 @@ private void maybeProcessNextIndex(
         reindexDataStreamIndexRequest.setParentTask(parentTaskId);
 
         SubscribableListener.<ReindexDataStreamIndexAction.Response>newForked(
-            l -> client.execute(ReindexDataStreamIndexAction.INSTANCE, reindexDataStreamIndexRequest, l)
+            l -> client.execute(
+                ReindexDataStreamIndexAction.INSTANCE,
+                reindexDataStreamIndexRequest,
+                l.delegateResponse(new BiConsumer<ActionListener<ReindexDataStreamIndexAction.Response>, Exception>() {
+                    @Override
+                    public void accept(ActionListener<ReindexDataStreamIndexAction.Response> responseActionListener, Exception e) {
+                        IndexMetadata sourceIndex = clusterService.state().getMetadata().index(index.getName());
+                        if (sourceIndex == null) {
+                            /*
+                             * One possible cause of exception here is that the source index no longer exists. This can happen if ILM
+                             * performs certain actions while reindex is happening, or if a user manually deletes it. In this case we don't
+                             * want to fail the task. We treat it as a success and move on. The updateDataStreamOrCleanup method will make
+                             * an attempt at deleting the destination index if it has been created.
+                             */
+                            logger.debug(
+                                "The source index {} in the data stream {} was removed during processing",
+                                index.getName(),
+                                sourceDataStream
+                            );
+                            // We need the destination index name in the response so that updateDataStreamOrCleanup can delete it
+                            l.onResponse(
+                                new ReindexDataStreamIndexAction.Response(
+                                    ReindexDataStreamIndexTransportAction.generateDestIndexName(index.getName())
+                                )
+                            );
+                        } else {
+                            l.onFailure(e);
+                        }
+                    }
+                })
+            )
         )
             .<AcknowledgedResponse>andThen(
-                (l, result) -> updateDataStream(sourceDataStream, index.getName(), result.getDestIndex(), l, parentTaskId)
+                (l, result) -> updateDataStreamOrCleanup(sourceDataStream, index.getName(), result.getDestIndex(), l, parentTaskId)
             )
             .<AcknowledgedResponse>andThen(l -> deleteIndex(index.getName(), parentTaskId, l))
             .addListener(ActionListener.wrap(unused -> {
                 reindexDataStreamTask.reindexSucceeded(index.getName());
                 listener.onResponse(null);
                 maybeProcessNextIndex(indicesRemaining, reindexDataStreamTask, sourceDataStream, listener, parentTaskId);
             }, e -> {
-                /*
-                 * One possible cause of exception here is that the source index no longer exists. This can happen if ILM performs certain
-                 * actions while reindex is happening, or if a user manually deletes it. In this case we don't want to fail the task. We
-                 * treat it as a success and move on after making an attempt at deleting the destination index if it has been created yet.
-                 */
-                IndexMetadata sourceIndex = clusterService.state().getMetadata().index(index.getName());
-                if (sourceIndex == null) {
-                    String destIndexName = ReindexDataStreamIndexTransportAction.generateDestIndexName(index.getName());
-                    logMissingSourceIndex(sourceDataStream, index.getName(), destIndexName);
-                    deleteIndex(destIndexName, parentTaskId, ActionListener.wrap(acknowledgedResponse -> {
-                        reindexDataStreamTask.reindexSucceeded(index.getName());
-                        listener.onResponse(null);
-                        maybeProcessNextIndex(indicesRemaining, reindexDataStreamTask, sourceDataStream, listener, parentTaskId);
-                    }, listener::onFailure));
-                } else {
-                    // The source index still exists, so this is a real problem we want to record
-                    reindexDataStreamTask.reindexFailed(index.getName(), e);
-                    listener.onResponse(null);
-                    maybeProcessNextIndex(indicesRemaining, reindexDataStreamTask, sourceDataStream, listener, parentTaskId);
-                }
+                reindexDataStreamTask.reindexFailed(index.getName(), e);
+                listener.onResponse(null);
+                maybeProcessNextIndex(indicesRemaining, reindexDataStreamTask, sourceDataStream, listener, parentTaskId);
             }));
     }
 
-    private void updateDataStream(
+    /*
+     * If the oldIndex still exists, this method swaps newIndex into the dataStream and swaps out oldIndex. If oldIndex no longer exists,
+     * this method cleans up by deleting newIndex (if it exists).
+     */
+    private void updateDataStreamOrCleanup(
         String dataStream,
         String oldIndex,
         String newIndex,
@@ -260,7 +278,12 @@ private void updateDataStream(
     ) {
         IndexMetadata sourceIndex = clusterService.state().getMetadata().index(oldIndex);
         if (sourceIndex == null) {
-            logMissingSourceIndex(dataStream, oldIndex, newIndex);
+            logger.debug(
+                "Index {} in data stream {} no longer exists after reindexing completed. Deleting reindexed index {}",
+                dataStream,
+                oldIndex,
+                newIndex
+            );
             deleteIndex(newIndex, parentTaskId, listener);
         } else {
             ModifyDataStreamsAction.Request modifyDataStreamRequest = new ModifyDataStreamsAction.Request(
@@ -273,15 +296,6 @@ private void updateDataStream(
         }
     }
 
-    private void logMissingSourceIndex(String dataStreamName, String sourceIndexName, String destinationIndexName) {
-        logger.debug(
-            "Index {} in data stream {} no longer exists after reindexing completed. Deleting reindexed index {}",
-            sourceIndexName,
-            dataStreamName,
-            destinationIndexName
-        );
-    }
-
     private void deleteIndex(String indexName, TaskId parentTaskId, ActionListener<AcknowledgedResponse> listener) {
         DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName).indicesOptions(IGNORE_MISSING_OPTIONS);
         deleteIndexRequest.setParentTask(parentTaskId);