Revert "Skip cluster state serialization to closed channel (#67413)"

This reverts commit 563d07f.

Author: DaveCTurner

qa/smoke-test-http/src/test/java/org/elasticsearch/http/ClusterStateRestCancellationIT.java

@@ -41,7 +41,6 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.network.CloseableChannel;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
@@ -458,8 +457,6 @@ public void sendResponse(RestResponse response) {
                     assertThat(response.content().utf8ToString(), not(containsString("verysecretpassword")));
                 } catch (AssertionError ex) {
                     clusterStateError.set(ex);
-                } finally {
-                    CloseableChannel.closeChannel(clusterStateRequest.getHttpChannel());
                 }
                 clusterStateLatch.countDown();
             }

server/src/internalClusterTest/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java

@@ -28,13 +28,8 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.tasks.CancellableTask;
-import org.elasticsearch.tasks.Task;
-import org.elasticsearch.tasks.TaskId;
 
 import java.io.IOException;
-import java.util.Arrays;
-import java.util.Map;
 
 public class ClusterStateRequest extends MasterNodeReadRequest<ClusterStateRequest> implements IndicesRequest.Replaceable {
 
@@ -201,47 +196,4 @@ public ClusterStateRequest waitForMetadataVersion(long waitForMetadataVersion) {
         this.waitForMetadataVersion = waitForMetadataVersion;
         return this;
     }
-
-    @Override
-    public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
-        return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers) {
-            @Override
-            public boolean shouldCancelChildrenOnCancellation() {
-                return true;
-            }
-        };
-    }
-
-    @Override
-    public String getDescription() {
-        final StringBuilder stringBuilder = new StringBuilder("cluster state [");
-        if (routingTable) {
-            stringBuilder.append("routing table, ");
-        }
-        if (nodes) {
-            stringBuilder.append("nodes, ");
-        }
-        if (metadata) {
-            stringBuilder.append("metadata, ");
-        }
-        if (blocks) {
-            stringBuilder.append("blocks, ");
-        }
-        if (customs) {
-            stringBuilder.append("customs, ");
-        }
-        if (local) {
-            stringBuilder.append("local, ");
-        }
-        if (waitForMetadataVersion != null) {
-            stringBuilder.append("wait for metadata version [").append(waitForMetadataVersion)
-                    .append("] with timeout [").append(waitForTimeout).append("], ");
-        }
-        if (indices.length > 0) {
-            stringBuilder.append("indices ").append(Arrays.toString(indices)).append(", ");
-        }
-        stringBuilder.append("master timeout [").append(masterNodeTimeout).append("]]");
-        return stringBuilder.toString();
-    }
-
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/state/ClusterStateRequest.java

