diff --git a/modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java b/modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java index 6ab68c7b75f83..bf56000693f96 100644 --- a/modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java +++ b/modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceIT.java @@ -62,6 +62,7 @@ import org.elasticsearch.core.TimeValue; import org.elasticsearch.datastreams.DataStreamsPlugin; import org.elasticsearch.datastreams.lifecycle.health.DataStreamLifecycleHealthIndicatorService; +import org.elasticsearch.dlm.DataStreamLifecycleErrorStore; import org.elasticsearch.health.Diagnosis; import org.elasticsearch.health.GetHealthAction; import org.elasticsearch.health.HealthIndicatorResult; @@ -146,7 +147,7 @@ protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) { settings.put(DataStreamLifecycleService.DATA_STREAM_LIFECYCLE_POLL_INTERVAL, "1s"); settings.put(DataStreamLifecycle.CLUSTER_LIFECYCLE_DEFAULT_ROLLOVER_SETTING.getKey(), "min_docs=1,max_docs=1"); // we'll test DSL errors reach the health node, so we're lowering the threshold over which we report errors - settings.put(DataStreamLifecycleService.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING.getKey(), "3"); + settings.put(DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING.getKey(), "3"); return settings.build(); } diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java index 914cac69d02f0..3cc1776577cc1 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java @@ -181,7 +181,6 @@ public List> getSettings() { pluginSettings.add(DataStreamLifecycleService.DATA_STREAM_LIFECYCLE_POLL_INTERVAL_SETTING); pluginSettings.add(DataStreamLifecycleService.DATA_STREAM_MERGE_POLICY_TARGET_FLOOR_SEGMENT_SETTING); pluginSettings.add(DataStreamLifecycleService.DATA_STREAM_MERGE_POLICY_TARGET_FACTOR_SETTING); - pluginSettings.add(DataStreamLifecycleService.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING); return pluginSettings; } diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java index 77dd16957e585..8860bef3452bb 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java @@ -142,18 +142,6 @@ public class DataStreamLifecycleService implements ClusterStateListener, Closeab Setting.Property.Dynamic, Setting.Property.NodeScope ); - /** - * This setting controls how often we signal that an index is in the error state when it comes to its data stream lifecycle - * progression. - * The signalling is currently logging at the `error` level but in the future it can signify other types of signalling. - */ - public static final Setting DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING = Setting.intSetting( - "data_streams.lifecycle.signalling.error_retry_interval", - 10, - 1, - Setting.Property.Dynamic, - Setting.Property.NodeScope - ); public static final String DOWNSAMPLED_INDEX_PREFIX = "downsample-"; @@ -243,7 +231,7 @@ public DataStreamLifecycleService( this.pollInterval = DATA_STREAM_LIFECYCLE_POLL_INTERVAL_SETTING.get(settings); this.targetMergePolicyFloorSegment = DATA_STREAM_MERGE_POLICY_TARGET_FLOOR_SEGMENT_SETTING.get(settings); this.targetMergePolicyFactor = DATA_STREAM_MERGE_POLICY_TARGET_FACTOR_SETTING.get(settings); - this.signallingErrorRetryInterval = DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING.get(settings); + this.signallingErrorRetryInterval = DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING.get(settings); this.rolloverConfiguration = clusterService.getClusterSettings() .get(DataStreamLifecycle.CLUSTER_LIFECYCLE_DEFAULT_ROLLOVER_SETTING); this.defaultRepository = RepositoriesService.DEFAULT_REPOSITORY_SETTING.get(settings); @@ -284,7 +272,10 @@ public void init() { clusterService.getClusterSettings() .addSettingsUpdateConsumer(DATA_STREAM_MERGE_POLICY_TARGET_FLOOR_SEGMENT_SETTING, this::updateMergePolicyFloorSegment); clusterService.getClusterSettings() - .addSettingsUpdateConsumer(DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING, this::updateSignallingRetryThreshold); + .addSettingsUpdateConsumer( + DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING, + this::updateSignallingRetryThreshold + ); clusterService.getClusterSettings() .addSettingsUpdateConsumer(RepositoriesService.DEFAULT_REPOSITORY_SETTING, this::updateDefaultRepository); } diff --git a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisher.java b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisher.java index 9f30d6fdc9f8e..5455d394e3dfd 100644 --- a/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisher.java +++ b/modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisher.java @@ -26,8 +26,6 @@ import java.util.List; -import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING; - /** * Provides the infrastructure to send errors encountered by indices managed by data stream lifecycle service to the health node. */ @@ -59,13 +57,16 @@ public DataStreamLifecycleHealthInfoPublisher( this.client = client; this.clusterService = clusterService; this.errorStore = errorStore; - this.signallingErrorRetryInterval = DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING.get(settings); + this.signallingErrorRetryInterval = DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING.get(settings); this.maxNumberOfErrorsToPublish = DATA_STREAM_LIFECYCLE_MAX_ERRORS_TO_PUBLISH_SETTING.get(settings); } public void init() { clusterService.getClusterSettings() - .addSettingsUpdateConsumer(DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING, this::updateSignallingRetryThreshold); + .addSettingsUpdateConsumer( + DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING, + this::updateSignallingRetryThreshold + ); clusterService.getClusterSettings() .addSettingsUpdateConsumer(DATA_STREAM_LIFECYCLE_MAX_ERRORS_TO_PUBLISH_SETTING, this::updateNumberOfErrorsToPublish); } @@ -80,7 +81,7 @@ private void updateNumberOfErrorsToPublish(int newValue) { /** * Publishes the DSL errors that have passed the signaling threshold (as defined by - * {@link org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService#DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING} + * {@link DataStreamLifecycleErrorStore#DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING} */ public void publishDslErrorEntries(ActionListener actionListener) { DiscoveryNode currentHealthNode = HealthNode.findHealthNode(clusterService.state()); diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java index 507e91078fcb2..2006a9064d76e 100644 --- a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java +++ b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleServiceTests.java @@ -115,7 +115,6 @@ import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleFixtures.createDataStream; import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.DATA_STREAM_MERGE_POLICY_TARGET_FACTOR_SETTING; import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.DATA_STREAM_MERGE_POLICY_TARGET_FLOOR_SEGMENT_SETTING; -import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING; import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.DOWNSAMPLED_INDEX_PREFIX; import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.FORCE_MERGE_COMPLETED_TIMESTAMP_METADATA_KEY; import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.ONE_HUNDRED_MB; @@ -157,7 +156,6 @@ public void setupServices() { builtInClusterSettings.add(DataStreamLifecycleService.DATA_STREAM_LIFECYCLE_POLL_INTERVAL_SETTING); builtInClusterSettings.add(DATA_STREAM_MERGE_POLICY_TARGET_FLOOR_SEGMENT_SETTING); builtInClusterSettings.add(DATA_STREAM_MERGE_POLICY_TARGET_FACTOR_SETTING); - builtInClusterSettings.add(DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING); ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, builtInClusterSettings); clusterService = createClusterService(threadPool, clusterSettings); diff --git a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisherTests.java b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisherTests.java index 23e00fad13d90..609159930406b 100644 --- a/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisherTests.java +++ b/modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/health/DataStreamLifecycleHealthInfoPublisherTests.java @@ -43,7 +43,6 @@ import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.DATA_STREAM_MERGE_POLICY_TARGET_FACTOR_SETTING; import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.DATA_STREAM_MERGE_POLICY_TARGET_FLOOR_SEGMENT_SETTING; -import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING; import static org.elasticsearch.test.ClusterServiceUtils.createClusterService; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.notNullValue; @@ -71,7 +70,6 @@ public void setupServices() { builtInClusterSettings.add(DataStreamLifecycleService.DATA_STREAM_LIFECYCLE_POLL_INTERVAL_SETTING); builtInClusterSettings.add(DATA_STREAM_MERGE_POLICY_TARGET_FLOOR_SEGMENT_SETTING); builtInClusterSettings.add(DATA_STREAM_MERGE_POLICY_TARGET_FACTOR_SETTING); - builtInClusterSettings.add(DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING); ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, builtInClusterSettings); clusterService = createClusterService(threadPool, clusterSettings); diff --git a/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java b/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java index 9943614869e48..ce6ea769751ec 100644 --- a/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java +++ b/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java @@ -81,6 +81,7 @@ import org.elasticsearch.discovery.PeerFinder; import org.elasticsearch.discovery.SeedHostsResolver; import org.elasticsearch.discovery.SettingsBasedSeedHostsProvider; +import org.elasticsearch.dlm.DataStreamLifecycleErrorStore; import org.elasticsearch.env.Environment; import org.elasticsearch.env.NodeEnvironment; import org.elasticsearch.gateway.GatewayService; @@ -669,6 +670,7 @@ public void apply(Settings value, Settings current, Settings previous) { DataStreamGlobalRetentionSettings.FAILURE_STORE_DEFAULT_RETENTION_SETTING, ShardsAvailabilityHealthIndicatorService.REPLICA_UNASSIGNED_BUFFER_TIME, DataStreamFailureStoreSettings.DATA_STREAM_FAILURE_STORED_ENABLED_SETTING, + DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING, IndexingStatsSettings.RECENT_WRITE_LOAD_HALF_LIFE_SETTING, SearchStatsSettings.RECENT_READ_LOAD_HALF_LIFE_SETTING, TransportGetAllocationStatsAction.CACHE_TTL_SETTING, diff --git a/server/src/main/java/org/elasticsearch/dlm/DataStreamLifecycleErrorStore.java b/server/src/main/java/org/elasticsearch/dlm/DataStreamLifecycleErrorStore.java index 0b628c3164146..66459fb49d52f 100644 --- a/server/src/main/java/org/elasticsearch/dlm/DataStreamLifecycleErrorStore.java +++ b/server/src/main/java/org/elasticsearch/dlm/DataStreamLifecycleErrorStore.java @@ -15,6 +15,7 @@ import org.elasticsearch.action.datastreams.lifecycle.ErrorEntry; import org.elasticsearch.cluster.metadata.ProjectId; import org.elasticsearch.common.Strings; +import org.elasticsearch.common.settings.Setting; import org.elasticsearch.core.Nullable; import org.elasticsearch.core.Tuple; import org.elasticsearch.health.node.DslErrorInfo; @@ -42,6 +43,19 @@ public class DataStreamLifecycleErrorStore { private static final Logger logger = getLogger(DataStreamLifecycleErrorStore.class); + /** + * This setting controls how often we signal that an index is in the error state when it comes to its data stream lifecycle + * progression. + * The signalling is currently logging at the `error` level but in the future it can signify other types of signalling. + */ + public static final Setting DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING = Setting.intSetting( + "data_streams.lifecycle.signalling.error_retry_interval", + 10, + 1, + Setting.Property.Dynamic, + Setting.Property.NodeScope + ); + public static final int MAX_ERROR_MESSAGE_LENGTH = 1000; private final ConcurrentMap> projectMap = new ConcurrentHashMap<>(); private final LongSupplier nowSupplier; diff --git a/x-pack/plugin/dlm-frozen-transition/src/main/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionExecutor.java b/x-pack/plugin/dlm-frozen-transition/src/main/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionExecutor.java index 545f51efd8789..21a17483d4c69 100644 --- a/x-pack/plugin/dlm-frozen-transition/src/main/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionExecutor.java +++ b/x-pack/plugin/dlm-frozen-transition/src/main/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionExecutor.java @@ -7,6 +7,7 @@ package org.elasticsearch.xpack.dlm.frozen; +import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.util.concurrent.EsExecutors; import org.elasticsearch.common.util.concurrent.ThreadContext; @@ -43,9 +44,17 @@ class DlmFrozenTransitionExecutor implements Closeable { private final ExecutorService executor; private final int maxConcurrency; private final int maxQueueSize; + private final ClusterService clusterService; private final DataStreamLifecycleErrorStore errorStore; - - DlmFrozenTransitionExecutor(int maxConcurrency, int maxQueueSize, Settings settings, DataStreamLifecycleErrorStore errorStore) { + private volatile int errorRetryInterval; + + DlmFrozenTransitionExecutor( + ClusterService clusterService, + int maxConcurrency, + int maxQueueSize, + Settings settings, + DataStreamLifecycleErrorStore errorStore + ) { this.maxConcurrency = maxConcurrency; this.maxQueueSize = maxQueueSize; this.submittedTransitions = new ConcurrentHashMap<>(maxQueueSize); @@ -58,7 +67,21 @@ class DlmFrozenTransitionExecutor implements Closeable { } return thread; }, new ThreadContext(settings), EsExecutors.TaskTrackingConfig.DEFAULT); + this.clusterService = clusterService; this.errorStore = errorStore; + this.errorRetryInterval = DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING.get(settings); + } + + public void init() { + this.clusterService.getClusterSettings() + .addSettingsUpdateConsumer( + DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING, + this::updateErrorInterval + ); + } + + private void updateErrorInterval(int newInterval) { + this.errorRetryInterval = newInterval; } public boolean transitionSubmitted(String indexName) { @@ -120,7 +143,7 @@ public void run() { indexName, ex, Strings.format("Error executing transition for index [%s]", indexName), - 1 + errorRetryInterval ); } finally { submittedTransitions.remove(indexName); diff --git a/x-pack/plugin/dlm-frozen-transition/src/main/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionService.java b/x-pack/plugin/dlm-frozen-transition/src/main/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionService.java index 689cfbecc0ec1..2e49bd783aab9 100644 --- a/x-pack/plugin/dlm-frozen-transition/src/main/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionService.java +++ b/x-pack/plugin/dlm-frozen-transition/src/main/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionService.java @@ -146,11 +146,13 @@ private void startThreadPools() { synchronized (this) { if (closing.get() == false) { transitionExecutor = new DlmFrozenTransitionExecutor( + clusterService, maxConcurrency, maxQueueSize, clusterService.getSettings(), errorStore ); + transitionExecutor.init(); schedulerThreadExecutor = Executors.newSingleThreadScheduledExecutor( EsExecutors.daemonThreadFactory(clusterService.getSettings(), "dlm-frozen-transition-scheduler") ); diff --git a/x-pack/plugin/dlm-frozen-transition/src/test/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionExecutorTests.java b/x-pack/plugin/dlm-frozen-transition/src/test/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionExecutorTests.java index eb8cfff5b3f95..9f108933c484b 100644 --- a/x-pack/plugin/dlm-frozen-transition/src/test/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionExecutorTests.java +++ b/x-pack/plugin/dlm-frozen-transition/src/test/java/org/elasticsearch/xpack/dlm/frozen/DlmFrozenTransitionExecutorTests.java @@ -9,10 +9,16 @@ import org.elasticsearch.action.datastreams.lifecycle.ErrorEntry; import org.elasticsearch.cluster.metadata.ProjectId; +import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.util.concurrent.WrappedRunnable; import org.elasticsearch.dlm.DataStreamLifecycleErrorStore; +import org.elasticsearch.test.ClusterServiceUtils; import org.elasticsearch.test.ESTestCase; +import org.elasticsearch.threadpool.TestThreadPool; +import org.elasticsearch.threadpool.ThreadPool; +import org.junit.After; +import org.junit.Before; import java.util.ArrayList; import java.util.HashSet; @@ -29,9 +35,29 @@ import static org.hamcrest.Matchers.containsString; public class DlmFrozenTransitionExecutorTests extends ESTestCase { + private ThreadPool threadPool; + private ClusterService clusterService; + + @Before + public void setup() throws Exception { + super.setUp(); + this.threadPool = new TestThreadPool("test-dlm-frozen-transition-executor"); + this.clusterService = ClusterServiceUtils.createClusterService(threadPool); + } + + @After + public void tearDown() throws Exception { + if (this.clusterService != null) { + this.clusterService.close(); + } + if (this.threadPool != null) { + terminate(threadPool); + } + super.tearDown(); + } public void testTransitionSubmitted() throws Exception { - try (var executor = new DlmFrozenTransitionExecutor(2, 10, Settings.EMPTY, makeErrorStore())) { + try (var executor = new DlmFrozenTransitionExecutor(clusterService, 2, 10, Settings.EMPTY, makeErrorStore())) { var task = new TestDlmFrozenTransitionRunnable("running-index"); task.blockUntil = new CountDownLatch(1); @@ -48,7 +74,7 @@ public void testTransitionSubmitted() throws Exception { } public void testTransitionRemovedAfterCompletion() throws Exception { - try (var executor = new DlmFrozenTransitionExecutor(2, 100, Settings.EMPTY, makeErrorStore())) { + try (var executor = new DlmFrozenTransitionExecutor(clusterService, 2, 100, Settings.EMPTY, makeErrorStore())) { var task = new TestDlmFrozenTransitionRunnable("done-index"); executor.submit(task).get(10, TimeUnit.SECONDS); @@ -59,7 +85,7 @@ public void testTransitionRemovedAfterCompletion() throws Exception { public void testTransitionRemovedAfterFailure() throws Exception { var errorStore = makeErrorStore(); - try (var executor = new DlmFrozenTransitionExecutor(2, 100, Settings.EMPTY, errorStore)) { + try (var executor = new DlmFrozenTransitionExecutor(clusterService, 2, 100, Settings.EMPTY, errorStore)) { var runtimeTask = new TestDlmFrozenTransitionRunnable("exception-index"); runtimeTask.throwOnRun = new IllegalStateException("simulated failure"); executor.submit(runtimeTask).get(10, TimeUnit.SECONDS); @@ -72,7 +98,7 @@ public void testTransitionRemovedAfterFailure() throws Exception { public void testHasCapacity() throws Exception { int maxQueue = randomIntBetween(2, 50); - try (var executor = new DlmFrozenTransitionExecutor(1, maxQueue, Settings.EMPTY, makeErrorStore())) { + try (var executor = new DlmFrozenTransitionExecutor(clusterService, 1, maxQueue, Settings.EMPTY, makeErrorStore())) { CountDownLatch tasksStarted = new CountDownLatch(1); CountDownLatch firstTaskBlock = new CountDownLatch(1); CountDownLatch taskBlock = new CountDownLatch(1); @@ -102,7 +128,7 @@ public void testHasCapacity() throws Exception { } public void testShutdownNow() throws Exception { - var executor = new DlmFrozenTransitionExecutor(1, 10, Settings.EMPTY, makeErrorStore()); + var executor = new DlmFrozenTransitionExecutor(clusterService, 1, 10, Settings.EMPTY, makeErrorStore()); var task = new TestDlmFrozenTransitionRunnable("block-index"); task.blockUntil = new CountDownLatch(1); @@ -121,7 +147,7 @@ public void testShutdownNow() throws Exception { * This is the invariant that {@code checkForFrozenIndices} relies on to prevent re-submission of queued tasks. */ public void testTransitionSubmittedReturnsTrueForQueuedTask() throws Exception { - try (var executor = new DlmFrozenTransitionExecutor(1, 2, Settings.EMPTY, makeErrorStore())) { + try (var executor = new DlmFrozenTransitionExecutor(clusterService, 1, 2, Settings.EMPTY, makeErrorStore())) { CountDownLatch firstStarted = new CountDownLatch(1); CountDownLatch block = new CountDownLatch(1); @@ -147,7 +173,7 @@ public void testTransitionSubmittedReturnsTrueForQueuedTask() throws Exception { * must remove the index from {@code submittedTransitions} before rethrowing, so that a future poll can retry. */ public void testSubmitCleansUpEntryOnRejectedExecution() throws Exception { - var executor = new DlmFrozenTransitionExecutor(1, 1, Settings.EMPTY, makeErrorStore()); + var executor = new DlmFrozenTransitionExecutor(clusterService, 1, 1, Settings.EMPTY, makeErrorStore()); try { CountDownLatch block = new CountDownLatch(1); CountDownLatch firstStarted = new CountDownLatch(1); @@ -180,7 +206,7 @@ public void testSubmitCleansUpEntryOnRejectedExecution() throws Exception { * and had not yet started, not only the currently-executing task. */ public void testShutdownNowReturnsQueuedTasks() throws Exception { - var executor = new DlmFrozenTransitionExecutor(1, 5, Settings.EMPTY, makeErrorStore()); + var executor = new DlmFrozenTransitionExecutor(clusterService, 1, 5, Settings.EMPTY, makeErrorStore()); CountDownLatch block = new CountDownLatch(1); CountDownLatch firstStarted = new CountDownLatch(1); @@ -216,7 +242,7 @@ public void testShutdownNowReturnsQueuedTasks() throws Exception { */ public void testSimultaneousSubmissionsFromMultipleThreads() throws Exception { int maxConcurrency = between(2, 50); - try (var executor = new DlmFrozenTransitionExecutor(maxConcurrency, 1, Settings.EMPTY, makeErrorStore())) { + try (var executor = new DlmFrozenTransitionExecutor(clusterService, maxConcurrency, 1, Settings.EMPTY, makeErrorStore())) { CyclicBarrier barrier = new CyclicBarrier(maxConcurrency + 1); List> futures = new CopyOnWriteArrayList<>(); List errors = new CopyOnWriteArrayList<>();