Merge remote-tracking branch 'origin/master' into feature/client_aggs_parsing

18 files changed, 287 insertions, 515 deletions

diff --git a/core/src/main/java/org/elasticsearch/Version.java b/core/src/main/java/org/elasticsearch/Version.java
index 44b989a378..1e60a4ac83 100644
--- a/core/src/main/java/org/elasticsearch/Version.java
+++ b/core/src/main/java/org/elasticsearch/Version.java
@@ -278,9 +278,12 @@ public class Version implements Comparable<Version> {
     public Version minimumCompatibilityVersion() {
         final int bwcMajor;
         final int bwcMinor;
-        if (this.onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
-            bwcMajor = major - 1;
-            bwcMinor = 4;
+        if (major == 6) { // we only specialize for current major here
+            bwcMajor = Version.V_5_4_0.major;
+            bwcMinor = Version.V_5_4_0.minor;
+        } else if (major > 6) { // all the future versions are compatible with first minor...
+            bwcMajor = major -1;
+            bwcMinor = 0;
         } else {
             bwcMajor = major;
             bwcMinor = 0;
diff --git a/core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java b/core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
index 5b99ed02cf..f0aede639e 100644
--- a/core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
+++ b/core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
@@ -23,7 +23,6 @@ import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.ExceptionsHelper;
-import org.elasticsearch.Version;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.delete.DeleteRequest;
@@ -31,8 +30,8 @@ import org.elasticsearch.action.delete.DeleteResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
 import org.elasticsearch.action.support.ActionFilters;
-import org.elasticsearch.action.support.replication.ReplicationOperation;
 import org.elasticsearch.action.support.TransportActions;
+import org.elasticsearch.action.support.replication.ReplicationOperation;
 import org.elasticsearch.action.support.replication.ReplicationResponse.ShardInfo;
 import org.elasticsearch.action.support.replication.TransportWriteAction;
 import org.elasticsearch.action.update.UpdateHelper;
@@ -56,7 +55,6 @@ import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
 import org.elasticsearch.index.get.GetResult;
 import org.elasticsearch.index.mapper.MapperParsingException;
-import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.Mapping;
 import org.elasticsearch.index.mapper.SourceToParse;
 import org.elasticsearch.index.seqno.SequenceNumbersService;
@@ -414,13 +412,6 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
         FAILURE
     }
 
-    static {
-        assert Version.CURRENT.minimumCompatibilityVersion().after(Version.V_6_0_0_alpha1_UNRELEASED) == false:
-                "Remove logic handling NoOp result from primary response; see TODO in replicaItemExecutionMode" +
-                        " as the current minimum compatible version [" +
-                        Version.CURRENT.minimumCompatibilityVersion() + "] is after 6.0";
-    }
-
     /**
      * Determines whether a bulk item request should be executed on the replica.
      * @return {@link ReplicaItemExecutionMode#NORMAL} upon normal primary execution with no failures
@@ -436,10 +427,11 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
                     ? ReplicaItemExecutionMode.FAILURE // we have a seq no generated with the failure, replicate as no-op
                     : ReplicaItemExecutionMode.NOOP; // no seq no generated, ignore replication
         } else {
-            // NOTE: write requests originating from pre-6.0 nodes can send a no-op operation to
-            // the replica; we ignore replication
-            // TODO: remove noOp result check from primary response, when pre-6.0 nodes are not supported
-            // we should return ReplicationItemExecutionMode.NORMAL instead
+            // TODO: once we know for sure that every operation that has been processed on the primary is assigned a seq#
+            // (i.e., all nodes on the cluster are on v6.0.0 or higher) we can use the existence of a seq# to indicate whether
+            // an operation should be processed or be treated as a noop. This means we could remove this method and the
+            // ReplicaItemExecutionMode enum and have a simple boolean check for seq != UNASSIGNED_SEQ_NO which will work for
+            // both failures and indexing operations.
             return primaryResponse.getResponse().getResult() != DocWriteResponse.Result.NOOP
                     ? ReplicaItemExecutionMode.NORMAL // execution successful on primary
                     : ReplicaItemExecutionMode.NOOP; // ignore replication
diff --git a/core/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java b/core/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java
index 9b96f407f3..d8ee93fe88 100644
--- a/core/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java
+++ b/core/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java
@@ -19,8 +19,6 @@
 package org.elasticsearch.common.settings;
 
 import org.elasticsearch.action.admin.indices.close.TransportCloseIndexAction;
-import org.elasticsearch.transport.RemoteClusterService;
-import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.support.AutoCreateIndex;
 import org.elasticsearch.action.support.DestructiveOperations;
@@ -88,6 +86,8 @@ import org.elasticsearch.search.SearchModule;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.fetch.subphase.highlight.FastVectorHighlighter;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.RemoteClusterAware;
+import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportService;
@@ -304,6 +304,8 @@ public final class ClusterSettings extends AbstractScopedSettings {
                     ScriptService.SCRIPT_CACHE_EXPIRE_SETTING,
                     ScriptService.SCRIPT_MAX_SIZE_IN_BYTES,
                     ScriptService.SCRIPT_MAX_COMPILATIONS_PER_MINUTE,
+                    ScriptService.TYPES_ALLOWED_SETTING,
+                    ScriptService.CONTEXTS_ALLOWED_SETTING,
                     IndicesService.INDICES_CACHE_CLEAN_INTERVAL_SETTING,
                     IndicesFieldDataCache.INDICES_FIELDDATA_CACHE_SIZE_KEY,
                     IndicesRequestCache.INDICES_CACHE_QUERY_SIZE,
@@ -339,6 +341,7 @@ public final class ClusterSettings extends AbstractScopedSettings {
                     ZenDiscovery.SEND_LEAVE_REQUEST_SETTING,
                     ZenDiscovery.MASTER_ELECTION_WAIT_FOR_JOINS_TIMEOUT_SETTING,
                     ZenDiscovery.MASTER_ELECTION_IGNORE_NON_MASTER_PINGS_SETTING,
+                    ZenDiscovery.MAX_PENDING_CLUSTER_STATES_SETTING,
                     UnicastZenPing.DISCOVERY_ZEN_PING_UNICAST_HOSTS_SETTING,
                     UnicastZenPing.DISCOVERY_ZEN_PING_UNICAST_CONCURRENT_CONNECTS_SETTING,
                     UnicastZenPing.DISCOVERY_ZEN_PING_UNICAST_HOSTS_RESOLVE_TIMEOUT,
diff --git a/core/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java b/core/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java
index 4150783a8f..ae469d162a 100644
--- a/core/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java
+++ b/core/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java
@@ -23,8 +23,8 @@ import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.util.IOUtils;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.ClusterChangedEvent;
-import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.cluster.IncompatibleClusterStateVersionException;
@@ -60,61 +60,53 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Supplier;
 
 public class PublishClusterStateAction extends AbstractComponent {
 
     public static final String SEND_ACTION_NAME = "internal:discovery/zen/publish/send";
     public static final String COMMIT_ACTION_NAME = "internal:discovery/zen/publish/commit";
 
-    public static final String SETTINGS_MAX_PENDING_CLUSTER_STATES = "discovery.zen.publish.max_pending_cluster_states";
+    public interface IncomingClusterStateListener {
 
-    public interface NewPendingClusterStateListener {
+        /**
+         * called when a new incoming cluster state has been received.
+         * Should validate the incoming state and throw an exception if it's not a valid successor state.
+         */
+        void onIncomingClusterState(ClusterState incomingState);
 
-        /** a new cluster state has been committed and is ready to process via {@link #pendingStatesQueue()} */
-        void onNewClusterState(String reason);
+        /**
+         * called when a cluster state has been committed and is ready to be processed
+         */
+        void onClusterStateCommitted(String stateUUID, ActionListener<Void> processedListener);
     }
 
     private final TransportService transportService;
     private final NamedWriteableRegistry namedWriteableRegistry;
-    private final Supplier<ClusterState> clusterStateSupplier;
-    private final NewPendingClusterStateListener newPendingClusterStatelistener;
+    private final IncomingClusterStateListener incomingClusterStateListener;
     private final DiscoverySettings discoverySettings;
-    private final ClusterName clusterName;
-    private final PendingClusterStatesQueue pendingStatesQueue;
 
     public PublishClusterStateAction(
             Settings settings,
             TransportService transportService,
             NamedWriteableRegistry namedWriteableRegistry,
-            Supplier<ClusterState> clusterStateSupplier,
-            NewPendingClusterStateListener listener,
-            DiscoverySettings discoverySettings,
-            ClusterName clusterName) {
+            IncomingClusterStateListener incomingClusterStateListener,
+            DiscoverySettings discoverySettings) {
         super(settings);
         this.transportService = transportService;
         this.namedWriteableRegistry = namedWriteableRegistry;
-        this.clusterStateSupplier = clusterStateSupplier;
-        this.newPendingClusterStatelistener = listener;
+        this.incomingClusterStateListener = incomingClusterStateListener;
         this.discoverySettings = discoverySettings;
-        this.clusterName = clusterName;
-        this.pendingStatesQueue = new PendingClusterStatesQueue(logger, settings.getAsInt(SETTINGS_MAX_PENDING_CLUSTER_STATES, 25));
         transportService.registerRequestHandler(SEND_ACTION_NAME, BytesTransportRequest::new, ThreadPool.Names.SAME, false, false,
             new SendClusterStateRequestHandler());
         transportService.registerRequestHandler(COMMIT_ACTION_NAME, CommitClusterStateRequest::new, ThreadPool.Names.SAME, false, false,
             new CommitClusterStateRequestHandler());
     }
 
-    public PendingClusterStatesQueue pendingStatesQueue() {
-        return pendingStatesQueue;
-    }
-
     /**
      * publishes a cluster change event to other nodes. if at least minMasterNodes acknowledge the change it is committed and will
      * be processed by the master and the other nodes.
@@ -387,7 +379,7 @@ public class PublishClusterStateAction extends AbstractComponent {
                 final ClusterState incomingState;
                 // If true we received full cluster state - otherwise diffs
                 if (in.readBoolean()) {
-                    incomingState = ClusterState.readFrom(in, clusterStateSupplier.get().nodes().getLocalNode());
+                    incomingState = ClusterState.readFrom(in, transportService.getLocalNode());
                     logger.debug("received full cluster state version [{}] with size [{}]", incomingState.version(),
                         request.bytes().length());
                 } else if (lastSeenClusterState != null) {
@@ -399,10 +391,7 @@ public class PublishClusterStateAction extends AbstractComponent {
                     logger.debug("received diff for but don't have any local cluster state - requesting full state");
                     throw new IncompatibleClusterStateVersionException("have no local cluster state");
                 }
-                // sanity check incoming state
-                validateIncomingState(incomingState, lastSeenClusterState);
-
-                pendingStatesQueue.addPending(incomingState);
+                incomingClusterStateListener.onIncomingClusterState(incomingState);
                 lastSeenClusterState = incomingState;
             }
         } finally {
@@ -411,56 +400,22 @@ public class PublishClusterStateAction extends AbstractComponent {
         channel.sendResponse(TransportResponse.Empty.INSTANCE);
     }
 
-    // package private for testing
-
-    /**
-     * does simple sanity check of the incoming cluster state. Throws an exception on rejections.
-     */
-    void validateIncomingState(ClusterState incomingState, ClusterState lastSeenClusterState) {
-        final ClusterName incomingClusterName = incomingState.getClusterName();
-        if (!incomingClusterName.equals(this.clusterName)) {
-            logger.warn("received cluster state from [{}] which is also master but with a different cluster name [{}]",
-                incomingState.nodes().getMasterNode(), incomingClusterName);
-            throw new IllegalStateException("received state from a node that is not part of the cluster");
-        }
-        final ClusterState clusterState = clusterStateSupplier.get();
-
-        if (clusterState.nodes().getLocalNode().equals(incomingState.nodes().getLocalNode()) == false) {
-            logger.warn("received a cluster state from [{}] and not part of the cluster, should not happen",
-                incomingState.nodes().getMasterNode());
-            throw new IllegalStateException("received state with a local node that does not match the current local node");
-        }
-
-        if (ZenDiscovery.shouldIgnoreOrRejectNewClusterState(logger, clusterState, incomingState)) {
-            String message = String.format(
-                    Locale.ROOT,
-                    "rejecting cluster state version [%d] uuid [%s] received from [%s]",
-                    incomingState.version(),
-                    incomingState.stateUUID(),
-                    incomingState.nodes().getMasterNodeId()
-            );
-            logger.warn(message);
-            throw new IllegalStateException(message);
-        }
-
-    }
-
     protected void handleCommitRequest(CommitClusterStateRequest request, final TransportChannel channel) {
-        final ClusterState state = pendingStatesQueue.markAsCommitted(request.stateUUID,
-            new PendingClusterStatesQueue.StateProcessedListener() {
+        incomingClusterStateListener.onClusterStateCommitted(request.stateUUID, new ActionListener<Void>() {
+
             @Override
-            public void onNewClusterStateProcessed() {
+            public void onResponse(Void ignore) {
                 try {
                     // send a response to the master to indicate that this cluster state has been processed post committing it.
                     channel.sendResponse(TransportResponse.Empty.INSTANCE);
                 } catch (Exception e) {
                     logger.debug("failed to send response on cluster state processed", e);
-                    onNewClusterStateFailed(e);
+                    onFailure(e);
                 }
             }
 
             @Override
-            public void onNewClusterStateFailed(Exception e) {
+            public void onFailure(Exception e) {
                 try {
                     channel.sendResponse(e);
                 } catch (Exception inner) {
@@ -469,10 +424,6 @@ public class PublishClusterStateAction extends AbstractComponent {
                 }
             }
         });
-        if (state != null) {
-            newPendingClusterStatelistener.onNewClusterState("master " + state.nodes().getMasterNode() +
-                " committed version [" + state.version() + "]");
-        }
     }
 
     private class SendClusterStateRequestHandler implements TransportRequestHandler<BytesTransportRequest> {
diff --git a/core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java b/core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java
index 09e6357ba5..dfbf3f780b 100644
--- a/core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java
+++ b/core/src/main/java/org/elasticsearch/discovery/zen/ZenDiscovery.java
@@ -25,6 +25,7 @@ import org.apache.logging.log4j.util.Supplier;
 import org.apache.lucene.util.IOUtils;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
@@ -56,6 +57,7 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.discovery.Discovery;
 import org.elasticsearch.discovery.DiscoverySettings;
 import org.elasticsearch.discovery.DiscoveryStats;
+import org.elasticsearch.discovery.zen.PublishClusterStateAction.IncomingClusterStateListener;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.EmptyTransportResponseHandler;
 import org.elasticsearch.transport.TransportChannel;
@@ -82,7 +84,7 @@ import java.util.stream.Collectors;
 import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds;
 import static org.elasticsearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
 
-public class ZenDiscovery extends AbstractLifecycleComponent implements Discovery, PingContextProvider {
+public class ZenDiscovery extends AbstractLifecycleComponent implements Discovery, PingContextProvider, IncomingClusterStateListener {
 
     public static final Setting<TimeValue> PING_TIMEOUT_SETTING =
         Setting.positiveTimeSetting("discovery.zen.ping_timeout", timeValueSeconds(3), Property.NodeScope);
@@ -104,6 +106,8 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
             Property.NodeScope);
     public static final Setting<Boolean> MASTER_ELECTION_IGNORE_NON_MASTER_PINGS_SETTING =
             Setting.boolSetting("discovery.zen.master_election.ignore_non_master_pings", false, Property.NodeScope);
+    public static final Setting<Integer> MAX_PENDING_CLUSTER_STATES_SETTING =
+        Setting.intSetting("discovery.zen.publish.max_pending_cluster_states", 25, 1, Property.NodeScope);
 
     public static final String DISCOVERY_REJOIN_ACTION_NAME = "internal:discovery/zen/rejoin";
 
@@ -139,6 +143,8 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
 
     private final JoinThreadControl joinThreadControl;
 
+    private final PendingClusterStatesQueue pendingStatesQueue;
+
     private final NodeJoinController nodeJoinController;
     private final NodeRemovalClusterStateTaskExecutor nodeRemovalExecutor;
 
@@ -197,16 +203,15 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
         this.masterFD.addListener(new MasterNodeFailureListener());
         this.nodesFD = new NodesFaultDetection(settings, threadPool, transportService, clusterName);
         this.nodesFD.addListener(new NodeFaultDetectionListener());
+        this.pendingStatesQueue = new PendingClusterStatesQueue(logger, MAX_PENDING_CLUSTER_STATES_SETTING.get(settings));
 
         this.publishClusterState =
                 new PublishClusterStateAction(
                         settings,
                         transportService,
                         namedWriteableRegistry,
-                        this::clusterState,
-                        new NewPendingClusterStateListener(),
-                        discoverySettings,
-                        clusterName);
+                        this,
+                        discoverySettings);
         this.membership = new MembershipAction(settings, transportService, new MembershipListener());
         this.joinThreadControl = new JoinThreadControl();
 
@@ -311,7 +316,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
             throw new FailedToCommitClusterStateException("state was mutated while calculating new CS update");
         }
 
-        publishClusterState.pendingStatesQueue().addPending(newState);
+        pendingStatesQueue.addPending(newState);
 
         try {
             publishClusterState.publish(clusterChangedEvent, electMaster.minimumMasterNodes(), ackListener);
@@ -321,7 +326,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
                 newState.version(), electMaster.minimumMasterNodes());
 
             synchronized (stateMutex) {
-                publishClusterState.pendingStatesQueue().failAllStatesAndClear(
+                pendingStatesQueue.failAllStatesAndClear(
                     new ElasticsearchException("failed to publish cluster state"));
 
                 rejoin("zen-disco-failed-to-publish");
@@ -332,7 +337,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
         final DiscoveryNode localNode = newState.getNodes().getLocalNode();
         final CountDownLatch latch = new CountDownLatch(1);
         final AtomicBoolean processedOrFailed = new AtomicBoolean();
-        publishClusterState.pendingStatesQueue().markAsCommitted(newState.stateUUID(),
+        pendingStatesQueue.markAsCommitted(newState.stateUUID(),
             new PendingClusterStatesQueue.StateProcessedListener() {
                 @Override
                 public void onNewClusterStateProcessed() {
@@ -391,7 +396,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
 
     @Override
     public DiscoveryStats stats() {
-        PendingClusterStateStats queueStats = publishClusterState.pendingStatesQueue().stats();
+        PendingClusterStateStats queueStats = pendingStatesQueue.stats();
         return new DiscoveryStats(queueStats);
     }
 
@@ -409,11 +414,11 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
 
     // used for testing
     public ClusterState[] pendingClusterStates() {
-        return publishClusterState.pendingStatesQueue().pendingClusterStates();
+        return pendingStatesQueue.pendingClusterStates();
     }
 
     PendingClusterStatesQueue pendingClusterStatesQueue() {
-        return publishClusterState.pendingStatesQueue();
+        return pendingStatesQueue;
     }
 
     /**
@@ -703,7 +708,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
         synchronized (stateMutex) {
             if (localNodeMaster() == false && masterNode.equals(committedState.get().nodes().getMasterNode())) {
                 // flush any pending cluster states from old master, so it will not be set as master again
-                publishClusterState.pendingStatesQueue().failAllStatesAndClear(new ElasticsearchException("master left [{}]", reason));
+                pendingStatesQueue.failAllStatesAndClear(new ElasticsearchException("master left [{}]", reason));
                 rejoin("master left (reason = " + reason + ")");
             }
         }
@@ -713,7 +718,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
     boolean processNextCommittedClusterState(String reason) {
         assert Thread.holdsLock(stateMutex);
 
-        final ClusterState newClusterState = publishClusterState.pendingStatesQueue().getNextClusterStateToProcess();
+        final ClusterState newClusterState = pendingStatesQueue.getNextClusterStateToProcess();
         final ClusterState currentState = committedState.get();
         final ClusterState adaptedNewClusterState;
         // all pending states have been processed
@@ -742,7 +747,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
             }
         } catch (Exception e) {
             try {
-                publishClusterState.pendingStatesQueue().markAsFailed(newClusterState, e);
+                pendingStatesQueue.markAsFailed(newClusterState, e);
             } catch (Exception inner) {
                 inner.addSuppressed(e);
                 logger.error((Supplier<?>) () -> new ParameterizedMessage("unexpected exception while failing [{}]", reason), inner);
@@ -811,7 +816,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
                 @Override
                 public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
                     try {
-                        publishClusterState.pendingStatesQueue().markAsProcessed(newClusterState);
+                        pendingStatesQueue.markAsProcessed(newClusterState);
                     } catch (Exception e) {
                         onFailure(source, e);
                     }
@@ -823,7 +828,7 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
                     try {
                         // TODO: use cluster state uuid instead of full cluster state so that we don't keep reference to CS around
                         // for too long.
-                        publishClusterState.pendingStatesQueue().markAsFailed(newClusterState, e);
+                        pendingStatesQueue.markAsFailed(newClusterState, e);
                     } catch (Exception inner) {
                         inner.addSuppressed(e);
                         logger.error((Supplier<?>) () -> new ParameterizedMessage("unexpected exception while failing [{}]", reason), inner);
@@ -1066,16 +1071,64 @@ public class ZenDiscovery extends AbstractLifecycleComponent implements Discover
         }
     }
 
-    private class NewPendingClusterStateListener implements PublishClusterStateAction.NewPendingClusterStateListener {
+    @Override
+    public void onIncomingClusterState(ClusterState incomingState) {
+        validateIncomingState(logger, incomingState, committedState.get());
+        pendingStatesQueue.addPending(incomingState);
+    }
 
-        @Override
-        public void onNewClusterState(String reason) {
+    @Override
+    public void onClusterStateCommitted(String stateUUID, ActionListener<Void> processedListener) {
+        final ClusterState state = pendingStatesQueue.markAsCommitted(stateUUID,
+            new PendingClusterStatesQueue.StateProcessedListener() {
+                @Override
+                public void onNewClusterStateProcessed() {
+                    processedListener.onResponse(null);
+                }
+
+                @Override
+                public void onNewClusterStateFailed(Exception e) {
+                    processedListener.onFailure(e);
+                }
+            });
+        if (state != null) {
             synchronized (stateMutex) {
-                processNextCommittedClusterState(reason);
+                processNextCommittedClusterState("master " + state.nodes().getMasterNode() +
+                    " committed version [" + state.version() + "]");
             }
         }
     }
 
+    /**
+     * does simple sanity check of the incoming cluster state. Throws an exception on rejections.
+     */
+    static void validateIncomingState(Logger logger, ClusterState incomingState, ClusterState lastState) {
+        final ClusterName incomingClusterName = incomingState.getClusterName();
+        if (!incomingClusterName.equals(lastState.getClusterName())) {
+            logger.warn("received cluster state from [{}] which is also master but with a different cluster name [{}]",
+                incomingState.nodes().getMasterNode(), incomingClusterName);
+            throw new IllegalStateException("received state from a node that is not part of the cluster");
+        }
+        if (lastState.nodes().getLocalNode().equals(incomingState.nodes().getLocalNode()) == false) {
+            logger.warn("received a cluster state from [{}] and not part of the cluster, should not happen",
+                incomingState.nodes().getMasterNode());
+            throw new IllegalStateException("received state with a local node that does not match the current local node");
+        }
+
+        if (shouldIgnoreOrRejectNewClusterState(logger, lastState, incomingState)) {
+            String message = String.format(
+                Locale.ROOT,
+                "rejecting cluster state version [%d] uuid [%s] received from [%s]",
+                incomingState.version(),
+                incomingState.stateUUID(),
+                incomingState.nodes().getMasterNodeId()
+            );
+            logger.warn(message);
+            throw new IllegalStateException(message);
+        }
+
+    }
+
     private class MembershipListener implements MembershipAction.MembershipListener {
         @Override
         public void onJoin(DiscoveryNode node, MembershipAction.JoinCallback callback) {
diff --git a/core/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java b/core/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
index 1b243bccd7..7371bde893 100644
--- a/core/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
+++ b/core/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
@@ -33,6 +33,7 @@ import org.elasticsearch.cluster.action.shard.ShardStateAction;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.cluster.routing.AllocationId;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.RecoverySource.Type;
 import org.elasticsearch.cluster.routing.RoutingNode;
@@ -551,18 +552,16 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent imple
         try {
             shard.updateRoutingEntry(shardRouting);
             if (shardRouting.primary()) {
-                IndexShardRoutingTable indexShardRoutingTable = routingTable.shardRoutingTable(shardRouting.shardId());
-                Set<String> activeIds = indexShardRoutingTable.activeShards().stream()
-                    // filter to shards that track seq# and should be taken into consideration for checkpoint tracking
-                    // shards on old nodes will go through a file based recovery which will also transfer seq# information.
-                    .filter(sr -> nodes.get(sr.currentNodeId()).getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED))
-                    .map(r -> r.allocationId().getId())
-                    .collect(Collectors.toSet());
-                Set<String> initializingIds = indexShardRoutingTable.getAllInitializingShards().stream()
-                    .filter(sr -> nodes.get(sr.currentNodeId()).getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED))
-                    .map(r -> r.allocationId().getId())
-                    .collect(Collectors.toSet());
-               shard.updateAllocationIdsFromMaster(activeIds, initializingIds);
+                final IndexShardRoutingTable indexShardRoutingTable = routingTable.shardRoutingTable(shardRouting.shardId());
+                /*
+                 * Filter to shards that track sequence numbers and should be taken into consideration for checkpoint tracking. Shards on
+                 * old nodes will go through a file-based recovery which will also transfer sequence number information.
+                 */
+                final Set<String> activeIds =
+                        allocationIdsForShardsOnNodesThatUnderstandSeqNos(indexShardRoutingTable.activeShards(), nodes);
+                final Set<String> initializingIds =
+                        allocationIdsForShardsOnNodesThatUnderstandSeqNos(indexShardRoutingTable.getAllInitializingShards(), nodes);
+                shard.updateAllocationIdsFromMaster(activeIds, initializingIds);
             }
         } catch (Exception e) {
             failAndRemoveShard(shardRouting, true, "failed updating shard routing entry", e, clusterState);
@@ -586,6 +585,17 @@ public class IndicesClusterStateService extends AbstractLifecycleComponent imple
         }
     }
 
+    private Set<String> allocationIdsForShardsOnNodesThatUnderstandSeqNos(
+            final List<ShardRouting> shardRoutings,
+            final DiscoveryNodes nodes) {
+        return shardRoutings
+                .stream()
+                .filter(sr -> nodes.get(sr.currentNodeId()).getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED))
+                .map(ShardRouting::allocationId)
+                .map(AllocationId::getId)
+                .collect(Collectors.toSet());
+    }
+
     /**
      * Finds the routing source node for peer recovery, return null if its not found. Note, this method expects the shard
      * routing to *require* peer recovery, use {@link ShardRouting#recoverySource()} to check if its needed or not.
diff --git a/core/src/main/java/org/elasticsearch/monitor/jvm/JvmInfo.java b/core/src/main/java/org/elasticsearch/monitor/jvm/JvmInfo.java
index c18a1c5e3f..1674ba33c0 100644
--- a/core/src/main/java/org/elasticsearch/monitor/jvm/JvmInfo.java
+++ b/core/src/main/java/org/elasticsearch/monitor/jvm/JvmInfo.java
@@ -299,19 +299,19 @@ public class JvmInfo implements Writeable, ToXContent {
     public int versionAsInteger() {
         try {
             int i = 0;
-            String sVersion = "";
+            StringBuilder sVersion = new StringBuilder();
             for (; i < version.length(); i++) {
                 if (!Character.isDigit(version.charAt(i)) && version.charAt(i) != '.') {
                     break;
                 }
                 if (version.charAt(i) != '.') {
-                    sVersion += version.charAt(i);
+                    sVersion.append(version.charAt(i));
                 }
             }
             if (i == 0) {
                 return -1;
             }
-            return Integer.parseInt(sVersion);
+            return Integer.parseInt(sVersion.toString());
         } catch (Exception e) {
             return -1;
         }
@@ -320,19 +320,19 @@ public class JvmInfo implements Writeable, ToXContent {
     public int versionUpdatePack() {
         try {
             int i = 0;
-            String sVersion = "";
+            StringBuilder sVersion = new StringBuilder();
             for (; i < version.length(); i++) {
                 if (!Character.isDigit(version.charAt(i)) && version.charAt(i) != '.') {
                     break;
                 }
                 if (version.charAt(i) != '.') {
-                    sVersion += version.charAt(i);
+                    sVersion.append(version.charAt(i));
                 }
             }
             if (i == 0) {
                 return -1;
             }
-            Integer.parseInt(sVersion);
+            Integer.parseInt(sVersion.toString());
             int from;
             if (version.charAt(i) == '_') {
                 // 1.7.0_4
diff --git a/core/src/main/java/org/elasticsearch/node/Node.java b/core/src/main/java/org/elasticsearch/node/Node.java
index bbda2869ff..dbf3d21181 100644
--- a/core/src/main/java/org/elasticsearch/node/Node.java
+++ b/core/src/main/java/org/elasticsearch/node/Node.java
@@ -327,7 +327,6 @@ public class Node implements Closeable {
             final ResourceWatcherService resourceWatcherService = new ResourceWatcherService(settings, threadPool);
             final ScriptModule scriptModule = ScriptModule.create(settings, pluginsService.filterPlugins(ScriptPlugin.class));
             AnalysisModule analysisModule = new AnalysisModule(this.environment, pluginsService.filterPlugins(AnalysisPlugin.class));
-            additionalSettings.addAll(scriptModule.getSettings());
             // this is as early as we can validate settings at this point. we already pass them to ScriptModule as well as ThreadPool
             // so we might be late here already
             final SettingsModule settingsModule = new SettingsModule(this.settings, additionalSettings, additionalSettingsFilter);
diff --git a/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java b/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java
index 81620ec8a7..3db635b044 100644
--- a/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java
+++ b/core/src/main/java/org/elasticsearch/rest/BaseRestHandler.java
@@ -85,12 +85,12 @@ public abstract class BaseRestHandler extends AbstractComponent implements RestH
         final Set<String> invalids,
         final Set<String> candidates,
         final String detail) {
-        String message = String.format(
+        StringBuilder message = new StringBuilder(String.format(
             Locale.ROOT,
             "request [%s] contains unrecognized %s%s: ",
             request.path(),
             detail,
-            invalids.size() > 1 ? "s" : "");
+            invalids.size() > 1 ? "s" : ""));
         boolean first = true;
         for (final String invalid : invalids) {
             final LevensteinDistance ld = new LevensteinDistance();
@@ -108,17 +108,23 @@ public abstract class BaseRestHandler extends AbstractComponent implements RestH
                 else return a.v2().compareTo(b.v2());
             });
             if (first == false) {
-                message += ", ";
+                message.append(", ");
             }
-            message += "[" + invalid + "]";
+            message.append("[").append(invalid).append("]");
             final List<String> keys = scoredParams.stream().map(Tuple::v2).collect(Collectors.toList());
             if (keys.isEmpty() == false) {
-                message += " -> did you mean " + (keys.size() == 1 ? "[" + keys.get(0) + "]" : "any of " + keys.toString()) + "?";
+                message.append(" -> did you mean ");
+                if (keys.size() == 1) {
+                    message.append("[").append(keys.get(0)).append("]");
+                } else {
+                    message.append("any of ").append(keys.toString());
+                }
+                message.append("?");
             }
             first = false;
         }
 
-        return message;
+        return message.toString();
     }
 
     /**
diff --git a/core/src/main/java/org/elasticsearch/script/ScriptModes.java b/core/src/main/java/org/elasticsearch/script/ScriptModes.java
deleted file mode 100644
index ef1355d678..0000000000
--- a/core/src/main/java/org/elasticsearch/script/ScriptModes.java
+++ /dev/null
@@ -1,150 +0,0 @@
-/*
- * 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.script;
-
-import org.apache.lucene.util.SetOnce;
-import org.elasticsearch.common.settings.Setting;
-import org.elasticsearch.common.settings.Settings;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.function.Function;
-
-/**
- * Holds the boolean indicating the enabled mode for each of the different scripting languages available, each script source and each
- * scripted operation.
- */
-public class ScriptModes {
-
-    private static final String SCRIPT_SETTINGS_PREFIX = "script";
-    private static final String ENGINE_SETTINGS_PREFIX = "script.engine";
-
-    final Map<String, Boolean> scriptEnabled;
-
-    private static final Setting<List<String>> TYPES_ALLOWED_SETTING =
-        Setting.listSetting("script.types_allowed", Collections.emptyList(), Function.identity(), Setting.Property.NodeScope);
-    private static final Setting<List<String>> CONTEXTS_ALLOWED_SETTING =
-        Setting.listSetting("script.contexts_allowed", Collections.emptyList(), Function.identity(), Setting.Property.NodeScope);
-
-    private final Set<String> typesAllowed;
-    private final Set<String> contextsAllowed;
-
-    ScriptModes(ScriptContextRegistry scriptContextRegistry, ScriptSettings scriptSettings, Settings settings) {
-        HashMap<String, Boolean> scriptModes = new HashMap<>();
-        for (Setting<Boolean> scriptModeSetting : scriptSettings.getScriptLanguageSettings()) {
-            scriptModes.put(scriptModeSetting.getKey(), scriptModeSetting.get(settings));
-        }
-        this.scriptEnabled = Collections.unmodifiableMap(scriptModes);
-
-        typesAllowed = TYPES_ALLOWED_SETTING.exists(settings) ? new HashSet<>() : null;
-
-        if (typesAllowed != null) {
-            for (String settingType : TYPES_ALLOWED_SETTING.get(settings)) {
-                boolean found = false;
-
-                for (ScriptType scriptType : ScriptType.values()) {
-                    if (scriptType.getName().equals(settingType)) {
-                        found = true;
-                        typesAllowed.add(settingType);
-
-                        break;
-                    }
-                }
-
-                if (!found) {
-                    throw new IllegalArgumentException(
-                        "unknown script type [" + settingType + "] found in setting [" + TYPES_ALLOWED_SETTING.getKey() + "].");
-                }
-            }
-        }
-
-        contextsAllowed = CONTEXTS_ALLOWED_SETTING.exists(settings) ? new HashSet<>() : null;
-
-        if (contextsAllowed != null) {
-            for (String settingContext : CONTEXTS_ALLOWED_SETTING.get(settings)) {
-                if (scriptContextRegistry.isSupportedContext(settingContext)) {
-                    contextsAllowed.add(settingContext);
-                } else {
-                    throw new IllegalArgumentException(
-                        "unknown script context [" + settingContext + "] found in setting [" + CONTEXTS_ALLOWED_SETTING.getKey() + "].");
-                }
-            }
-        }
-    }
-
-    /**
-     * Returns the script mode for a script of a certain written in a certain language,
-     * of a certain type and executing as part of a specific operation/api.
-     *
-     * @param lang the language that the script is written in
-     * @param scriptType the type of the script
-     * @param scriptContext the operation that requires the execution of the script
-     * @return whether scripts are enabled (true) or disabled (false)
-     */
-    public boolean getScriptEnabled(String lang, ScriptType scriptType, ScriptContext scriptContext) {
-        if (typesAllowed != null && typesAllowed.contains(scriptType.getName()) == false) {
-            throw new IllegalArgumentException("[" + scriptType.getName() + "] scripts cannot be executed");
-        }
-
-        if (contextsAllowed != null && contextsAllowed.contains(scriptContext.getKey()) == false) {
-            throw new IllegalArgumentException("[" + scriptContext.getKey() + "] scripts cannot be executed");
-        }
-
-        Boolean scriptMode = scriptEnabled.get(getKey(lang, scriptType, scriptContext));
-        if (scriptMode == null) {
-            throw new IllegalArgumentException("script mode not found for lang [" + lang + "], script_type [" + scriptType + "], operation [" + scriptContext.getKey() + "]");
-        }
-        return scriptMode;
-    }
-
-    static String operationKey(ScriptContext scriptContext) {
-        return SCRIPT_SETTINGS_PREFIX + "." + scriptContext.getKey();
-    }
-
-    static String sourceKey(ScriptType scriptType) {
-        return SCRIPT_SETTINGS_PREFIX + "." + scriptType.getName();
-    }
-
-    static String getGlobalKey(String lang, ScriptType scriptType) {
-        return ENGINE_SETTINGS_PREFIX + "." + lang + "." + scriptType;
-    }
-
-    static String getKey(String lang, ScriptType scriptType, ScriptContext scriptContext) {
-        return ENGINE_SETTINGS_PREFIX + "." + lang + "." + scriptType + "." + scriptContext.getKey();
-    }
-
-    @Override
-    public String toString() {
-        //order settings by key before printing them out, for readability
-        TreeMap<String, Boolean> scriptModesTreeMap = new TreeMap<>();
-        scriptModesTreeMap.putAll(scriptEnabled);
-        StringBuilder stringBuilder = new StringBuilder();
-        for (Map.Entry<String, Boolean> stringScriptModeEntry : scriptModesTreeMap.entrySet()) {
-            stringBuilder.append(stringScriptModeEntry.getKey()).append(": ").append(stringScriptModeEntry.getValue()).append("\n");
-        }
-        return stringBuilder.toString();
-    }
-}
diff --git a/core/src/main/java/org/elasticsearch/script/ScriptModule.java b/core/src/main/java/org/elasticsearch/script/ScriptModule.java
index 29c9c90e76..1d7ecf119e 100644
--- a/core/src/main/java/org/elasticsearch/script/ScriptModule.java
+++ b/core/src/main/java/org/elasticsearch/script/ScriptModule.java
@@ -20,24 +20,18 @@
 package org.elasticsearch.script;
 
 import org.elasticsearch.common.settings.ClusterSettings;
-import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.env.Environment;
 import org.elasticsearch.plugins.ScriptPlugin;
-import org.elasticsearch.watcher.ResourceWatcherService;
 
 import java.io.IOException;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /**
- * Manages building {@link ScriptService} and {@link ScriptSettings} from a list of plugins.
+ * Manages building {@link ScriptService}.
  */
 public class ScriptModule {
-    private final ScriptSettings scriptSettings;
     private final ScriptService scriptService;
 
     /**
@@ -59,22 +53,14 @@ public class ScriptModule {
                         List<ScriptContext.Plugin> customScriptContexts) {
         ScriptContextRegistry scriptContextRegistry = new ScriptContextRegistry(customScriptContexts);
         ScriptEngineRegistry scriptEngineRegistry = new ScriptEngineRegistry(scriptEngines);
-        scriptSettings = new ScriptSettings(scriptEngineRegistry, scriptContextRegistry);
         try {
-            scriptService = new ScriptService(settings, scriptEngineRegistry, scriptContextRegistry, scriptSettings);
+            scriptService = new ScriptService(settings, scriptEngineRegistry, scriptContextRegistry);
         } catch (IOException e) {
             throw new RuntimeException("Couldn't setup ScriptService", e);
         }
     }
 
     /**
-     * Extra settings for scripts.
-     */
-    public List<Setting<?>> getSettings() {
-        return scriptSettings.getSettings();
-    }
-
-    /**
      * Service responsible for managing scripts.
      */
     public ScriptService getScriptService() {
diff --git a/core/src/main/java/org/elasticsearch/script/ScriptService.java b/core/src/main/java/org/elasticsearch/script/ScriptService.java
index 860bb31560..712fa5d3ea 100644
--- a/core/src/main/java/org/elasticsearch/script/ScriptService.java
+++ b/core/src/main/java/org/elasticsearch/script/ScriptService.java
@@ -19,12 +19,6 @@
 
 package org.elasticsearch.script;
 
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Map;
-import java.util.Objects;
-
 import org.apache.lucene.util.IOUtils;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
@@ -41,7 +35,6 @@ import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.breaker.CircuitBreakingException;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.cache.Cache;
 import org.elasticsearch.common.cache.CacheBuilder;
 import org.elasticsearch.common.cache.RemovalListener;
@@ -55,6 +48,16 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.search.lookup.SearchLookup;
 import org.elasticsearch.template.CompiledTemplate;
 
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+
 public class ScriptService extends AbstractComponent implements Closeable, ClusterStateListener {
 
     static final String DISABLE_DYNAMIC_SCRIPTING_SETTING = "script.disable_dynamic";
@@ -68,11 +71,20 @@ public class ScriptService extends AbstractComponent implements Closeable, Clust
     public static final Setting<Integer> SCRIPT_MAX_COMPILATIONS_PER_MINUTE =
         Setting.intSetting("script.max_compilations_per_minute", 15, 0, Property.Dynamic, Property.NodeScope);
 
+    public static final String ALLOW_NONE = "none";
+
+    public static final Setting<List<String>> TYPES_ALLOWED_SETTING =
+        Setting.listSetting("script.types_allowed", Collections.emptyList(), Function.identity(), Setting.Property.NodeScope);
+    public static final Setting<List<String>> CONTEXTS_ALLOWED_SETTING =
+        Setting.listSetting("script.contexts_allowed", Collections.emptyList(), Function.identity(), Setting.Property.NodeScope);
+
+    private final Set<String> typesAllowed;
+    private final Set<String> contextsAllowed;
+
     private final Map<String, ScriptEngine> engines;
 
     private final Cache<CacheKey, CompiledScript> cache;
 
-    private final ScriptModes scriptModes;
     private final ScriptContextRegistry scriptContextRegistry;
 
     private final ScriptMetrics scriptMetrics = new ScriptMetrics();
@@ -84,18 +96,87 @@ public class ScriptService extends AbstractComponent implements Closeable, Clust
     private double scriptsPerMinCounter;
     private double compilesAllowedPerNano;
 
-    public ScriptService(Settings settings, ScriptEngineRegistry scriptEngineRegistry,
-                         ScriptContextRegistry scriptContextRegistry, ScriptSettings scriptSettings) throws IOException {
+    public ScriptService(Settings settings, ScriptEngineRegistry scriptEngineRegistry, ScriptContextRegistry scriptContextRegistry) throws IOException {
         super(settings);
+
+        Objects.requireNonNull(settings);
         Objects.requireNonNull(scriptEngineRegistry);
         Objects.requireNonNull(scriptContextRegistry);
-        Objects.requireNonNull(scriptSettings);
+
         if (Strings.hasLength(settings.get(DISABLE_DYNAMIC_SCRIPTING_SETTING))) {
             throw new IllegalArgumentException(DISABLE_DYNAMIC_SCRIPTING_SETTING + " is not a supported setting, replace with fine-grained script settings. \n" +
-                    "Dynamic scripts can be enabled for all languages and all operations by replacing `script.disable_dynamic: false` with `script.inline: true` and `script.stored: true` in elasticsearch.yml");
+                    "Dynamic scripts can be enabled for all languages and all operations not using `script.disable_dynamic: false` in elasticsearch.yml");
+        }
+
+        this.typesAllowed = TYPES_ALLOWED_SETTING.exists(settings) ? new HashSet<>() : null;
+
+        if (this.typesAllowed != null) {
+            List<String> typesAllowedList = TYPES_ALLOWED_SETTING.get(settings);
+
+            if (typesAllowedList.isEmpty()) {
+                throw new IllegalArgumentException(
+                    "must specify at least one script type or none for setting [" + TYPES_ALLOWED_SETTING.getKey() + "].");
+            }
+
+            for (String settingType : typesAllowedList) {
+                if (ALLOW_NONE.equals(settingType)) {
+                    if (typesAllowedList.size() != 1) {
+                        throw new IllegalArgumentException("cannot specify both [" + ALLOW_NONE + "]" +
+                            " and other script types for setting [" + TYPES_ALLOWED_SETTING.getKey() + "].");
+                    } else {
+                        break;
+                    }
+                }
+
+                boolean found = false;
+
+                for (ScriptType scriptType : ScriptType.values()) {
+                    if (scriptType.getName().equals(settingType)) {
+                        found = true;
+                        this.typesAllowed.add(settingType);
+
+                        break;
+                    }
+                }
+
+                if (found == false) {
+                    throw new IllegalArgumentException(
+                        "unknown script type [" + settingType + "] found in setting [" + TYPES_ALLOWED_SETTING.getKey() + "].");
+                }
+            }
+        }
+
+        this.contextsAllowed = CONTEXTS_ALLOWED_SETTING.exists(settings) ? new HashSet<>() : null;
+
+        if (this.contextsAllowed != null) {
+            List<String> contextsAllowedList = CONTEXTS_ALLOWED_SETTING.get(settings);
+
+            if (contextsAllowedList.isEmpty()) {
+                throw new IllegalArgumentException(
+                    "must specify at least one script context or none for setting [" + CONTEXTS_ALLOWED_SETTING.getKey() + "].");
+            }
+
+            for (String settingContext : contextsAllowedList) {
+                if (ALLOW_NONE.equals(settingContext)) {
+                    if (contextsAllowedList.size() != 1) {
+                        throw new IllegalArgumentException("cannot specify both [" + ALLOW_NONE + "]" +
+                            " and other script contexts for setting [" + CONTEXTS_ALLOWED_SETTING.getKey() + "].");
+                    } else {
+                        break;
+                    }
+                }
+
+                if (scriptContextRegistry.isSupportedContext(settingContext)) {
+                    this.contextsAllowed.add(settingContext);
+                } else {
+                    throw new IllegalArgumentException(
+                        "unknown script context [" + settingContext + "] found in setting [" + CONTEXTS_ALLOWED_SETTING.getKey() + "].");
+                }
+            }
         }
 
         this.scriptContextRegistry = scriptContextRegistry;
+
         int cacheMaxSize = SCRIPT_CACHE_SIZE_SETTING.get(settings);
 
         CacheBuilder<CacheKey, CompiledScript> cacheBuilder = CacheBuilder.builder();
@@ -110,8 +191,9 @@ public class ScriptService extends AbstractComponent implements Closeable, Clust
 
         logger.debug("using script cache with max_size [{}], expire [{}]", cacheMaxSize, cacheExpire);
         this.cache = cacheBuilder.removalListener(new ScriptCacheRemovalListener()).build();
+
         this.engines = scriptEngineRegistry.getRegisteredLanguages();
-        this.scriptModes = new ScriptModes(scriptContextRegistry, scriptSettings, settings);
+
         this.lastInlineCompileTime = System.nanoTime();
         this.setMaxCompilationsPerMinute(SCRIPT_MAX_COMPILATIONS_PER_MINUTE.get(settings));
     }
@@ -194,9 +276,16 @@ public class ScriptService extends AbstractComponent implements Closeable, Clust
 
         ScriptEngine scriptEngine = getEngine(lang);
 
-        if (canExecuteScript(lang, type, scriptContext) == false) {
-            throw new IllegalStateException("scripts of type [" + script.getType() + "]," +
-                " operation [" + scriptContext.getKey() + "] and lang [" + lang + "] are disabled");
+        if (isTypeEnabled(type) == false) {
+            throw new IllegalArgumentException("cannot execute [" + type + "] scripts");
+        }
+
+        if (scriptContextRegistry.isSupportedContext(scriptContext.getKey()) == false) {
+            throw new IllegalArgumentException("script context [" + scriptContext.getKey() + "] not supported");
+        }
+
+        if (isContextEnabled(scriptContext) == false) {
+            throw new IllegalArgumentException("cannot execute scripts using [" + scriptContext.getKey() + "] context");
         }
 
         if (logger.isTraceEnabled()) {
@@ -288,6 +377,18 @@ public class ScriptService extends AbstractComponent implements Closeable, Clust
         return engines.containsKey(lang);
     }
 
+    public boolean isTypeEnabled(ScriptType scriptType) {
+        return typesAllowed == null || typesAllowed.contains(scriptType.getName());
+    }
+
+    public boolean isContextEnabled(ScriptContext scriptContext) {
+        return contextsAllowed == null || contextsAllowed.contains(scriptContext.getKey());
+    }
+
+    public boolean isAnyContextEnabled() {
+        return contextsAllowed == null || contextsAllowed.isEmpty() == false;
+    }
+
     StoredScriptSource getScriptFromClusterState(String id, String lang) {
         if (lang != null && isLangSupported(lang) == false) {
             throw new IllegalArgumentException("unable to get stored script with unsupported lang [" + lang + "]");
@@ -328,16 +429,19 @@ public class ScriptService extends AbstractComponent implements Closeable, Clust
         try {
             ScriptEngine scriptEngine = getEngine(source.getLang());
 
-            if (isAnyScriptContextEnabled(source.getLang(), ScriptType.STORED)) {
+            if (isTypeEnabled(ScriptType.STORED) == false) {
+                throw new IllegalArgumentException(
+                    "cannot put [" + ScriptType.STORED + "] script, [" + ScriptType.STORED + "] scripts are not enabled");
+            } else if (isAnyContextEnabled() == false) {
+                throw new IllegalArgumentException(
+                    "cannot put [" + ScriptType.STORED + "] script, no script contexts are enabled");
+            } else {
                 Object compiled = scriptEngine.compile(request.id(), source.getCode(), Collections.emptyMap());
 
                 if (compiled == null) {
                     throw new IllegalArgumentException("failed to parse/compile stored script [" + request.id() + "]" +
                         (source.getCode() == null ? "" : " using code [" + source.getCode() + "]"));
                 }
-            } else {
-                throw new IllegalArgumentException(
-                    "cannot put stored script [" + request.id() + "], stored scripts cannot be run under any context");
             }
         } catch (ScriptException good) {
             throw good;
@@ -422,23 +526,6 @@ public class ScriptService extends AbstractComponent implements Closeable, Clust
         return getEngine(compiledScript.lang()).search(compiledScript, lookup, params);
     }
 
-    private boolean isAnyScriptContextEnabled(String lang, ScriptType scriptType) {
-        for (ScriptContext scriptContext : scriptContextRegistry.scriptContexts()) {
-            if (canExecuteScript(lang, scriptType, scriptContext)) {
-                return true;
-            }
-        }
-        return false;
-    }
-
-    private boolean canExecuteScript(String lang, ScriptType scriptType, ScriptContext scriptContext) {
-        assert lang != null;
-        if (scriptContextRegistry.isSupportedContext(scriptContext.getKey()) == false) {
-            throw new IllegalArgumentException("script context [" + scriptContext.getKey() + "] not supported");
-        }
-        return scriptModes.getScriptEnabled(lang, scriptType, scriptContext);
-    }
-
     public ScriptStats stats() {
         return scriptMetrics.stats();
     }
diff --git a/core/src/main/java/org/elasticsearch/script/ScriptSettings.java b/core/src/main/java/org/elasticsearch/script/ScriptSettings.java
deleted file mode 100644
index 306f5dc068..0000000000
--- a/core/src/main/java/org/elasticsearch/script/ScriptSettings.java
+++ /dev/null
@@ -1,155 +0,0 @@
-/*
- * 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.script;
-
-import org.elasticsearch.common.settings.Setting;
-import org.elasticsearch.common.settings.Setting.Property;
-import org.elasticsearch.common.settings.Settings;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.EnumMap;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Function;
-
-public class ScriptSettings {
-
-    private static final Map<ScriptType, Setting<Boolean>> SCRIPT_TYPE_SETTING_MAP;
-
-    static {
-        Map<ScriptType, Setting<Boolean>> scriptTypeSettingMap = new EnumMap<>(ScriptType.class);
-        for (ScriptType scriptType : ScriptType.values()) {
-            scriptTypeSettingMap.put(scriptType, Setting.boolSetting(
-                ScriptModes.sourceKey(scriptType),
-                scriptType.isDefaultEnabled(),
-                Property.NodeScope,
-                Property.Deprecated));
-        }
-        SCRIPT_TYPE_SETTING_MAP = Collections.unmodifiableMap(scriptTypeSettingMap);
-    }
-
-    private final Map<ScriptContext, Setting<Boolean>> scriptContextSettingMap;
-    private final List<Setting<Boolean>> scriptLanguageSettings;
-
-    public ScriptSettings(ScriptEngineRegistry scriptEngineRegistry, ScriptContextRegistry scriptContextRegistry) {
-        Map<ScriptContext, Setting<Boolean>> scriptContextSettingMap = contextSettings(scriptContextRegistry);
-        this.scriptContextSettingMap = Collections.unmodifiableMap(scriptContextSettingMap);
-
-        List<Setting<Boolean>> scriptLanguageSettings = languageSettings(SCRIPT_TYPE_SETTING_MAP, scriptContextSettingMap, scriptEngineRegistry, scriptContextRegistry);
-        this.scriptLanguageSettings = Collections.unmodifiableList(scriptLanguageSettings);
-    }
-
-    private static Map<ScriptContext, Setting<Boolean>> contextSettings(ScriptContextRegistry scriptContextRegistry) {
-        Map<ScriptContext, Setting<Boolean>> scriptContextSettingMap = new HashMap<>();
-        for (ScriptContext scriptContext : scriptContextRegistry.scriptContexts()) {
-            scriptContextSettingMap.put(scriptContext,
-                    Setting.boolSetting(ScriptModes.operationKey(scriptContext), false, Property.NodeScope, Property.Deprecated));
-        }
-        return scriptContextSettingMap;
-    }
-
-    private static List<Setting<Boolean>> languageSettings(Map<ScriptType, Setting<Boolean>> scriptTypeSettingMap,
-                                                              Map<ScriptContext, Setting<Boolean>> scriptContextSettingMap,
-                                                              ScriptEngineRegistry scriptEngineRegistry,
-                                                              ScriptContextRegistry scriptContextRegistry) {
-        final List<Setting<Boolean>> scriptModeSettings = new ArrayList<>();
-
-        for (final Class<? extends ScriptEngine> scriptEngineService : scriptEngineRegistry.getRegisteredScriptEngineServices()) {
-            final String language = scriptEngineRegistry.getLanguage(scriptEngineService);
-            for (final ScriptType scriptType : ScriptType.values()) {
-                // Top level, like "script.engine.groovy.inline"
-                final boolean defaultNonFileScriptMode = scriptEngineRegistry.getDefaultInlineScriptEnableds().get(language);
-                boolean defaultLangAndType = defaultNonFileScriptMode;
-                // Files are treated differently because they are never default-deny
-                final boolean defaultIfNothingSet = defaultLangAndType;
-
-                Function<Settings, String> defaultLangAndTypeFn = settings -> {
-                    final Setting<Boolean> globalTypeSetting = scriptTypeSettingMap.get(scriptType);
-                    final Setting<Boolean> langAndTypeSetting = Setting.boolSetting(ScriptModes.getGlobalKey(language, scriptType),
-                            defaultIfNothingSet, Property.NodeScope, Property.Deprecated);
-
-                    if (langAndTypeSetting.exists(settings)) {
-                        // fine-grained e.g. script.engine.groovy.inline
-                        return langAndTypeSetting.get(settings).toString();
-                    } else if (globalTypeSetting.exists(settings)) {
-                        // global type - script.inline
-                        return globalTypeSetting.get(settings).toString();
-                    } else {
-                        return Boolean.toString(defaultIfNothingSet);
-                    }
-                };
-
-                // Setting for something like "script.engine.groovy.inline"
-                final Setting<Boolean> langAndTypeSetting = Setting.boolSetting(ScriptModes.getGlobalKey(language, scriptType),
-                        defaultLangAndTypeFn, Property.NodeScope, Property.Deprecated);
-                scriptModeSettings.add(langAndTypeSetting);
-
-                for (ScriptContext scriptContext : scriptContextRegistry.scriptContexts()) {
-                    final String langAndTypeAndContextName = ScriptModes.getKey(language, scriptType, scriptContext);
-                    // A function that, given a setting, will return what the default should be. Since the fine-grained script settings
-                    // read from a bunch of different places this is implemented in this way.
-                    Function<Settings, String> defaultSettingFn = settings -> {
-                        final Setting<Boolean> globalOpSetting = scriptContextSettingMap.get(scriptContext);
-                        final Setting<Boolean> globalTypeSetting = scriptTypeSettingMap.get(scriptType);
-                        final Setting<Boolean> langAndTypeAndContextSetting = Setting.boolSetting(langAndTypeAndContextName,
-                                defaultIfNothingSet, Property.NodeScope, Property.Deprecated);
-
-                        // fallback logic for script mode settings
-                        if (langAndTypeAndContextSetting.exists(settings)) {
-                            // like: "script.engine.groovy.inline.aggs: true"
-                            return langAndTypeAndContextSetting.get(settings).toString();
-                        } else if (langAndTypeSetting.exists(settings)) {
-                            // like: "script.engine.groovy.inline: true"
-                            return langAndTypeSetting.get(settings).toString();
-                        } else if (globalOpSetting.exists(settings)) {
-                            // like: "script.aggs: true"
-                            return globalOpSetting.get(settings).toString();
-                        } else if (globalTypeSetting.exists(settings)) {
-                            // like: "script.inline: true"
-                            return globalTypeSetting.get(settings).toString();
-                        } else {
-                            // Nothing is set!
-                            return Boolean.toString(defaultIfNothingSet);
-                        }
-                    };
-                    // The actual setting for finest grained script settings
-                    Setting<Boolean> setting =
-                        Setting.boolSetting(langAndTypeAndContextName, defaultSettingFn, Property.NodeScope, Property.Deprecated);
-                    scriptModeSettings.add(setting);
-                }
-            }
-        }
-        return scriptModeSettings;
-    }
-
-    public List<Setting<?>> getSettings() {
-        List<Setting<?>> settings = new ArrayList<>();
-        settings.addAll(SCRIPT_TYPE_SETTING_MAP.values());
-        settings.addAll(scriptContextSettingMap.values());
-        settings.addAll(scriptLanguageSettings);
-        return settings;
-    }
-
-    public Iterable<Setting<Boolean>> getScriptLanguageSettings() {
-        return scriptLanguageSettings;
-    }
-}
diff --git a/core/src/main/java/org/elasticsearch/script/ScriptType.java b/core/src/main/java/org/elasticsearch/script/ScriptType.java
index cb77e37404..c076ccfd88 100644
--- a/core/src/main/java/org/elasticsearch/script/ScriptType.java
+++ b/core/src/main/java/org/elasticsearch/script/ScriptType.java
@@ -28,8 +28,8 @@ import java.io.IOException;
 
 /**
  * ScriptType represents the way a script is stored and retrieved from the {@link ScriptService}.
- * It's also used to by {@link ScriptSettings} and {@link ScriptModes} to determine whether or not
- * a {@link Script} is allowed to be executed based on both default and user-defined settings.
+ * It's also used to by {@link ScriptService} to determine whether or not a {@link Script} is
+ * allowed to be executed based on both default and user-defined settings.
  */
 public enum ScriptType implements Writeable {
 
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java
index c3eab06f28..6082366094 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalDateHistogram.java
@@ -369,8 +369,8 @@ public final class InternalDateHistogram extends InternalMultiBucketAggregation<
             Bucket firstBucket = iter.hasNext() ? list.get(iter.nextIndex()) : null;
             if (firstBucket == null) {
                 if (bounds.getMin() != null && bounds.getMax() != null) {
-                    long key = bounds.getMin();
-                    long max = bounds.getMax();
+                    long key = bounds.getMin() + offset;
+                    long max = bounds.getMax() + offset;
                     while (key <= max) {
                         iter.add(new InternalDateHistogram.Bucket(key, 0, keyed, format, reducedEmptySubAggs));
                         key = nextKey(key).longValue();
@@ -378,7 +378,7 @@ public final class InternalDateHistogram extends InternalMultiBucketAggregation<
                 }
             } else {
                 if (bounds.getMin() != null) {
-                    long key = bounds.getMin();
+                    long key = bounds.getMin() + offset;
                     if (key < firstBucket.key) {
                         while (key < firstBucket.key) {
                             iter.add(new InternalDateHistogram.Bucket(key, 0, keyed, format, reducedEmptySubAggs));
@@ -405,12 +405,12 @@ public final class InternalDateHistogram extends InternalMultiBucketAggregation<
         }
 
         // finally, adding the empty buckets *after* the actual data (based on the extended_bounds.max requested by the user)
-        if (bounds != null && lastBucket != null && bounds.getMax() != null && bounds.getMax() > lastBucket.key) {
-            long key = emptyBucketInfo.rounding.nextRoundingValue(lastBucket.key);
-            long max = bounds.getMax();
+        if (bounds != null && lastBucket != null && bounds.getMax() != null && bounds.getMax() + offset > lastBucket.key) {
+            long key = nextKey(lastBucket.key).longValue();
+            long max = bounds.getMax() + offset;
             while (key <= max) {
                 iter.add(new InternalDateHistogram.Bucket(key, 0, keyed, format, reducedEmptySubAggs));
-                key = emptyBucketInfo.rounding.nextRoundingValue(key);
+                key = nextKey(key).longValue();
             }
         }
     }
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantLongTerms.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantLongTerms.java
index c88c7c8e1f..29a8e16d9b 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantLongTerms.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantLongTerms.java
@@ -77,11 +77,6 @@ public class SignificantLongTerms extends InternalMappedSignificantTerms<Signifi
         }
 
         @Override
-        public int compareTerm(SignificantTerms.Bucket other) {
-            return Long.compare(term, ((Number) other.getKey()).longValue());
-        }
-
-        @Override
         public String getKeyAsString() {
             return format.format(term);
         }
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantStringTerms.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantStringTerms.java
index 6627a2295d..b170171f5c 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantStringTerms.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantStringTerms.java
@@ -28,7 +28,6 @@ import org.elasticsearch.search.aggregations.bucket.significant.heuristics.Signi
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 
 import java.io.IOException;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -83,11 +82,6 @@ public class SignificantStringTerms extends InternalMappedSignificantTerms<Signi
         }
 
         @Override
-        public int compareTerm(SignificantTerms.Bucket other) {
-            return termBytes.compareTo(((Bucket) other).termBytes);
-        }
-
-        @Override
         public String getKeyAsString() {
             return format.format(termBytes);
         }
diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantTerms.java b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantTerms.java
index ae09feef93..8c0da8b890 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantTerms.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantTerms.java
@@ -40,8 +40,6 @@ public interface SignificantTerms extends MultiBucketsAggregation, Iterable<Sign
         long getSupersetSize();
 
         long getSubsetSize();
-
-        int compareTerm(SignificantTerms.Bucket other);
     }
 
     @Override