Unmute flaky csv-spec tests with more logging on warnings context (#137089)

These tests do not reproduce, even with reported random seeds. We are unmuting them in the hope that the additional logging will bring some clarity to the situation under which they fail.

Author: craigtaverner

muted-tests.yml

@@ -360,9 +360,6 @@ tests:
 - class: org.elasticsearch.xpack.security.authc.AuthenticationServiceTests
   method: testInvalidToken
   issue: https://github.com/elastic/elasticsearch/issues/133328
-- class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
-  method: test {csv-spec:change_point.Values null column}
-  issue: https://github.com/elastic/elasticsearch/issues/133334
 - class: org.elasticsearch.xpack.search.CrossClusterAsyncSearchIT
   method: testCancelViaExpirationOnRemoteResultsWithMinimizeRoundtrips
   issue: https://github.com/elastic/elasticsearch/issues/127302
@@ -372,21 +369,12 @@ tests:
 - class: org.elasticsearch.xpack.ml.integration.ClassificationIT
   method: testWithCustomFeatureProcessors
   issue: https://github.com/elastic/elasticsearch/issues/134001
-- class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
-  method: test {csv-spec:spatial.ConvertFromStringParseError}
-  issue: https://github.com/elastic/elasticsearch/issues/134104
-- class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeForkIT
-  method: test {csv-spec:spatial_shapes.ConvertFromStringParseError}
-  issue: https://github.com/elastic/elasticsearch/issues/134254
 - class: org.elasticsearch.action.admin.cluster.state.TransportClusterStateActionTests
   method: testGetClusterStateWithDefaultProjectOnly
   issue: https://github.com/elastic/elasticsearch/issues/134450
 - class: org.elasticsearch.xpack.esql.plugin.CanMatchIT
   method: testAliasFilters
   issue: https://github.com/elastic/elasticsearch/issues/134512
-- class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
-  method: test {csv-spec:spatial.ConvertCartesianFromStringParseError}
-  issue: https://github.com/elastic/elasticsearch/issues/134635
 - class: org.elasticsearch.xpack.esql.analysis.AnalyzerTests
   method: testDenseVectorImplicitCastingKnnQueryParams
   issue: https://github.com/elastic/elasticsearch/issues/134639
@@ -444,9 +432,6 @@ tests:
 - class: org.elasticsearch.gradle.TestClustersPluginFuncTest
   method: override jdk usage via ES_JAVA_HOME for known jdk os incompatibilities
   issue: https://github.com/elastic/elasticsearch/issues/135413
-- class: org.elasticsearch.xpack.esql.qa.single_node.EsqlSpecIT
-  method: test {csv-spec:spatial_shapes.ConvertCartesianShapeFromStringParseError}
-  issue: https://github.com/elastic/elasticsearch/issues/135455
 - class: org.elasticsearch.xpack.esql.heap_attack.HeapAttackIT
   method: testAggTooManyMvLongs
   issue: https://github.com/elastic/elasticsearch/issues/135585
@@ -471,9 +456,6 @@ tests:
 - class: org.elasticsearch.test.rest.yaml.CcsCommonYamlTestSuiteIT
   method: test {p0=field_caps/10_basic/Field caps for number field with only doc values}
   issue: https://github.com/elastic/elasticsearch/issues/136244
-- class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeForkIT
-  method: test {csv-spec:string.SpaceNegative}
-  issue: https://github.com/elastic/elasticsearch/issues/136249
 - class: org.elasticsearch.xpack.restart.FullClusterRestartIT
   method: testDataStreams {cluster=UPGRADED}
   issue: https://github.com/elastic/elasticsearch/issues/136353

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/RestEsqlTestCase.java

@@ -405,10 +405,10 @@ public void testCSVNoHeaderMode() throws IOException {
         options.addHeader("Accept", "text/csv; header=absent");
         request.setOptions(options);
         Response response = performRequest(request);
-        assertWarnings(response, new AssertWarnings.NoWarnings());
         HttpEntity entity = response.getEntity();
         String actual = Streams.copyToString(new InputStreamReader(entity.getContent(), StandardCharsets.UTF_8));
         assertEquals("keyword0,0\r\n", actual);
+        assertWarnings(response, new AssertWarnings.NoWarnings(), actual);
     }
 
     public void testOutOfRangeComparisons() throws IOException {
@@ -1433,7 +1433,7 @@ public static Map<String, Object> runEsqlSync(
         if (supportsAsyncHeadersFix) {
             assertNoAsyncHeaders(response);
         }
-        assertWarnings(response, assertWarnings);
+        assertWarnings(response, assertWarnings, json);
 
         return json;
     }
@@ -1485,7 +1485,7 @@ public static Map<String, Object> runEsqlAsync(
             if (profileLogger != null) {
                 profileLogger.extractProfile(json, profileEnabled);
             }
-            assertWarnings(response, assertWarnings);
+            assertWarnings(response, assertWarnings, json);
             json.remove("is_running"); // remove this to not mess up later map assertions
             return Collections.unmodifiableMap(json);
         } else {
@@ -1498,7 +1498,7 @@ public static Map<String, Object> runEsqlAsync(
                 if (profileLogger != null) {
                     profileLogger.extractProfile(json, profileEnabled);
                 }
-                assertWarnings(response, assertWarnings);
+                assertWarnings(response, assertWarnings, json);
                 // we already have the results, but let's remember them so that we can compare to async get
                 initialColumns = json.get("columns");
                 initialValues = json.get("values");
@@ -1538,7 +1538,7 @@ public static Map<String, Object> runEsqlAsync(
         if (profileLogger != null) {
             profileLogger.extractProfile(result, profileEnabled);
         }
-        assertWarnings(response, assertWarnings);
+        assertWarnings(response, assertWarnings, result);
         assertDeletable(id);
         return removeAsyncProperties(result);
     }
@@ -1801,12 +1801,12 @@ static String runEsqlAsTextWithFormat(RequestObjectBuilder builder, String forma
         }
 
         Response response = performRequest(request);
-        assertWarnings(response, new AssertWarnings.NoWarnings());
         HttpEntity entity = response.getEntity();
 
         // get the content, it could be empty because the request might have not completed
         String initialValue = Streams.copyToString(new InputStreamReader(entity.getContent(), StandardCharsets.UTF_8));
         String id = response.getHeader("X-Elasticsearch-Async-Id");
+        assertWarnings(response, new AssertWarnings.NoWarnings(), initialValue);
 
         if (mode == SYNC) {
             assertThat(id, is(emptyOrNullString()));
@@ -1855,10 +1855,10 @@ static String runEsqlAsTextWithFormat(RequestObjectBuilder builder, String forma
             // if `addParam` is false, `options` will already have an `Accept` header
             getRequest.setOptions(options);
             response = performRequest(getRequest);
-            assertWarnings(response, new AssertWarnings.NoWarnings());
             entity = response.getEntity();
         }
         String newValue = Streams.copyToString(new InputStreamReader(entity.getContent(), StandardCharsets.UTF_8));
+        assertWarnings(response, new AssertWarnings.NoWarnings(), newValue);
 
         // assert initial contents, if any, are the same as async get contents
         if (initialValue != null && initialValue.isEmpty() == false) {
@@ -1911,13 +1911,13 @@ static void assertNotPartial(Map<String, Object> answer) {
         assertThat(reason, answer.get("is_partial"), anyOf(nullValue(), is(false)));
     }
 
-    private static void assertWarnings(Response response, AssertWarnings assertWarnings) {
+    private static void assertWarnings(Response response, AssertWarnings assertWarnings, Object context) {
         List<String> warnings = new ArrayList<>(response.getWarnings());
         warnings.removeAll(mutedWarnings());
         if (shouldLog()) {
             LOGGER.info("RESPONSE warnings (after muted)={}", warnings);
         }
-        assertWarnings.assertWarnings(warnings);
+        assertWarnings.assertWarnings(warnings, context);
     }
 
     private static Set<String> mutedWarnings() {

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/AssertWarnings.java

@@ -18,34 +18,55 @@
  * How should we assert the warnings returned by ESQL.
  */
 public interface AssertWarnings {
-    void assertWarnings(List<String> warnings);
+    void assertWarnings(List<String> warnings, Object context);
+
+    default String contextMessage(Object context) {
+        if (context == null) {
+            return "";
+        }
+        StringBuilder contextBuilder = new StringBuilder();
+        contextBuilder.append("Context: ").append(context);
+        if (contextBuilder.length() > 1000) {
+            contextBuilder.setLength(1000);
+            contextBuilder.append("...(truncated)");
+        }
+        contextBuilder.append("\n");
+        return contextBuilder.toString();
+    }
 
     record NoWarnings() implements AssertWarnings {
         @Override
-        public void assertWarnings(List<String> warnings) {
-            assertMap(warnings.stream().sorted().toList(), matchesList());
+        public void assertWarnings(List<String> warnings, Object context) {
+            assertMap(contextMessage(context), warnings.stream().sorted().toList(), matchesList());
         }
     }
 
     record ExactStrings(List<String> expected) implements AssertWarnings {
         @Override
-        public void assertWarnings(List<String> warnings) {
-            assertMap(warnings.stream().sorted().toList(), matchesList(expected.stream().sorted().toList()));
+        public void assertWarnings(List<String> warnings, Object context) {
+            assertMap(contextMessage(context), warnings.stream().sorted().toList(), matchesList(expected.stream().sorted().toList()));
         }
     }
 
     record DeduplicatedStrings(List<String> expected) implements AssertWarnings {
         @Override
-        public void assertWarnings(List<String> warnings) {
-            assertMap(warnings.stream().sorted().distinct().toList(), matchesList(expected.stream().sorted().toList()));
+        public void assertWarnings(List<String> warnings, Object context) {
+            assertMap(
+                contextMessage(context),
+                warnings.stream().sorted().distinct().toList(),
+                matchesList(expected.stream().sorted().toList())
+            );
         }
     }
 
     record AllowedRegexes(List<Pattern> expected) implements AssertWarnings {
         @Override
-        public void assertWarnings(List<String> warnings) {
+        public void assertWarnings(List<String> warnings, Object context) {
             for (String warning : warnings) {
-                assertTrue("Unexpected warning: " + warning, expected.stream().anyMatch(x -> x.matcher(warning).matches()));
+                assertTrue(
+                    contextMessage(context) + "Unexpected warning: " + warning,
+                    expected.stream().anyMatch(x -> x.matcher(warning).matches())
+                );
             }
         }
     }

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

@@ -391,7 +391,7 @@ private void doTest() throws Exception {
 
             var log = logResults() ? LOGGER : null;
             assertResults(expected, actualResults, testCase.ignoreOrder, log);
-            assertWarnings(actualResults.responseHeaders().getOrDefault("Warning", List.of()));
+            assertWarnings(actualResults.responseHeaders().getOrDefault("Warning", List.of()), actualResults);
         } finally {
             Releasables.close(() -> Iterators.map(actualResults.pages().iterator(), p -> p::releaseBlocks));
             // Give the breaker service some time to clear in case we got results before the rest of the driver had cleaned up
@@ -717,7 +717,7 @@ private void opportunisticallyAssertPlanSerialization(PhysicalPlan plan) {
         SerializationTestUtils.assertSerialization(plan, configuration);
     }
 
-    private void assertWarnings(List<String> warnings) {
+    private void assertWarnings(List<String> warnings, Object context) {
         List<String> normalized = new ArrayList<>(warnings.size());
         for (String w : warnings) {
             String normW = HeaderWarning.extractWarningValueFromWarningHeader(w, false);
@@ -726,7 +726,7 @@ private void assertWarnings(List<String> warnings) {
                 normalized.add(normW);
             }
         }
-        testCase.assertWarnings(false).assertWarnings(normalized);
+        testCase.assertWarnings(false).assertWarnings(normalized, context);
     }
 
     PlanRunner planRunner(BigArrays bigArrays, TestPhysicalOperationProviders physicalOperationProviders) {