diff --git a/plugins/workload-management/src/javaRestTest/java/org/opensearch/rest/WorkloadManagementRestIT.java b/plugins/workload-management/src/javaRestTest/java/org/opensearch/rest/WorkloadManagementRestIT.java index 826faa393f8fc..aa72dbfaacc4b 100644 --- a/plugins/workload-management/src/javaRestTest/java/org/opensearch/rest/WorkloadManagementRestIT.java +++ b/plugins/workload-management/src/javaRestTest/java/org/opensearch/rest/WorkloadManagementRestIT.java @@ -148,14 +148,17 @@ public void testOperationWhenWlmDisabled() throws Exception { } public void testSearchSettings() throws Exception { - // Create with settings + // Create with all search settings String createJson = """ { "name": "search_test", "resiliency_mode": "enforced", "resource_limits": {"cpu": 0.3, "memory": 0.3}, "settings": { - "search.default_search_timeout": "30s" + "search.default_search_timeout": "30s", + "search.cancel_after_time_interval": "1m", + "search.max_concurrent_shard_requests": "5", + "search.batched_reduce_size": "512" } }"""; Response response = performOperation("PUT", "_wlm/workload_group", createJson); @@ -165,13 +168,20 @@ public void testSearchSettings() throws Exception { Response getResponse = performOperation("GET", "_wlm/workload_group/search_test", null); String responseBody = EntityUtils.toString(getResponse.getEntity()); assertTrue(responseBody.contains("\"settings\"")); + assertFalse(responseBody.contains("\"override_request_values\"")); assertTrue(responseBody.contains("\"search.default_search_timeout\":\"30s\"")); + assertTrue(responseBody.contains("\"search.cancel_after_time_interval\":\"1m\"")); + assertTrue(responseBody.contains("\"search.max_concurrent_shard_requests\":\"5\"")); + assertTrue(responseBody.contains("\"search.batched_reduce_size\":\"512\"")); - // Update settings + // Update search settings String updateJson = """ { "settings": { - "search.default_search_timeout": "1m" + "search.default_search_timeout": "1m", + "search.cancel_after_time_interval": "5m", + "search.max_concurrent_shard_requests": "10", + "search.batched_reduce_size": "256" } }"""; Response updateResponse = performOperation("PUT", "_wlm/workload_group/search_test", updateJson); @@ -181,10 +191,183 @@ public void testSearchSettings() throws Exception { Response getResponse2 = performOperation("GET", "_wlm/workload_group/search_test", null); String responseBody2 = EntityUtils.toString(getResponse2.getEntity()); assertTrue(responseBody2.contains("\"search.default_search_timeout\":\"1m\"")); + assertTrue(responseBody2.contains("\"search.cancel_after_time_interval\":\"5m\"")); + assertTrue(responseBody2.contains("\"search.max_concurrent_shard_requests\":\"10\"")); + assertTrue(responseBody2.contains("\"search.batched_reduce_size\":\"256\"")); performOperation("DELETE", "_wlm/workload_group/search_test", null); } + public void testSearchSettingsOverrideRequestValues() throws Exception { + // Create a WLM group with override_request_values=true so WLM settings win over request params + String createJson = """ + { + "name": "override_test", + "resiliency_mode": "enforced", + "resource_limits": {"cpu": 0.3, "memory": 0.3}, + "settings": { + "search.default_search_timeout": "30s", + "search.max_concurrent_shard_requests": "3", + "search.batched_reduce_size": "64", + "override_request_values": "true" + } + }"""; + Response response = performOperation("PUT", "_wlm/workload_group", createJson); + assertEquals(200, response.getStatusLine().getStatusCode()); + + // Verify override_request_values is "true" in GET response + Response getResponse = performOperation("GET", "_wlm/workload_group/override_test", null); + assertTrue(EntityUtils.toString(getResponse.getEntity()).contains("\"override_request_values\":\"true\"")); + + // Toggle to false and verify + String toggleJson = """ + {"settings": {"override_request_values": "false"}}"""; + Response toggleResponse = performOperation("PUT", "_wlm/workload_group/override_test", toggleJson); + assertEquals(200, toggleResponse.getStatusLine().getStatusCode()); + Response getResponse2 = performOperation("GET", "_wlm/workload_group/override_test", null); + assertTrue(EntityUtils.toString(getResponse2.getEntity()).contains("\"override_request_values\":\"false\"")); + + // Exercise the full request path: create an index, run a search with the WLM header. + // This confirms the listener is wired into the request flow without errors. Override + // semantics themselves are verified in WorkloadGroupRequestOperationListenerTests. + performOperation("PUT", "wlm-test-idx", "{\"settings\":{\"number_of_shards\":1,\"number_of_replicas\":0}}"); + performOperation("POST", "wlm-test-idx/_doc", "{\"msg\":\"hello\"}"); + performOperation("POST", "wlm-test-idx/_refresh", null); + + Request searchRequest = new Request("POST", "wlm-test-idx/_search"); + searchRequest.setJsonEntity("{\"query\":{\"match_all\":{}},\"timeout\":\"1m\"}"); + searchRequest.setOptions(searchRequest.getOptions().toBuilder().addHeader("X-opaque-id", "wlm=override_test")); + Response searchResponse = client().performRequest(searchRequest); + assertEquals(200, searchResponse.getStatusLine().getStatusCode()); + assertTrue(EntityUtils.toString(searchResponse.getEntity()).contains("\"hits\"")); + + performOperation("DELETE", "wlm-test-idx", null); + performOperation("DELETE", "_wlm/workload_group/override_test", null); + } + + public void testSearchSettingsInvalidSettingsRejected() throws Exception { + // Unknown setting key should be rejected + String unknownKeyJson = """ + { + "name": "invalid_test", + "resiliency_mode": "enforced", + "resource_limits": {"cpu": 0.3, "memory": 0.3}, + "settings": { + "unknown_setting": "value" + } + }"""; + ResponseException unknownKeyException = expectThrows( + ResponseException.class, + () -> performOperation("PUT", "_wlm/workload_group", unknownKeyJson) + ); + assertTrue(EntityUtils.toString(unknownKeyException.getResponse().getEntity()).contains("Unknown WLM setting: unknown_setting")); + + // Invalid value for max_concurrent_shard_requests (must be >= 1) + String invalidIntJson = """ + { + "name": "invalid_test", + "resiliency_mode": "enforced", + "resource_limits": {"cpu": 0.3, "memory": 0.3}, + "settings": { + "search.max_concurrent_shard_requests": "0" + } + }"""; + ResponseException invalidIntException = expectThrows( + ResponseException.class, + () -> performOperation("PUT", "_wlm/workload_group", invalidIntJson) + ); + String invalidIntBody = EntityUtils.toString(invalidIntException.getResponse().getEntity()); + assertTrue(invalidIntBody.contains("search.max_concurrent_shard_requests")); + assertTrue(invalidIntBody.contains("must be >= 1")); + + // Invalid value for batched_reduce_size (must be >= 2) + String invalidBatchJson = """ + { + "name": "invalid_test", + "resiliency_mode": "enforced", + "resource_limits": {"cpu": 0.3, "memory": 0.3}, + "settings": { + "search.batched_reduce_size": "1" + } + }"""; + ResponseException invalidBatchException = expectThrows( + ResponseException.class, + () -> performOperation("PUT", "_wlm/workload_group", invalidBatchJson) + ); + String invalidBatchBody = EntityUtils.toString(invalidBatchException.getResponse().getEntity()); + assertTrue(invalidBatchBody.contains("search.batched_reduce_size")); + assertTrue(invalidBatchBody.contains("must be >= 2")); + + // Invalid time value + String invalidTimeJson = """ + { + "name": "invalid_test", + "resiliency_mode": "enforced", + "resource_limits": {"cpu": 0.3, "memory": 0.3}, + "settings": { + "search.cancel_after_time_interval": "not_a_time" + } + }"""; + ResponseException invalidTimeException = expectThrows( + ResponseException.class, + () -> performOperation("PUT", "_wlm/workload_group", invalidTimeJson) + ); + String invalidTimeBody = EntityUtils.toString(invalidTimeException.getResponse().getEntity()); + assertTrue(invalidTimeBody.contains("search.cancel_after_time_interval")); + assertTrue(invalidTimeBody.contains("Invalid value")); + } + + public void testSearchSettingsMergeSemantics() throws Exception { + // Create with multiple settings + String createJson = """ + { + "name": "merge_test", + "resiliency_mode": "enforced", + "resource_limits": {"cpu": 0.3, "memory": 0.3}, + "settings": { + "search.default_search_timeout": "30s", + "search.max_concurrent_shard_requests": "5", + "override_request_values": "true" + } + }"""; + Response response = performOperation("PUT", "_wlm/workload_group", createJson); + assertEquals(200, response.getStatusLine().getStatusCode()); + + // Update only timeout — other settings should persist + String updateJson = """ + { + "settings": { + "search.default_search_timeout": "1m" + } + }"""; + Response updateResponse = performOperation("PUT", "_wlm/workload_group/merge_test", updateJson); + assertEquals(200, updateResponse.getStatusLine().getStatusCode()); + + // Verify merge: timeout updated, max_concurrent persists, override persists + Response getResponse = performOperation("GET", "_wlm/workload_group/merge_test", null); + String responseBody = EntityUtils.toString(getResponse.getEntity()); + assertTrue(responseBody.contains("\"search.default_search_timeout\":\"1m\"")); + assertTrue(responseBody.contains("\"search.max_concurrent_shard_requests\":\"5\"")); + assertTrue(responseBody.contains("\"override_request_values\":\"true\"")); + + // Clear all settings with empty object + String clearJson = """ + { + "settings": {} + }"""; + Response clearResponse = performOperation("PUT", "_wlm/workload_group/merge_test", clearJson); + assertEquals(200, clearResponse.getStatusLine().getStatusCode()); + + // Verify cleared — all settings should be gone + Response getResponse2 = performOperation("GET", "_wlm/workload_group/merge_test", null); + String responseBody2 = EntityUtils.toString(getResponse2.getEntity()); + assertFalse(responseBody2.contains("\"override_request_values\"")); + assertFalse(responseBody2.contains("\"search.default_search_timeout\"")); + assertFalse(responseBody2.contains("\"search.max_concurrent_shard_requests\"")); + + performOperation("DELETE", "_wlm/workload_group/merge_test", null); + } + static String getCreateJson(String name, String resiliencyMode, double cpu, double memory) { return String.format(Locale.ROOT, """ { diff --git a/server/src/main/java/org/opensearch/action/search/SearchRequest.java b/server/src/main/java/org/opensearch/action/search/SearchRequest.java index a1e6e7605cbdb..0adcd25c3a168 100644 --- a/server/src/main/java/org/opensearch/action/search/SearchRequest.java +++ b/server/src/main/java/org/opensearch/action/search/SearchRequest.java @@ -39,6 +39,7 @@ import org.opensearch.action.IndicesRequest; import org.opensearch.action.support.IndicesOptions; import org.opensearch.common.Nullable; +import org.opensearch.common.annotation.ExperimentalApi; import org.opensearch.common.annotation.PublicApi; import org.opensearch.common.io.stream.BytesStreamOutput; import org.opensearch.common.unit.TimeValue; @@ -649,6 +650,15 @@ public int getMaxConcurrentShardRequests() { return maxConcurrentShardRequests == 0 ? 5 : maxConcurrentShardRequests; } + /** + * Returns the raw value of maxConcurrentShardRequests without applying the default. + * A value of {@code 0} means the user has not explicitly set this parameter. + */ + @ExperimentalApi + public int getMaxConcurrentShardRequestsRaw() { + return maxConcurrentShardRequests; + } + /** * Sets the number of shard requests that should be executed concurrently on a single node. This value should be used as a * protection mechanism to reduce the number of shard requests fired per high level search request. Searches that hit the entire diff --git a/server/src/main/java/org/opensearch/cluster/metadata/WorkloadGroup.java b/server/src/main/java/org/opensearch/cluster/metadata/WorkloadGroup.java index 294c05ff17701..d2e14948c297e 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/WorkloadGroup.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/WorkloadGroup.java @@ -114,18 +114,31 @@ public static WorkloadGroup updateExistingWorkloadGroup( } final ResiliencyMode mode = Optional.ofNullable(mutableWorkloadGroupFragment.getResiliencyMode()) .orElse(existingGroup.getResiliencyMode()); - // Handle settings update: - // null = not specified (keep existing) - // empty Settings = explicitly clear (set to empty) - // non-empty Settings = replace with new values + // Handle settings update with merge semantics: + // null settings = not specified in request (keep existing) + // empty Settings = clear all settings + // non-empty Settings = merge with existing; keys with null values are removed final Settings mutableFragmentSettings = mutableWorkloadGroupFragment.getSettings(); final Settings updatedSettings; if (mutableFragmentSettings == null) { // Not specified - keep existing updatedSettings = Settings.builder().put(existingGroup.getSettings()).build(); + } else if (mutableFragmentSettings.isEmpty()) { + // Explicitly empty - clear all settings + updatedSettings = Settings.EMPTY; } else { - // Specified (empty or non-empty) - use the new value - updatedSettings = Settings.builder().put(mutableFragmentSettings).build(); + // Merge: start with existing settings, overlay new values, remove null-valued keys + Settings.Builder builder = Settings.builder().put(existingGroup.getSettings()); + for (String key : mutableFragmentSettings.keySet()) { + String value = mutableFragmentSettings.get(key); + if (value == null) { + // null value means "clear this setting" + builder.remove(key); + } else { + builder.put(key, value); + } + } + updatedSettings = builder.build(); } return new WorkloadGroup( existingGroup.getName(), @@ -284,6 +297,11 @@ public static Builder fromXContent(XContentParser parser) throws IOException { throw new IllegalArgumentException(fieldName + " is not a valid object in WorkloadGroup"); } mutableWorkloadGroupFragment1.parseField(parser, fieldName); + } else if (token == XContentParser.Token.VALUE_NULL) { + if (fieldName.equals(MutableWorkloadGroupFragment.SETTINGS_STRING)) { + // "settings": null means clear all settings + mutableWorkloadGroupFragment1.parseField(parser, fieldName); + } } } return builder.mutableWorkloadGroupFragment(mutableWorkloadGroupFragment1); diff --git a/server/src/main/java/org/opensearch/wlm/MutableWorkloadGroupFragment.java b/server/src/main/java/org/opensearch/wlm/MutableWorkloadGroupFragment.java index c87658cf72b40..0d767a5b9fa7b 100644 --- a/server/src/main/java/org/opensearch/wlm/MutableWorkloadGroupFragment.java +++ b/server/src/main/java/org/opensearch/wlm/MutableWorkloadGroupFragment.java @@ -56,7 +56,7 @@ public MutableWorkloadGroupFragment(ResiliencyMode resiliencyMode, Map parseField(XContentParser parser) throws IOExce static class SearchSettingsParser implements FieldParser { public Settings parseField(XContentParser parser) throws IOException { + // "settings": null means clear all settings + if (parser.currentToken() == XContentParser.Token.VALUE_NULL) { + return Settings.EMPTY; + } Settings settings = Settings.fromXContent(parser); WorkloadGroupSearchSettings.validate(settings); return settings; @@ -292,6 +296,7 @@ void setResourceLimits(Map resourceLimits) { void setSettings(Settings settings) { WorkloadGroupSearchSettings.validate(settings); - this.settings = settings; + this.settings = settings != null ? settings : Settings.EMPTY; } + } diff --git a/server/src/main/java/org/opensearch/wlm/WorkloadGroupSearchSettings.java b/server/src/main/java/org/opensearch/wlm/WorkloadGroupSearchSettings.java index 3974140e21a70..4ad771c5d3152 100644 --- a/server/src/main/java/org/opensearch/wlm/WorkloadGroupSearchSettings.java +++ b/server/src/main/java/org/opensearch/wlm/WorkloadGroupSearchSettings.java @@ -37,10 +37,54 @@ private WorkloadGroupSearchSettings() { */ public static final Setting WLM_SEARCH_TIMEOUT = Setting.timeSetting("search.default_search_timeout", TimeValue.MINUS_ONE); + /** + * The WLM cancel after time interval setting. Specifies the time after which a search + * request should be cancelled if it has not completed. + */ + public static final Setting WLM_CANCEL_AFTER_TIME_INTERVAL = Setting.timeSetting( + "search.cancel_after_time_interval", + TimeValue.MINUS_ONE + ); + + /** + * The WLM max concurrent shard requests setting. Controls the number of shard requests + * that should be executed concurrently on a single node. Must be >= 1. + */ + public static final Setting WLM_MAX_CONCURRENT_SHARD_REQUESTS = Setting.intSetting( + "search.max_concurrent_shard_requests", + 5, + 1 + ); + + /** + * The WLM batched reduce size setting. Controls the number of shard results to reduce + * at once on the coordinating node. Must be >= 2. + */ + public static final Setting WLM_BATCHED_REDUCE_SIZE = Setting.intSetting("search.batched_reduce_size", 512, 2); + + /** + * Controls whether WLM search settings should override values explicitly set in the + * search request query parameters. When {@code false} (default), WLM settings are only + * applied when the request does not have an explicit value. When {@code true}, WLM + * settings always take precedence over request-level values. + */ + public static final Setting WLM_OVERRIDE_REQUEST_VALUES = Setting.boolSetting("override_request_values", false); + /** * All registered WLM settings, keyed by their canonical key name. */ - private static final Map> REGISTERED_SETTINGS = Map.of("search.default_search_timeout", WLM_SEARCH_TIMEOUT); + private static final Map> REGISTERED_SETTINGS = Map.of( + "search.default_search_timeout", + WLM_SEARCH_TIMEOUT, + "search.cancel_after_time_interval", + WLM_CANCEL_AFTER_TIME_INTERVAL, + "search.max_concurrent_shard_requests", + WLM_MAX_CONCURRENT_SHARD_REQUESTS, + "search.batched_reduce_size", + WLM_BATCHED_REDUCE_SIZE, + "override_request_values", + WLM_OVERRIDE_REQUEST_VALUES + ); /** * Validates a {@link Settings} object against registered WLM settings. @@ -59,6 +103,10 @@ public static void validate(Settings settings) { if (setting == null) { throw new IllegalArgumentException("Unknown WLM setting: " + key); } + // null value means "clear this setting" — skip type validation + if (value == null) { + continue; + } try { Settings testSettings = Settings.builder().put(key, value).build(); setting.get(testSettings); @@ -67,13 +115,4 @@ public static void validate(Settings settings) { } } } - - /** - * Returns an unmodifiable view of the registered settings. - * - * @return map of canonical key names to their {@link Setting} objects - */ - public static Map> getRegisteredSettings() { - return REGISTERED_SETTINGS; - } } diff --git a/server/src/main/java/org/opensearch/wlm/listeners/WorkloadGroupRequestOperationListener.java b/server/src/main/java/org/opensearch/wlm/listeners/WorkloadGroupRequestOperationListener.java index 31221f95113eb..73f29b09776e1 100644 --- a/server/src/main/java/org/opensearch/wlm/listeners/WorkloadGroupRequestOperationListener.java +++ b/server/src/main/java/org/opensearch/wlm/listeners/WorkloadGroupRequestOperationListener.java @@ -56,32 +56,99 @@ protected void onRequestFailure(SearchPhaseContext context, SearchRequestContext /** * Applies workload group-specific search settings to the search request. * Settings are only applied for workload groups that exist in cluster state. + *

+ * When {@code override_request_values} is {@code false} (default), WLM settings are only + * applied when the request does not already have an explicit value set. + * When {@code true}, WLM settings always take precedence over request-level values. * * @param workloadGroupId the workload group identifier from thread context * @param searchRequest the search request to modify */ private void applyWorkloadGroupSearchSettings(String workloadGroupId, SearchRequest searchRequest) { if (workloadGroupId == null) { - // Return if request contains no WLM group assignment (default group is added later) return; } WorkloadGroup workloadGroup = workloadGroupService.getWorkloadGroupById(workloadGroupId); - if (workloadGroup == null) { return; } Settings wlmSettings = workloadGroup.getSettings(); - if (wlmSettings != null && wlmSettings.hasValue(WorkloadGroupSearchSettings.WLM_SEARCH_TIMEOUT.getKey())) { - try { - TimeValue timeout = WorkloadGroupSearchSettings.WLM_SEARCH_TIMEOUT.get(wlmSettings); - if (searchRequest.source() != null && searchRequest.source().timeout() == null) { - searchRequest.source().timeout(timeout); - } - } catch (Exception e) { - logger.error("Failed to apply workload group settings", e); + if (wlmSettings == null || wlmSettings.isEmpty()) { + return; + } + + boolean overrideRequestValues = WorkloadGroupSearchSettings.WLM_OVERRIDE_REQUEST_VALUES.get(wlmSettings); + + applyTimeout(wlmSettings, searchRequest, overrideRequestValues); + applyCancelAfterTimeInterval(wlmSettings, searchRequest, overrideRequestValues); + applyMaxConcurrentShardRequests(wlmSettings, searchRequest, overrideRequestValues); + applyBatchedReduceSize(wlmSettings, searchRequest, overrideRequestValues); + } + + private void applyTimeout(Settings wlmSettings, SearchRequest searchRequest, boolean overrideRequestValues) { + if (wlmSettings.hasValue(WorkloadGroupSearchSettings.WLM_SEARCH_TIMEOUT.getKey()) == false) { + return; + } + try { + TimeValue timeout = WorkloadGroupSearchSettings.WLM_SEARCH_TIMEOUT.get(wlmSettings); + if (searchRequest.source() == null) { + return; + } + if (overrideRequestValues || searchRequest.source().timeout() == null) { + searchRequest.source().timeout(timeout); + } + } catch (Exception e) { + logger.error("Failed to apply workload group setting [search.default_search_timeout]", e); + } + } + + private void applyCancelAfterTimeInterval(Settings wlmSettings, SearchRequest searchRequest, boolean overrideRequestValues) { + if (wlmSettings.hasValue(WorkloadGroupSearchSettings.WLM_CANCEL_AFTER_TIME_INTERVAL.getKey()) == false) { + return; + } + try { + TimeValue cancelAfter = WorkloadGroupSearchSettings.WLM_CANCEL_AFTER_TIME_INTERVAL.get(wlmSettings); + if (overrideRequestValues || searchRequest.getCancelAfterTimeInterval() == null) { + searchRequest.setCancelAfterTimeInterval(cancelAfter); + } + } catch (Exception e) { + logger.error("Failed to apply workload group setting [search.cancel_after_time_interval]", e); + } + } + + private void applyMaxConcurrentShardRequests(Settings wlmSettings, SearchRequest searchRequest, boolean overrideRequestValues) { + if (wlmSettings.hasValue(WorkloadGroupSearchSettings.WLM_MAX_CONCURRENT_SHARD_REQUESTS.getKey()) == false) { + return; + } + try { + int maxConcurrent = WorkloadGroupSearchSettings.WLM_MAX_CONCURRENT_SHARD_REQUESTS.get(wlmSettings); + // Raw value 0 means not explicitly set by the user + if (overrideRequestValues || searchRequest.getMaxConcurrentShardRequestsRaw() == 0) { + searchRequest.setMaxConcurrentShardRequests(maxConcurrent); + } + } catch (Exception e) { + logger.error("Failed to apply workload group setting [search.max_concurrent_shard_requests]", e); + } + } + + private void applyBatchedReduceSize(Settings wlmSettings, SearchRequest searchRequest, boolean overrideRequestValues) { + if (wlmSettings.hasValue(WorkloadGroupSearchSettings.WLM_BATCHED_REDUCE_SIZE.getKey()) == false) { + return; + } + try { + int batchedReduceSize = WorkloadGroupSearchSettings.WLM_BATCHED_REDUCE_SIZE.get(wlmSettings); + // Only apply WLM batched reduce size when the request uses the default value. + // Note: batchedReduceSize is a primitive int with no sentinel value, so we cannot + // distinguish between "not set" and "explicitly set to 512 (the default)". If a user + // explicitly sets batched_reduce_size=512, WLM will still override it when + // override_request_values is false. + if (overrideRequestValues || searchRequest.getBatchedReduceSize() == SearchRequest.DEFAULT_BATCHED_REDUCE_SIZE) { + searchRequest.setBatchedReduceSize(batchedReduceSize); } + } catch (Exception e) { + logger.error("Failed to apply workload group setting [search.batched_reduce_size]", e); } } } diff --git a/server/src/test/java/org/opensearch/cluster/metadata/WorkloadGroupTests.java b/server/src/test/java/org/opensearch/cluster/metadata/WorkloadGroupTests.java index a18f57e7667ab..0e57b739a5cd2 100644 --- a/server/src/test/java/org/opensearch/cluster/metadata/WorkloadGroupTests.java +++ b/server/src/test/java/org/opensearch/cluster/metadata/WorkloadGroupTests.java @@ -259,4 +259,133 @@ public void testLegacySearchSettingsFieldRejected() throws IOException { IllegalArgumentException exception = expectThrows(IllegalArgumentException.class, () -> WorkloadGroup.fromXContent(parser)); assertTrue(exception.getMessage().contains("search_settings")); } + + public void testUpdateWithEmptySettingsClearsExisting() { + WorkloadGroup existing = new WorkloadGroup( + "test", + "test_id", + new MutableWorkloadGroupFragment( + ResiliencyMode.ENFORCED, + Map.of(ResourceType.MEMORY, 0.5), + Settings.builder().put("search.default_search_timeout", "30s").build() + ), + System.currentTimeMillis() + ); + + // Empty settings should clear all search settings + MutableWorkloadGroupFragment updateFragment = new MutableWorkloadGroupFragment(null, Map.of(), Settings.EMPTY); + + WorkloadGroup updated = WorkloadGroup.updateExistingWorkloadGroup(existing, updateFragment); + // All settings should be cleared + assertTrue(updated.getSettings().isEmpty()); + } + + public void testUpdateMergesSettings() { + WorkloadGroup existing = new WorkloadGroup( + "test", + "test_id", + new MutableWorkloadGroupFragment( + ResiliencyMode.ENFORCED, + Map.of(ResourceType.MEMORY, 0.5), + Settings.builder().put("search.default_search_timeout", "30s").put("search.max_concurrent_shard_requests", "5").build() + ), + System.currentTimeMillis() + ); + + // Update only timeout — max_concurrent_shard_requests should persist + MutableWorkloadGroupFragment updateFragment = new MutableWorkloadGroupFragment( + null, + Map.of(), + Settings.builder().put("search.default_search_timeout", "1m").build() + ); + + WorkloadGroup updated = WorkloadGroup.updateExistingWorkloadGroup(existing, updateFragment); + assertEquals("1m", updated.getSettings().get("search.default_search_timeout")); + assertEquals("5", updated.getSettings().get("search.max_concurrent_shard_requests")); + } + + public void testUpdateWithNullValueRemovesSetting() { + WorkloadGroup existing = new WorkloadGroup( + "test", + "test_id", + new MutableWorkloadGroupFragment( + ResiliencyMode.ENFORCED, + Map.of(ResourceType.MEMORY, 0.5), + Settings.builder().put("search.default_search_timeout", "30s").put("search.max_concurrent_shard_requests", "5").build() + ), + System.currentTimeMillis() + ); + + // Send null for timeout — should remove it, keep max_concurrent + MutableWorkloadGroupFragment updateFragment = new MutableWorkloadGroupFragment( + null, + Map.of(), + Settings.builder().putNull("search.default_search_timeout").build() + ); + + WorkloadGroup updated = WorkloadGroup.updateExistingWorkloadGroup(existing, updateFragment); + assertNull(updated.getSettings().get("search.default_search_timeout")); + assertEquals("5", updated.getSettings().get("search.max_concurrent_shard_requests")); + } + + public void testUpdateWithNullFragmentSettingsKeepsExisting() throws IOException { + WorkloadGroup existing = new WorkloadGroup( + "test", + "test_id", + new MutableWorkloadGroupFragment( + ResiliencyMode.ENFORCED, + Map.of(ResourceType.MEMORY, 0.5), + Settings.builder().put("search.default_search_timeout", "30s").build() + ), + System.currentTimeMillis() + ); + + // Parse an update request that doesn't include "settings" key + String json = "{\"resiliency_mode\":\"soft\",\"resource_limits\":{\"memory\":0.6}}"; + XContentParser parser = createParser(JsonXContent.jsonXContent, json); + WorkloadGroup.Builder builder = WorkloadGroup.Builder.fromXContent(parser); + MutableWorkloadGroupFragment updateFragment = builder.getMutableWorkloadGroupFragment(); + + WorkloadGroup updated = WorkloadGroup.updateExistingWorkloadGroup(existing, updateFragment); + // Settings should be preserved + assertEquals("30s", updated.getSettings().get("search.default_search_timeout")); + assertEquals(ResiliencyMode.SOFT, updated.getResiliencyMode()); + } + + public void testUpdateOverrideRequestValuesPersistsThroughMerge() { + WorkloadGroup existing = new WorkloadGroup( + "test", + "test_id", + new MutableWorkloadGroupFragment( + ResiliencyMode.ENFORCED, + Map.of(ResourceType.MEMORY, 0.5), + Settings.builder().put("search.default_search_timeout", "30s").put("override_request_values", "true").build() + ), + System.currentTimeMillis() + ); + + // Update only timeout — override_request_values should persist as "true" + MutableWorkloadGroupFragment updateFragment = new MutableWorkloadGroupFragment( + null, + Map.of(), + Settings.builder().put("search.default_search_timeout", "1m").build() + ); + + WorkloadGroup updated = WorkloadGroup.updateExistingWorkloadGroup(existing, updateFragment); + assertEquals("1m", updated.getSettings().get("search.default_search_timeout")); + assertEquals("true", updated.getSettings().get("override_request_values")); + } + + public void testSettingsNullFromXContentClearsSettings() throws IOException { + // Simulate parsing {"settings": null} via XContent + String json = "{\"_id\":\"test_id\",\"name\":\"test\",\"resiliency_mode\":\"enforced\"," + + "\"resource_limits\":{\"memory\":0.5}," + + "\"settings\":null," + + "\"updated_at\":1720047207}"; + XContentParser parser = createParser(JsonXContent.jsonXContent, json); + WorkloadGroup.Builder builder = WorkloadGroup.Builder.fromXContent(parser); + MutableWorkloadGroupFragment fragment = builder.getMutableWorkloadGroupFragment(); + // Settings should be empty (cleared) + assertTrue(fragment.getSettings().isEmpty()); + } } diff --git a/server/src/test/java/org/opensearch/wlm/WorkloadGroupSearchSettingsTests.java b/server/src/test/java/org/opensearch/wlm/WorkloadGroupSearchSettingsTests.java index 9f8d9cbdc843f..fe9efc2615766 100644 --- a/server/src/test/java/org/opensearch/wlm/WorkloadGroupSearchSettingsTests.java +++ b/server/src/test/java/org/opensearch/wlm/WorkloadGroupSearchSettingsTests.java @@ -18,6 +18,26 @@ public void testWlmSearchTimeoutSettingExists() { assertEquals("search.default_search_timeout", WorkloadGroupSearchSettings.WLM_SEARCH_TIMEOUT.getKey()); } + public void testWlmCancelAfterTimeIntervalSettingExists() { + assertNotNull(WorkloadGroupSearchSettings.WLM_CANCEL_AFTER_TIME_INTERVAL); + assertEquals("search.cancel_after_time_interval", WorkloadGroupSearchSettings.WLM_CANCEL_AFTER_TIME_INTERVAL.getKey()); + } + + public void testWlmMaxConcurrentShardRequestsSettingExists() { + assertNotNull(WorkloadGroupSearchSettings.WLM_MAX_CONCURRENT_SHARD_REQUESTS); + assertEquals("search.max_concurrent_shard_requests", WorkloadGroupSearchSettings.WLM_MAX_CONCURRENT_SHARD_REQUESTS.getKey()); + } + + public void testWlmBatchedReduceSizeSettingExists() { + assertNotNull(WorkloadGroupSearchSettings.WLM_BATCHED_REDUCE_SIZE); + assertEquals("search.batched_reduce_size", WorkloadGroupSearchSettings.WLM_BATCHED_REDUCE_SIZE.getKey()); + } + + public void testWlmOverrideRequestValuesSettingExists() { + assertNotNull(WorkloadGroupSearchSettings.WLM_OVERRIDE_REQUEST_VALUES); + assertEquals("override_request_values", WorkloadGroupSearchSettings.WLM_OVERRIDE_REQUEST_VALUES.getKey()); + } + public void testValidateSettingsValid() { Settings settings = Settings.builder().put("search.default_search_timeout", "30s").build(); WorkloadGroupSearchSettings.validate(settings); @@ -30,6 +50,79 @@ public void testValidateSettingsValidTimeValues() { } } + public void testValidateCancelAfterTimeInterval() { + Settings settings = Settings.builder().put("search.cancel_after_time_interval", "1m").build(); + WorkloadGroupSearchSettings.validate(settings); + + settings = Settings.builder().put("search.cancel_after_time_interval", "30s").build(); + WorkloadGroupSearchSettings.validate(settings); + } + + public void testValidateMaxConcurrentShardRequests() { + Settings settings = Settings.builder().put("search.max_concurrent_shard_requests", "1").build(); + WorkloadGroupSearchSettings.validate(settings); + + settings = Settings.builder().put("search.max_concurrent_shard_requests", "100").build(); + WorkloadGroupSearchSettings.validate(settings); + } + + public void testValidateMaxConcurrentShardRequestsInvalid() { + Settings settings = Settings.builder().put("search.max_concurrent_shard_requests", "0").build(); + IllegalArgumentException exception = expectThrows( + IllegalArgumentException.class, + () -> WorkloadGroupSearchSettings.validate(settings) + ); + assertTrue(exception.getMessage().contains("Invalid value")); + assertTrue(exception.getMessage().contains("search.max_concurrent_shard_requests")); + + Settings settings2 = Settings.builder().put("search.max_concurrent_shard_requests", "-1").build(); + exception = expectThrows(IllegalArgumentException.class, () -> WorkloadGroupSearchSettings.validate(settings2)); + assertTrue(exception.getMessage().contains("Invalid value")); + assertTrue(exception.getMessage().contains("search.max_concurrent_shard_requests")); + } + + public void testValidateBatchedReduceSize() { + Settings settings = Settings.builder().put("search.batched_reduce_size", "2").build(); + WorkloadGroupSearchSettings.validate(settings); + + settings = Settings.builder().put("search.batched_reduce_size", "512").build(); + WorkloadGroupSearchSettings.validate(settings); + } + + public void testValidateBatchedReduceSizeInvalid() { + Settings settings = Settings.builder().put("search.batched_reduce_size", "1").build(); + IllegalArgumentException exception = expectThrows( + IllegalArgumentException.class, + () -> WorkloadGroupSearchSettings.validate(settings) + ); + assertTrue(exception.getMessage().contains("Invalid value")); + assertTrue(exception.getMessage().contains("search.batched_reduce_size")); + + Settings settings2 = Settings.builder().put("search.batched_reduce_size", "0").build(); + exception = expectThrows(IllegalArgumentException.class, () -> WorkloadGroupSearchSettings.validate(settings2)); + assertTrue(exception.getMessage().contains("Invalid value")); + assertTrue(exception.getMessage().contains("search.batched_reduce_size")); + } + + public void testValidateOverrideRequestValues() { + Settings settings = Settings.builder().put("override_request_values", "true").build(); + WorkloadGroupSearchSettings.validate(settings); + + settings = Settings.builder().put("override_request_values", "false").build(); + WorkloadGroupSearchSettings.validate(settings); + } + + public void testValidateMultipleSettings() { + Settings settings = Settings.builder() + .put("search.default_search_timeout", "30s") + .put("search.cancel_after_time_interval", "1m") + .put("search.max_concurrent_shard_requests", "5") + .put("search.batched_reduce_size", "256") + .put("override_request_values", "true") + .build(); + WorkloadGroupSearchSettings.validate(settings); + } + public void testValidateSettingsUnknownKey() { Settings settings = Settings.builder().put("unknown_key", "value").build(); IllegalArgumentException exception = expectThrows( @@ -57,11 +150,6 @@ public void testValidateSettingsEmpty() { WorkloadGroupSearchSettings.validate(Settings.EMPTY); } - public void testGetRegisteredSettings() { - assertNotNull(WorkloadGroupSearchSettings.getRegisteredSettings()); - assertTrue(WorkloadGroupSearchSettings.getRegisteredSettings().containsKey("search.default_search_timeout")); - } - public void testLegacyTimeoutKeyRejected() { Settings settings = Settings.builder().put("timeout", "30s").build(); IllegalArgumentException exception = expectThrows( @@ -70,4 +158,39 @@ public void testLegacyTimeoutKeyRejected() { ); assertTrue(exception.getMessage().contains("Unknown WLM setting: timeout")); } + + public void testValidateNonNumericIntSetting() { + Settings settings = Settings.builder().put("search.max_concurrent_shard_requests", "abc").build(); + IllegalArgumentException exception = expectThrows( + IllegalArgumentException.class, + () -> WorkloadGroupSearchSettings.validate(settings) + ); + assertTrue(exception.getMessage().contains("Invalid value")); + assertTrue(exception.getMessage().contains("search.max_concurrent_shard_requests")); + + Settings settings2 = Settings.builder().put("search.batched_reduce_size", "xyz").build(); + exception = expectThrows(IllegalArgumentException.class, () -> WorkloadGroupSearchSettings.validate(settings2)); + assertTrue(exception.getMessage().contains("Invalid value")); + assertTrue(exception.getMessage().contains("search.batched_reduce_size")); + } + + public void testValidateInvalidCancelAfterTimeInterval() { + Settings settings = Settings.builder().put("search.cancel_after_time_interval", "not_a_time").build(); + IllegalArgumentException exception = expectThrows( + IllegalArgumentException.class, + () -> WorkloadGroupSearchSettings.validate(settings) + ); + assertTrue(exception.getMessage().contains("Invalid value")); + assertTrue(exception.getMessage().contains("search.cancel_after_time_interval")); + } + + public void testValidateInvalidOverrideRequestValues() { + Settings settings = Settings.builder().put("override_request_values", "not_a_boolean").build(); + IllegalArgumentException exception = expectThrows( + IllegalArgumentException.class, + () -> WorkloadGroupSearchSettings.validate(settings) + ); + assertTrue(exception.getMessage().contains("Invalid value")); + assertTrue(exception.getMessage().contains("override_request_values")); + } } diff --git a/server/src/test/java/org/opensearch/wlm/listeners/WorkloadGroupRequestOperationListenerTests.java b/server/src/test/java/org/opensearch/wlm/listeners/WorkloadGroupRequestOperationListenerTests.java index 61e9b6e70e9cb..5594c77fd5d58 100644 --- a/server/src/test/java/org/opensearch/wlm/listeners/WorkloadGroupRequestOperationListenerTests.java +++ b/server/src/test/java/org/opensearch/wlm/listeners/WorkloadGroupRequestOperationListenerTests.java @@ -358,6 +358,222 @@ public void testApplySearchSettings_Timeout_NullSource() { assertNull(mockSearchRequest.source()); // Should not throw, source remains null } + public void testApplySearchSettings_CancelAfterTimeInterval_WlmAppliedWhenNull() { + assertNull(mockSearchRequest.getCancelAfterTimeInterval()); + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup(wgId, Settings.builder().put("search.cancel_after_time_interval", "30s").build()); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + assertEquals(TimeValue.timeValueSeconds(30), mockSearchRequest.getCancelAfterTimeInterval()); + } + + public void testApplySearchSettings_CancelAfterTimeInterval_RequestAlreadySet() { + mockSearchRequest.setCancelAfterTimeInterval(TimeValue.timeValueSeconds(10)); + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup(wgId, Settings.builder().put("search.cancel_after_time_interval", "30s").build()); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + assertEquals(TimeValue.timeValueSeconds(10), mockSearchRequest.getCancelAfterTimeInterval()); // Request value preserved + } + + public void testApplySearchSettings_MaxConcurrentShardRequests_WlmAppliedWhenDefault() { + // Request has default value (not explicitly set), raw field is 0 + assertEquals(0, mockSearchRequest.getMaxConcurrentShardRequestsRaw()); + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup(wgId, Settings.builder().put("search.max_concurrent_shard_requests", "10").build()); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + assertEquals(10, mockSearchRequest.getMaxConcurrentShardRequests()); // WLM applied + } + + public void testApplySearchSettings_MaxConcurrentShardRequests_RequestAlreadySet() { + mockSearchRequest.setMaxConcurrentShardRequests(20); // explicitly set by user + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup(wgId, Settings.builder().put("search.max_concurrent_shard_requests", "5").build()); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + assertEquals(20, mockSearchRequest.getMaxConcurrentShardRequests()); // Request value preserved + } + + public void testApplySearchSettings_BatchedReduceSize_WlmAppliedWhenDefault() { + // Request uses default value (512) + assertEquals(SearchRequest.DEFAULT_BATCHED_REDUCE_SIZE, mockSearchRequest.getBatchedReduceSize()); + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup(wgId, Settings.builder().put("search.batched_reduce_size", "100").build()); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + assertEquals(100, mockSearchRequest.getBatchedReduceSize()); // WLM applied + } + + public void testApplySearchSettings_BatchedReduceSize_RequestAlreadySet() { + mockSearchRequest.setBatchedReduceSize(50); // explicitly set by user + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup(wgId, Settings.builder().put("search.batched_reduce_size", "100").build()); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + assertEquals(50, mockSearchRequest.getBatchedReduceSize()); // Request value preserved + } + + public void testApplySearchSettings_OverrideRequestValues_True() { + // Set explicit values on the request + mockSearchRequest.source(new SearchSourceBuilder().timeout(TimeValue.timeValueSeconds(5))); + mockSearchRequest.setCancelAfterTimeInterval(TimeValue.timeValueSeconds(10)); + mockSearchRequest.setMaxConcurrentShardRequests(20); + mockSearchRequest.setBatchedReduceSize(50); + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup( + wgId, + Settings.builder() + .put("search.default_search_timeout", "1m") + .put("search.cancel_after_time_interval", "2m") + .put("search.max_concurrent_shard_requests", "3") + .put("search.batched_reduce_size", "256") + .put("override_request_values", "true") + .build() + ); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + // All values should be overridden by WLM settings + assertEquals(TimeValue.timeValueMinutes(1), mockSearchRequest.source().timeout()); + assertEquals(TimeValue.timeValueMinutes(2), mockSearchRequest.getCancelAfterTimeInterval()); + assertEquals(3, mockSearchRequest.getMaxConcurrentShardRequests()); + assertEquals(256, mockSearchRequest.getBatchedReduceSize()); + } + + public void testApplySearchSettings_OverrideRequestValues_False() { + // Set explicit values on the request + mockSearchRequest.source(new SearchSourceBuilder().timeout(TimeValue.timeValueSeconds(5))); + mockSearchRequest.setCancelAfterTimeInterval(TimeValue.timeValueSeconds(10)); + mockSearchRequest.setMaxConcurrentShardRequests(20); + mockSearchRequest.setBatchedReduceSize(50); + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup( + wgId, + Settings.builder() + .put("search.default_search_timeout", "1m") + .put("search.cancel_after_time_interval", "2m") + .put("search.max_concurrent_shard_requests", "3") + .put("search.batched_reduce_size", "256") + .put("override_request_values", "false") + .build() + ); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + // All request values should be preserved + assertEquals(TimeValue.timeValueSeconds(5), mockSearchRequest.source().timeout()); + assertEquals(TimeValue.timeValueSeconds(10), mockSearchRequest.getCancelAfterTimeInterval()); + assertEquals(20, mockSearchRequest.getMaxConcurrentShardRequests()); + assertEquals(50, mockSearchRequest.getBatchedReduceSize()); + } + + public void testApplySearchSettings_MultipleSettings() { + mockSearchRequest.source(new SearchSourceBuilder()); + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup( + wgId, + Settings.builder() + .put("search.default_search_timeout", "30s") + .put("search.cancel_after_time_interval", "1m") + .put("search.max_concurrent_shard_requests", "5") + .put("search.batched_reduce_size", "100") + .build() + ); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + assertEquals(TimeValue.timeValueSeconds(30), mockSearchRequest.source().timeout()); + assertEquals(TimeValue.timeValueMinutes(1), mockSearchRequest.getCancelAfterTimeInterval()); + assertEquals(5, mockSearchRequest.getMaxConcurrentShardRequests()); + assertEquals(100, mockSearchRequest.getBatchedReduceSize()); + } + + public void testApplySearchSettings_OverrideRequestValues_DefaultsToFalseWhenAbsent() { + // Set explicit values on the request + mockSearchRequest.source(new SearchSourceBuilder().timeout(TimeValue.timeValueSeconds(5))); + mockSearchRequest.setCancelAfterTimeInterval(TimeValue.timeValueSeconds(10)); + + String wgId = "test-wg"; + // No override_request_values key in settings — should default to false + WorkloadGroup wg = createWorkloadGroup( + wgId, + Settings.builder().put("search.default_search_timeout", "1m").put("search.cancel_after_time_interval", "2m").build() + ); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + // Request values should be preserved (override defaults to false) + assertEquals(TimeValue.timeValueSeconds(5), mockSearchRequest.source().timeout()); + assertEquals(TimeValue.timeValueSeconds(10), mockSearchRequest.getCancelAfterTimeInterval()); + } + + public void testApplySearchSettings_OverrideRequestValues_TrueWithRequestUnset() { + // Request has no explicit values set + mockSearchRequest.source(new SearchSourceBuilder()); + assertNull(mockSearchRequest.source().timeout()); + assertNull(mockSearchRequest.getCancelAfterTimeInterval()); + assertEquals(0, mockSearchRequest.getMaxConcurrentShardRequestsRaw()); + assertEquals(SearchRequest.DEFAULT_BATCHED_REDUCE_SIZE, mockSearchRequest.getBatchedReduceSize()); + + String wgId = "test-wg"; + WorkloadGroup wg = createWorkloadGroup( + wgId, + Settings.builder() + .put("search.default_search_timeout", "1m") + .put("search.cancel_after_time_interval", "2m") + .put("search.max_concurrent_shard_requests", "3") + .put("search.batched_reduce_size", "256") + .put("override_request_values", "true") + .build() + ); + when(workloadGroupService.getWorkloadGroupById(wgId)).thenReturn(wg); + testThreadPool.getThreadContext().putHeader(WorkloadGroupTask.WORKLOAD_GROUP_ID_HEADER, wgId); + + sut.onRequestStart(mockSearchRequestContext); + + // WLM values applied (override=true, but nothing to override - same result as override=false) + assertEquals(TimeValue.timeValueMinutes(1), mockSearchRequest.source().timeout()); + assertEquals(TimeValue.timeValueMinutes(2), mockSearchRequest.getCancelAfterTimeInterval()); + assertEquals(3, mockSearchRequest.getMaxConcurrentShardRequests()); + assertEquals(256, mockSearchRequest.getBatchedReduceSize()); + } + private WorkloadGroup createWorkloadGroup(String id, Settings searchSettings) { return new WorkloadGroup( "test-name",