Merge branch 'master' into snapshot-lifecycle-management

Author: elasticmachine

buildSrc/src/main/groovy/org/elasticsearch/gradle/BuildPlugin.groovy

@@ -825,7 +825,7 @@ class BuildPlugin implements Plugin<Project> {
 
                 test.executable = "${ext.get('runtimeJavaHome')}/bin/java"
                 test.workingDir = project.file("${project.buildDir}/testrun/${test.name}")
-                test.maxParallelForks = project.rootProject.extensions.getByType(ExtraPropertiesExtension).get('defaultParallel') as Integer
+                test.maxParallelForks = System.getProperty('tests.jvms', project.rootProject.extensions.extraProperties.get('defaultParallel').toString()) as Integer
 
                 test.exclude '**/*$*.class'
 

buildSrc/src/main/groovy/org/elasticsearch/gradle/doc/RestTestsFromSnippetsTask.groovy

@@ -353,7 +353,7 @@ public class RestTestsFromSnippetsTask extends SnippetsTask {
 
         private void testSetup(Snippet snippet) {
             if (lastDocsPath == snippet.path) {
-                throw new InvalidUserDataException("$snippet: wasn't first")
+                throw new InvalidUserDataException("$snippet: wasn't first. TESTSETUP can only be used in the first snippet of a document.")
             }
             setupCurrent(snippet)
             current.println('---')

buildSrc/src/main/java/org/elasticsearch/gradle/testclusters/ElasticsearchNode.java

@@ -46,6 +46,8 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.LineNumberReader;
+
 import java.io.UncheckedIOException;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
@@ -61,6 +63,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -82,13 +85,22 @@ public class ElasticsearchNode implements TestClusterConfiguration {
     private static final Logger LOGGER = Logging.getLogger(ElasticsearchNode.class);
     private static final int ES_DESTROY_TIMEOUT = 20;
     private static final TimeUnit ES_DESTROY_TIMEOUT_UNIT = TimeUnit.SECONDS;
+
     private static final int NODE_UP_TIMEOUT = 2;
     private static final TimeUnit NODE_UP_TIMEOUT_UNIT = TimeUnit.MINUTES;
     private static final int ADDITIONAL_CONFIG_TIMEOUT = 15;
     private static final TimeUnit ADDITIONAL_CONFIG_TIMEOUT_UNIT = TimeUnit.SECONDS;
     private static final List<String> OVERRIDABLE_SETTINGS = Arrays.asList(
         "path.repo",
         "discovery.seed_providers"
+
+    );
+
+    private static final int TAIL_LOG_MESSAGES_COUNT = 40;
+    private static final List<String> MESSAGES_WE_DONT_CARE_ABOUT = Arrays.asList(
+        "Option UseConcMarkSweepGC was deprecated",
+        "is a pre-release version of Elasticsearch",
+        "max virtual memory areas vm.max_map_count"
     );
 
     private final String path;
@@ -693,14 +705,73 @@ private void logProcessInfo(String prefix, ProcessHandle.Info info) {
     }
 
     private void logFileContents(String description, Path from) {
-        LOGGER.error("{} `{}`", description, this);
-        try (Stream<String> lines = Files.lines(from, StandardCharsets.UTF_8)) {
-            lines
-                .map(line -> "  " + line)
-                .forEach(LOGGER::error);
+        final Map<String, Integer> errorsAndWarnings = new LinkedHashMap<>();
+        LinkedList<String> ring = new LinkedList<>();
+        try (LineNumberReader reader = new LineNumberReader(Files.newBufferedReader(from))) {
+            for (String line = reader.readLine(); line != null ; line = reader.readLine()) {
+                final String lineToAdd;
+                if (ring.isEmpty()) {
+                    lineToAdd = line;
+                } else {
+                    if (line.startsWith("[")) {
+                        lineToAdd = line;
+                        // check to see if the previous message (possibly combined from multiple lines) was an error or
+                        // warning as we want to show all of them
+                        String previousMessage = normalizeLogLine(ring.getLast());
+                        if (MESSAGES_WE_DONT_CARE_ABOUT.stream().noneMatch(previousMessage::contains) &&
+                            (previousMessage.contains("ERROR") || previousMessage.contains("WARN"))) {
+                            errorsAndWarnings.put(
+                                previousMessage,
+                                errorsAndWarnings.getOrDefault(previousMessage, 0) + 1
+                            );
+                        }
+                    } else {
+                        // We combine multi line log messages to make sure we never break exceptions apart
+                        lineToAdd = ring.removeLast() + "\n" + line;
+                    }
+                }
+                ring.add(lineToAdd);
+                if (ring.size() >= TAIL_LOG_MESSAGES_COUNT) {
+                    ring.removeFirst();
+                }
+            }
         } catch (IOException e) {
             throw new UncheckedIOException("Failed to tail log " + this, e);
         }
+
+        if (errorsAndWarnings.isEmpty() == false || ring.isEmpty() == false) {
+            LOGGER.error("\n=== {} `{}` ===", description, this);
+        }
+        if (errorsAndWarnings.isEmpty() == false) {
+            LOGGER.lifecycle("\n»    ↓ errors and warnings from " + from + " ↓");
+            errorsAndWarnings.forEach((message, count) -> {
+                LOGGER.lifecycle("» " + message.replace("\n", "\n»  "));
+                if (count > 1) {
+                    LOGGER.lifecycle("»   ↑ repeated " + count + " times ↑");
+                }
+            });
+        }
+
+        ring.removeIf(line -> MESSAGES_WE_DONT_CARE_ABOUT.stream().anyMatch(line::contains));
+
+        if (ring.isEmpty() == false) {
+            LOGGER.lifecycle("»   ↓ last " + TAIL_LOG_MESSAGES_COUNT + " non error or warning messages from " + from + " ↓");
+            ring.forEach(message -> {
+                if (errorsAndWarnings.containsKey(normalizeLogLine(message)) == false) {
+                    LOGGER.lifecycle("» " + message.replace("\n", "\n»  "));
+                }
+            });
+        }
+    }
+
+    private String normalizeLogLine(String line) {
+        if (line.contains("ERROR")) {
+            return line.substring(line.indexOf("ERROR"));
+        }
+        if (line.contains("WARN")) {
+            return line.substring(line.indexOf("WARN"));
+        }
+        return line;
     }
 
     private void waitForProcessToExit(ProcessHandle processHandle) {

buildSrc/src/main/java/org/elasticsearch/gradle/testfixtures/TestFixturesPlugin.java

@@ -122,7 +122,7 @@ public void apply(Project project) {
 
         extension.fixtures
             .matching(fixtureProject -> fixtureProject.equals(project) == false)
-            .all(fixtureProject ->  project.evaluationDependsOn(fixtureProject.getPath()));
+            .all(fixtureProject -> project.evaluationDependsOn(fixtureProject.getPath()));
 
         conditionTaskByType(tasks, extension, Test.class);
         conditionTaskByType(tasks, extension, getTaskClass("org.elasticsearch.gradle.test.RestIntegTestTask"));

buildSrc/src/test/java/org/elasticsearch/gradle/EmptyDirTaskTests.java

@@ -21,6 +21,8 @@
 import java.io.File;
 import java.io.IOException;
 
+import com.carrotsearch.randomizedtesting.RandomizedTest;
+import org.apache.tools.ant.taskdefs.condition.Os;
 import org.elasticsearch.gradle.test.GradleUnitTestCase;
 import org.gradle.api.Project;
 import org.gradle.testfixtures.ProjectBuilder;
@@ -49,6 +51,8 @@ public void testCreateEmptyDir() throws Exception {
     }
 
     public void testCreateEmptyDirNoPermissions() throws Exception {
+        RandomizedTest.assumeFalse("Functionality is Unix specific", Os.isFamily(Os.FAMILY_WINDOWS));
+
         Project project = ProjectBuilder.builder().build();
         EmptyDirTask emptyDirTask = project.getTasks().create("emptyDirTask", EmptyDirTask.class);
         emptyDirTask.setDirMode(0000);

buildSrc/src/test/java/org/elasticsearch/gradle/testclusters/TestClustersPluginIT.java

@@ -68,9 +68,9 @@ public void testUseClusterByTwo() {
     public void testUseClusterByUpToDateTask() {
         // Run it once, ignoring the result and again to make sure it's considered up to date.
         // Gradle randomly considers tasks without inputs and outputs as as up-to-date or success on the first run
-        getTestClustersRunner(":upToDate1", ":upToDate2").build();
-        BuildResult result = getTestClustersRunner(":upToDate1", ":upToDate2").build();
-        assertTaskUpToDate(result, ":upToDate1", ":upToDate2");
+        getTestClustersRunner(":upToDate1").build();
+        BuildResult result = getTestClustersRunner(":upToDate1").build();
+        assertTaskUpToDate(result, ":upToDate1");
         assertNotStarted(result);
     }
 

buildSrc/src/testKit/testclusters/build.gradle

@@ -105,10 +105,10 @@ task printLog {
 
 task upToDate1 {
     useCluster testClusters.myTestCluster
-}
-
-task upToDate2 {
-    useCluster testClusters.myTestCluster
+    outputs.upToDateWhen { true }
+    doLast {
+        println "Some task action"
+    }
 }
 
 task skipped1 {

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/TransportNoopBulkAction.java

@@ -32,15 +32,13 @@
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 
-import java.util.function.Supplier;
-
 public class TransportNoopBulkAction extends HandledTransportAction<BulkRequest, BulkResponse> {
     private static final BulkItemResponse ITEM_RESPONSE = new BulkItemResponse(1, DocWriteRequest.OpType.UPDATE,
         new UpdateResponse(new ShardId("mock", "", 1), "mock_type", "1", 1L, DocWriteResponse.Result.CREATED));
 
     @Inject
     public TransportNoopBulkAction(TransportService transportService, ActionFilters actionFilters) {
-        super(NoopBulkAction.NAME, transportService, actionFilters, (Supplier<BulkRequest>) BulkRequest::new);
+        super(NoopBulkAction.NAME, transportService, BulkRequest::new, actionFilters);
     }
 
     @Override

client/rest-high-level/src/main/java/org/elasticsearch/client/dataframe/transforms/DataFrameIndexerPosition.java

@@ -0,0 +1,99 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.client.dataframe.transforms;
+
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ObjectParser.ValueType;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
+
+/**
+ * Holds state of the cursors:
+ *
+ *  indexer_position: the position of the indexer querying the source
+ *  bucket_position: the position used for identifying changes
+ */
+public class DataFrameIndexerPosition {
+    public static final ParseField INDEXER_POSITION = new ParseField("indexer_position");
+    public static final ParseField BUCKET_POSITION = new ParseField("bucket_position");
+
+    private final Map<String, Object> indexerPosition;
+    private final Map<String, Object> bucketPosition;
+
+    @SuppressWarnings("unchecked")
+    public static final ConstructingObjectParser<DataFrameIndexerPosition, Void> PARSER = new ConstructingObjectParser<>(
+            "data_frame_indexer_position",
+            true,
+            args -> new DataFrameIndexerPosition((Map<String, Object>) args[0],(Map<String, Object>) args[1]));
+
+    static {
+        PARSER.declareField(optionalConstructorArg(), XContentParser::mapOrdered, INDEXER_POSITION, ValueType.OBJECT);
+        PARSER.declareField(optionalConstructorArg(), XContentParser::mapOrdered, BUCKET_POSITION, ValueType.OBJECT);
+    }
+
+    public DataFrameIndexerPosition(Map<String, Object> indexerPosition, Map<String, Object> bucketPosition) {
+        this.indexerPosition = indexerPosition == null ? null : Collections.unmodifiableMap(indexerPosition);
+        this.bucketPosition = bucketPosition == null ? null : Collections.unmodifiableMap(bucketPosition);
+    }
+
+    public Map<String, Object> getIndexerPosition() {
+        return indexerPosition;
+    }
+
+    public Map<String, Object> getBucketsPosition() {
+        return bucketPosition;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (this == other) {
+            return true;
+        }
+
+        if (other == null || getClass() != other.getClass()) {
+            return false;
+        }
+
+        DataFrameIndexerPosition that = (DataFrameIndexerPosition) other;
+
+        return Objects.equals(this.indexerPosition, that.indexerPosition) &&
+            Objects.equals(this.bucketPosition, that.bucketPosition);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(indexerPosition, bucketPosition);
+    }
+
+    public static DataFrameIndexerPosition fromXContent(XContentParser parser) {
+        try {
+            return PARSER.parse(parser, null);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}