Reduce inactive sink interval in VectorSimilarityFunctionsIT (elastic#132288)

testDifferentDimensions fails the final driver on the coordinator, 
causing the entire request to be cancelled. If the exchange sink is
opened on a remote node but the compute request hasn't been sent, we
currently cannot remove the exchange sink. To address this, we reduce
the inactive sinks interval to remove inactive sinks more quickly. I
also considered bypassing cancellation to close or remove the exchange
sink in these cases, but since it's a larger change, I decided to defer
it.

Closes elastic#131361
Closes elastic#131362
Closes elastic#131363
Closes elastic#131845

Author: dnhatn

muted-tests.yml

@@ -413,15 +413,6 @@ tests:
 - class: org.elasticsearch.test.rest.yaml.RcsCcsCommonYamlTestSuiteIT
   method: test {p0=search/110_field_collapsing/field collapsing, inner_hits and maxConcurrentGroupRequests}
   issue: https://github.com/elastic/elasticsearch/issues/131348
-- class: org.elasticsearch.xpack.esql.vector.VectorSimilarityFunctionsIT
-  method: testSimilarityBetweenConstantVectors {functionName=v_cosine similarityFunction=COSINE}
-  issue: https://github.com/elastic/elasticsearch/issues/131361
-- class: org.elasticsearch.xpack.esql.vector.VectorSimilarityFunctionsIT
-  method: testDifferentDimensions {functionName=v_cosine similarityFunction=COSINE}
-  issue: https://github.com/elastic/elasticsearch/issues/131362
-- class: org.elasticsearch.xpack.esql.vector.VectorSimilarityFunctionsIT
-  method: testSimilarityBetweenConstantVectorAndField {functionName=v_cosine similarityFunction=COSINE}
-  issue: https://github.com/elastic/elasticsearch/issues/131363
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=ml/delete_expired_data/Test delete expired data with body parameters}
   issue: https://github.com/elastic/elasticsearch/issues/131364
@@ -485,9 +476,6 @@ tests:
 - class: org.elasticsearch.index.mapper.vectors.DenseVectorFieldMapperTests
   method: testUpdates
   issue: https://github.com/elastic/elasticsearch/issues/131795
-- class: org.elasticsearch.xpack.esql.vector.VectorSimilarityFunctionsIT
-  method: testDifferentDimensions {functionName=v_dot_product similarityFunction=DOT_PRODUCT}
-  issue: https://github.com/elastic/elasticsearch/issues/131845
 - class: org.elasticsearch.xpack.restart.FullClusterRestartIT
   method: testWatcherWithApiKey {cluster=UPGRADED}
   issue: https://github.com/elastic/elasticsearch/issues/131964

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/vector/VectorSimilarityFunctionsIT.java

@@ -14,6 +14,10 @@
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.compute.operator.exchange.ExchangeService;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentFactory;
 import org.elasticsearch.xpack.esql.EsqlClientException;
@@ -28,6 +32,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.Locale;
 
@@ -55,6 +60,22 @@ public static Iterable<Object[]> parameters() throws Exception {
         return params;
     }
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal, otherSettings))
+            // testDifferentDimensions fails the final driver on the coordinator, leading to cancellation of the entire request.
+            // If the exchange sink is opened on a remote node but the compute request hasn't been sent, we cannot close the exchange
+            // sink (for now).Here, we reduce the inactive sinks interval to ensure those inactive sinks are removed quickly.
+            .put(ExchangeService.INACTIVE_SINKS_INTERVAL_SETTING, TimeValue.timeValueMillis(between(3000, 4000)))
+            .build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return CollectionUtils.appendToCopy(super.nodePlugins(), InternalExchangePlugin.class);
+    }
+
     private final String functionName;
     private final SimilarityEvaluatorFunction similarityFunction;
     private int numDims;