Introduce ElectionScheduler (#32846)

The ElectionScheduler runs while there is no known elected master and is
responsible for scheduling elections randomly, backing off on failure, to
balance the desire to elect a master quickly with the desire to avoid more than
one node starting an election at once.

server/src/main/java/org/elasticsearch/cluster/coordination/ElectionSchedulerFactory.java

@@ -0,0 +1,194 @@
+/*
+ * 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.cluster.coordination;
+
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.common.SuppressForbidden;
+import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Setting.Property;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.threadpool.ThreadPool.Names;
+
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * It's provably impossible to guarantee that any leader election algorithm ever elects a leader, but they generally work (with probability
+ * that approaches 1 over time) as long as elections occur sufficiently infrequently, compared to the time it takes to send a message to
+ * another node and receive a response back. We do not know the round-trip latency here, but we can approximate it by attempting elections
+ * randomly at reasonably high frequency and backing off (linearly) until one of them succeeds. We also place an upper bound on the backoff
+ * so that if elections are failing due to a network partition that lasts for a long time then when the partition heals there is an election
+ * attempt reasonably quickly.
+ */
+public class ElectionSchedulerFactory extends AbstractComponent {
+
+    private static final String ELECTION_INITIAL_TIMEOUT_SETTING_KEY = "cluster.election.initial_timeout";
+    private static final String ELECTION_BACK_OFF_TIME_SETTING_KEY = "cluster.election.back_off_time";
+    private static final String ELECTION_MAX_TIMEOUT_SETTING_KEY = "cluster.election.max_timeout";
+
+    /*
+     * The first election is scheduled to occur a random number of milliseconds after the scheduler is started, where the random number of
+     * milliseconds is chosen uniformly from
+     *
+     *     (0, min(ELECTION_INITIAL_TIMEOUT_SETTING, ELECTION_MAX_TIMEOUT_SETTING)]
+     *
+     * For `n > 1`, the `n`th election is scheduled to occur a random number of milliseconds after the `n - 1`th election, where the random
+     * number of milliseconds is chosen uniformly from
+     *
+     *     (0, min(ELECTION_INITIAL_TIMEOUT_SETTING + (n-1) * ELECTION_BACK_OFF_TIME_SETTING, ELECTION_MAX_TIMEOUT_SETTING)]
+     */
+
+    public static final Setting<TimeValue> ELECTION_INITIAL_TIMEOUT_SETTING = Setting.timeSetting(ELECTION_INITIAL_TIMEOUT_SETTING_KEY,
+        TimeValue.timeValueMillis(100), TimeValue.timeValueMillis(1), TimeValue.timeValueSeconds(10), Property.NodeScope);
+
+    public static final Setting<TimeValue> ELECTION_BACK_OFF_TIME_SETTING = Setting.timeSetting(ELECTION_BACK_OFF_TIME_SETTING_KEY,
+        TimeValue.timeValueMillis(100), TimeValue.timeValueMillis(1), TimeValue.timeValueSeconds(60), Property.NodeScope);
+
+    public static final Setting<TimeValue> ELECTION_MAX_TIMEOUT_SETTING = Setting.timeSetting(ELECTION_MAX_TIMEOUT_SETTING_KEY,
+        TimeValue.timeValueSeconds(10), TimeValue.timeValueMillis(200), TimeValue.timeValueSeconds(300), Property.NodeScope);
+
+    private final TimeValue initialTimeout;
+    private final TimeValue backoffTime;
+    private final TimeValue maxTimeout;
+    private final ThreadPool threadPool;
+    private final Random random;
+
+    public ElectionSchedulerFactory(Settings settings, Random random, ThreadPool threadPool) {
+        super(settings);
+
+        this.random = random;
+        this.threadPool = threadPool;
+
+        initialTimeout = ELECTION_INITIAL_TIMEOUT_SETTING.get(settings);
+        backoffTime = ELECTION_BACK_OFF_TIME_SETTING.get(settings);
+        maxTimeout = ELECTION_MAX_TIMEOUT_SETTING.get(settings);
+
+        if (maxTimeout.millis() < initialTimeout.millis()) {
+            throw new IllegalArgumentException(new ParameterizedMessage("[{}] is [{}], but must be at least [{}] which is [{}]",
+                ELECTION_MAX_TIMEOUT_SETTING_KEY, maxTimeout, ELECTION_INITIAL_TIMEOUT_SETTING_KEY, initialTimeout).getFormattedMessage());
+        }
+    }
+
+    /**
+     * Start the process to schedule repeated election attempts.
+     *
+     * @param gracePeriod       An initial period to wait before attempting the first election.
+     * @param scheduledRunnable The action to run each time an election should be attempted.
+     */
+    public Releasable startElectionScheduler(TimeValue gracePeriod, Runnable scheduledRunnable) {
+        final ElectionScheduler scheduler = new ElectionScheduler();
+        scheduler.scheduleNextElection(gracePeriod, scheduledRunnable);
+        return scheduler;
+    }
+
+    @SuppressForbidden(reason = "Argument to Math.abs() is definitely not Long.MIN_VALUE")
+    private static long nonNegative(long n) {
+        return n == Long.MIN_VALUE ? 0 : Math.abs(n);
+    }
+
+    /**
+     * @param randomNumber a randomly-chosen long
+     * @param upperBound   inclusive upper bound
+     * @return a number in the range (0, upperBound]
+     */
+    // package-private for testing
+    static long toPositiveLongAtMost(long randomNumber, long upperBound) {
+        assert 0 < upperBound : upperBound;
+        return nonNegative(randomNumber) % upperBound + 1;
+    }
+
+    @Override
+    public String toString() {
+        return "ElectionSchedulerFactory{" +
+            "initialTimeout=" + initialTimeout +
+            ", backoffTime=" + backoffTime +
+            ", maxTimeout=" + maxTimeout +
+            '}';
+    }
+
+    private class ElectionScheduler implements Releasable {
+        private final AtomicBoolean isClosed = new AtomicBoolean();
+        private final AtomicLong attempt = new AtomicLong();
+
+        void scheduleNextElection(final TimeValue gracePeriod, final Runnable scheduledRunnable) {
+            if (isClosed.get()) {
+                logger.debug("{} not scheduling election", this);
+                return;
+            }
+
+            final long thisAttempt = attempt.getAndIncrement();
+            // to overflow here would take over a million years of failed election attempts, so we won't worry about that:
+            final long maxDelayMillis = Math.min(maxTimeout.millis(), initialTimeout.millis() + thisAttempt * backoffTime.millis());
+            final long delayMillis = toPositiveLongAtMost(random.nextLong(), maxDelayMillis) + gracePeriod.millis();
+            final Runnable runnable = new AbstractRunnable() {
+                @Override
+                public void onFailure(Exception e) {
+                    logger.debug(new ParameterizedMessage("unexpected exception in wakeup of {}", this), e);
+                    assert false : e;
+                }
+
+                @Override
+                protected void doRun() {
+                    if (isClosed.get()) {
+                        logger.debug("{} not starting election", this);
+                        return;
+                    }
+                    logger.debug("{} starting election", this);
+                    scheduledRunnable.run();
+                }
+
+                @Override
+                public void onAfter() {
+                    scheduleNextElection(TimeValue.ZERO, scheduledRunnable);
+                }
+
+                @Override
+                public String toString() {
+                    return "scheduleNextElection{gracePeriod=" + gracePeriod
+                        + ", thisAttempt=" + thisAttempt
+                        + ", maxDelayMillis=" + maxDelayMillis
+                        + ", delayMillis=" + delayMillis
+                        + ", " + ElectionScheduler.this + "}";
+                }
+            };
+
+            logger.debug("scheduling {}", runnable);
+            threadPool.schedule(TimeValue.timeValueMillis(delayMillis), Names.GENERIC, runnable);
+        }
+
+        @Override
+        public String toString() {
+            return "ElectionScheduler{attempt=" + attempt
+                + ", " + ElectionSchedulerFactory.this + "}";
+        }
+
+        @Override
+        public void close() {
+            boolean wasNotPreviouslyClosed = isClosed.compareAndSet(false, true);
+            assert wasNotPreviouslyClosed;
+        }
+    }
+}

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -30,6 +30,7 @@ import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.InternalClusterInfoService;
 import org.elasticsearch.cluster.NodeConnectionsService;
 import org.elasticsearch.cluster.action.index.MappingUpdatedAction;
+import org.elasticsearch.cluster.coordination.ElectionSchedulerFactory;
 import org.elasticsearch.cluster.metadata.IndexGraveyard;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.routing.OperationRouting;
@@ -425,6 +426,9 @@ public final class ClusterSettings extends AbstractScopedSettings {
                     OperationRouting.USE_ADAPTIVE_REPLICA_SELECTION_SETTING,
                     IndexGraveyard.SETTING_MAX_TOMBSTONES,
                     EnableAssignmentDecider.CLUSTER_TASKS_ALLOCATION_ENABLE_SETTING,
-                    PeerFinder.DISCOVERY_FIND_PEERS_INTERVAL_SETTING
+                    PeerFinder.DISCOVERY_FIND_PEERS_INTERVAL_SETTING,
+                    ElectionSchedulerFactory.ELECTION_INITIAL_TIMEOUT_SETTING,
+                    ElectionSchedulerFactory.ELECTION_BACK_OFF_TIME_SETTING,
+                    ElectionSchedulerFactory.ELECTION_MAX_TIMEOUT_SETTING
             )));
 }

