Merge remote-tracking branch 'elastic/master' into minimum-java-11

* elastic/master:
  Fix Failing to Handle Ex. in TransportShardBulkAction (elastic#40923)
  Be lenient when parsing build flavor and type on the wire (elastic#40734)
  Make Transport Shard Bulk Action Async (elastic#39793)

Author: jasontedor

server/src/main/java/org/elasticsearch/Build.java

@@ -57,7 +57,7 @@ public String displayName() {
             return displayName;
         }
 
-        public static Flavor fromDisplayName(final String displayName) {
+        public static Flavor fromDisplayName(final String displayName, final boolean strict) {
             switch (displayName) {
                 case "default":
                     return Flavor.DEFAULT;
@@ -66,7 +66,12 @@ public static Flavor fromDisplayName(final String displayName) {
                 case "unknown":
                     return Flavor.UNKNOWN;
                 default:
-                    throw new IllegalStateException("unexpected distribution flavor [" + displayName + "]; your distribution is broken");
+                    if (strict) {
+                        final String message = "unexpected distribution flavor [" + displayName + "]; your distribution is broken";
+                        throw new IllegalStateException(message);
+                    } else {
+                        return Flavor.UNKNOWN;
+                    }
             }
         }
 
@@ -91,7 +96,7 @@ public String displayName() {
             this.displayName = displayName;
         }
 
-        public static Type fromDisplayName(final String displayName) {
+        public static Type fromDisplayName(final String displayName, final boolean strict) {
             switch (displayName) {
                 case "deb":
                     return Type.DEB;
@@ -106,9 +111,14 @@ public static Type fromDisplayName(final String displayName) {
                 case "unknown":
                     return Type.UNKNOWN;
                 default:
-                    throw new IllegalStateException("unexpected distribution type [" + displayName + "]; your distribution is broken");
+                    if (strict) {
+                        throw new IllegalStateException("unexpected distribution type [" + displayName + "]; your distribution is broken");
+                    } else {
+                        return Type.UNKNOWN;
+                    }
             }
         }
+
     }
 
     static {
@@ -119,8 +129,9 @@ public static Type fromDisplayName(final String displayName) {
         final boolean isSnapshot;
         final String version;
 
-        flavor = Flavor.fromDisplayName(System.getProperty("es.distribution.flavor", "unknown"));
-        type = Type.fromDisplayName(System.getProperty("es.distribution.type", "unknown"));
+        // these are parsed at startup, and we require that we are able to recognize the values passed in by the startup scripts
+        flavor = Flavor.fromDisplayName(System.getProperty("es.distribution.flavor", "unknown"), true);
+        type = Type.fromDisplayName(System.getProperty("es.distribution.type", "unknown"), true);
 
         final String esPrefix = "elasticsearch-" + Version.CURRENT;
         final URL url = getElasticsearchCodeSourceLocation();
@@ -214,12 +225,14 @@ public static Build readBuild(StreamInput in) throws IOException {
         final Flavor flavor;
         final Type type;
         if (in.getVersion().onOrAfter(Version.V_6_3_0)) {
-            flavor = Flavor.fromDisplayName(in.readString());
+            // be lenient when reading on the wire, the enumeration values from other versions might be different than what we know
+            flavor = Flavor.fromDisplayName(in.readString(), false);
         } else {
             flavor = Flavor.OSS;
         }
         if (in.getVersion().onOrAfter(Version.V_6_3_0)) {
-            type = Type.fromDisplayName(in.readString());
+            // be lenient when reading on the wire, the enumeration values from other versions might be different than what we know
+            type = Type.fromDisplayName(in.readString(), false);
         } else {
             type = Type.UNKNOWN;
         }

server/src/main/java/org/elasticsearch/action/admin/indices/close/TransportVerifyShardBeforeCloseAction.java

@@ -85,14 +85,16 @@ protected void acquireReplicaOperationPermit(final IndexShard replica,
     }
 
     @Override
-    protected PrimaryResult<ShardRequest, ReplicationResponse> shardOperationOnPrimary(final ShardRequest shardRequest,
-                                                                                       final IndexShard primary) throws Exception {
-        executeShardOperation(shardRequest, primary);
-        return new PrimaryResult<>(shardRequest, new ReplicationResponse());
+    protected void shardOperationOnPrimary(final ShardRequest shardRequest, final IndexShard primary,
+            ActionListener<PrimaryResult<ShardRequest, ReplicationResponse>> listener) {
+        ActionListener.completeWith(listener, () -> {
+            executeShardOperation(shardRequest, primary);
+            return new PrimaryResult<>(shardRequest, new ReplicationResponse());
+        });
     }
 
     @Override
-    protected ReplicaResult shardOperationOnReplica(final ShardRequest shardRequest, final IndexShard replica) throws Exception {
+    protected ReplicaResult shardOperationOnReplica(final ShardRequest shardRequest, final IndexShard replica) {
         executeShardOperation(shardRequest, replica);
         return new ReplicaResult();
     }

server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
@@ -51,11 +52,13 @@ protected ReplicationResponse newResponseInstance() {
     }
 
     @Override
-    protected PrimaryResult<ShardFlushRequest, ReplicationResponse> shardOperationOnPrimary(ShardFlushRequest shardRequest,
-            IndexShard primary) {
-        primary.flush(shardRequest.getRequest());
-        logger.trace("{} flush request executed on primary", primary.shardId());
-        return new PrimaryResult<>(shardRequest, new ReplicationResponse());
+    protected void shardOperationOnPrimary(ShardFlushRequest shardRequest, IndexShard primary,
+            ActionListener<PrimaryResult<ShardFlushRequest, ReplicationResponse>> listener) {
+        ActionListener.completeWith(listener, () -> {
+            primary.flush(shardRequest.getRequest());
+            logger.trace("{} flush request executed on primary", primary.shardId());
+            return new PrimaryResult<>(shardRequest, new ReplicationResponse());
+        });
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.refresh;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.BasicReplicationRequest;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
@@ -53,11 +54,13 @@ protected ReplicationResponse newResponseInstance() {
     }
 
     @Override
-    protected PrimaryResult<BasicReplicationRequest, ReplicationResponse> shardOperationOnPrimary(
-            BasicReplicationRequest shardRequest, IndexShard primary) {
-        primary.refresh("api");
-        logger.trace("{} refresh request executed on primary", primary.shardId());
-        return new PrimaryResult<>(shardRequest, new ReplicationResponse());
+    protected void shardOperationOnPrimary(BasicReplicationRequest shardRequest, IndexShard primary,
+            ActionListener<PrimaryResult<BasicReplicationRequest, ReplicationResponse>> listener) {
+        ActionListener.completeWith(listener, () -> {
+            primary.refresh("api");
+            logger.trace("{} refresh request executed on primary", primary.shardId());
+            return new PrimaryResult<>(shardRequest, new ReplicationResponse());
+        });
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/bulk/MappingUpdatePerformer.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.bulk;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.index.mapper.Mapping;
 import org.elasticsearch.index.shard.ShardId;
 
@@ -27,6 +28,6 @@ public interface MappingUpdatePerformer {
     /**
      * Update the mappings on the master.
      */
-    void updateMappings(Mapping update, ShardId shardId, String type);
+    void updateMappings(Mapping update, ShardId shardId, String type, ActionListener<Void> listener);
 
 }