Merge remote-tracking branch 'es/6.x' into ccr-6.x

* es/6.x:
  Enable skipping fetching latest for BWC builds (#29497)
  Add remote cluster client (#29495)
  Ensure flush happens on shard idle
  Adds SpanGapQueryBuilder in the query DSL (#28636)
  Fix auto-generated ID example format (#29461)
  Fix typo in max number of threads check docs (#29469)
  Add primary term to translog header (#29227)
  Add a helper method to get a random java.util.TimeZone (#29487)
  Move TimeValue into elasticsearch-core project (#29486)
  Fix NPE in InternalGeoCentroidTests#testReduceRandom (#29481)
  Build: introduce keystoreFile for cluster config (#29491)
  test: Index more docs, so that it is less likely the search request does not time out.

TESTING.asciidoc

@@ -498,6 +498,13 @@ will contain your change.
 . Push both branches to your remote repository.
 . Run the tests with `./gradlew check -Dtests.bwc.remote=${remote} -Dtests.bwc.refspec.5.x=index_req_bwc_5.x`.
 
+== Skip fetching latest
+
+For some BWC testing scenarios, you want to use the local clone of the
+repository without fetching latest. For these use cases, you can set the system
+property `tests.bwc.git_fetch_latest` to `false` and the BWC builds will skip
+fetching the latest from the remote.
+
 == Test coverage analysis
 
 Generating test coverage reports for Elasticsearch is currently not possible through Gradle.

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

@@ -141,6 +141,8 @@ class ClusterConfiguration {
 
     Map<String, String> keystoreSettings = new HashMap<>()
 
+    Map<String, Object> keystoreFiles = new HashMap<>()
+
     // map from destination path, to source file
     Map<String, Object> extraConfigFiles = new HashMap<>()
 
@@ -167,6 +169,15 @@ class ClusterConfiguration {
         keystoreSettings.put(name, value)
     }
 
+    /**
+     * Adds a file to the keystore. The name is the secure setting name, and the sourceFile
+     * is anything accepted by project.file()
+     */
+    @Input
+    void keystoreFile(String name, Object sourceFile) {
+        keystoreFiles.put(name, sourceFile)
+    }
+
     @Input
     void plugin(String path) {
         Project pluginProject = project.project(path)

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

@@ -180,6 +180,7 @@ class ClusterFormationTasks {
         setup = configureWriteConfigTask(taskName(prefix, node, 'configure'), project, setup, node, seedNode)
         setup = configureCreateKeystoreTask(taskName(prefix, node, 'createKeystore'), project, setup, node)
         setup = configureAddKeystoreSettingTasks(prefix, project, setup, node)
+        setup = configureAddKeystoreFileTasks(prefix, project, setup, node)
 
         if (node.config.plugins.isEmpty() == false) {
             if (node.nodeVersion == VersionProperties.elasticsearch) {
@@ -327,7 +328,7 @@ class ClusterFormationTasks {
 
     /** Adds a task to create keystore */
     static Task configureCreateKeystoreTask(String name, Project project, Task setup, NodeInfo node) {
-        if (node.config.keystoreSettings.isEmpty()) {
+        if (node.config.keystoreSettings.isEmpty() && node.config.keystoreFiles.isEmpty()) {
             return setup
         } else {
             /*
@@ -361,6 +362,37 @@ class ClusterFormationTasks {
         return parentTask
     }
 
+    /** Adds tasks to add files to the keystore */
+    static Task configureAddKeystoreFileTasks(String parent, Project project, Task setup, NodeInfo node) {
+        Map<String, Object> kvs = node.config.keystoreFiles
+        if (kvs.isEmpty()) {
+            return setup
+        }
+        Task parentTask = setup
+        /*
+         * We have to delay building the string as the path will not exist during configuration which will fail on Windows due to getting
+         * the short name requiring the path to already exist.
+         */
+        final Object esKeystoreUtil = "${-> node.binPath().resolve('elasticsearch-keystore').toString()}"
+        for (Map.Entry<String, Object> entry in kvs) {
+            String key = entry.getKey()
+            String name = taskName(parent, node, 'addToKeystore#' + key)
+            String srcFileName = entry.getValue()
+            Task t = configureExecTask(name, project, parentTask, node, esKeystoreUtil, 'add-file', key, srcFileName)
+            t.doFirst {
+                File srcFile = project.file(srcFileName)
+                if (srcFile.isDirectory()) {
+                    throw new GradleException("Source for keystoreFile must be a file: ${srcFile}")
+                }
+                if (srcFile.exists() == false) {
+                    throw new GradleException("Source file for keystoreFile does not exist: ${srcFile}")
+                }
+            }
+            parentTask = t
+        }
+        return parentTask
+    }
+
     static Task configureExtraConfigFilesTask(String name, Project project, Task setup, NodeInfo node) {
         if (node.config.extraConfigFiles.isEmpty()) {
             return setup

distribution/bwc/build.gradle

@@ -54,6 +54,16 @@ subprojects {
 
   final String remote = System.getProperty("tests.bwc.remote", "elastic")
 
+  final boolean gitFetchLatest
+  final String gitFetchLatestProperty = System.getProperty("tests.bwc.git_fetch_latest", "true")
+  if ("true".equals(gitFetchLatestProperty)) {
+    gitFetchLatest = true
+  } else if ("false".equals(gitFetchLatestProperty)) {
+    gitFetchLatest = false
+  } else {
+    throw new GradleException("tests.bwc.git_fetch_latest must be [true] or [false] but was [" + gitFetchLatestProperty + "]")
+  }
+
   task createClone(type: LoggedExec) {
     onlyIf { checkoutDir.exists() == false }
     commandLine = ['git', 'clone', rootDir, checkoutDir]
@@ -83,7 +93,7 @@ subprojects {
   }
 
   task fetchLatest(type: LoggedExec) {
-    onlyIf { project.gradle.startParameter.isOffline() == false }
+    onlyIf { project.gradle.startParameter.isOffline() == false && gitFetchLatest }
     dependsOn addRemote
     workingDir = checkoutDir
     commandLine = ['git', 'fetch', '--all']

docs/reference/docs/index_.asciidoc

@@ -229,14 +229,14 @@ The result of the above index operation is:
     },
     "_index" : "twitter",
     "_type" : "_doc",
-    "_id" : "6a8ca01c-7896-48e9-81cc-9f70661fcb32",
+    "_id" : "W0tpsmIBdwcYyG50zbta",
     "_version" : 1,
     "_seq_no" : 0,
     "_primary_term" : 1,
     "result": "created"
 }
 --------------------------------------------------
-// TESTRESPONSE[s/6a8ca01c-7896-48e9-81cc-9f70661fcb32/$body._id/ s/"successful" : 2/"successful" : 1/]
+// TESTRESPONSE[s/W0tpsmIBdwcYyG50zbta/$body._id/ s/"successful" : 2/"successful" : 1/]
 
 [float]
 [[index-routing]]

docs/reference/setup/bootstrap-checks.asciidoc

@@ -114,7 +114,7 @@ that the Elasticsearch process has the rights to create enough threads
 under normal use. This check is enforced only on Linux. If you are on
 Linux, to pass the maximum number of threads check, you must configure
 your system to allow the Elasticsearch process the ability to create at
-least 2048 threads. This can be done via `/etc/security/limits.conf`
+least 4096 threads. This can be done via `/etc/security/limits.conf`
 using the `nproc` setting (note that you might have to increase the
 limits for the `root` user too).
 

server/src/main/java/org/elasticsearch/common/unit/TimeValue.java → libs/elasticsearch-core/src/main/java/org/elasticsearch/common/unit/TimeValue.java

@@ -19,15 +19,12 @@
 
 package org.elasticsearch.common.unit;
 
-import org.elasticsearch.common.xcontent.ToXContentFragment;
-import org.elasticsearch.common.xcontent.XContentBuilder;
-
 import java.io.IOException;
 import java.util.Locale;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
-public class TimeValue implements Comparable<TimeValue>, ToXContentFragment {
+public class TimeValue implements Comparable<TimeValue> {
 
     /** How many nano-seconds in one milli-second */
     public static final long NSEC_PER_MSEC = TimeUnit.NANOSECONDS.convert(1, TimeUnit.MILLISECONDS);
@@ -352,9 +349,4 @@ public int compareTo(TimeValue timeValue) {
         double otherValue = ((double) timeValue.duration) * timeValue.timeUnit.toNanos(1);
         return Double.compare(thisValue, otherValue);
     }
-
-    @Override
-    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        return builder.value(toString());
-    }
 }

server/src/test/java/org/elasticsearch/common/unit/TimeValueTests.java → libs/elasticsearch-core/src/test/java/org/elasticsearch/common/unit/TimeValueTests.java

@@ -19,15 +19,10 @@
 
 package org.elasticsearch.common.unit;
 
-import org.elasticsearch.common.io.stream.BytesStreamOutput;
-import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.test.ESTestCase;
 
-import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
-import static org.elasticsearch.common.unit.TimeValue.timeValueNanos;
-import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.Matchers.containsString;
@@ -154,31 +149,6 @@ private String randomTimeUnit() {
         return randomFrom("nanos", "micros", "ms", "s", "m", "h", "d");
     }
 
-    private void assertEqualityAfterSerialize(TimeValue value, int expectedSize) throws IOException {
-        BytesStreamOutput out = new BytesStreamOutput();
-        out.writeTimeValue(value);
-        assertEquals(expectedSize, out.size());
-
-        StreamInput in = out.bytes().streamInput();
-        TimeValue inValue = in.readTimeValue();
-
-        assertThat(inValue, equalTo(value));
-        assertThat(inValue.duration(), equalTo(value.duration()));
-        assertThat(inValue.timeUnit(), equalTo(value.timeUnit()));
-    }
-
-    public void testSerialize() throws Exception {
-        assertEqualityAfterSerialize(new TimeValue(100, TimeUnit.DAYS), 3);
-        assertEqualityAfterSerialize(timeValueNanos(-1), 2);
-        assertEqualityAfterSerialize(timeValueNanos(1), 2);
-        assertEqualityAfterSerialize(timeValueSeconds(30), 2);
-
-        final TimeValue timeValue = new TimeValue(randomIntBetween(0, 1024), randomFrom(TimeUnit.values()));
-        BytesStreamOutput out = new BytesStreamOutput();
-        out.writeZLong(timeValue.duration());
-        assertEqualityAfterSerialize(timeValue, 1 + out.bytes().length());
-    }
-
     public void testFailOnUnknownUnits() {
         try {
             TimeValue.parseTimeValue("23tw", null, "test");

plugins/repository-gcs/build.gradle

@@ -94,9 +94,7 @@ task createServiceAccountFile() {
 
 integTestCluster {
   dependsOn createServiceAccountFile, googleCloudStorageFixture
-  setupCommand 'create-elasticsearch-keystore', 'bin/elasticsearch-keystore', 'create'
-  setupCommand 'add-credentials-to-elasticsearch-keystore',
-          'bin/elasticsearch-keystore', 'add-file', 'gcs.client.integration_test.credentials_file', "${serviceAccountFile.absolutePath}"
+  keystoreFile 'gcs.client.integration_test.credentials_file', "${serviceAccountFile.absolutePath}"
 
   /* Use a closure on the string to delay evaluation until tests are executed */
   setting 'gcs.client.integration_test.endpoint', "http://${ -> googleCloudStorageFixture.addressAndPort }"

server/src/main/java/org/elasticsearch/client/Client.java

@@ -477,4 +477,14 @@ public interface Client extends ElasticsearchClient, Releasable {
      * issued from it.
      */
     Client filterWithHeader(Map<String, String> headers);
+
+    /**
+     * Returns a client to a remote cluster with the given cluster alias.
+     *
+     * @throws IllegalArgumentException if the given clusterAlias doesn't exist
+     * @throws UnsupportedOperationException if this functionality is not available on this client.
+     */
+    default Client getRemoteClusterClient(String clusterAlias) {
+        throw new UnsupportedOperationException("this client doesn't support remote cluster connections");
+    }
 }

server/src/main/java/org/elasticsearch/client/FilterClient.java

@@ -73,4 +73,9 @@ protected <Request extends ActionRequest, Response extends ActionResponse, Reque
     protected Client in() {
         return in;
     }
+
+    @Override
+    public Client getRemoteClusterClient(String clusterAlias) {
+        return in.getRemoteClusterClient(clusterAlias);
+    }
 }

server/src/main/java/org/elasticsearch/client/node/NodeClient.java

@@ -33,6 +33,7 @@
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskListener;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.RemoteClusterService;
 
 import java.util.Map;
 import java.util.function.Supplier;
@@ -48,14 +49,17 @@ public class NodeClient extends AbstractClient {
      * {@link #executeLocally(GenericAction, ActionRequest, TaskListener)}.
      */
     private Supplier<String> localNodeId;
+    private RemoteClusterService remoteClusterService;
 
     public NodeClient(Settings settings, ThreadPool threadPool) {
         super(settings, threadPool);
     }
 
-    public void initialize(Map<GenericAction, TransportAction> actions, Supplier<String> localNodeId) {
+    public void initialize(Map<GenericAction, TransportAction> actions, Supplier<String> localNodeId,
+                           RemoteClusterService remoteClusterService) {
         this.actions = actions;
         this.localNodeId = localNodeId;
+        this.remoteClusterService = remoteClusterService;
     }
 
     @Override
@@ -117,4 +121,9 @@ > TransportAction<Request, Response> transportAction(GenericAction<Request, Resp
         }
         return transportAction;
     }
+
+    @Override
+    public Client getRemoteClusterClient(String clusterAlias) {
+        return remoteClusterService.getRemoteClusterClient(threadPool(), clusterAlias);
+    }
 }

server/src/main/java/org/elasticsearch/common/xcontent/XContentElasticsearchExtension.java

@@ -61,6 +61,7 @@ public Map<Class<?>, XContentBuilder.Writer> getXContentWriters() {
         writers.put(FixedDateTimeZone.class, (b, v) -> b.value(Objects.toString(v)));
         writers.put(MutableDateTime.class, XContentBuilder::timeValue);
         writers.put(DateTime.class, XContentBuilder::timeValue);
+        writers.put(TimeValue.class, (b, v) -> b.value(v.toString()));
 
         writers.put(BytesReference.class, (b, v) -> {
             if (v == null) {

server/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -188,7 +188,7 @@ public final class IndexSettings {
     public static final Setting<ByteSizeValue> INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING =
         Setting.byteSizeSetting("index.translog.flush_threshold_size", new ByteSizeValue(512, ByteSizeUnit.MB),
             /*
-             * An empty translog occupies 43 bytes on disk. If the flush threshold is below this, the flush thread
+             * An empty translog occupies 55 bytes on disk. If the flush threshold is below this, the flush thread
              * can get stuck in an infinite loop as the shouldPeriodicallyFlush can still be true after flushing.
              * However, small thresholds are useful for testing so we do not add a large lower bound here.
              */
@@ -223,7 +223,7 @@ public final class IndexSettings {
                     "index.translog.generation_threshold_size",
                     new ByteSizeValue(64, ByteSizeUnit.MB),
                     /*
-                     * An empty translog occupies 43 bytes on disk. If the generation threshold is
+                     * An empty translog occupies 55 bytes on disk. If the generation threshold is
                      * below this, the flush thread can get stuck in an infinite loop repeatedly
                      * rolling the generation as every new generation will already exceed the
                      * generation threshold. However, small thresholds are useful for testing so we

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -1066,14 +1066,13 @@ public Index(Term uid, ParsedDocument doc, long seqNo, long primaryTerm, long ve
             this.autoGeneratedIdTimestamp = autoGeneratedIdTimestamp;
         }
 
-        public Index(Term uid, ParsedDocument doc) {
-            this(uid, doc, Versions.MATCH_ANY);
+        public Index(Term uid, long primaryTerm, ParsedDocument doc) {
+            this(uid, primaryTerm, doc, Versions.MATCH_ANY);
         } // TEST ONLY
 
-        Index(Term uid, ParsedDocument doc, long version) {
-            // use a primary term of 2 to allow tests to reduce it to a valid >0 term
-            this(uid, doc, SequenceNumbers.UNASSIGNED_SEQ_NO, 2, version, VersionType.INTERNAL,
-                    Origin.PRIMARY, System.nanoTime(), -1, false);
+        Index(Term uid, long primaryTerm, ParsedDocument doc, long version) {
+            this(uid, doc, SequenceNumbers.UNASSIGNED_SEQ_NO, primaryTerm, version, VersionType.INTERNAL,
+                Origin.PRIMARY, System.nanoTime(), -1, false);
         } // TEST ONLY
 
         public ParsedDocument parsedDoc() {
@@ -1147,8 +1146,8 @@ public Delete(String type, String id, Term uid, long seqNo, long primaryTerm, lo
             this.id = Objects.requireNonNull(id);
         }
 
-        public Delete(String type, String id, Term uid) {
-            this(type, id, uid, SequenceNumbers.UNASSIGNED_SEQ_NO, 0, Versions.MATCH_ANY, VersionType.INTERNAL, Origin.PRIMARY, System.nanoTime());
+        public Delete(String type, String id, Term uid, long primaryTerm) {
+            this(type, id, uid, SequenceNumbers.UNASSIGNED_SEQ_NO, primaryTerm, Versions.MATCH_ANY, VersionType.INTERNAL, Origin.PRIMARY, System.nanoTime());
         }
 
         public Delete(Delete template, VersionType versionType) {

server/src/main/java/org/elasticsearch/index/engine/EngineConfig.java

@@ -79,6 +79,7 @@ public final class EngineConfig {
     @Nullable
     private final CircuitBreakerService circuitBreakerService;
     private final LongSupplier globalCheckpointSupplier;
+    private final LongSupplier primaryTermSupplier;
 
     /**
      * Index setting to change the low level lucene codec used for writing new segments.
@@ -125,7 +126,7 @@ public EngineConfig(ShardId shardId, String allocationId, ThreadPool threadPool,
                         List<ReferenceManager.RefreshListener> externalRefreshListener,
                         List<ReferenceManager.RefreshListener> internalRefreshListener, Sort indexSort,
                         TranslogRecoveryRunner translogRecoveryRunner, CircuitBreakerService circuitBreakerService,
-                        LongSupplier globalCheckpointSupplier) {
+                        LongSupplier globalCheckpointSupplier, LongSupplier primaryTermSupplier) {
         this.shardId = shardId;
         this.allocationId = allocationId;
         this.indexSettings = indexSettings;
@@ -152,6 +153,7 @@ public EngineConfig(ShardId shardId, String allocationId, ThreadPool threadPool,
         this.translogRecoveryRunner = translogRecoveryRunner;
         this.circuitBreakerService = circuitBreakerService;
         this.globalCheckpointSupplier = globalCheckpointSupplier;
+        this.primaryTermSupplier = primaryTermSupplier;
     }
 
     /**
@@ -354,4 +356,11 @@ public Sort getIndexSort() {
     public CircuitBreakerService getCircuitBreakerService() {
         return this.circuitBreakerService;
     }
+
+    /**
+     * Returns a supplier that supplies the latest primary term value of the associated shard.
+     */
+    public LongSupplier getPrimaryTermSupplier() {
+        return primaryTermSupplier;
+    }
 }

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -432,7 +432,7 @@ private Translog openTranslog(EngineConfig engineConfig, TranslogDeletionPolicy
             translogDeletionPolicy.setMinTranslogGenerationForRecovery(minRequiredTranslogGen);
         }
         // We expect that this shard already exists, so it must already have an existing translog else something is badly wrong!
-        return new Translog(translogConfig, translogUUID, translogDeletionPolicy, globalCheckpointSupplier);
+        return new Translog(translogConfig, translogUUID, translogDeletionPolicy, globalCheckpointSupplier, engineConfig.getPrimaryTermSupplier());
     }
 
     @Override