server/src/main/java/org/elasticsearch/common/settings/Setting.java

@@ -1242,6 +1242,20 @@ public class Setting<T> implements ToXContentObject {
         return new GroupSetting(key, validator, properties);
     }
 
+    public static Setting<TimeValue> timeSetting(String key, TimeValue defaultValue, TimeValue minValue, TimeValue maxValue,
+                                                 Property... properties) {
+        return new Setting<>(key, (s) -> defaultValue.getStringRep(), (s) -> {
+            TimeValue timeValue = TimeValue.parseTimeValue(s, null, key);
+            if (timeValue.millis() < minValue.millis()) {
+                throw new IllegalArgumentException("Failed to parse value [" + s + "] for setting [" + key + "] must be >= " + minValue);
+            }
+            if (maxValue.millis() < timeValue.millis()) {
+                throw new IllegalArgumentException("Failed to parse value [" + s + "] for setting [" + key + "] must be <= " + maxValue);
+            }
+            return timeValue;
+        }, properties);
+    }
+
     public static Setting<TimeValue> timeSetting(String key, Function<Settings, TimeValue> defaultValue, TimeValue minValue,
                                                  Property... properties) {
         return new Setting<>(key, (s) -> defaultValue.apply(s).getStringRep(), (s) -> {

server/src/test/java/org/elasticsearch/cluster/coordination/DeterministicTaskQueue.java

@@ -74,6 +74,16 @@ public class DeterministicTaskQueue extends AbstractComponent {
         }
     }
 
+    public void runAllTasks(Random random) {
+        while (hasDeferredTasks() || hasRunnableTasks()) {
+            if (hasDeferredTasks() && random.nextBoolean()) {
+                advanceTime();
+            } else if (hasRunnableTasks()) {
+                runRandomTask(random);
+            }
+        }
+    }
+
     /**
      * @return whether there are any runnable tasks.
      */

server/src/test/java/org/elasticsearch/cluster/coordination/ElectionSchedulerFactoryTests.java

@@ -0,0 +1,225 @@
+/*
+ * 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.cluster.coordination;
+
+import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.settings.Settings.Builder;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.elasticsearch.cluster.coordination.ElectionSchedulerFactory.ELECTION_BACK_OFF_TIME_SETTING;
+import static org.elasticsearch.cluster.coordination.ElectionSchedulerFactory.ELECTION_INITIAL_TIMEOUT_SETTING;
+import static org.elasticsearch.cluster.coordination.ElectionSchedulerFactory.ELECTION_MAX_TIMEOUT_SETTING;
+import static org.elasticsearch.cluster.coordination.ElectionSchedulerFactory.toPositiveLongAtMost;
+import static org.elasticsearch.node.Node.NODE_NAME_SETTING;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
+
+public class ElectionSchedulerFactoryTests extends ESTestCase {
+
+    private TimeValue randomGracePeriod() {
+        return TimeValue.timeValueMillis(randomLongBetween(0, 10000));
+    }
+
+    private void assertElectionSchedule(final DeterministicTaskQueue deterministicTaskQueue,
+                                        final ElectionSchedulerFactory electionSchedulerFactory,
+                                        final long initialTimeout, final long backOffTime, final long maxTimeout) {
+
+        final TimeValue initialGracePeriod = randomGracePeriod();
+        final AtomicBoolean electionStarted = new AtomicBoolean();
+
+        try (Releasable ignored = electionSchedulerFactory.startElectionScheduler(initialGracePeriod,
+            () -> assertTrue(electionStarted.compareAndSet(false, true)))) {
+
+            long lastElectionTime = deterministicTaskQueue.getCurrentTimeMillis();
+            int electionCount = 0;
+
+            while (true) {
+                electionCount++;
+
+                while (electionStarted.get() == false) {
+                    if (deterministicTaskQueue.hasRunnableTasks() == false) {
+                        deterministicTaskQueue.advanceTime();
+                    }
+                    deterministicTaskQueue.runAllRunnableTasks(random());
+                }
+                assertTrue(electionStarted.compareAndSet(true, false));
+
+                final long thisElectionTime = deterministicTaskQueue.getCurrentTimeMillis();
+
+                if (electionCount == 1) {
+                    final long electionDelay = thisElectionTime - lastElectionTime;
+
+                    // Check grace period
+                    assertThat(electionDelay, greaterThanOrEqualTo(initialGracePeriod.millis()));
+
+                    // Check upper bound
+                    assertThat(electionDelay, lessThanOrEqualTo(initialTimeout + initialGracePeriod.millis()));
+                    assertThat(electionDelay, lessThanOrEqualTo(maxTimeout + initialGracePeriod.millis()));
+
+                } else {
+
+                    final long electionDelay = thisElectionTime - lastElectionTime;
+
+                    // Check upper bound
+                    assertThat(electionDelay, lessThanOrEqualTo(initialTimeout + backOffTime * (electionCount - 1)));
+                    assertThat(electionDelay, lessThanOrEqualTo(maxTimeout));
+
+                    // Run until we get a delay close to the maximum to show that backing off does work
+                    if (electionCount >= 1000) {
+                        if (electionDelay >= maxTimeout * 0.99) {
+                            break;
+                        }
+                    }
+                }
+
+                lastElectionTime = thisElectionTime;
+            }
+        }
+        deterministicTaskQueue.runAllTasks(random());
+        assertFalse(electionStarted.get());
+    }
+
+    public void testRetriesOnCorrectSchedule() {
+        final Builder settingsBuilder = Settings.builder();
+
+        final long initialTimeoutMillis;
+        if (randomBoolean()) {
+            initialTimeoutMillis = randomLongBetween(1, 10000);
+            settingsBuilder.put(ELECTION_INITIAL_TIMEOUT_SETTING.getKey(), initialTimeoutMillis + "ms");
+        } else {
+            initialTimeoutMillis = ELECTION_INITIAL_TIMEOUT_SETTING.get(Settings.EMPTY).millis();
+        }
+
+        if (randomBoolean()) {
+            settingsBuilder.put(ELECTION_BACK_OFF_TIME_SETTING.getKey(), randomLongBetween(1, 60000) + "ms");
+        }
+
+        if (ELECTION_MAX_TIMEOUT_SETTING.get(Settings.EMPTY).millis() < initialTimeoutMillis || randomBoolean()) {
+            settingsBuilder.put(ELECTION_MAX_TIMEOUT_SETTING.getKey(),
+                randomLongBetween(Math.max(200, initialTimeoutMillis), 180000) + "ms");
+        }
+
+        final Settings settings = settingsBuilder.put(NODE_NAME_SETTING.getKey(), "node").build();
+        final long initialTimeout = ELECTION_INITIAL_TIMEOUT_SETTING.get(settings).millis();
+        final long backOffTime = ELECTION_BACK_OFF_TIME_SETTING.get(settings).millis();
+        final long maxTimeout = ELECTION_MAX_TIMEOUT_SETTING.get(settings).millis();
+
+        final DeterministicTaskQueue deterministicTaskQueue = new DeterministicTaskQueue(settings);
+        final ElectionSchedulerFactory electionSchedulerFactory
+            = new ElectionSchedulerFactory(settings, random(), deterministicTaskQueue.getThreadPool());
+
+        assertElectionSchedule(deterministicTaskQueue, electionSchedulerFactory, initialTimeout, backOffTime, maxTimeout);
+
+        // do it again to show that the max is reset when the scheduler is restarted
+        assertElectionSchedule(deterministicTaskQueue, electionSchedulerFactory, initialTimeout, backOffTime, maxTimeout);
+    }
+
+    public void testSettingsValidation() {
+        {
+            final Settings settings = Settings.builder().put(ELECTION_INITIAL_TIMEOUT_SETTING.getKey(), "0s").build();
+            IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> ELECTION_INITIAL_TIMEOUT_SETTING.get(settings));
+            assertThat(e.getMessage(), is("Failed to parse value [0s] for setting [cluster.election.initial_timeout] must be >= 1ms"));
+        }
+
+        {
+            final Settings settings = Settings.builder().put(ELECTION_INITIAL_TIMEOUT_SETTING.getKey(), "10001ms").build();
+            IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> ELECTION_INITIAL_TIMEOUT_SETTING.get(settings));
+            assertThat(e.getMessage(), is("Failed to parse value [10001ms] for setting [cluster.election.initial_timeout] must be <= 10s"));
+        }
+
+        {
+            final Settings settings = Settings.builder().put(ELECTION_BACK_OFF_TIME_SETTING.getKey(), "0s").build();
+            IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> ELECTION_BACK_OFF_TIME_SETTING.get(settings));
+            assertThat(e.getMessage(), is("Failed to parse value [0s] for setting [cluster.election.back_off_time] must be >= 1ms"));
+        }
+
+        {
+            final Settings settings = Settings.builder().put(ELECTION_BACK_OFF_TIME_SETTING.getKey(), "60001ms").build();
+            IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> ELECTION_BACK_OFF_TIME_SETTING.get(settings));
+            assertThat(e.getMessage(), is("Failed to parse value [60001ms] for setting [cluster.election.back_off_time] must be <= 1m"));
+        }
+
+        {
+            final Settings settings = Settings.builder().put(ELECTION_MAX_TIMEOUT_SETTING.getKey(), "199ms").build();
+            IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> ELECTION_MAX_TIMEOUT_SETTING.get(settings));
+            assertThat(e.getMessage(), is("Failed to parse value [199ms] for setting [cluster.election.max_timeout] must be >= 200ms"));
+        }
+
+        {
+            final Settings settings = Settings.builder().put(ELECTION_MAX_TIMEOUT_SETTING.getKey(), "301s").build();
+            IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> ELECTION_MAX_TIMEOUT_SETTING.get(settings));
+            assertThat(e.getMessage(), is("Failed to parse value [301s] for setting [cluster.election.max_timeout] must be <= 5m"));
+        }
+
+        {
+            final long initialTimeoutMillis = randomLongBetween(1, 10000);
+            final long backOffMillis = randomLongBetween(1, 60000);
+            final long maxTimeoutMillis = randomLongBetween(Math.max(200, initialTimeoutMillis), 180000);
+
+            final Settings settings = Settings.builder()
+                .put(ELECTION_INITIAL_TIMEOUT_SETTING.getKey(), initialTimeoutMillis + "ms")
+                .put(ELECTION_BACK_OFF_TIME_SETTING.getKey(), backOffMillis + "ms")
+                .put(ELECTION_MAX_TIMEOUT_SETTING.getKey(), maxTimeoutMillis + "ms")
+                .build();
+
+            assertThat(ELECTION_INITIAL_TIMEOUT_SETTING.get(settings), is(TimeValue.timeValueMillis(initialTimeoutMillis)));
+            assertThat(ELECTION_BACK_OFF_TIME_SETTING.get(settings), is(TimeValue.timeValueMillis(backOffMillis)));
+            assertThat(ELECTION_MAX_TIMEOUT_SETTING.get(settings), is(TimeValue.timeValueMillis(maxTimeoutMillis)));
+
+            assertThat(new ElectionSchedulerFactory(settings, random(), null), not(nullValue())); // doesn't throw an IAE
+        }
+
+        {
+            final long initialTimeoutMillis = randomLongBetween(201, 10000);
+            final long maxTimeoutMillis = randomLongBetween(200, initialTimeoutMillis - 1);
+
+            final Settings settings = Settings.builder()
+                .put(ELECTION_INITIAL_TIMEOUT_SETTING.getKey(), initialTimeoutMillis + "ms")
+                .put(ELECTION_MAX_TIMEOUT_SETTING.getKey(), maxTimeoutMillis + "ms")
+                .build();
+
+            IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
+                () -> new ElectionSchedulerFactory(settings, random(), null));
+            assertThat(e.getMessage(), equalTo("[cluster.election.max_timeout] is ["
+                + TimeValue.timeValueMillis(maxTimeoutMillis)
+                + "], but must be at least [cluster.election.initial_timeout] which is ["
+                + TimeValue.timeValueMillis(initialTimeoutMillis) + "]"));
+        }
+    }
+
+    public void testRandomPositiveLongLessThan() {
+        for (long input : new long[]{0, 1, -1, Long.MIN_VALUE, Long.MAX_VALUE, randomLong()}) {
+            for (long upperBound : new long[]{1, 2, 3, 100, Long.MAX_VALUE}) {
+                long l = toPositiveLongAtMost(input, upperBound);
+                assertThat(l, greaterThan(0L));
+                assertThat(l, lessThanOrEqualTo(upperBound));
+            }
+        }
+    }
+}

server/src/test/java/org/elasticsearch/common/settings/SettingTests.java

@@ -754,6 +754,35 @@ public class SettingTests extends ESTestCase {
         assertThat(setting.get(Settings.EMPTY).getMillis(), equalTo(random.getMillis() * factor));
     }
 
+    public void testTimeValueBounds() {
+        Setting<TimeValue> settingWithLowerBound
+            = Setting.timeSetting("foo", TimeValue.timeValueSeconds(10), TimeValue.timeValueSeconds(5));
+        assertThat(settingWithLowerBound.get(Settings.EMPTY), equalTo(TimeValue.timeValueSeconds(10)));
+
+        assertThat(settingWithLowerBound.get(Settings.builder().put("foo", "5000ms").build()), equalTo(TimeValue.timeValueSeconds(5)));
+        IllegalArgumentException illegalArgumentException
+            = expectThrows(IllegalArgumentException.class,
+            () -> settingWithLowerBound.get(Settings.builder().put("foo", "4999ms").build()));
+
+        assertThat(illegalArgumentException.getMessage(), equalTo("Failed to parse value [4999ms] for setting [foo] must be >= 5s"));
+
+        Setting<TimeValue> settingWithBothBounds = Setting.timeSetting("bar",
+            TimeValue.timeValueSeconds(10), TimeValue.timeValueSeconds(5), TimeValue.timeValueSeconds(20));
+        assertThat(settingWithBothBounds.get(Settings.EMPTY), equalTo(TimeValue.timeValueSeconds(10)));
+
+        assertThat(settingWithBothBounds.get(Settings.builder().put("bar", "5000ms").build()), equalTo(TimeValue.timeValueSeconds(5)));
+        assertThat(settingWithBothBounds.get(Settings.builder().put("bar", "20000ms").build()), equalTo(TimeValue.timeValueSeconds(20)));
+        illegalArgumentException
+            = expectThrows(IllegalArgumentException.class,
+            () -> settingWithBothBounds.get(Settings.builder().put("bar", "4999ms").build()));
+        assertThat(illegalArgumentException.getMessage(), equalTo("Failed to parse value [4999ms] for setting [bar] must be >= 5s"));
+
+        illegalArgumentException
+            = expectThrows(IllegalArgumentException.class,
+            () -> settingWithBothBounds.get(Settings.builder().put("bar", "20001ms").build()));
+        assertThat(illegalArgumentException.getMessage(), equalTo("Failed to parse value [20001ms] for setting [bar] must be <= 20s"));
+    }
+
     public void testSettingsGroupUpdater() {
         Setting<Integer> intSetting = Setting.intSetting("prefix.foo", 1, Property.NodeScope, Property.Dynamic);
         Setting<Integer> intSetting2 = Setting.intSetting("prefix.same", 1, Property.NodeScope, Property.Dynamic);