Pull feedback

Author: smalyshev

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeHandler.java

@@ -29,7 +29,7 @@
 import org.elasticsearch.xpack.esql.plan.physical.ExchangeSinkExec;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
 import org.elasticsearch.xpack.esql.session.Configuration;
-import org.elasticsearch.xpack.esql.session.EsqlSessionCCSUtils;
+import org.elasticsearch.xpack.esql.session.EsqlCCSUtils;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -130,8 +130,8 @@ void startComputeOnRemoteCluster(
                         new ActionListenerResponseHandler<>(clusterListener, ComputeResponse::new, esqlExecutor)
                     );
                 }, e -> {
-                    if (EsqlSessionCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
-                        EsqlSessionCCSUtils.markClusterWithFinalStateAndNoShards(
+                    if (EsqlCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
+                        EsqlCCSUtils.markClusterWithFinalStateAndNoShards(
                             executionInfo,
                             clusterAlias,
                             EsqlExecutionInfo.Cluster.Status.SKIPPED,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/RemoteListenerGroup.java

@@ -17,16 +17,18 @@
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.esql.action.EsqlExecutionInfo;
-import org.elasticsearch.xpack.esql.session.EsqlSessionCCSUtils;
+import org.elasticsearch.xpack.esql.session.EsqlCCSUtils;
 
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
 
-// Create group task for this cluster. This group task ensures that two branches of the computation:
-// the exchange sink and the cluster request, belong to the same group and each of them can cancel the other.
-// runAfter listeners below ensure that the group is finalized when both branches are done.
-// The group task is the child of the root task, so if the root task is cancelled, the group task is cancelled too.
+/**
+ * Create group task for this cluster. This group task ensures that two branches of the computation:
+ * the exchange sink and the cluster request, belong to the same group and each of them can cancel the other.
+ * runAfter listeners below ensure that the group is finalized when both branches are done.
+ * The group task is the child of the root task, so if the root task is cancelled, the group task is cancelled too.
+ */
 class RemoteListenerGroup {
     private final CancellableTask groupTask;
     private final ActionListener<Void> exchangeRequestListener;
@@ -72,8 +74,8 @@ class RemoteListenerGroup {
     private <T> ActionListener<T> createCancellingListener(String reason, ActionListener<T> delegate, Runnable finishGroup) {
         return ActionListener.runAfter(delegate.delegateResponse((inner, e) -> {
             taskManager.cancelTaskAndDescendants(groupTask, reason, true, ActionListener.running(() -> {
-                if (EsqlSessionCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
-                    EsqlSessionCCSUtils.markClusterWithFinalStateAndNoShards(
+                if (EsqlCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
+                    EsqlCCSUtils.markClusterWithFinalStateAndNoShards(
                         executionInfo,
                         clusterAlias,
                         EsqlExecutionInfo.Cluster.Status.PARTIAL,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSessionCCSUtils.java renamed to x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlCCSUtils.java

@@ -41,9 +41,9 @@
 import java.util.Objects;
 import java.util.Set;
 
-public class EsqlSessionCCSUtils {
+public class EsqlCCSUtils {
 
-    private EsqlSessionCCSUtils() {}
+    private EsqlCCSUtils() {}
 
     static Map<String, FieldCapabilitiesFailure> determineUnavailableRemoteClusters(List<FieldCapabilitiesFailure> failures) {
         Map<String, FieldCapabilitiesFailure> unavailableRemotes = new HashMap<>();
@@ -335,9 +335,9 @@ public static void checkForCcsLicense(
     }
 
     /**
-     * Mark cluster with a default cluster state with the given status and potentially failure from exception.
-     * Most metrics are set to 0 except for "took" which is set to the total time taken so far.
-     * The status must be the final state of the cluster, not RUNNING.
+     * Mark cluster with a final status (success or failure).
+     * Most metrics are set to 0 if not set yet, except for "took" which is set to the total time taken so far.
+     * The status must be the final status of the cluster, not RUNNING.
      */
     public static void markClusterWithFinalStateAndNoShards(
         EsqlExecutionInfo executionInfo,
@@ -366,6 +366,6 @@ public static boolean shouldIgnoreRuntimeError(EsqlExecutionInfo executionInfo,
         }
 
         return ExceptionsHelper.isRemoteUnavailableException(e)
-            || (e instanceof RemoteTransportException && e.getCause() instanceof TaskCancelledException);
+            || (e instanceof RemoteTransportException && ExceptionsHelper.unwrap(e, TaskCancelledException.class) != null);
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java

@@ -161,7 +161,7 @@ public void execute(EsqlQueryRequest request, EsqlExecutionInfo executionInfo, P
             parse(request.query(), request.params()),
             executionInfo,
             request.filter(),
-            new EsqlSessionCCSUtils.CssPartialErrorsActionListener(executionInfo, listener) {
+            new EsqlCCSUtils.CssPartialErrorsActionListener(executionInfo, listener) {
                 @Override
                 public void onResponse(LogicalPlan analyzedPlan) {
                     preMapper.preMapper(
@@ -188,7 +188,7 @@ public void executeOptimizedPlan(
     ) {
         PhysicalPlan physicalPlan = logicalPlanToPhysicalPlan(optimizedPlan, request);
         // TODO: this could be snuck into the underlying listener
-        EsqlSessionCCSUtils.updateExecutionInfoAtEndOfPlanning(executionInfo);
+        EsqlCCSUtils.updateExecutionInfoAtEndOfPlanning(executionInfo);
         // execute any potential subplans
         executeSubPlans(physicalPlan, planRunner, executionInfo, request, listener);
     }
@@ -315,7 +315,7 @@ public void analyzedPlan(
             .collect(Collectors.toSet());
         final List<TableInfo> indices = preAnalysis.indices;
 
-        EsqlSessionCCSUtils.checkForCcsLicense(executionInfo, indices, indicesExpressionGrouper, verifier.licenseState());
+        EsqlCCSUtils.checkForCcsLicense(executionInfo, indices, indicesExpressionGrouper, verifier.licenseState());
 
         final Set<String> targetClusters = enrichPolicyResolver.groupIndicesPerCluster(
             indices.stream()
@@ -430,7 +430,7 @@ private void preAnalyzeIndices(
             }
             // if the preceding call to the enrich policy API found unavailable clusters, recreate the index expression to search
             // based only on available clusters (which could now be an empty list)
-            String indexExpressionToResolve = EsqlSessionCCSUtils.createIndexExpressionFromAvailableClusters(executionInfo);
+            String indexExpressionToResolve = EsqlCCSUtils.createIndexExpressionFromAvailableClusters(executionInfo);
             if (indexExpressionToResolve.isEmpty()) {
                 // if this was a pure remote CCS request (no local indices) and all remotes are offline, return an empty IndexResolution
                 listener.onResponse(
@@ -464,8 +464,8 @@ private boolean analyzeCCSIndices(
         ActionListener<PreAnalysisResult> l
     ) {
         IndexResolution indexResolution = result.indices;
-        EsqlSessionCCSUtils.updateExecutionInfoWithClustersWithNoMatchingIndices(executionInfo, indexResolution);
-        EsqlSessionCCSUtils.updateExecutionInfoWithUnavailableClusters(executionInfo, indexResolution.unavailableClusters());
+        EsqlCCSUtils.updateExecutionInfoWithClustersWithNoMatchingIndices(executionInfo, indexResolution);
+        EsqlCCSUtils.updateExecutionInfoWithUnavailableClusters(executionInfo, indexResolution.unavailableClusters());
         if (executionInfo.isCrossClusterSearch()
             && executionInfo.getClusterStates(EsqlExecutionInfo.Cluster.Status.RUNNING).findAny().isEmpty()) {
             // for a CCS, if all clusters have been marked as SKIPPED, nothing to search so send a sentinel Exception

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/IndexResolver.java

@@ -139,7 +139,7 @@ public IndexResolution mergedMappings(String indexPattern, FieldCapabilitiesResp
             fields.put(name, field);
         }
 
-        Map<String, FieldCapabilitiesFailure> unavailableRemotes = EsqlSessionCCSUtils.determineUnavailableRemoteClusters(
+        Map<String, FieldCapabilitiesFailure> unavailableRemotes = EsqlCCSUtils.determineUnavailableRemoteClusters(
             fieldCapsResponse.getFailures()
         );