Fix AntFixture waiting condition

buildSrc/src/main/groovy/org/elasticsearch/gradle/test/AntFixture.groovy

@@ -149,11 +149,10 @@ public class AntFixture extends AntTask implements Fixture {
         }
 
         // the process is started (has a pid) and is bound to a network interface
-        // so now wait undil the waitCondition has been met
-        // TODO: change this to a loop?
+        // so now evaluates if the waitCondition is successful
         boolean success
         try {
-            success = waitCondition(this, ant) == false
+            success = waitCondition(this, ant)
         } catch (Exception e) {
             String msg = "Wait condition caught exception for ${name}"
             logger.error(msg, e)

modules/reindex/build.gradle

@@ -121,6 +121,11 @@ if (Os.isFamily(Os.FAMILY_WINDOWS)) {
            baseDir,
            unzip.temporaryDir,
            version == '090'
+      waitCondition = { fixture, ant ->
+        // the fixture writes the ports file when Elasticsearch's HTTP service
+        // is ready, so we can just wait for the file to exist
+        return fixture.portsFile.exists()
+      }
     }
     integTest.dependsOn fixture
     integTestRunner {

modules/repository-url/src/test/java/org/elasticsearch/repositories/url/URLFixture.java

@@ -39,6 +39,7 @@
 import java.util.Map;
 import java.util.Objects;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singleton;
 import static java.util.Collections.singletonMap;
@@ -67,7 +68,6 @@ public static void main(String[] args) throws Exception {
             writeFile(workingDirectory, "ports", addressAndPort);
 
             // Exposes the repository over HTTP
-            final String url = "http://" + addressAndPort;
             httpServer.createContext("/", new ResponseHandler(dir(args[1])));
             httpServer.start();
 
@@ -110,7 +110,13 @@ static class ResponseHandler implements HttpHandler {
         @Override
         public void handle(HttpExchange exchange) throws IOException {
             Response response;
-            if ("GET".equalsIgnoreCase(exchange.getRequestMethod())) {
+
+            final String userAgent = exchange.getRequestHeaders().getFirst("User-Agent");
+            if (userAgent != null && userAgent.startsWith("Apache Ant")) {
+                // This is a request made by the AntFixture, just reply "OK"
+                response = new Response(RestStatus.OK, emptyMap(), "text/plain; charset=utf-8", "OK".getBytes(UTF_8));
+
+            } else if ("GET".equalsIgnoreCase(exchange.getRequestMethod())) {
                 String path = exchange.getRequestURI().toString();
                 if (path.length() > 0 && path.charAt(0) == '/') {
                     path = path.substring(1);
@@ -125,13 +131,13 @@ public void handle(HttpExchange exchange) throws IOException {
                         Map<String, String> headers = singletonMap("Content-Length", String.valueOf(content.length));
                         response = new Response(RestStatus.OK, headers, "application/octet-stream", content);
                     } else {
-                        response = new Response(RestStatus.NOT_FOUND, emptyMap(), "text/plain", new byte[0]);
+                        response = new Response(RestStatus.NOT_FOUND, emptyMap(), "text/plain; charset=utf-8", new byte[0]);
                     }
                 } else {
-                    response = new Response(RestStatus.FORBIDDEN, emptyMap(), "text/plain", new byte[0]);
+                    response = new Response(RestStatus.FORBIDDEN, emptyMap(), "text/plain; charset=utf-8", new byte[0]);
                 }
             } else {
-                response = new Response(RestStatus.INTERNAL_SERVER_ERROR, emptyMap(), "text/plain",
+                response = new Response(RestStatus.INTERNAL_SERVER_ERROR, emptyMap(), "text/plain; charset=utf-8",
                     "Unsupported HTTP method".getBytes(StandardCharsets.UTF_8));
             }
             exchange.sendResponseHeaders(response.status.getStatus(), response.body.length);

plugins/examples/rest-handler/src/test/java/org/elasticsearch/example/resthandler/ExampleFixtureIT.java

@@ -23,25 +23,41 @@
 import org.elasticsearch.test.ESTestCase;
 
 import java.io.BufferedReader;
+import java.io.BufferedWriter;
 import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
 import java.net.InetAddress;
 import java.net.Socket;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
-import java.util.Objects;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.Matchers.hasItems;
 
 public class ExampleFixtureIT extends ESTestCase {
 
     public void testExample() throws Exception {
-        final String stringAddress = Objects.requireNonNull(System.getProperty("external.address"));
-        final URL url = new URL("http://" + stringAddress);
+        final String externalAddress = System.getProperty("external.address");
+        assertNotNull("External address must not be null", externalAddress);
 
+        final URL url = new URL("http://" + externalAddress);
         final InetAddress address = InetAddress.getByName(url.getHost());
         try (
             Socket socket = new MockSocket(address, url.getPort());
+            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(), StandardCharsets.UTF_8));
             BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8))
         ) {
-            assertEquals("TEST", reader.readLine());
+            writer.write("GET / HTTP/1.1\r\n");
+            writer.write("Host: elastic.co\r\n\r\n");
+            writer.flush();
+
+            final List<String> lines = new ArrayList<>();
+            String line;
+            while ((line = reader.readLine()) != null) {
+                lines.add(line);
+            }
+            assertThat(lines, hasItems("HTTP/1.1 200 OK", "TEST"));
         }
     }
 }

plugins/repository-azure/qa/microsoft-azure-storage/src/test/java/org/elasticsearch/repositories/azure/AzureStorageFixture.java

@@ -24,6 +24,8 @@
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.mocksocket.MockHttpServer;
+import org.elasticsearch.repositories.azure.AzureStorageTestServer.Response;
+import org.elasticsearch.rest.RestStatus;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -39,6 +41,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Collections.emptyMap;
 import static java.util.Collections.singleton;
 import static java.util.Collections.singletonList;
 
@@ -121,7 +125,16 @@ public void handle(HttpExchange exchange) throws IOException {
             ByteArrayOutputStream out = new ByteArrayOutputStream();
             Streams.copy(exchange.getRequestBody(), out);
 
-            final AzureStorageTestServer.Response response = server.handle(method, path, query, headers, out.toByteArray());
+            Response response = null;
+
+            final String userAgent = exchange.getRequestHeaders().getFirst("User-Agent");
+            if (userAgent != null && userAgent.startsWith("Apache Ant")) {
+                // This is a request made by the AntFixture, just reply "OK"
+                response = new Response(RestStatus.OK, emptyMap(), "text/plain; charset=utf-8", "OK".getBytes(UTF_8));
+            } else {
+                // Otherwise simulate a S3 response
+                response = server.handle(method, path, query, headers, out.toByteArray());
+            }
 
             Map<String, List<String>> responseHeaders = exchange.getResponseHeaders();
             responseHeaders.put("Content-Type", singletonList(response.contentType));

plugins/repository-gcs/qa/google-cloud-storage/src/test/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageFixture.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.core.internal.io.Streams;
 import org.elasticsearch.mocksocket.MockHttpServer;
 import org.elasticsearch.repositories.gcs.GoogleCloudStorageTestServer.Response;
+import org.elasticsearch.rest.RestStatus;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -40,6 +41,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Collections.emptyMap;
 import static java.util.Collections.singleton;
 import static java.util.Collections.singletonList;
 
@@ -123,7 +126,16 @@ public void handle(HttpExchange exchange) throws IOException {
             ByteArrayOutputStream out = new ByteArrayOutputStream();
             Streams.copy(exchange.getRequestBody(), out);
 
-            final Response storageResponse = storageServer.handle(method, path, query, headers, out.toByteArray());
+            Response storageResponse = null;
+
+            final String userAgent = exchange.getRequestHeaders().getFirst("User-Agent");
+            if (userAgent != null && userAgent.startsWith("Apache Ant")) {
+                // This is a request made by the AntFixture, just reply "OK"
+                storageResponse = new Response(RestStatus.OK, emptyMap(), "text/plain; charset=utf-8", "OK".getBytes(UTF_8));
+            } else {
+                // Otherwise simulate a S3 response
+                storageResponse = storageServer.handle(method, path, query, headers, out.toByteArray());
+            }
 
             Map<String, List<String>> responseHeaders = exchange.getResponseHeaders();
             responseHeaders.put("Content-Type", singletonList(storageResponse.contentType));

plugins/repository-hdfs/build.gradle

@@ -116,6 +116,11 @@ for (String fixtureName : ['hdfsFixture', 'haHdfsFixture', 'secureHdfsFixture',
     dependsOn project.configurations.hdfsFixture
     executable = new File(project.runtimeJavaHome, 'bin/java')
     env 'CLASSPATH', "${ -> project.configurations.hdfsFixture.asPath }"
+    waitCondition = { fixture, ant ->
+      // the hdfs.MiniHDFS fixture writes the ports file when
+      // it's ready, so we can just wait for the file to exist
+      return fixture.portsFile.exists()
+    }
 
     final List<String> miniHDFSArgs = []
 

plugins/repository-s3/qa/amazon-s3/src/test/java/org/elasticsearch/repositories/s3/AmazonS3Fixture.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.mocksocket.MockHttpServer;
 import org.elasticsearch.repositories.s3.AmazonS3TestServer.Response;
+import org.elasticsearch.rest.RestStatus;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -40,6 +41,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Collections.emptyMap;
 import static java.util.Collections.singleton;
 import static java.util.Collections.singletonList;
 
@@ -122,7 +125,16 @@ public void handle(HttpExchange exchange) throws IOException {
             ByteArrayOutputStream out = new ByteArrayOutputStream();
             Streams.copy(exchange.getRequestBody(), out);
 
-            final Response storageResponse = storageServer.handle(method, path, query, headers, out.toByteArray());
+            Response storageResponse = null;
+
+            final String userAgent = exchange.getRequestHeaders().getFirst("User-Agent");
+            if (userAgent != null && userAgent.startsWith("Apache Ant")) {
+                // This is a request made by the AntFixture, just reply "OK"
+                storageResponse = new Response(RestStatus.OK, emptyMap(), "text/plain; charset=utf-8", "OK".getBytes(UTF_8));
+            } else {
+                // Otherwise simulate a S3 response
+                storageResponse = storageServer.handle(method, path, query, headers, out.toByteArray());
+            }
 
             Map<String, List<String>> responseHeaders = exchange.getResponseHeaders();
             responseHeaders.put("Content-Type", singletonList(storageResponse.contentType));

test/fixtures/example-fixture/src/main/java/example/ExampleTestFixture.java

@@ -19,14 +19,12 @@
 
 package example;
 
+import com.sun.net.httpserver.HttpServer;
+
 import java.lang.management.ManagementFactory;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.AsynchronousServerSocketChannel;
-import java.nio.channels.AsynchronousSocketChannel;
-import java.nio.channels.CompletionHandler;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -41,9 +39,9 @@ public static void main(String args[]) throws Exception {
             throw new IllegalArgumentException("ExampleTestFixture <logDirectory>");
         }
         Path dir = Paths.get(args[0]);
-        AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel
-                .open()
-                .bind(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0));
+
+        final InetSocketAddress socketAddress = new InetSocketAddress(InetAddress.getLoopbackAddress(), 0);
+        final HttpServer httpServer = HttpServer.create(socketAddress, 0);
 
         // write pid file
         Path tmp = Files.createTempFile(dir, null, null);
@@ -53,29 +51,26 @@ public static void main(String args[]) throws Exception {
 
         // write port file
         tmp = Files.createTempFile(dir, null, null);
-        InetSocketAddress bound = (InetSocketAddress) server.getLocalAddress();
+        InetSocketAddress bound = httpServer.getAddress();
         if (bound.getAddress() instanceof Inet6Address) {
             Files.write(tmp, Collections.singleton("[" + bound.getHostString() + "]:" + bound.getPort()));
         } else {
             Files.write(tmp, Collections.singleton(bound.getHostString() + ":" + bound.getPort()));
         }
         Files.move(tmp, dir.resolve("ports"), StandardCopyOption.ATOMIC_MOVE);
 
+        final byte[] response = "TEST\n".getBytes(StandardCharsets.UTF_8);
+
         // go time
-        server.accept(null, new CompletionHandler<AsynchronousSocketChannel,Void>() {
-            @Override
-            public void completed(AsynchronousSocketChannel socket, Void attachment) {
-                server.accept(null, this);
-                try (AsynchronousSocketChannel ch = socket) {
-                    ch.write(ByteBuffer.wrap("TEST\n".getBytes(StandardCharsets.UTF_8))).get();
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                }
+        httpServer.createContext("/", exchange -> {
+            try {
+                exchange.sendResponseHeaders(200, response.length);
+                exchange.getResponseBody().write(response);
+            } finally {
+                exchange.close();
             }
-
-            @Override
-            public void failed(Throwable exc, Void attachment) {}
         });
+        httpServer.start();
 
         // wait forever, until you kill me
         Thread.sleep(Long.MAX_VALUE);