Refresh source index before reindexing data stream index (#120752)

Add step to the ReindexDatastreamIndexAction which refreshes the source index after setting it to read-only but before calling reindex. Without doing a refresh it is possible for docs from the source index to be missing from the destination index. This happens because the docs arrived before the source index is set to read-only, but because the index hasn't refreshed, the reindex action cannot see these updates.

Author: parkertimmins

docs/changelog/120752.yaml

@@ -0,0 +1,6 @@
+pr: 120752
+summary: Refresh source index before reindexing data stream index
+area: Data streams
+type: bug
+issues:
+ - 120314

muted-tests.yml

@@ -205,9 +205,6 @@ tests:
 - class: org.elasticsearch.oldrepos.OldRepositoryAccessIT
   method: testOldSourceOnlyRepoAccess
   issue: https://github.com/elastic/elasticsearch/issues/120080
-- class: org.elasticsearch.xpack.migrate.action.ReindexDatastreamIndexTransportActionIT
-  method: testTsdbStartEndSet
-  issue: https://github.com/elastic/elasticsearch/issues/120314
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=snapshot/10_basic/Failed to snapshot indices with synthetic source}
   issue: https://github.com/elastic/elasticsearch/issues/120332

x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/xpack/migrate/action/ReindexDatastreamIndexTransportActionIT.java

@@ -87,6 +87,7 @@ public void testDestIndexDeletedIfExists() throws Exception {
         var destIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
         indicesAdmin().create(new CreateIndexRequest(destIndex)).actionGet();
         indexDocs(destIndex, 10);
+        indicesAdmin().refresh(new RefreshRequest(destIndex)).actionGet();
         assertHitCount(prepareSearch(destIndex).setSize(0), 10);
 
         // call reindex
@@ -195,19 +196,7 @@ public void testMappingsAddedToDestIndex() throws Exception {
         assumeTrue("requires the migration reindex feature flag", REINDEX_DATA_STREAM_FEATURE_FLAG.isEnabled());
 
         var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
-        String mapping = """
-            {
-              "_doc":{
-                "dynamic":"strict",
-                "properties":{
-                  "foo1":{
-                    "type":"text"
-                  }
-                }
-              }
-            }
-            """;
-        indicesAdmin().create(new CreateIndexRequest(sourceIndex).mapping(mapping)).actionGet();
+        indicesAdmin().create(new CreateIndexRequest(sourceIndex).mapping(MAPPING)).actionGet();
 
         // call reindex
         var destIndex = client().execute(ReindexDataStreamIndexAction.INSTANCE, new ReindexDataStreamIndexAction.Request(sourceIndex))
@@ -337,6 +326,12 @@ public void testSettingsAndMappingsFromTemplate() throws IOException {
         var sourceIndex = "logs-" + randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
         indicesAdmin().create(new CreateIndexRequest(sourceIndex)).actionGet();
 
+        {
+            var indexRequest = new IndexRequest(sourceIndex);
+            indexRequest.source("{ \"foo1\": \"cheese\" }", XContentType.JSON);
+            client().index(indexRequest).actionGet();
+        }
+
         // call reindex
         var destIndex = client().execute(ReindexDataStreamIndexAction.INSTANCE, new ReindexDataStreamIndexAction.Request(sourceIndex))
             .actionGet()
@@ -359,6 +354,9 @@ public void testSettingsAndMappingsFromTemplate() throws IOException {
             // sanity check specific value from dest mapping
             assertEquals("text", XContentMapValues.extractValue("properties.foo1.type", destMappings));
         }
+
+        // verify doc was successfully added
+        assertHitCount(prepareSearch(destIndex).setSize(0), 1);
     }
 
     private static final String TSDB_MAPPING = """
@@ -455,12 +453,10 @@ public void testTsdbStartEndSet() throws Exception {
 
         assertEquals(startTime, destStart);
         assertEquals(endTime, destEnd);
-    }
 
-    // TODO more logsdb/tsdb specific tests
-    // TODO more data stream specific tests (how are data streams indices are different from regular indices?)
-    // TODO check other IndexMetadata fields that need to be fixed after the fact
-    // TODO what happens if don't have necessary perms for a given index?
+        // verify doc was successfully added
+        assertHitCount(prepareSearch(destIndex).setSize(0), 1);
+    }
 
     private static void cleanupMetadataBlocks(String index) {
         var settings = Settings.builder()
@@ -483,7 +479,6 @@ private static void indexDocs(String index, int numDocs) {
         }
         BulkResponse bulkResponse = client().bulk(bulkRequest).actionGet();
         assertThat(bulkResponse.getItems().length, equalTo(numDocs));
-        indicesAdmin().refresh(new RefreshRequest(index)).actionGet();
     }
 
     private static String formatInstant(Instant instant) {

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportAction.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.SubscribableListener;
+import org.elasticsearch.action.support.broadcast.BroadcastResponse;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -140,6 +141,7 @@ protected void doExecute(
         }
 
         SubscribableListener.<AcknowledgedResponse>newForked(l -> setBlockWrites(sourceIndexName, l, taskId))
+            .<BroadcastResponse>andThen(l -> refresh(sourceIndexName, l, taskId))
             .<AcknowledgedResponse>andThen(l -> deleteDestIfExists(destIndexName, l, taskId))
             .<AcknowledgedResponse>andThen(l -> createIndex(sourceIndex, destIndexName, l, taskId))
             .<BulkByScrollResponse>andThen(l -> reindex(sourceIndexName, destIndexName, l, taskId))
@@ -175,6 +177,13 @@ public void onFailure(Exception e) {
         }, parentTaskId);
     }
 
+    private void refresh(String sourceIndexName, ActionListener<BroadcastResponse> listener, TaskId parentTaskId) {
+        logger.debug("Refreshing source index [{}]", sourceIndexName);
+        var refreshRequest = new RefreshRequest(sourceIndexName);
+        refreshRequest.setParentTask(parentTaskId);
+        client.execute(RefreshAction.INSTANCE, refreshRequest, listener);
+    }
+
     private void deleteDestIfExists(String destIndexName, ActionListener<AcknowledgedResponse> listener, TaskId parentTaskId) {
         logger.debug("Attempting to delete index [{}]", destIndexName);
         var deleteIndexRequest = new DeleteIndexRequest(destIndexName).indicesOptions(IGNORE_MISSING_OPTIONS)