[CCI] [BUG] Fixing extension settings update consumers

server/src/main/java/org/opensearch/extensions/AddSettingsUpdateConsumerRequestHandler.java

@@ -15,6 +15,7 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.SettingsException;
+import org.opensearch.common.settings.SettingsModule;
 import org.opensearch.common.settings.WriteableSetting;
 import org.opensearch.transport.TransportResponse;
 import org.opensearch.transport.TransportService;
@@ -31,6 +32,7 @@ public class AddSettingsUpdateConsumerRequestHandler {
     private final ClusterService clusterService;
     private final TransportService transportService;
     private final String updateSettingsRequestType;
+    private final SettingsModule settingsModule;
 
     /**
      * Instantiates a new Add Settings Update Consumer Request Handler with the {@link ClusterService} and {@link TransportService}
@@ -42,11 +44,13 @@ public class AddSettingsUpdateConsumerRequestHandler {
     public AddSettingsUpdateConsumerRequestHandler(
         ClusterService clusterService,
         TransportService transportService,
-        String updateSettingsRequestType
+        String updateSettingsRequestType,
+        SettingsModule settingsModule
     ) {
         this.clusterService = clusterService;
         this.transportService = transportService;
         this.updateSettingsRequestType = updateSettingsRequestType;
+        this.settingsModule = settingsModule;
     }
 
     /**
@@ -68,25 +72,51 @@ TransportResponse handleAddSettingsUpdateConsumerRequest(AddSettingsUpdateConsum
 
                 // Extract setting and type from writeable setting
                 Setting<?> setting = extensionComponentSetting.getSetting();
+
+                // we need to get the actual setting from nodeSetting or indexsetting maps in SettingsModule
+                // use conditional based on setting properties
+                Setting<?> settingForUpdateConsumer = null;
+                if (setting.hasIndexScope()) {
+                    settingForUpdateConsumer = settingsModule.getIndexScopedSettings().get(setting.getKey());
+                } else if (setting.hasNodeScope()) {
+                    settingForUpdateConsumer = settingsModule.getClusterSettings().get(setting.getKey());
+                }
+                // do a null check and throw IllegalArgument exception here if neither index or node scope
+
                 WriteableSetting.SettingType settingType = extensionComponentSetting.getType();
 
                 // Register setting update consumer with callback method to extension
-                clusterService.getClusterSettings().addSettingsUpdateConsumer(setting, (data) -> {
-                    logger.debug("Sending extension request type: " + updateSettingsRequestType);
-                    UpdateSettingsResponseHandler updateSettingsResponseHandler = new UpdateSettingsResponseHandler();
-                    transportService.sendRequest(
-                        extensionNode,
-                        updateSettingsRequestType,
-                        new UpdateSettingsRequest(settingType, setting, data),
-                        updateSettingsResponseHandler
-                    );
-                });
+                if (setting.hasIndexScope()) {
+                    clusterService.getClusterSettings().addSettingsUpdateConsumer(settingForUpdateConsumer, (data) -> {
+                        logger.debug("Sending extension request type: " + updateSettingsRequestType);
+                        UpdateSettingsResponseHandler updateSettingsResponseHandler = new UpdateSettingsResponseHandler();
+                        transportService.sendRequest(
+                            extensionNode,
+                            updateSettingsRequestType,
+                            new UpdateSettingsRequest(settingType, setting, data),
+                            updateSettingsResponseHandler
+                        );
+                    });
+                }
+                if (setting.hasNodeScope()) {
+                    clusterService.getClusterSettings()
+                        // Register setting update consumer with callback method to extension
+                        .addSettingsUpdateConsumer(settingForUpdateConsumer, (data) -> {
+                            logger.debug("Sending extension request type: " + updateSettingsRequestType);
+                            UpdateSettingsResponseHandler updateSettingsResponseHandler = new UpdateSettingsResponseHandler();
+                            transportService.sendRequest(
+                                extensionNode,
+                                updateSettingsRequestType,
+                                new UpdateSettingsRequest(settingType, setting, data),
+                                updateSettingsResponseHandler
+                            );
+                        });
+                }
             }
         } catch (SettingsException e) {
             logger.error(e.toString());
             status = false;
         }
-
         return new AcknowledgedResponse(status);
     }
 }

server/src/main/java/org/opensearch/extensions/ExtensionsManager.java

@@ -169,7 +169,8 @@ public void initializeServicesAndRestHandler(
         this.addSettingsUpdateConsumerRequestHandler = new AddSettingsUpdateConsumerRequestHandler(
             clusterService,
             transportService,
-            REQUEST_EXTENSION_UPDATE_SETTINGS
+            REQUEST_EXTENSION_UPDATE_SETTINGS,
+            settingsModule
         );
         this.client = client;
         this.extensionTransportActionsHandler = new ExtensionTransportActionsHandler(

server/src/test/java/org/opensearch/common/settings/ScopedSettingsTests.java

@@ -1462,4 +1462,20 @@ public List<String> getListValue(final List<String> value) {
         );
     }
 
+    public void testAddSettingsUpdateConsumer() {
+        Setting<Integer> testSetting = Setting.intSetting("foo.bar", 1, Property.Dynamic, Property.NodeScope);
+        Setting<Integer> testSetting2 = Setting.intSetting("foo.bar.baz", 1, Property.Dynamic, Property.NodeScope);
+        AbstractScopedSettings service = new ClusterSettings(Settings.EMPTY, new HashSet<>(Arrays.asList(testSetting, testSetting2)));
+        AtomicInteger consumer2 = new AtomicInteger();
+        service.addSettingsUpdateConsumer(testSetting2, consumer2::set, (s) -> assertTrue(s > 0));
+        Setting<Integer> wrongKeySetting = Setting.intSetting("foo.bar.wrong", 1, Property.Dynamic, Property.NodeScope);
+
+        expectThrows(SettingsException.class, () -> service.addSettingsUpdateConsumer(wrongKeySetting, consumer2::set, (i) -> {
+            if (i == 42) throw new AssertionError("wrong key");
+        }));
+
+        expectThrows(NullPointerException.class, () -> service.addSettingsUpdateConsumer(null, consumer2::set, (i) -> {
+            if (i == 42) throw new AssertionError("empty key");
+        }));
+    }
 }