From 5fa6ec40306a98a6e99d7b08b2e88d5281a4e7a9 Mon Sep 17 00:00:00 2001 From: Lee Hinman Date: Tue, 31 Mar 2026 16:43:14 -0600 Subject: [PATCH 1/3] Move DLM signalling error retry setting into core This moves the `DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING` setting into the core module so that it can be used by both the regular DLM service and the frozen conversion bits to set up the retry interval in the DLM error store. Relates to #145243 --- .../DataStreamLifecycleServiceIT.java | 3 +- .../datastreams/DataStreamsPlugin.java | 1 - .../lifecycle/DataStreamLifecycleService.java | 19 +++----- ...ataStreamLifecycleHealthInfoPublisher.java | 11 ++--- .../DataStreamLifecycleServiceTests.java | 3 +- ...reamLifecycleHealthInfoPublisherTests.java | 3 +- .../common/settings/ClusterSettings.java | 2 + .../dlm/DataStreamLifecycleErrorStore.java | 14 ++++++ .../frozen/DlmFrozenTransitionExecutor.java | 29 ++++++++++-- .../frozen/DlmFrozenTransitionService.java | 2 + .../DlmFrozenTransitionExecutorTests.java | 44 +++++++++++++++---- 11 files changed, 94 insertions(+), 37 deletions(-) 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..902f0414c0b9f 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,7 @@ 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); + builtInClusterSettings.add(DataStreamLifecycleErrorStore.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..09b75984be4a9 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,7 @@ 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); + builtInClusterSettings.add(DataStreamLifecycleErrorStore.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..ac3e38fe84c5b 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) { + this.threadPool.shutdownNow(); + } + 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<>(); From c74cb2346412dce532ad3bc4f888b56d1ed35e43 Mon Sep 17 00:00:00 2001 From: Lee Hinman Date: Wed, 1 Apr 2026 07:58:09 -0600 Subject: [PATCH 2/3] Remove extra unneeded setting registration --- .../datastreams/lifecycle/DataStreamLifecycleServiceTests.java | 1 - .../health/DataStreamLifecycleHealthInfoPublisherTests.java | 1 - 2 files changed, 2 deletions(-) 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 902f0414c0b9f..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 @@ -156,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(DataStreamLifecycleErrorStore.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 09b75984be4a9..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 @@ -70,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(DataStreamLifecycleErrorStore.DATA_STREAM_SIGNALLING_ERROR_RETRY_INTERVAL_SETTING); ClusterSettings clusterSettings = new ClusterSettings(Settings.EMPTY, builtInClusterSettings); clusterService = createClusterService(threadPool, clusterSettings); From ca54255bf233fa8a6c784c0c1194b73060c8ac15 Mon Sep 17 00:00:00 2001 From: Lee Hinman Date: Wed, 1 Apr 2026 07:58:22 -0600 Subject: [PATCH 3/3] Use `terminate` instead of shutdownNow --- .../xpack/dlm/frozen/DlmFrozenTransitionExecutorTests.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 ac3e38fe84c5b..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 @@ -51,7 +51,7 @@ public void tearDown() throws Exception { this.clusterService.close(); } if (this.threadPool != null) { - this.threadPool.shutdownNow(); + terminate(threadPool); } super.tearDown(); }