[Transform] ResolveIndex from TransformUpdate (#144208)

prwhelan · web-flow · commit cde98aaa5086 · 2026-03-19T21:17:58.000+01:00
Both CCS and CPS need to call ResolveIndexAction in order to get source
indices from a remote cluster or project.
diff --git a/x-pack/plugin/transform/src/internalClusterTest/java/org/elasticsearch/xpack/transform/checkpoint/TransformCCSCanMatchIT.java b/x-pack/plugin/transform/src/internalClusterTest/java/org/elasticsearch/xpack/transform/checkpoint/TransformCCSCanMatchIT.java
@@ -14,6 +14,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateRequest;
 import org.elasticsearch.action.admin.cluster.snapshots.features.TransportResetFeatureStateAction;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
@@ -37,6 +38,7 @@
 import org.elasticsearch.node.NodeRoleSettings;
 import org.elasticsearch.plugins.EnginePlugin;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.reindex.ReindexPlugin;
 import org.elasticsearch.search.SearchModule;
 import org.elasticsearch.search.aggregations.BaseAggregationBuilder;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
@@ -52,14 +54,19 @@
 import org.elasticsearch.xpack.core.transform.MockDeprecatedAggregationBuilder;
 import org.elasticsearch.xpack.core.transform.MockDeprecatedQueryBuilder;
 import org.elasticsearch.xpack.core.transform.TransformNamedXContentProvider;
+import org.elasticsearch.xpack.core.transform.action.DeleteTransformAction;
 import org.elasticsearch.xpack.core.transform.action.GetCheckpointAction;
 import org.elasticsearch.xpack.core.transform.action.GetTransformStatsAction;
 import org.elasticsearch.xpack.core.transform.action.PutTransformAction;
 import org.elasticsearch.xpack.core.transform.action.StartTransformAction;
+import org.elasticsearch.xpack.core.transform.action.StopTransformAction;
+import org.elasticsearch.xpack.core.transform.action.UpdateTransformAction;
 import org.elasticsearch.xpack.core.transform.transforms.DestConfig;
 import org.elasticsearch.xpack.core.transform.transforms.QueryConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SourceConfig;
+import org.elasticsearch.xpack.core.transform.transforms.TimeSyncConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
+import org.elasticsearch.xpack.core.transform.transforms.TransformConfigUpdate;
 import org.elasticsearch.xpack.core.transform.transforms.TransformStats;
 import org.elasticsearch.xpack.core.transform.transforms.latest.LatestConfig;
 import org.elasticsearch.xpack.transform.LocalStateTransform;
@@ -84,6 +91,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 
 public class TransformCCSCanMatchIT extends AbstractMultiClustersTestCase {
@@ -392,6 +400,72 @@ private void testTransformLifecycle(QueryBuilder query, long expectedHitCount) t
         });
     }
 
