NO-ISSUE Fix which keys are used to update dynamic properties

Author: Yang-33

centraldogma/src/main/java/com/linecorp/decaton/centraldogma/CentralDogmaPropertySupplier.java

@@ -121,15 +121,14 @@ public CentralDogmaPropertySupplier(CentralDogmaRepository centralDogmaRepositor
         }
 
         rootWatcher.watch(node -> {
-            for (PropertyDefinition<?> definition : PROPERTY_DEFINITIONS) {
-                DynamicProperty<?> p = cachedProperties.get(definition.name());
-                if (p != null && node.has(definition.name())) {
+            for(ConcurrentHashMap.Entry<String, DynamicProperty<?>> cachedProperty : cachedProperties.entrySet()) {
+                if (node.has(cachedProperty.getKey())) {
                     try {
-                        setValue(p, node.get(definition.name()));
+                        setValue(cachedProperty.getValue(), node.get(cachedProperty.getKey()));
                     } catch (Exception e) {
                         // Catching Exception instead of RuntimeException, since
                         // Kotlin-implemented DynamicProperty would throw checked exceptions
-                        logger.warn("Failed to set value from CentralDogma for {}", definition.name(), e);
+                        logger.warn("Failed to set value updatedfrom CentralDogma for {}", cachedProperty.getKey(), e);
                     }
                 }
             }
@@ -165,7 +164,7 @@ public <T> Optional<Property<T>> getProperty(PropertyDefinition<T> definition) {
             } catch (Exception e) {
                 // Catching Exception instead of RuntimeException, since
                 // Kotlin-implemented DynamicProperty would throw checked exceptions
-                logger.warn("Failed to set value from CentralDogma for {}", definition.name(), e);
+                logger.warn("Failed to set initial value from CentralDogma for {}", definition.name(), e);
             }
             return prop;
         });

centraldogma/src/test/java/com/linecorp/decaton/centraldogma/CentralDogmaPropertySupplierIntegrationTest.java

@@ -41,6 +41,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import com.linecorp.decaton.processor.runtime.PropertyDefinition;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -222,6 +223,98 @@ void testCDIntegrationYaml() throws Exception {
                 }).get().get().value().intValue());
     }
 
+    @Test
+    @Timeout(50)
+    public void testCDIntegrationDynamicPropertyJson() throws InterruptedException {
+        final String FILE = "/subscription.json";
+        CentralDogma client = extension.client();
+
+        client.createProject(PROJECT_NAME).join();
+        CentralDogmaRepository repo =
+                client.createRepository(PROJECT_NAME, REPOSITORY_NAME).join();
+
+        final String topic = "orders";
+        final String dynamicName =
+                "decaton.shaping.topic.processing.rate.per.partition." + topic;
+        assertFalse(ProcessorProperties.defaultProperties().stream()
+                .anyMatch(p -> p.definition().name().equals(dynamicName)));
+
+        final String ORIGINAL =
+                "{\n"
+                        + "  \"" + dynamicName + "\": 7,\n"
+                        + "  \"decaton.partition.concurrency\": 10\n"
+                        + "}\n";
+
+        repo.commit("init-json", Change.ofJsonUpsert(FILE, ORIGINAL)).push().join();
+
+        CentralDogmaPropertySupplier supplier = new CentralDogmaPropertySupplier(repo, FILE);
+
+        PropertyDefinition<Integer> DYNAMIC_INT =
+                PropertyDefinition.define(dynamicName, Integer.class, 0, v -> v instanceof Integer);
+
+        Property<Integer> prop = supplier.getProperty(DYNAMIC_INT).get();
+        assertEquals(7, prop.value().intValue());
+
+        final String UPDATED =
+                "{\n"
+                        + "  \"" + dynamicName + "\": 11,\n" // This is changed
+                        + "  \"decaton.partition.concurrency\": 10\n"
+                        + "}\n";
+
+        CountDownLatch latch = new CountDownLatch(2);
+        prop.listen((oldV, newV) -> latch.countDown());
+
+        repo.commit("patch-json", Change.ofJsonPatch(FILE, ORIGINAL, UPDATED)).push().join();
+
+        latch.await();
+        assertEquals(11, prop.value().intValue());
+    }
+
+    @Test
+    @Timeout(50)
+    public void testCDIntegrationDynamicPropertyYaml() throws Exception {
+        final String FILE = "/subscription.yaml";
+        CentralDogma client = extension.client();
+
+        client.createProject(PROJECT_NAME).join();
+        CentralDogmaRepository repo =
+                client.createRepository(PROJECT_NAME, REPOSITORY_NAME).join();
+
+        final String topic = "payments";
+        final String dynamicName =
+                "decaton.shaping.topic.processing.rate.per.partition." + topic;
+        assertFalse(ProcessorProperties.defaultProperties().stream()
+                .anyMatch(p -> p.definition().name().equals(dynamicName)));
+
+        final String ORIGINAL_YAML =
+                "decaton.partition.concurrency: 10\n"
+                        + dynamicName + ": 3\n";
+
+        repo.commit("init-yaml", Change.ofTextUpsert(FILE, ORIGINAL_YAML))
+                .push().join();
+
+        CentralDogmaPropertySupplier supplier = new CentralDogmaPropertySupplier(repo, FILE);
+
+        PropertyDefinition<Integer> DYNAMIC_INT =
+                PropertyDefinition.define(dynamicName, Integer.class, 0, v -> v instanceof Integer);
+
+        Property<Integer> prop = supplier.getProperty(DYNAMIC_INT).get();
+        assertEquals(3, prop.value().intValue());
+
+        final String UPDATED_YAML =
+                "decaton.partition.concurrency: 10\n"
+                        + dynamicName + ": 9\n"; // This is changed
+
+        CountDownLatch latch = new CountDownLatch(2);
+        prop.listen((o, n) -> latch.countDown());
+
+        repo.commit("patch-yaml", Change.ofTextPatch(FILE, ORIGINAL_YAML, UPDATED_YAML))
+                .push().join();
+
+        latch.await();
+        assertEquals(9, prop.value().intValue());
+    }
+
     @Test
     @Timeout(10)
     public void testCDRegisterSuccessJson() {
@@ -414,7 +507,7 @@ static Stream<FormatCase> formats() {
         return Stream.of(JSON, YAML);
     }
 
-    @ParameterizedTest(name = "registerTimeout‑{0}")
+    @ParameterizedTest()
     @MethodSource("formats")
     @Timeout(15)
     void testCDRegisterTimeout(FormatCase testCase) {
@@ -433,7 +526,7 @@ void testCDRegisterTimeout(FormatCase testCase) {
         });
     }
 
