Add support for passing down if restricted params are used

# Conflicts:
#	server/src/main/java/org/elasticsearch/TransportVersions.java

Author: lukewhiting

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -365,6 +365,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion SIMULATE_INGEST_MAPPING_MERGE_TYPE = def(9_138_0_00);
     public static final TransportVersion ESQL_LOOKUP_JOIN_ON_MANY_FIELDS = def(9_139_0_00);
     public static final TransportVersion SIMULATE_INGEST_EFFECTIVE_MAPPING = def(9_140_0_00);
+    public static final TransportVersion STREAMS_ENDPOINT_PARAM_RESTRICTIONS = def(9_141_00_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java

@@ -86,6 +86,7 @@ public class BulkRequest extends LegacyActionRequest
     private Boolean globalRequireAlias;
     private Boolean globalRequireDatsStream;
     private boolean includeSourceOnError = true;
+    private boolean streamsRestrictedParamsUsed = false;
 
     private long sizeInBytes = 0;
 
@@ -107,7 +108,10 @@ public BulkRequest(StreamInput in) throws IOException {
         }
         if (in.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
             includeSourceOnError = in.readBoolean();
-        } // else default value is true
+        }
+        if (in.getTransportVersion().onOrAfter(TransportVersions.STREAMS_ENDPOINT_PARAM_RESTRICTIONS)) {
+            streamsRestrictedParamsUsed = in.readBoolean();
+        }
     }
 
     public BulkRequest(@Nullable String globalIndex) {
@@ -474,6 +478,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
             out.writeBoolean(includeSourceOnError);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.STREAMS_ENDPOINT_PARAM_RESTRICTIONS)) {
+            out.writeBoolean(streamsRestrictedParamsUsed);
+        }
     }
 
     @Override
@@ -516,6 +523,14 @@ public boolean isSimulated() {
         return false; // Always false, but may be overridden by a subclass
     }
 
