[Transform] Fail checkpoint on missing clusters (#106793) (#106842)

When there are no remote or local clusters for a given source index, we
call the listener's `onFailure` method with a `CheckpointException`.
A running transform will fail and retry, eventually moving into an
unhealthy and failed state.  Any call to the stats API will note the
checkpoint failure and return.

This fixes a timeout issue calling the Transform stats API and prevents
the Transform from being stuck in indexing.

Fix #106790
Fix #104533

Author: prwhelan

docs/changelog/106793.yaml

@@ -0,0 +1,7 @@
+pr: 106793
+summary: Fail checkpoint on missing clusters
+area: Transform
+type: bug
+issues:
+ - 104533
+ - 106790

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/checkpoint/CheckpointException.java

@@ -10,6 +10,10 @@
 import org.elasticsearch.ElasticsearchException;
 
 class CheckpointException extends ElasticsearchException {
+    CheckpointException(String msg, Object... params) {
+        super(msg, params);
+    }
+
     CheckpointException(String msg, Throwable cause, Object... params) {
         super(msg, cause, params);
     }

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/checkpoint/DefaultCheckpointProvider.java

@@ -116,6 +116,12 @@ protected void getIndexCheckpoints(TimeValue timeout, ActionListener<Map<String,
             ResolvedIndices resolvedIndexes = remoteClusterResolver.resolve(transformConfig.getSource().getIndex());
             ActionListener<Map<String, long[]>> groupedListener = listener;
 
+            if (resolvedIndexes.numClusters() == 0) {
+                var indices = String.join(",", transformConfig.getSource().getIndex());
+                listener.onFailure(new CheckpointException("No clusters exist for [{}]", indices));
+                return;
+            }
+
             if (resolvedIndexes.numClusters() > 1) {
                 ActionListener<Collection<Map<String, long[]>>> mergeMapsListener = ActionListener.wrap(indexCheckpoints -> {
                     listener.onResponse(
@@ -228,10 +234,7 @@ private static void getCheckpointsFromOneClusterV2(
         );
         ActionListener<GetCheckpointAction.Response> checkpointListener;
         if (RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY.equals(cluster)) {
-            checkpointListener = ActionListener.wrap(
-                checkpointResponse -> listener.onResponse(checkpointResponse.getCheckpoints()),
-                listener::onFailure
-            );
+            checkpointListener = listener.safeMap(GetCheckpointAction.Response::getCheckpoints);
         } else {
             checkpointListener = ActionListener.wrap(
                 checkpointResponse -> listener.onResponse(
@@ -395,12 +398,12 @@ public void getCheckpointingInfo(
 
         long timestamp = clock.millis();
 
-        getIndexCheckpoints(timeout, ActionListener.wrap(checkpointsByIndex -> {
+        getIndexCheckpoints(timeout, listener.delegateFailure((l, checkpointsByIndex) -> {
             TransformCheckpoint sourceCheckpoint = new TransformCheckpoint(transformConfig.getId(), timestamp, -1L, checkpointsByIndex, 0L);
             checkpointingInfoBuilder.setSourceCheckpoint(sourceCheckpoint);
             checkpointingInfoBuilder.setOperationsBehind(TransformCheckpoint.getBehind(lastCheckpoint, sourceCheckpoint));
-            listener.onResponse(checkpointingInfoBuilder);
-        }, listener::onFailure));
+            l.onResponse(checkpointingInfoBuilder);
+        }));
     }
 
     @Override

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/checkpoint/DefaultCheckpointProviderTests.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.test.MockLogAppender;
 import org.elasticsearch.test.MockLogAppender.LoggingExpectation;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.ActionNotFoundTransportException;
 import org.elasticsearch.xpack.core.transform.action.GetCheckpointAction;
 import org.elasticsearch.xpack.core.transform.transforms.SourceConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformCheckpoint;
@@ -67,7 +68,7 @@
 
 public class DefaultCheckpointProviderTests extends ESTestCase {
 
-    private static Logger checkpointProviderLogger = LogManager.getLogger(DefaultCheckpointProvider.class);
+    private static final Logger checkpointProviderLogger = LogManager.getLogger(DefaultCheckpointProvider.class);
 
     private Clock clock;
     private Client client;
@@ -96,7 +97,7 @@ public void setUpMocks() {
         transformAuditor = MockTransformAuditor.createMockAuditor();
     }
 
-    public void testReportSourceIndexChangesRunsEmpty() throws Exception {
+    public void testReportSourceIndexChangesRunsEmpty() {
         String transformId = getTestName();
         TransformConfig transformConfig = TransformConfigTests.randomTransformConfig(transformId);
         DefaultCheckpointProvider provider = newCheckpointProvider(transformConfig);
@@ -138,7 +139,7 @@ public void testReportSourceIndexChangesRunsEmpty() throws Exception {
         );
     }
 
-    public void testReportSourceIndexChangesAddDelete() throws Exception {
+    public void testReportSourceIndexChangesAddDelete() {
         String transformId = getTestName();
         TransformConfig transformConfig = TransformConfigTests.randomTransformConfig(transformId);
         DefaultCheckpointProvider provider = newCheckpointProvider(transformConfig);
@@ -197,7 +198,7 @@ public void testReportSourceIndexChangesAddDelete() throws Exception {
         );
     }
 
-    public void testReportSourceIndexChangesAddDeleteMany() throws Exception {
+    public void testReportSourceIndexChangesAddDeleteMany() {
         String transformId = getTestName();
         TransformConfig transformConfig = TransformConfigTests.randomTransformConfig(transformId);
         DefaultCheckpointProvider provider = newCheckpointProvider(transformConfig);
@@ -231,29 +232,88 @@ public void testReportSourceIndexChangesAddDeleteMany() throws Exception {
     }
 
     public void testHandlingShardFailures() throws Exception {
-        String transformId = getTestName();
-        String indexName = "some-index";
+        var transformId = getTestName();
+        var indexName = "some-index";
         TransformConfig transformConfig = new TransformConfig.Builder(TransformConfigTests.randomTransformConfig(transformId)).setSource(
             new SourceConfig(indexName)
         ).build();
 
-        RemoteClusterResolver remoteClusterResolver = mock(RemoteClusterResolver.class);
+        var remoteClusterResolver = mock(RemoteClusterResolver.class);
         doReturn(new RemoteClusterResolver.ResolvedIndices(Collections.emptyMap(), Collections.singletonList(indexName))).when(
             remoteClusterResolver
         ).resolve(transformConfig.getSource().getIndex());
 
+        mockGetIndexResponse(indexName);
+        mockIndicesStatsResponse(indexName);
+        mockGetCheckpointAction();
+
+        var provider = new DefaultCheckpointProvider(
+            clock,
+            parentTaskClient,
+            remoteClusterResolver,
+            transformConfigManager,
+            transformAuditor,
+            transformConfig
+        );
+
+        var latch = new CountDownLatch(1);
+        provider.createNextCheckpoint(
+            null,
+            new LatchedActionListener<>(
+                ActionListener.wrap(
+                    response -> fail("This test case must fail"),
+                    e -> assertThat(
+                        e.getMessage(),
+                        startsWith(
+                            "Source has [7] failed shards, first shard failure: [some-index][3] failed, "
+                                + "reason [java.lang.Exception: something's wrong"
+                        )
+                    )
+                ),
+                latch
+            )
+        );
+        assertTrue(latch.await(1, TimeUnit.MILLISECONDS));
+    }
+
+    private void mockGetIndexResponse(String indexName) {
         GetIndexResponse getIndexResponse = new GetIndexResponse(new String[] { indexName }, null, null, null, null, null);
         doAnswer(withResponse(getIndexResponse)).when(client).execute(eq(GetIndexAction.INSTANCE), any(), any());
+    }
 
+    private void mockIndicesStatsResponse(String indexName) {
         IndicesStatsResponse indicesStatsResponse = mock(IndicesStatsResponse.class);
         doReturn(7).when(indicesStatsResponse).getFailedShards();
         doReturn(
             new DefaultShardOperationFailedException[] {
                 new DefaultShardOperationFailedException(indexName, 3, new Exception("something's wrong")) }
         ).when(indicesStatsResponse).getShardFailures();
         doAnswer(withResponse(indicesStatsResponse)).when(client).execute(eq(IndicesStatsAction.INSTANCE), any(), any());
+    }
 
-        DefaultCheckpointProvider provider = new DefaultCheckpointProvider(
+    private void mockGetCheckpointAction() {
+        doAnswer(invocationOnMock -> {
+            ActionListener<?> listener = invocationOnMock.getArgument(2);
+            listener.onFailure(new ActionNotFoundTransportException("This should fail."));
+            return null;
+        }).when(client).execute(eq(GetCheckpointAction.INSTANCE), any(), any());
+    }
+
+    public void testHandlingNoClusters() throws Exception {
+        var transformId = getTestName();
+        var indexName = "some-missing-index";
+        var transformConfig = new TransformConfig.Builder(TransformConfigTests.randomTransformConfig(transformId)).setSource(
+            new SourceConfig(indexName)
+        ).build();
+
+        var remoteClusterResolver = mock(RemoteClusterResolver.class);
+        doReturn(new RemoteClusterResolver.ResolvedIndices(Map.of(), List.of())).when(remoteClusterResolver)
+            .resolve(transformConfig.getSource().getIndex());
+
+        mockGetIndexResponse(indexName);
+        mockIndicesStatsResponse(indexName);
+
+        var provider = new DefaultCheckpointProvider(
             clock,
             parentTaskClient,
             remoteClusterResolver,
@@ -262,24 +322,18 @@ public void testHandlingShardFailures() throws Exception {
             transformConfig
         );
 
-        CountDownLatch latch = new CountDownLatch(1);
+        var latch = new CountDownLatch(1);
         provider.createNextCheckpoint(
             null,
             new LatchedActionListener<>(
                 ActionListener.wrap(
                     response -> fail("This test case must fail"),
-                    e -> assertThat(
-                        e.getMessage(),
-                        startsWith(
-                            "Source has [7] failed shards, first shard failure: [some-index][3] failed, "
-                                + "reason [java.lang.Exception: something's wrong"
-                        )
-                    )
+                    e -> assertThat(e.getMessage(), equalTo("No clusters exist for [some-missing-index]"))
                 ),
                 latch
             )
         );
-        latch.await(10, TimeUnit.SECONDS);
+        assertTrue(latch.await(1, TimeUnit.MILLISECONDS));
     }
 
     public void testSourceHasChanged() throws InterruptedException {
@@ -407,8 +461,7 @@ private DefaultCheckpointProvider newCheckpointProvider(TransformConfig transfor
         );
     }
 
-    private void assertExpectation(LoggingExpectation loggingExpectation, AuditExpectation auditExpectation, Runnable codeBlock)
-        throws IllegalAccessException {
+    private void assertExpectation(LoggingExpectation loggingExpectation, AuditExpectation auditExpectation, Runnable codeBlock) {
         MockLogAppender mockLogAppender = new MockLogAppender();
         mockLogAppender.start();
 
@@ -429,10 +482,9 @@ private void assertExpectation(LoggingExpectation loggingExpectation, AuditExpec
         }
     }
 
-    @SuppressWarnings("unchecked")
     private static <Response> Answer<Response> withResponse(Response response) {
         return invocationOnMock -> {
-            ActionListener<Response> listener = (ActionListener<Response>) invocationOnMock.getArguments()[2];
+            ActionListener<Response> listener = invocationOnMock.getArgument(2);
             listener.onResponse(response);
             return null;
         };