make jvm heap usage a dynamic setting

src/main/java/org/opensearch/ad/task/ADBatchTaskRunner.java

@@ -19,7 +19,6 @@
 import static org.opensearch.ad.settings.AnomalyDetectorSettings.MAX_TOP_ENTITIES_FOR_HISTORICAL_ANALYSIS;
 import static org.opensearch.ad.settings.AnomalyDetectorSettings.MAX_TOP_ENTITIES_LIMIT_FOR_HISTORICAL_ANALYSIS;
 import static org.opensearch.timeseries.TimeSeriesAnalyticsPlugin.AD_BATCH_TASK_THREAD_POOL_NAME;
-import static org.opensearch.timeseries.breaker.MemoryCircuitBreaker.DEFAULT_JVM_HEAP_USAGE_THRESHOLD;
 import static org.opensearch.timeseries.stats.InternalStatNames.JVM_HEAP_USAGE;
 import static org.opensearch.timeseries.stats.StatNames.AD_EXECUTING_BATCH_TASK_COUNT;
 
@@ -76,6 +75,7 @@
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.timeseries.AnalysisType;
 import org.opensearch.timeseries.breaker.CircuitBreakerService;
+import org.opensearch.timeseries.breaker.MemoryCircuitBreakerNumericSetting;
 import org.opensearch.timeseries.caching.PriorityTracker;
 import org.opensearch.timeseries.cluster.HashRing;
 import org.opensearch.timeseries.common.exception.EndRunException;
@@ -704,12 +704,15 @@ private void dispatchTask(ADTask adTask, ActionListener<DiscoveryNode> listener)
                 List<StatsNodeResponse> candidateNodeResponse = adStatsResponse
                     .getNodes()
                     .stream()