@@ -41,9 +41,6 @@
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.node.NodeClosedException;
-import org.elasticsearch.tasks.CancellableTask;
-import org.elasticsearch.tasks.Task;
-import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
@@ -81,38 +78,25 @@ protected ClusterBlockException checkBlock(ClusterStateRequest request, ClusterS
     @Override
     protected void masterOperation(final ClusterStateRequest request, final ClusterState state,
                                    final ActionListener<ClusterStateResponse> listener) throws IOException {
-        assert false : "task is required";
-        throw new UnsupportedOperationException("task is required");
-    }
-
-    @Override
-    protected void masterOperation(final Task task, final ClusterStateRequest request, final ClusterState state,
-                                   final ActionListener<ClusterStateResponse> listener) throws IOException {
-
-        assert task instanceof CancellableTask : task + " not cancellable";
-        final CancellableTask cancellableTask = (CancellableTask) task;
 
         final Predicate<ClusterState> acceptableClusterStatePredicate
             = request.waitForMetadataVersion() == null ? clusterState -> true
             : clusterState -> clusterState.metadata().version() >= request.waitForMetadataVersion();
 
-        final Predicate<ClusterState> acceptableClusterStateOrFailedPredicate = request.local()
+        final Predicate<ClusterState> acceptableClusterStateOrNotMasterPredicate = request.local()
             ? acceptableClusterStatePredicate
-            : acceptableClusterStatePredicate.or(clusterState ->
-                cancellableTask.isCancelled() || clusterState.nodes().isLocalNodeElectedMaster() == false);
+            : acceptableClusterStatePredicate.or(clusterState -> clusterState.nodes().isLocalNodeElectedMaster() == false);
 
         if (acceptableClusterStatePredicate.test(state)) {
             ActionListener.completeWith(listener, () -> buildResponse(request, state));
         } else {
-            assert acceptableClusterStateOrFailedPredicate.test(state) == false;
+            assert acceptableClusterStateOrNotMasterPredicate.test(state) == false;
             new ClusterStateObserver(state, clusterService, request.waitForTimeout(), logger, threadPool.getThreadContext())
                 .waitForNextChange(new ClusterStateObserver.Listener() {
 
                 @Override
                 public void onNewClusterState(ClusterState newState) {
-                    if (cancellableTask.isCancelled()) {
-                        listener.onFailure(new TaskCancelledException("task cancelled"));
-                    } else if (acceptableClusterStatePredicate.test(newState)) {
+                    if (acceptableClusterStatePredicate.test(newState)) {
                         ActionListener.completeWith(listener, () -> buildResponse(request, newState));
                     } else {
                         listener.onFailure(new NotMasterException(
@@ -128,16 +112,12 @@ public void onClusterServiceClose() {
                 @Override
                 public void onTimeout(TimeValue timeout) {
                     try {
-                        if (cancellableTask.isCancelled()) {
-                            listener.onFailure(new TaskCancelledException("task cancelled"));
-                        } else {
-                            listener.onResponse(new ClusterStateResponse(state.getClusterName(), null, true));
-                        }
+                        listener.onResponse(new ClusterStateResponse(state.getClusterName(), null, true));
                     } catch (Exception e) {
                         listener.onFailure(e);
                     }
                 }
-            }, acceptableClusterStateOrFailedPredicate);
+            }, acceptableClusterStateOrNotMasterPredicate);
         }
     }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/state/TransportClusterStateAction.java

@@ -21,7 +21,6 @@
 
 import org.elasticsearch.ElasticsearchTimeoutException;
 import org.elasticsearch.action.ActionRunnable;
-import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -41,8 +40,6 @@
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.rest.action.RestActionListener;
 import org.elasticsearch.rest.action.RestBuilderListener;
-import org.elasticsearch.rest.action.RestCancellableNodeClient;
-import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
@@ -119,18 +116,10 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         }
         settingsFilter.addFilterSettingParams(request);
 
-        return channel -> new RestCancellableNodeClient(client, request.getHttpChannel())
-                .execute(ClusterStateAction.INSTANCE, clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
-
-                    private void ensureOpen() {
-                        if (request.getHttpChannel().isOpen() == false) {
-                            throw new TaskCancelledException("response channel [" + request.getHttpChannel() + "] closed");
-                        }
-                    }
+        return channel -> client.admin().cluster().state(clusterStateRequest, new RestActionListener<ClusterStateResponse>(channel) {
 
                     @Override
                     protected void processResponse(ClusterStateResponse response) {
-                        ensureOpen();
                         final long startTimeMs = threadPool.relativeTimeInMillis();
                         // Process serialization on MANAGEMENT pool since the serialization of the cluster state to XContent
                         // can be too slow to execute on an IO thread
@@ -139,7 +128,6 @@ protected void processResponse(ClusterStateResponse response) {
                                     @Override
                                     public RestResponse buildResponse(final ClusterStateResponse response,
                                                                       final XContentBuilder builder) throws Exception {
-                                        ensureOpen();
                                         if (clusterStateRequest.local() == false &&
                                                 threadPool.relativeTimeInMillis() - startTimeMs >
                                                         clusterStateRequest.masterNodeTimeout().millis()) {
@@ -152,16 +140,13 @@ public RestResponse buildResponse(final ClusterStateResponse response,
                                         builder.field(Fields.CLUSTER_NAME, response.getClusterName().value());
                                         ToXContent.Params params = new ToXContent.DelegatingMapParams(
                                                 singletonMap(Metadata.CONTEXT_MODE_PARAM, Metadata.CONTEXT_MODE_API), request);
-                                        final ClusterState responseState = response.getState();
-                                        if (responseState != null) {
-                                            responseState.toXContent(builder, params);
-                                        }
+                                        response.getState().toXContent(builder, params);
                                         builder.endObject();
                                         return new BytesRestResponse(RestStatus.OK, builder);
                                     }
                                 }.onResponse(response)));
                     }
-                });
+        });
     }
 
     private static final Set<String> RESPONSE_PARAMS;

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStateAction.java

@@ -27,7 +27,6 @@
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.VersionUtils;
 
-import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.equalTo;
 
 /**
@@ -88,19 +87,4 @@ private static void assertOptionsMatch(IndicesOptions in, IndicesOptions out) {
         assertThat(in.expandWildcardsOpen(), equalTo(out.expandWildcardsOpen()));
         assertThat(in.allowNoIndices(), equalTo(out.allowNoIndices()));
     }
-
-    public void testDescription() {
-        assertThat(new ClusterStateRequest().clear().getDescription(), equalTo("cluster state [master timeout [30s]]"));
-        assertThat(new ClusterStateRequest().masterNodeTimeout("5m").getDescription(),
-                equalTo("cluster state [routing table, nodes, metadata, blocks, customs, master timeout [5m]]"));
-        assertThat(new ClusterStateRequest().clear().routingTable(true).getDescription(), containsString("routing table"));
-        assertThat(new ClusterStateRequest().clear().nodes(true).getDescription(), containsString("nodes"));
-        assertThat(new ClusterStateRequest().clear().metadata(true).getDescription(), containsString("metadata"));
-        assertThat(new ClusterStateRequest().clear().blocks(true).getDescription(), containsString("blocks"));
-        assertThat(new ClusterStateRequest().clear().customs(true).getDescription(), containsString("customs"));
-        assertThat(new ClusterStateRequest().local(true).getDescription(), containsString("local"));
-        assertThat(new ClusterStateRequest().waitForMetadataVersion(23L).getDescription(),
-                containsString("wait for metadata version [23] with timeout [1m]"));
-        assertThat(new ClusterStateRequest().indices("foo", "bar").getDescription(), containsString("indices [foo, bar]"));
-    }
 }