Make sure doc written to all shards before adding mapping

Author: parkertimmins

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

@@ -49,6 +49,7 @@
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.migrate.MigratePlugin;
 import org.elasticsearch.xpack.migrate.MigrateTemplateRegistry;
+import org.junit.After;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -67,6 +68,14 @@
 
 public class ReindexDatastreamIndexTransportActionIT extends ESIntegTestCase {
 
+    @After
+    private void cleanup() throws Exception {
+        deletePipeline(MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME);
+        assertBusy(() -> {
+            assertTrue(getPipelines(MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME).isFound());
+        });
+    }
+
     private static final String MAPPING = """
         {
           "_doc":{
@@ -103,16 +112,19 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
         """;
 
     public void testTimestamp0AddedIfMissing() {
-        // Delete pipeline in case is a custom pipeline from another test
-        // MigrateTemplateRegistry will immediately rebuild pipeline
-        deletePipeline(MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME);
-
         var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
         safeGet(indicesAdmin().create(new CreateIndexRequest(sourceIndex)));
 
         // add doc without timestamp
         addDoc(sourceIndex, "{\"foo\":\"baz\"}");
 
+        // wait until doc is written to all shards before adding mapping
+        if (cluster().numDataNodes() > 1) {
+            ensureGreen(sourceIndex);
+        } else {
+            ensureYellow(sourceIndex);
+        }
+
         // add timestamp to source mapping
         indicesAdmin().preparePutMapping(sourceIndex).setSource(DATA_STREAM_MAPPING, XContentType.JSON).get();
 
@@ -128,9 +140,6 @@ public void testTimestamp0AddedIfMissing() {
     }
 
     public void testTimestampNotAddedIfExists() {
-        // Delete pipeline in case is a custom pipeline from another test
-        // MigrateTemplateRegistry will immediately rebuild pipeline
-        deletePipeline(MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME);
 
         var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
         safeGet(indicesAdmin().create(new CreateIndexRequest(sourceIndex)));
@@ -296,7 +305,7 @@ public void testMissingSourceIndex() {
         );
     }
 
-    public void testSettingsAddedBeforeReindex() throws Exception {
+    public void testSettingsAddedBeforeReindex() {
         // start with a static setting
         var numShards = randomIntBetween(1, 10);
         var staticSettings = Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, numShards).build();