Revert stop, more lenient error-handling in handle, comment to explain why

Author: ldematte

test/external-modules/apm-integration/src/javaRestTest/java/org/elasticsearch/test/apmintegration/MetricsApmIT.java

@@ -129,7 +129,6 @@ public void testApmIntegration() throws Exception {
         var remainingAssertions = Stream.concat(valueAssertions.keySet().stream(), histogramAssertions.keySet().stream())
             .collect(Collectors.joining(","));
         assertTrue("Timeout when waiting for assertions to complete. Remaining assertions to match: " + remainingAssertions, completed);
-        mockApmServer.stop();
     }
 
     private <T> Map.Entry<String, Predicate<Map<String, Object>>> assertion(

test/external-modules/apm-integration/src/javaRestTest/java/org/elasticsearch/test/apmintegration/RecordingApmServer.java

@@ -15,7 +15,6 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.core.SuppressForbidden;
-import org.elasticsearch.xcontent.spi.XContentProvider;
 import org.junit.rules.ExternalResource;
 
 import java.io.BufferedReader;
@@ -34,8 +33,6 @@
 public class RecordingApmServer extends ExternalResource {
     private static final Logger logger = LogManager.getLogger(RecordingApmServer.class);
 
-    private static final XContentProvider.FormatProvider XCONTENT = XContentProvider.provider().getJsonXContent();
-
     final ArrayBlockingQueue<String> received = new ArrayBlockingQueue<>(1000);
 
     private static HttpServer server;
@@ -73,14 +70,11 @@ private Thread consumerThread() {
 
     @Override
     protected void after() {
+        running = false;
         server.stop(1);
         consumer = null;
     }
 
-    void stop() {
-        running = false;
-    }
-
     private void handle(HttpExchange exchange) throws IOException {
         try (exchange) {
             if (running) {
@@ -92,15 +86,22 @@ private void handle(HttpExchange exchange) throws IOException {
                         }
                     }
 
-                } catch (RuntimeException e) {
-                    logger.warn("failed to parse request", e);
+                } catch (Throwable t) {
+                    // The lifetime of HttpServer makes message handling "brittle": we need to start handling and recording received
+                    // messages before the test starts running. We should also stop handling them before the test ends (and the test
+                    // cluster is torn down), or we may run into IOException as the communication channel is interrupted.
+                    // Coordinating the lifecycle of the mock HttpServer and of the test ES cluster is difficult and error-prone, so
+                    // we just handle Throwable and don't care (log, but don't care): if we have an error in communicating to/from
+                    // the mock server while the test is running, the test would fail anyway as the expected messages will not arrive, and
+                    // if we have an error outside the test scope (before or after) that is OK.
+                    logger.warn("failed to parse request", t);
                 }
             }
             exchange.sendResponseHeaders(201, 0);
         }
     }
 
-    private List<String> readJsonMessages(InputStream input) throws IOException {
+    private List<String> readJsonMessages(InputStream input) {
         // parse NDJSON
         return new BufferedReader(new InputStreamReader(input, StandardCharsets.UTF_8)).lines().toList();
     }

test/external-modules/apm-integration/src/javaRestTest/java/org/elasticsearch/test/apmintegration/TracesApmIT.java

@@ -94,8 +94,6 @@ public void testApmIntegration() throws Exception {
         var completed = finished.await(30, TimeUnit.SECONDS);
         assertTrue("Timeout when waiting for assertions to complete", completed);
         assertThat(assertions, equalTo(Collections.emptySet()));
-
-        mockApmServer.stop();
     }
 
     private boolean isTransactionTraceMessage(Map<String, Object> apmMessage) {