Reduce inactive sink interval in VectorSimilarityFunctionsIT

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
@@ -488,9 +479,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;