-                    .filter(stat -> (long) stat.getStatsMap().get(JVM_HEAP_USAGE.getName()) < DEFAULT_JVM_HEAP_USAGE_THRESHOLD)
+                    .filter(
+                        stat -> (long) stat.getStatsMap().get(JVM_HEAP_USAGE.getName()) < MemoryCircuitBreakerNumericSetting
+                            .getJVMHeapUsageThreshold()
+                    )
                     .collect(Collectors.toList());
 
                 if (candidateNodeResponse.size() == 0) {
                     StringBuilder errorMessageBuilder = new StringBuilder("All nodes' memory usage exceeds limitation ")
-                        .append(DEFAULT_JVM_HEAP_USAGE_THRESHOLD)
+                        .append(MemoryCircuitBreakerNumericSetting.getJVMHeapUsageThreshold())
                         .append("%. ")
                         .append(NO_ELIGIBLE_NODE_TO_RUN_DETECTOR)
                         .append(adTask.getConfigId());

src/main/java/org/opensearch/timeseries/breaker/MemoryCircuitBreaker.java

@@ -16,17 +16,16 @@
 /**
  * A circuit breaker for memory usage.
  */
-public class MemoryCircuitBreaker extends ThresholdCircuitBreaker<Short> {
+public class MemoryCircuitBreaker extends ThresholdCircuitBreaker<Integer> {
 
-    public static final short DEFAULT_JVM_HEAP_USAGE_THRESHOLD = 95;
     private final JvmService jvmService;
 
     public MemoryCircuitBreaker(JvmService jvmService) {
-        super(DEFAULT_JVM_HEAP_USAGE_THRESHOLD);
+        super(MemoryCircuitBreakerNumericSetting.getJVMHeapUsageThreshold());
         this.jvmService = jvmService;
     }
 
-    public MemoryCircuitBreaker(short threshold, JvmService jvmService) {
+    public MemoryCircuitBreaker(int threshold, JvmService jvmService) {
         super(threshold);
         this.jvmService = jvmService;
     }

src/main/java/org/opensearch/timeseries/breaker/MemoryCircuitBreakerNumericSetting.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.timeseries.breaker;
+
+import static java.util.Collections.unmodifiableMap;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.opensearch.common.settings.Setting;
+import org.opensearch.timeseries.settings.DynamicNumericSetting;
+
+public class MemoryCircuitBreakerNumericSetting extends DynamicNumericSetting {
+
+    /**
+     * Singleton instance
+     */
+    private static MemoryCircuitBreakerNumericSetting INSTANCE;
+
+    /**
+     * Setting name
+     */
+    public static final String JVM_HEAP_USAGE_THRESHOLD = "plugins.ad.jvm_heap_usage.threshold";
+
+    public MemoryCircuitBreakerNumericSetting(Map<String, Setting<?>> settings) {
+        super(settings);
+    }
+
+    public static synchronized MemoryCircuitBreakerNumericSetting getInstance() {
+        if (INSTANCE == null) {
+            INSTANCE = new MemoryCircuitBreakerNumericSetting(settings);
+        }
+        return INSTANCE;
+    }
+
+    /**
+     * Get the jvm_heap_usage threshold setting value
+     * @return jvm_heap_usage threshold setting value
+     */
+    public static int getJVMHeapUsageThreshold() {
+        return MemoryCircuitBreakerNumericSetting
+            .getInstance()
+            .getSettingValue(MemoryCircuitBreakerNumericSetting.JVM_HEAP_USAGE_THRESHOLD);
+    }
+
+    private static final Map<String, Setting<?>> settings = unmodifiableMap(new HashMap<String, Setting<?>>() {
+        {
+            put(
+                JVM_HEAP_USAGE_THRESHOLD,
+                Setting.intSetting(JVM_HEAP_USAGE_THRESHOLD, 95, 0, 98, Setting.Property.NodeScope, Setting.Property.Dynamic)
+            );
+        }
+    });
+}

src/test/java/org/opensearch/ad/breaker/MemoryCircuitBreakerNumericSettingTests.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.ad.breaker;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.opensearch.common.settings.Setting;
+import org.opensearch.test.OpenSearchTestCase;
+import org.opensearch.timeseries.breaker.MemoryCircuitBreakerNumericSetting;
+
+public class MemoryCircuitBreakerNumericSettingTests extends OpenSearchTestCase {
+    private MemoryCircuitBreakerNumericSetting memoryCircuitBreakerNumericSetting;
+
+    @Override
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+        memoryCircuitBreakerNumericSetting = MemoryCircuitBreakerNumericSetting.getInstance();
+    }
+
+    @Test
+    public void testGetThresholdValue_shouldReturnThresholdValue() {
+        memoryCircuitBreakerNumericSetting.setSettingValue(MemoryCircuitBreakerNumericSetting.JVM_HEAP_USAGE_THRESHOLD, 96);
+        int value = MemoryCircuitBreakerNumericSetting.getJVMHeapUsageThreshold();
+        assertEquals(96, value);
+    }
+
+    @Test
+    public void testGetSettingValue_shouldReturnSettingValue() {
+        Map<String, Setting<?>> settingsMap = new HashMap<>();
+        Setting<Integer> testSetting = Setting.intSetting("test.setting", 1, Setting.Property.NodeScope);
+        settingsMap.put("test.setting", testSetting);
+        memoryCircuitBreakerNumericSetting = new MemoryCircuitBreakerNumericSetting(settingsMap);
+
+        memoryCircuitBreakerNumericSetting.setSettingValue("test.setting", 2);
+        Integer value = memoryCircuitBreakerNumericSetting.getSettingValue("test.setting");
+        assertEquals(2, value.intValue());
+    }
+
+    @Test
+    public void testGetSettingValue_withNonexistent_shouldThrowException() {
+        try {
+            memoryCircuitBreakerNumericSetting.getSettingValue("nonexistent.key");
+            fail("Expected an IllegalArgumentException to be thrown");
+        } catch (IllegalArgumentException e) {
+            assertEquals("Cannot find setting by key [nonexistent.key]", e.getMessage());
+        }
+    }
+}

src/test/java/org/opensearch/ad/breaker/MemoryCircuitBreakerTests.java

@@ -13,6 +13,7 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.when;
 
 import org.junit.Before;
@@ -44,32 +45,33 @@ public void setup() {
     }
 
     @Test
-    public void testIsOpen() {
+    public void testIsOpen_whenUsageIsBelowDefaultValue_shouldReturnFalse() {
         CircuitBreaker breaker = new MemoryCircuitBreaker(jvmService);
 
         assertThat(breaker.isOpen(), equalTo(false));
     }
 
     @Test
-    public void testIsOpen1() {
-        CircuitBreaker breaker = new MemoryCircuitBreaker((short) 90, jvmService);
+    public void testIsOpen_whenUsageIsAboveDefaultValue_shouldReturnTrue() {
+        CircuitBreaker breaker = new MemoryCircuitBreaker(jvmService);
 
-        assertThat(breaker.isOpen(), equalTo(false));
+        doReturn((short) 96).when(mem).getHeapUsedPercent();
+        assertThat(breaker.isOpen(), equalTo(true));
     }
 
     @Test
-    public void testIsOpen2() {
-        CircuitBreaker breaker = new MemoryCircuitBreaker(jvmService);
+    public void testIsOpen_whenUsageIsAboveSettingValue_shouldReturnTrue() {
+        CircuitBreaker breaker = new MemoryCircuitBreaker(90, jvmService);
 
-        when(mem.getHeapUsedPercent()).thenReturn((short) 96);
+        doReturn((short) 96).when(mem).getHeapUsedPercent();
         assertThat(breaker.isOpen(), equalTo(true));
     }
 
     @Test
-    public void testIsOpen3() {
-        CircuitBreaker breaker = new MemoryCircuitBreaker((short) 90, jvmService);
+    public void testIsOpen_whenUsageIsBelowSettingValue_butAboveDefaultValue_shouldReturnFalse() {
+        CircuitBreaker breaker = new MemoryCircuitBreaker(97, jvmService);
 
-        when(mem.getHeapUsedPercent()).thenReturn((short) 96);
-        assertThat(breaker.isOpen(), equalTo(true));
+        doReturn((short) 96).when(mem).getHeapUsedPercent();
+        assertThat(breaker.isOpen(), equalTo(false));
     }
 }