+    public boolean streamsRestrictedParamsUsed() {
+        return streamsRestrictedParamsUsed;
+    }
+
+    public void streamsRestrictedParamsUsed(boolean streamsRestrictedParamsUsed) {
+        this.streamsRestrictedParamsUsed = streamsRestrictedParamsUsed;
+    }
+
     /*
      * Returns any component template substitutions that are to be used as part of this bulk request. We would likely only have
      * substitutions in the event of a simulated request.
@@ -554,6 +569,7 @@ public BulkRequest shallowClone() {
         bulkRequest.routing(routing());
         bulkRequest.requireAlias(requireAlias());
         bulkRequest.requireDataStream(requireDataStream());
+        bulkRequest.streamsRestrictedParamsUsed(streamsRestrictedParamsUsed());
         return bulkRequest;
     }
 }

server/src/main/java/org/elasticsearch/action/bulk/IncrementalBulkService.java

@@ -62,12 +62,25 @@ public IncrementalBulkService(Client client, IndexingPressure indexingPressure,
 
     public Handler newBulkRequest() {
         ensureEnabled();
-        return newBulkRequest(null, null, null);
+        return newBulkRequest(null, null, null, false);
     }
 
-    public Handler newBulkRequest(@Nullable String waitForActiveShards, @Nullable TimeValue timeout, @Nullable String refresh) {
+    public Handler newBulkRequest(
+        @Nullable String waitForActiveShards,
+        @Nullable TimeValue timeout,
+        @Nullable String refresh,
+        boolean usesStreamsRestrictedParams
+    ) {
         ensureEnabled();
-        return new Handler(client, indexingPressure, waitForActiveShards, timeout, refresh, chunkWaitTimeMillisHistogram);
+        return new Handler(
+            client,
+            indexingPressure,
+            waitForActiveShards,
+            timeout,
+            refresh,
+            chunkWaitTimeMillisHistogram,
+            usesStreamsRestrictedParams
+        );
     }
 
     private void ensureEnabled() {
@@ -105,6 +118,7 @@ public static class Handler implements Releasable {
         private final Client client;
         private final ActiveShardCount waitForActiveShards;
         private final TimeValue timeout;
+        private final boolean usesStreamsRestrictedParams;
         private final String refresh;
 
         private final ArrayList<Releasable> releasables = new ArrayList<>(4);
@@ -125,12 +139,14 @@ protected Handler(
             @Nullable String waitForActiveShards,
             @Nullable TimeValue timeout,
             @Nullable String refresh,
-            LongHistogram chunkWaitTimeMillisHistogram
+            LongHistogram chunkWaitTimeMillisHistogram,
+            boolean usesStreamsRestrictedParams
         ) {
             this.client = client;
             this.waitForActiveShards = waitForActiveShards != null ? ActiveShardCount.parseString(waitForActiveShards) : null;
             this.timeout = timeout;
             this.refresh = refresh;
+            this.usesStreamsRestrictedParams = usesStreamsRestrictedParams;
             this.incrementalOperation = indexingPressure.startIncrementalCoordinating(0, 0, false);
             this.chunkWaitTimeMillisHistogram = chunkWaitTimeMillisHistogram;
             createNewBulkRequest(EMPTY_STATE);
@@ -310,6 +326,7 @@ private void createNewBulkRequest(BulkRequest.IncrementalState incrementalState)
             if (refresh != null) {
                 bulkRequest.setRefreshPolicy(refresh);
             }
+            bulkRequest.streamsRestrictedParamsUsed(usesStreamsRestrictedParams);
         }
     }
 }

server/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.TimeValue;
@@ -62,6 +63,7 @@ public class RestBulkAction extends BaseRestHandler {
 
     public static final String TYPES_DEPRECATION_MESSAGE = "[types removal] Specifying types in bulk requests is deprecated.";
     public static final String FAILURE_STORE_STATUS_CAPABILITY = "failure_store_status";
+    public static final Set<String> STREAMS_ALLOWED_PARAMS = Set.of("timeout");
 
     private final boolean allowExplicitIndex;
     private final IncrementalBulkService bulkHandler;
@@ -113,6 +115,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             bulkRequest.timeout(request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT));
             bulkRequest.setRefreshPolicy(request.param("refresh"));
             bulkRequest.includeSourceOnError(RestUtils.getIncludeSourceOnError(request));
+            bulkRequest.streamsRestrictedParamsUsed(usesStreamsRestrictedParams(request));
             ReleasableBytesReference content = request.requiredContent();
 
             try {
@@ -140,10 +143,18 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             String waitForActiveShards = request.param("wait_for_active_shards");
             TimeValue timeout = request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT);
             String refresh = request.param("refresh");
-            return new ChunkHandler(allowExplicitIndex, request, () -> bulkHandler.newBulkRequest(waitForActiveShards, timeout, refresh));
+            return new ChunkHandler(
+                allowExplicitIndex,
+                request,
+                () -> bulkHandler.newBulkRequest(waitForActiveShards, timeout, refresh, usesStreamsRestrictedParams(request))
+            );
         }
     }
 
+    private boolean usesStreamsRestrictedParams(RestRequest request) {
+        return Sets.difference(request.params().keySet(), STREAMS_ALLOWED_PARAMS).isEmpty() == false;
+    }
+
     private static Exception parseFailureException(Exception e) {
         if (e instanceof IllegalArgumentException) {
             return e;

server/src/test/java/org/elasticsearch/rest/action/document/RestBulkActionTests.java

@@ -240,7 +240,8 @@ public void next() {
                 null,
                 null,
                 null,
-                MeterRegistry.NOOP.getLongHistogram(IncrementalBulkService.CHUNK_WAIT_TIME_HISTOGRAM_NAME)
+                MeterRegistry.NOOP.getLongHistogram(IncrementalBulkService.CHUNK_WAIT_TIME_HISTOGRAM_NAME),
+                false
             ) {
 
                 @Override