Merge remote-tracking branch 'elastic/master' into ack-as-publish-timeout

Author: ywelsch

build.gradle

@@ -131,7 +131,7 @@ task verifyVersions {
     new URL('https://repo1.maven.org/maven2/org/elasticsearch/elasticsearch/maven-metadata.xml').openStream().withStream { s ->
         xml = new XmlParser().parse(s)
     }
-    Set<Version> knownVersions = new TreeSet<>(xml.versioning.versions.version.collect { it.text() }.findAll { it ==~ /\d\.\d\.\d/ }.collect { Version.fromString(it) })
+    Set<Version> knownVersions = new TreeSet<>(xml.versioning.versions.version.collect { it.text() }.findAll { it ==~ /\d+\.\d+\.\d+/ }.collect { Version.fromString(it) })
 
     // Limit the known versions to those that should be index compatible, and are not future versions
     knownVersions = knownVersions.findAll { it.major >= bwcVersions.currentVersion.major - 1 && it.before(VersionProperties.elasticsearch) }

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

@@ -149,11 +149,11 @@ 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
+        // so now evaluates if the waitCondition is successful
         // TODO: change this to a loop?
         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)

client/rest-high-level/src/test/java/org/elasticsearch/client/ClusterClientIT.java

@@ -129,7 +129,6 @@ public void testClusterHealthYellowClusterLevel() throws IOException {
         createIndex("index2", Settings.EMPTY);
         ClusterHealthRequest request = new ClusterHealthRequest();
         request.timeout("5s");
-        request.level(ClusterHealthRequest.Level.CLUSTER);
         ClusterHealthResponse response = execute(request, highLevelClient().cluster()::health, highLevelClient().cluster()::healthAsync);
 
         assertYellowShards(response);
@@ -170,6 +169,7 @@ public void testClusterHealthYellowSpecificIndex() throws IOException {
         createIndex("index", Settings.EMPTY);
         createIndex("index2", Settings.EMPTY);
         ClusterHealthRequest request = new ClusterHealthRequest("index");
+        request.level(ClusterHealthRequest.Level.SHARDS);
         request.timeout("5s");
         ClusterHealthResponse response = execute(request, highLevelClient().cluster()::health, highLevelClient().cluster()::healthAsync);
 

client/rest-high-level/src/test/java/org/elasticsearch/client/RequestConvertersTests.java

@@ -1568,7 +1568,7 @@ public void testClusterHealth() {
             healthRequest.level(level);
             expectedParams.put("level", level.name().toLowerCase(Locale.ROOT));
         } else {
-            expectedParams.put("level", "shards");
+            expectedParams.put("level", "cluster");
         }
         if (randomBoolean()) {
             Priority priority = randomFrom(Priority.values());

distribution/src/bin/elasticsearch-cli.bat

@@ -2,7 +2,7 @@ call "%~dp0elasticsearch-env.bat" || exit /b 1
 
 if defined ES_ADDITIONAL_SOURCES (
   for %%a in ("%ES_ADDITIONAL_SOURCES:;=","%") do (
-    call %~dp0%%a
+    call "%~dp0%%a"
   )
 )
 

docs/java-rest/high-level/cluster/health.asciidoc

@@ -67,6 +67,7 @@ include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[health-request-wai
 include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[health-request-level]
 --------------------------------------------------
 <1> The level of detail of the returned health information. Accepts a `ClusterHealthRequest.Level` value.
+Default value is `cluster`.
 
 ["source","java",subs="attributes,callouts,macros"]
 --------------------------------------------------

docs/reference/aggregations/metrics/scripted-metric-aggregation.asciidoc

@@ -1,8 +1,6 @@
 [[search-aggregations-metrics-scripted-metric-aggregation]]
 === Scripted Metric Aggregation
 
-experimental[]
-
 A metric aggregation that executes using scripts to provide a metric output.
 
 Example:

docs/reference/migration/migrate_7_0/restclient.asciidoc

@@ -10,4 +10,11 @@ header, e.g. `client.index(indexRequest)` becomes
 `client.index(indexRequest, RequestOptions.DEFAULT)`.
 In case you are specifying headers
 e.g. `client.index(indexRequest, new Header("name" "value"))` becomes
-`client.index(indexRequest, RequestOptions.DEFAULT.toBuilder().addHeader("name", "value").build());`
+`client.index(indexRequest, RequestOptions.DEFAULT.toBuilder().addHeader("name", "value").build());`
+
+==== Cluster Health API default to `cluster` level
+
+The Cluster Health API used to default to `shards` level to ease migration
+from transport client that doesn't support the `level` parameter and always
+returns information including indices and shards details. The level default
+value has been aligned with the Elasticsearch default level: `cluster`.

libs/nio/src/main/java/org/elasticsearch/nio/ChannelContext.java

@@ -66,7 +66,7 @@ protected void setSelectionKey(SelectionKey selectionKey) {
      * @throws IOException during channel / context close
      */
     public void closeFromSelector() throws IOException {
-        if (closeContext.isDone() == false) {
+        if (isOpen()) {
             try {
                 rawChannel.close();
                 closeContext.complete(null);

libs/nio/src/main/java/org/elasticsearch/nio/EventHandler.java

@@ -159,8 +159,7 @@ protected void listenerException(Exception exception) {
     }
 
     /**
-     * This method is called after ready events (READ, ACCEPT, WRITE, CONNECT) have been handled for a
-     * channel.
+     * This method is called after events (READ, WRITE, CONNECT) have been handled for a channel.
      *
      * @param context that was handled
      */

libs/nio/src/main/java/org/elasticsearch/nio/NioSelector.java

@@ -43,9 +43,6 @@
  * {@link #runLoop()}, the selector will run until {@link #close()} is called. This instance handles closing
  * of channels. Users should call {@link #queueChannelClose(NioChannel)} to schedule a channel for close by
  * this selector.
- * <p>
- * Children of this class should implement the specific {@link #processKey(SelectionKey)},
- * {@link #preSelect()}, and {@link #cleanup()} functionality.
  */
 public class NioSelector implements Closeable {
 
@@ -65,7 +62,7 @@ public NioSelector(EventHandler eventHandler) throws IOException {
         this(eventHandler, Selector.open());
     }
 
-    public NioSelector(EventHandler eventHandler, Selector selector) throws IOException {
+    public NioSelector(EventHandler eventHandler, Selector selector) {
         this.selector = selector;
         this.eventHandler = eventHandler;
     }
@@ -165,7 +162,7 @@ void singleLoop() {
     }
 
     void cleanupAndCloseChannels() {
-        cleanup();
+        cleanupPendingWrites();
         channelsToClose.addAll(channelsToRegister);
         channelsToRegister.clear();
         channelsToClose.addAll(selector.keys().stream().map(sk -> (ChannelContext<?>) sk.attachment()).collect(Collectors.toList()));
@@ -234,16 +231,6 @@ void preSelect() {
         handleQueuedWrites();
     }
 
-    /**
-     * Called once as the selector is being closed.
-     */
-    void cleanup() {
-        WriteOperation op;
-        while ((op = queuedWrites.poll()) != null) {
-            executeFailedListener(op.getListener(), new ClosedSelectorException());
-        }
-    }
-
     /**
      * Queues a write operation to be handled by the event loop. This can be called by any thread and is the
      * api available for non-selector threads to schedule writes.
@@ -284,20 +271,31 @@ public void scheduleForRegistration(NioChannel channel) {
     }
 
     /**
-     * Queues a write operation directly in a channel's buffer. Channel buffers are only safe to be accessed
-     * by the selector thread. As a result, this method should only be called by the selector thread.
+     * Queues a write operation directly in a channel's buffer. If this channel does not have pending writes
+     * already, the channel will be flushed. Channel buffers are only safe to be accessed by the selector
+     * thread. As a result, this method should only be called by the selector thread. If this channel does
+     * not have pending writes already, the channel will be flushed.
      *
      * @param writeOperation to be queued in a channel's buffer
      */
-    public void queueWriteInChannelBuffer(WriteOperation writeOperation) {
+    public void writeToChannel(WriteOperation writeOperation) {
         assertOnSelectorThread();
         SocketChannelContext context = writeOperation.getChannel();
+        // If the channel does not currently have anything that is ready to flush, we should flush after
+        // the write operation is queued.
+        boolean shouldFlushAfterQueuing = context.readyForFlush() == false;
         try {
             SelectionKeyUtils.setWriteInterested(context.getSelectionKey());
             context.queueWriteOperation(writeOperation);
         } catch (Exception e) {
+            shouldFlushAfterQueuing = false;
             executeFailedListener(writeOperation.getListener(), e);
         }
+
+        if (shouldFlushAfterQueuing) {
+            handleWrite(context);
+            eventHandler.postHandling(context);
+        }
     }
 
     /**
@@ -332,6 +330,13 @@ public <V> void executeFailedListener(BiConsumer<V, Exception> listener, Excepti
         }
     }
 
+    private void cleanupPendingWrites() {
+        WriteOperation op;
+        while ((op = queuedWrites.poll()) != null) {
+            executeFailedListener(op.getListener(), new ClosedSelectorException());
+        }
+    }
+
     private void wakeup() {
         // TODO: Do we need the wakeup optimizations that some other libraries use?
         selector.wakeup();
@@ -394,7 +399,7 @@ private void handleQueuedWrites() {
         WriteOperation writeOperation;
         while ((writeOperation = queuedWrites.poll()) != null) {
             if (writeOperation.getChannel().isOpen()) {
-                queueWriteInChannelBuffer(writeOperation);
+                writeToChannel(writeOperation);
             } else {
                 executeFailedListener(writeOperation.getListener(), new ClosedChannelException());
             }

libs/nio/src/main/java/org/elasticsearch/nio/SocketChannelContext.java

@@ -135,7 +135,7 @@ public void sendMessage(Object message, BiConsumer<Void, Exception> listener) {
             return;
         }
 
-        selector.queueWriteInChannelBuffer(writeOperation);
+        selector.writeToChannel(writeOperation);
     }
 
     public void queueWriteOperation(WriteOperation writeOperation) {
@@ -164,7 +164,7 @@ protected FlushOperation getPendingFlush() {
     @Override
     public void closeFromSelector() throws IOException {
         getSelector().assertOnSelectorThread();
-        if (channel.isOpen()) {
+        if (isOpen()) {
             ArrayList<IOException> closingExceptions = new ArrayList<>(3);
             try {
                 super.closeFromSelector();

libs/nio/src/test/java/org/elasticsearch/nio/NioSelectorTests.java

@@ -262,11 +262,28 @@ public void testQueueWriteSuccessful() throws Exception {
     public void testQueueDirectlyInChannelBufferSuccessful() throws Exception {
         WriteOperation writeOperation = new FlushReadyWrite(channelContext, buffers, listener);
 
-        assertTrue((selectionKey.interestOps() & SelectionKey.OP_WRITE) == 0);
+        assertEquals(0, (selectionKey.interestOps() & SelectionKey.OP_WRITE));
 
-        selector.queueWriteInChannelBuffer(writeOperation);
+        when(channelContext.readyForFlush()).thenReturn(true);
+        selector.writeToChannel(writeOperation);
 
         verify(channelContext).queueWriteOperation(writeOperation);
+        verify(eventHandler, times(0)).handleWrite(channelContext);
+        verify(eventHandler, times(0)).postHandling(channelContext);
+        assertTrue((selectionKey.interestOps() & SelectionKey.OP_WRITE) != 0);
+    }
+
+    public void testShouldFlushIfNoPendingFlushes() throws Exception {
+        WriteOperation writeOperation = new FlushReadyWrite(channelContext, buffers, listener);
+
+        assertEquals(0, (selectionKey.interestOps() & SelectionKey.OP_WRITE));
+
+        when(channelContext.readyForFlush()).thenReturn(false);
+        selector.writeToChannel(writeOperation);
+
+        verify(channelContext).queueWriteOperation(writeOperation);
+        verify(eventHandler).handleWrite(channelContext);
+        verify(eventHandler).postHandling(channelContext);
         assertTrue((selectionKey.interestOps() & SelectionKey.OP_WRITE) != 0);
     }
 
@@ -277,10 +294,13 @@ public void testQueueDirectlyInChannelBufferSelectionKeyThrowsException() throws
         CancelledKeyException cancelledKeyException = new CancelledKeyException();
 
         when(channelContext.getSelectionKey()).thenReturn(selectionKey);
+        when(channelContext.readyForFlush()).thenReturn(false);
         when(selectionKey.interestOps(anyInt())).thenThrow(cancelledKeyException);
-        selector.queueWriteInChannelBuffer(writeOperation);
+        selector.writeToChannel(writeOperation);
 
         verify(channelContext, times(0)).queueWriteOperation(writeOperation);
+        verify(eventHandler, times(0)).handleWrite(channelContext);
+        verify(eventHandler, times(0)).postHandling(channelContext);
         verify(listener).accept(null, cancelledKeyException);
     }
 

libs/nio/src/test/java/org/elasticsearch/nio/SocketChannelContextTests.java

@@ -170,7 +170,7 @@ public void testSendMessageFromSameThreadIsQueuedInChannel() {
         when(readWriteHandler.createWriteOperation(context, buffers, listener)).thenReturn(writeOperation);
         context.sendMessage(buffers, listener);
 
-        verify(selector).queueWriteInChannelBuffer(writeOpCaptor.capture());
+        verify(selector).writeToChannel(writeOpCaptor.capture());
         WriteOperation writeOp = writeOpCaptor.getValue();
 
         assertSame(writeOperation, writeOp);

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"));
         }
     }
 }