-    @ParameterizedTest(name = "registerNonExistentProject‑{0}")
+    @ParameterizedTest()
     @MethodSource("formats")
     void testCDRegisterNonExistentProject(FormatCase testCase) {
         assertThrows(RuntimeException.class, () -> {
@@ -442,7 +535,7 @@ void testCDRegisterNonExistentProject(FormatCase testCase) {
         });
     }
 
-    @ParameterizedTest(name = "fileExists‑{0}")
+    @ParameterizedTest()
     @MethodSource("formats")
     void testFileExist(FormatCase testCase) {
         CentralDogma client = extension.client();
@@ -457,7 +550,7 @@ void testFileExist(FormatCase testCase) {
                 .fileExists(centralDogmaRepository, testCase.file(), Revision.HEAD));
     }
 
-    @ParameterizedTest(name = "fileNonExistent‑{0}")
+    @ParameterizedTest()
     @MethodSource("formats")
     void testFileNonExistent(FormatCase testCase) {
         CentralDogma client = extension.client();

docs/dynamic-property-configuration.adoc

@@ -1,6 +1,7 @@
 = Dynamic Property Configuration
 :base_version: 9.0.0
 :modules: centraldogma,processor
+:toc:
 
 == Property Supplier
 Decaton provides some properties for you to configure how it processes tasks. These properties don't need to be hard-coded. Decaton lets you configure some of the properties so they can be loaded dynamically.
@@ -79,7 +80,7 @@ public class CentralDogmaSupplierMain {
 }
 ----
 
-== YAML Support
+== Use YAML instead of JSON
 
 You can store the property file in YAML as well as JSON.
 Nothing changes in your code except the file‐name extension.
@@ -98,8 +99,6 @@ CentralDogmaPropertySupplier supplier =
 <1> Use `.yaml` (or `.yml`) instead of `.json`.
 All other APIs and behaviours remain exactly the same.
 
-=== Authoring the YAML file
-
 You may keep the flat, dot‑separated keys.
 
 [source,yaml]
@@ -110,11 +109,11 @@ decaton.processing.rate.per.partition: 50
 ----
 
 As with JSON, you cannot use nested structures in YAML.
-Therefore, the following is not allowed:
+Therefore, the following is NOT allowed:
 
 [source,yaml]
 ----
-# This is not supported.
+# This style is not supported.
 decaton:
   partition:
     concurrency: 8
@@ -143,7 +142,7 @@ Hence you should put supplier with higher priority before others.
 ----
 
 
-== JSON Schema
+== Validate your configuration file with JSON Schema
 
 Decaton ships a set of https://json-schema.org/[JSON Schema] files that precisely describe every key available in `CentralDogmaPropertySupplier` including each key’s type and default value.
 Leveraging these schemas in your configuration files gives you two immediate benefits:
@@ -157,6 +156,7 @@ Leveraging these schemas in your configuration files gives you two immediate ben
 
 
 [source,json]
+.decaton-processor-dynamic-configuration.json
 ----
 {
   "$schema": "https://raw.githubusercontent.com/line/decaton/vX.Y.Z/centraldogma/src/jsonschema/dist/decaton-processor-properties-central-dogma-schema-draft_7.json",
@@ -165,7 +165,19 @@ Leveraging these schemas in your configuration files gives you two immediate ben
   ...
 }
 ----
-For example, you can use JSON Schema by adding a `$schema` directive at the top of the file as shown above.follows.
+
+[source,yaml]
+.decaton-processor-dynamic-configuration.yaml
+----
+# $schema: https://raw.githubusercontent.com/line/decaton/vX.Y.Z/centraldogma/src/jsonschema/dist/decaton-processor-properties-central-dogma-schema-draft_7.json[https://raw.githubusercontent.com/Yang-33/decaton/support-decaton-processor-property-jsonschema-9-1-2-test/centraldogma/src/jsonschema/dist/decaton-processor-properties-central-dogma-schema-draft_2019_09-allow-additional-properties.json]
+# yaml-language-server: $schema=https://raw.githubusercontent.com/line/decaton/vX.Y.Z/centraldogma/src/jsonschema/dist/decaton-processor-properties-central-dogma-schema-draft_7.json
+decaton.partition.concurrency: 10000
+decaton.processing.rate.per.partition: -1
+...
+----
+
+
+For example, you can use JSON Schema by adding a `$schema` directive at the top of the file as shown above follows.
 Of course, there may be other ways to use it.
 Replace `vX.Y.Z` with the exact Decaton version your application depends on.
 If you prefer living at HEAD, you can also reference `master`, but pinning to a release tag guarantees repeatable builds.