Refresh dest before detected extracted fields

Author: dimitris-athanasiou

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/DataFrameAnalyticsManager.java

@@ -15,6 +15,9 @@
 import org.elasticsearch.action.admin.indices.get.GetIndexAction;
 import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
 import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
+import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
 import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.cluster.ClusterState;
@@ -42,6 +45,7 @@
 import java.util.function.Supplier;
 
 import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
+import static org.elasticsearch.xpack.core.ClientHelper.executeWithHeadersAsync;
 
 public class DataFrameAnalyticsManager {
 
@@ -225,9 +229,6 @@ private void reindexDataframeAndStartAnalysis(DataFrameAnalyticsTask task, DataF
     }
 
     private void startAnalytics(DataFrameAnalyticsTask task, DataFrameAnalyticsConfig config) {
-        // Ensure we mark reindexing is finished for the case we are recovering a task that had finished reindexing
-        task.setReindexingFinished();
-
         // Update state to ANALYZING and start process
         ActionListener<DataFrameDataExtractorFactory> dataExtractorFactoryListener = ActionListener.wrap(
             dataExtractorFactory -> {
@@ -247,10 +248,23 @@ private void startAnalytics(DataFrameAnalyticsTask task, DataFrameAnalyticsConfi
             error -> task.updateState(DataFrameAnalyticsState.FAILED, error.getMessage())
         );
 
-        // TODO This could fail with errors. In that case we get stuck with the copied index.
-        // We could delete the index in case of failure or we could try building the factory before reindexing
-        // to catch the error early on.
-        DataFrameDataExtractorFactory.createForDestinationIndex(client, config, dataExtractorFactoryListener);
+        ActionListener<RefreshResponse> refreshListener = ActionListener.wrap(
+            refreshResponse -> {
+                // Ensure we mark reindexing is finished for the case we are recovering a task that had finished reindexing
+                task.setReindexingFinished();
+
+                // TODO This could fail with errors. In that case we get stuck with the copied index.
+                // We could delete the index in case of failure or we could try building the factory before reindexing
+                // to catch the error early on.
+                DataFrameDataExtractorFactory.createForDestinationIndex(client, config, dataExtractorFactoryListener);
+            },
+            dataExtractorFactoryListener::onFailure
+        );
+
+        // First we need to refresh the dest index to ensure data is searchable in case the job
+        // was stopped after reindexing was complete but before the index was refreshed.
+        executeWithHeadersAsync(config.getHeaders(), ML_ORIGIN, client, RefreshAction.INSTANCE,
+            new RefreshRequest(config.getDest().getIndex()), refreshListener);
     }
 
     public void stop(DataFrameAnalyticsTask task) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/process/AnalyticsProcessManager.java

@@ -109,9 +109,6 @@ public void runJob(DataFrameAnalyticsTask task, DataFrameAnalyticsConfig config,
                 }
             }
 
-            // Refresh the dest index to ensure data is searchable
-            refreshDest(config);
-
             // Fetch existing model state (if any)
             BytesReference state = getModelState(config);