Merge branch 'master' into use-reader-for-doc-stats

* master: (22 commits)
  Support negative numbers in writeVLong (elastic#22314)
  UnicastZenPing's PingingRound should prevent opening connections after being closed
  Add task to clean idea build directory. Make cleanIdea task invoke it.
  add trace logging to UnicastZenPingTests.testResolveReuseExistingNodeConnections
  Adds ingest processor headers to exception for unknown processor. (elastic#22315)
  Remove much ceremony from parsing client yaml test suites (elastic#22311)
  Support numeric bounds with decimal parts for long/integer/short/byte datatypes (elastic#21972)
  inner hits: Don't inline inner hits if the query the inner hits is inlined into can't resolve mappings and ignore_unmapped has been set to true
  Fix stackoverflow error on InternalNumericMetricAggregation
  Date detection should not rely on a hardcoded set of characters. (elastic#22171)
  `value_type` is useful regardless of scripting. (elastic#22160)
  Improve concurrency of ShardCoreKeyMap. (elastic#22316)
  fixed jdocs and removed already fixed norelease
  Adds abstract test classes for serialisation (elastic#22281)
  Introduce translog no-op
  Provide helpful error message if a plugin exists
  Clear static variable after suite
  Repeated language analyzers (elastic#22240)
  Restore deprecation warning for invalid match_mapping_type values (elastic#22304)
  Make `-0` compare less than `+0` consistently. (elastic#22173)
  ...

build.gradle

@@ -204,6 +204,14 @@ allprojects {
       }
     }
   }
+
+  task cleanIdeaBuildDir(type: Delete) {
+    delete 'build-idea'
+  }
+  cleanIdeaBuildDir.setGroup("ide")
+  cleanIdeaBuildDir.setDescription("Deletes the IDEA build directory.")
+
+  tasks.cleanIdea.dependsOn(cleanIdeaBuildDir)
 }
 
 idea {

core/src/main/java/org/elasticsearch/common/Strings.java

@@ -276,26 +276,6 @@ public static boolean substringMatch(CharSequence str, int index, CharSequence s
         return true;
     }
 
-    /**
-     * Count the occurrences of the substring in string s.
-     *
-     * @param str string to search in. Return 0 if this is null.
-     * @param sub string to search for. Return 0 if this is null.
-     */
-    public static int countOccurrencesOf(String str, String sub) {
-        if (str == null || sub == null || str.length() == 0 || sub.length() == 0) {
-            return 0;
-        }
-        int count = 0;
-        int pos = 0;
-        int idx;
-        while ((idx = str.indexOf(sub, pos)) != -1) {
-            ++count;
-            pos = idx + sub.length();
-        }
-        return count;
-    }
-
     /**
      * Replace all occurrences of a substring within a string with
      * another string.

core/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -214,9 +214,8 @@ public long readLong() throws IOException {
     }
 
     /**
-     * Reads a long stored in variable-length format.  Reads between one and
-     * nine bytes.  Smaller values take fewer bytes.  Negative numbers are not
-     * supported.
+     * Reads a long stored in variable-length format. Reads between one and ten bytes. Smaller values take fewer bytes. Negative numbers
+     * are encoded in ten bytes so prefer {@link #readLong()} or {@link #readZLong()} for negative numbers.
      */
     public long readVLong() throws IOException {
         byte b = readByte();
@@ -260,8 +259,16 @@ public long readVLong() throws IOException {
             return i;
         }
         b = readByte();
-        assert (b & 0x80) == 0;
-        return i | ((b & 0x7FL) << 56);
+        i |= ((b & 0x7FL) << 56);
+        if ((b & 0x80) == 0) {
+            return i;
+        }
+        b = readByte();
+        if (b != 0 && b != 1) {
+            throw new IOException("Invalid vlong (" + Integer.toHexString(b) + " << 63) | " + Long.toHexString(i));
+        }
+        i |= ((long) b) << 63;
+        return i;
     }
 
     public long readZLong() throws IOException {

core/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -28,7 +28,6 @@
 import org.apache.lucene.util.BitUtil;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefBuilder;
-import org.elasticsearch.Assertions;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.common.Nullable;
@@ -211,16 +210,22 @@ public void writeLong(long i) throws IOException {
     }
 
     /**
-     * Writes a non-negative long in a variable-length format.
-     * Writes between one and nine bytes. Smaller values take fewer bytes.
-     * Negative numbers are not supported.
+     * Writes a non-negative long in a variable-length format. Writes between one and ten bytes. Smaller values take fewer bytes. Negative
+     * numbers use ten bytes and trip assertions (if running in tests) so prefer {@link #writeLong(long)} or {@link #writeZLong(long)} for
+     * negative numbers.
      */
     public void writeVLong(long i) throws IOException {
-        if (Assertions.ENABLED) {
-            if (i < 0) {
-                throw new IllegalStateException(Long.toString(i));
-            }
+        if (i < 0) {
+            throw new IllegalStateException("Negative longs unsupported, use writeLong or writeZLong for negative numbers [" + i + "]");
         }
+        writeVLongNoCheck(i);
+    }
+
+    /**
+     * Writes a long in a variable-length format without first checking if it is negative. Package private for testing. Use
+     * {@link #writeVLong(long)} instead.
+     */
+    void writeVLongNoCheck(long i) throws IOException {
         while ((i & ~0x7F) != 0) {
             writeByte((byte) ((i & 0x7f) | 0x80));
             i >>>= 7;
@@ -333,7 +338,7 @@ public void writeString(String str) throws IOException {
             // make sure any possible char can fit into the buffer in any possible iteration
             // we need at most 3 bytes so we flush the buffer once we have less than 3 bytes
             // left before we start another iteration
-            if (offset > buffer.length-3) {
+            if (offset > buffer.length - 3) {
                 writeBytes(buffer, offset);
                 offset = 0;
             }

core/src/main/java/org/elasticsearch/common/lucene/ShardCoreKeyMap.java

@@ -29,9 +29,9 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.IdentityHashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * A map between segment core cache keys and the shard that these segments
@@ -50,7 +50,7 @@ public final class ShardCoreKeyMap {
     private final Map<String, Set<Object>> indexToCoreKey;
 
     public ShardCoreKeyMap() {
-        coreKeyToShard = new IdentityHashMap<>();
+        coreKeyToShard = new ConcurrentHashMap<>();
         indexToCoreKey = new HashMap<>();
     }
 
@@ -64,9 +64,17 @@ public void add(LeafReader reader) {
             throw new IllegalArgumentException("Could not extract shard id from " + reader);
         }
         final Object coreKey = reader.getCoreCacheKey();
+
+        if (coreKeyToShard.containsKey(coreKey)) {
+            // Do this check before entering the synchronized block in order to
+            // avoid taking the mutex if possible (which should happen most of
+            // the time).
+            return;
+        }
+
         final String index = shardId.getIndexName();
         synchronized (this) {
-            if (coreKeyToShard.put(coreKey, shardId) == null) {
+            if (coreKeyToShard.containsKey(coreKey) == false) {
                 Set<Object> objects = indexToCoreKey.get(index);
                 if (objects == null) {
                     objects = new HashSet<>();
@@ -90,6 +98,14 @@ public void add(LeafReader reader) {
                 try {
                     reader.addCoreClosedListener(listener);
                     addedListener = true;
+
+                    // Only add the core key to the map as a last operation so that
+                    // if another thread sees that the core key is already in the
+                    // map (like the check just before this synchronized block),
+                    // then it means that the closed listener has already been
+                    // registered.
+                    ShardId previous = coreKeyToShard.put(coreKey, shardId);
+                    assert previous == null;
                 } finally {
                     if (false == addedListener) {
                         try {

core/src/main/java/org/elasticsearch/discovery/zen/UnicastZenPing.java

@@ -385,12 +385,14 @@ public Connection getOrConnect(DiscoveryNode node) throws IOException {
             try (Releasable ignore = connectionLock.acquire(node.getAddress())) {
                 result = tempConnections.get(node.getAddress());
                 if (result == null) {
+                    ensureOpen();
                     boolean success = false;
                     result = transportService.openConnection(node, connectionProfile);
                     try {
                         transportService.handshake(result, connectionProfile.getHandshakeTimeout().millis());
                         synchronized (this) {
-                            // acquire lock to prevent concurrent closing
+                            // acquire lock and check if closed, to prevent leaving an open connection after closing
+                            ensureOpen();
                             Connection existing = tempConnections.put(node.getAddress(), result);
                             assert existing == null;
                             success = true;

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

@@ -294,6 +294,8 @@ public Condition newCondition() {
      */
     public abstract DeleteResult delete(final Delete delete);
 
+    public abstract NoOpResult noOp(final NoOp noOp);
+
     /**
      * Base class for index and delete operation results
      * Holds result meta data (e.g. translog location, updated version)
@@ -380,6 +382,7 @@ void freeze() {
     }
 
     public static class IndexResult extends Result {
+
         private final boolean created;
 
         public IndexResult(long version, long seqNo, boolean created) {
@@ -395,9 +398,11 @@ public IndexResult(Exception failure, long version, long seqNo) {
         public boolean isCreated() {
             return created;
         }
+
     }
 
     public static class DeleteResult extends Result {
+
         private final boolean found;
 
         public DeleteResult(long version, long seqNo, boolean found) {
@@ -413,6 +418,19 @@ public DeleteResult(Exception failure, long version, long seqNo) {
         public boolean isFound() {
             return found;
         }
+
+    }
+
+    static class NoOpResult extends Result {
+
+        NoOpResult(long seqNo) {
+            super(Operation.TYPE.NO_OP, 0, seqNo);
+        }
+
+        NoOpResult(long seqNo, Exception failure) {
+            super(Operation.TYPE.NO_OP, failure, 0, seqNo);
+        }
+
     }
 
     /**
@@ -908,7 +926,7 @@ public abstract static class Operation {
 
         /** type of operation (index, delete), subclasses use static types */
         public enum TYPE {
-            INDEX, DELETE;
+            INDEX, DELETE, NO_OP;
 
             private final String lowercase;
 
@@ -1112,6 +1130,50 @@ TYPE operationType() {
         public int estimatedSizeInBytes() {
             return (uid().field().length() + uid().text().length()) * 2 + 20;
         }
+
+    }
+
+    public static class NoOp extends Operation {
+
+        private final String reason;
+
+        public String reason() {
+            return reason;
+        }
+
+        public NoOp(
+            final Term uid,
+            final long seqNo,
+            final long primaryTerm,
+            final long version,
+            final VersionType versionType,
+            final Origin origin,
+            final long startTime,
+            final String reason) {
+            super(uid, seqNo, primaryTerm, version, versionType, origin, startTime);
+            this.reason = reason;
+        }
+
+        @Override
+        public String type() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        String id() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        TYPE operationType() {
+            return TYPE.NO_OP;
+        }
+
+        @Override
+        public int estimatedSizeInBytes() {
+            return 2 * reason.length() + 2 * Long.BYTES;
+        }
+
     }
 
     public static class Get {