+    public void testUpdateTransformCreatesDestIndexWhenRunningAndDestMissingWithRemoteSource() throws Exception {
+        String transformId = "test-ccs-transform-update-dest";
+        String destIndex = transformId + "-dest";
+
+        // Create a continuous transform sourced from both local and remote indices (CCS).
+        // ZERO delay ensures the already-indexed documents are included in the first checkpoint.
+        TransformConfig transformConfig = TransformConfig.builder()
+            .setId(transformId)
+            .setSource(new SourceConfig(new String[] { "local_*", "*:remote_*" }, QueryConfig.matchAll(), Map.of(), null))
+            .setDest(new DestConfig(destIndex, null, null))
+            .setFrequency(TimeValue.timeValueMinutes(1))
+            .setSyncConfig(new TimeSyncConfig("@timestamp", TimeValue.ZERO))
+            .setLatestConfig(new LatestConfig(List.of("position"), "@timestamp"))
+            .build();
+
+        client().execute(PutTransformAction.INSTANCE, new PutTransformAction.Request(transformConfig, false, TIMEOUT)).actionGet(TIMEOUT);
+        client().execute(StartTransformAction.INSTANCE, new StartTransformAction.Request(transformId, null, TIMEOUT)).actionGet(TIMEOUT);
+
+        // Wait for the first checkpoint to create the destination index.
+        assertBusy(
+            () -> assertThat(
+                "dest index should be created after the first checkpoint",
+                cluster(LOCAL_CLUSTER).clusterService().state().metadata().getProject().index(destIndex),
+                is(notNullValue())
+            )
+        );
+
+        // Delete the destination index to simulate it going missing while the transform is still running.
+        client(LOCAL_CLUSTER).admin().indices().delete(new DeleteIndexRequest(destIndex)).actionGet(TIMEOUT);
+        assertThat(cluster(LOCAL_CLUSTER).clusterService().state().metadata().getProject().index(destIndex), is(nullValue()));
+
+        // Update the transform. A description change ensures we go through updateTransformConfiguration
+        // (an EMPTY update against a current-version config returns NONE early without touching the dest index).
+        client().execute(
+            UpdateTransformAction.INSTANCE,
+            new UpdateTransformAction.Request(
+                new TransformConfigUpdate(null, null, null, null, "post-delete-update", null, null, null),
+                transformId,
+                false,
+                TIMEOUT
+            )
+        ).actionGet(TIMEOUT);
+
+        // The update should have invoked resolveSourceIndicesAndCreateDestIfNeeded, which calls ResolveIndexAction
+        // to resolve the remote CCS source indices and then recreates the destination index.
+        assertThat(
+            "update should have recreated the destination index using ResolveIndexAction for CCS source resolution",
+            cluster(LOCAL_CLUSTER).clusterService().state().metadata().getProject().index(destIndex),
+            is(notNullValue())
+        );
+
+        // Cleanup
+        stopTransform(transformId);
+        deleteTransform(transformId);
+    }
+
+    private void stopTransform(String transformId) {
+        client().execute(StopTransformAction.INSTANCE, new StopTransformAction.Request(transformId, true, true, TIMEOUT, false, false))
+            .actionGet(TIMEOUT);
+    }
+
+    private void deleteTransform(String transformId) {
+        client().execute(DeleteTransformAction.INSTANCE, new DeleteTransformAction.Request(transformId, true, false, TIMEOUT))
+            .actionGet(TIMEOUT);
+    }
+
     @Override
     protected NamedXContentRegistry xContentRegistry() {
         return namedXContentRegistry;
@@ -405,7 +479,10 @@ protected List<String> remoteClusterAlias() {
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
         return CollectionUtils.appendToCopy(
-            CollectionUtils.appendToCopy(super.nodePlugins(clusterAlias), LocalStateTransform.class),
+            CollectionUtils.appendToCopy(
+                CollectionUtils.appendToCopy(super.nodePlugins(clusterAlias), LocalStateTransform.class),
+                ReindexPlugin.class
+            ),
             ExposingTimestampEnginePlugin.class
         );
     }
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransformUpdater.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransformUpdater.java
@@ -11,6 +11,7 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.resolve.ResolveIndexAction;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterState;
@@ -21,7 +22,6 @@
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
 import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
-import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.XPackSettings;
 import org.elasticsearch.xpack.core.security.SecurityContext;
@@ -37,7 +37,6 @@
 import org.elasticsearch.xpack.transform.persistence.TransformConfigManager;
 import org.elasticsearch.xpack.transform.persistence.TransformIndex;
 
-import java.util.Arrays;
 import java.util.Map;
 
 /**
@@ -337,21 +336,13 @@ private static void updateTransformConfiguration(
         final String destinationIndex = config.getDestination().getIndex();
         String[] dest = indexNameExpressionResolver.concreteIndexNames(clusterState, IndicesOptions.lenientExpandOpen(), destinationIndex);
 
-        // FIXME: what do we do with remote indices?
-        String[] sourceIndices = Arrays.stream(config.getSource().getIndex())
-            .filter(ind -> RemoteClusterAware.isRemoteIndexName(ind) == false)
-            .toArray(String[]::new);
-        String[] src = sourceIndices.length > 0
-            ? indexNameExpressionResolver.concreteIndexNames(clusterState, IndicesOptions.lenientExpandOpen(), true, sourceIndices)
-            : null;
         // If we are running, we should verify that the destination index exists and create it if it does not
-        if (PersistentTasksCustomMetadata.getTaskWithId(clusterState, config.getId()) != null && dest.length == 0
-        // Verify we have source indices. The user could defer_validations and if the task is already running
-        // we allow source indices to disappear. If the source and destination indices do not exist, don't do anything
-        // the transform will just have to dynamically create the destination index without special mapping.
-            && src != null
-            && src.length > 0) {
-            TransformIndex.createDestinationIndex(
+        if (PersistentTasksCustomMetadata.getTaskWithId(clusterState, config.getId()) != null && dest.length == 0) {
+            // Resolve source indices (including remote) to verify they exist before creating the dest index.
+            // The user could defer_validations and if the task is already running we allow source indices to
+            // disappear. If the source and destination indices do not exist, don't do anything -- the transform
+            // will just have to dynamically create the destination index without special mapping.
+            resolveSourceIndicesAndCreateDestIfNeeded(
                 client,
                 auditor,
                 indexNameExpressionResolver,
@@ -366,5 +357,52 @@ private static void updateTransformConfiguration(
         }
     }
 
+    private static void resolveSourceIndicesAndCreateDestIfNeeded(
+        Client client,
+        TransformAuditor auditor,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        ClusterState clusterState,
+        TransformConfig config,
+        Settings destIndexSettings,
+        Map<String, String> destIndexMappings,
+        ActionListener<Boolean> listener
+    ) {
+        ResolveIndexAction.Request resolveRequest = new ResolveIndexAction.Request(
+            config.getSource().getIndex(),
+            config.getSource().indicesOptions()
+        );
+        ClientHelper.executeAsyncWithOrigin(
+            client,
+            ClientHelper.TRANSFORM_ORIGIN,
+            ResolveIndexAction.INSTANCE,
+            resolveRequest,
+            ActionListener.wrap(resolveResponse -> {
+                boolean hasSourceIndices = resolveResponse.getIndices().isEmpty() == false
+                    || resolveResponse.getAliases().isEmpty() == false
+                    || resolveResponse.getDataStreams().isEmpty() == false;
+                if (hasSourceIndices) {
+                    TransformIndex.createDestinationIndex(
+                        client,
+                        auditor,
+                        indexNameExpressionResolver,
+                        clusterState,
+                        config,
+                        destIndexSettings,
+                        destIndexMappings,
+                        listener
+                    );
+                } else {
+                    listener.onResponse(null);
+                }
+            }, e -> {
+                logger.debug(
+                    () -> "[" + config.getId() + "] failed to resolve source indices during update, skipping dest index creation",
+                    e
+                );
+                listener.onResponse(null);
+            })
+        );
+    }
+
     private TransformUpdater() {}
 }
diff --git a/x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/action/TransformUpdaterTests.java b/x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/action/TransformUpdaterTests.java
