remove skipping

Author: idegtiarenko

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

@@ -113,7 +113,7 @@ protected void sendRequest(
                 NodeListener nodeListener
             ) {
                 if (exchangeSource.isFinished()) {
-                    nodeListener.onSkip(true);
+                    nodeListener.onSkip();
                     return;
                 }
 

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

@@ -66,7 +66,6 @@ abstract class DataNodeRequestSender {
     private final AtomicInteger skippedShards = new AtomicInteger();
     private final AtomicBoolean changed = new AtomicBoolean();
     private boolean reportedFailure = false; // guarded by sendingLock
-    private volatile boolean skipRemaining = false;
 
     DataNodeRequestSender(
         TransportService transportService,
@@ -93,13 +92,12 @@ final void startComputeOnDataNodes(
         final long startTimeInNanos = System.nanoTime();
         searchShards(rootTask, clusterAlias, requestFilter, concreteIndices, originalIndices, ActionListener.wrap(targetShards -> {
             try (var computeListener = new ComputeListener(transportService.getThreadPool(), runOnTaskFailure, listener.map(profiles -> {
-                var skipped = skippedShards.get() + pendingShardIds.size();
                 return new ComputeResponse(
                     profiles,
                     TimeValue.timeValueNanos(System.nanoTime() - startTimeInNanos),
                     targetShards.totalShards(),
-                    targetShards.totalShards() - shardFailures.size() - skipped,
-                    targetShards.skippedShards() + skipped,
+                    targetShards.totalShards() - shardFailures.size() - skippedShards.get(),
+                    targetShards.skippedShards() + skippedShards.get(),
                     shardFailures.size()
                 );
             }))) {
@@ -138,7 +136,7 @@ private void trySendingRequestsForPendingShards(TargetShards targetShards, Compu
                         || (allowPartialResults == false && shardFailures.values().stream().anyMatch(shardFailure -> shardFailure.fatal))) {
                         reportedFailure = true;
                         reportFailures(computeListener);
-                    } else if (skipRemaining == false) {
+                    } else {
                         for (NodeRequest request : selectNodeRequests(targetShards)) {
                             sendOneNodeRequest(targetShards, computeListener, request);
                         }
@@ -204,11 +202,8 @@ public void onFailure(Exception e, boolean receivedData) {
             }
 
             @Override
-            public void onSkip(boolean skipRemaining) {
-                DataNodeRequestSender.this.skippedShards.incrementAndGet();
-                if (skipRemaining) {
-                    DataNodeRequestSender.this.skipRemaining = true;
-                }
+            public void onSkip() {
+                skippedShards.incrementAndGet();
                 onAfter(List.of());
             }
         });
@@ -221,7 +216,7 @@ interface NodeListener {
 
         void onFailure(Exception e, boolean receivedData);
 
-        void onSkip(boolean skipRemaining);
+        void onSkip();
     }
 
     private static Exception unwrapFailure(Exception e) {

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequestSenderTests.java

@@ -295,7 +295,7 @@ public void testLimitConcurrentNodes() {
             targetShard(shard5, node5)
         );
 
-        var concurrency = randomIntBetween(1, targetShards.size() - 1);
+        var concurrency = randomIntBetween(1, 2);
         AtomicInteger maxConcurrentRequests = new AtomicInteger(0);
         AtomicInteger concurrentRequests = new AtomicInteger(0);
         Queue<NodeRequest> sent = ConcurrentCollections.newQueue();
@@ -340,11 +340,11 @@ public void testDoesNotSendMoreRequestsAfterNodeIsSkipped() {
                 if (processed.incrementAndGet() == 1) {
                     listener.onResponse(new DataNodeComputeResponse(List.of(), Map.of()));
                 } else {
-                    listener.onSkip(true);
+                    listener.onSkip();
                 }
             });
         }));
-        assertThat(sent.size(), equalTo(2));// onResponse() + onSkip()
+        assertThat(sent.size(), equalTo(5));
         assertThat(response.totalShards, equalTo(5));
         assertThat(response.successfulShards, equalTo(1));
         assertThat(response.skippedShards, equalTo(4));