Cache CD dynamic properties

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

@@ -18,6 +18,8 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -44,6 +46,9 @@
 import com.linecorp.decaton.processor.runtime.PropertyDefinition;
 import com.linecorp.decaton.processor.runtime.PropertySupplier;
 
+import lombok.AccessLevel;
+import lombok.Getter;
+
 /**
  * A {@link PropertySupplier} implementation with Central Dogma backend.
  *
@@ -72,6 +77,9 @@ public class CentralDogmaPropertySupplier implements PropertySupplier, AutoClose
 
     private final Watcher<JsonNode> rootWatcher;
 
+    @Getter(AccessLevel.PACKAGE)// visible for testing
+    private final ConcurrentMap<String, DynamicProperty<?>> cachedProperties = new ConcurrentHashMap<>();
+
     /**
      * Creates a new {@link CentralDogmaPropertySupplier}.
      * @param centralDogma a {@link CentralDogma} instance to use to access Central Dogma server.
@@ -118,32 +126,43 @@ public <T> Optional<Property<T>> getProperty(PropertyDefinition<T> definition) {
             return Optional.empty();
         }
 
-        DynamicProperty<T> prop = new DynamicProperty<>(definition);
-        Watcher<JsonNode> child = rootWatcher.newChild(jsonNode -> jsonNode.path(definition.name()));
-        child.watch(node -> {
+        // note: cache DynamicProperties to avoid using too many child watchers if getProperty is called repeatedly.
+        // for most use cases though, this cache is only filled/read once.
+        final DynamicProperty<?> cachedProp = cachedProperties.computeIfAbsent(definition.name(), name -> {
+            DynamicProperty<T> prop = new DynamicProperty<>(definition);
+            Watcher<JsonNode> child = rootWatcher.newChild(jsonNode -> jsonNode.path(definition.name()));
+            child.watch(node -> {
+                try {
+                    setValue(prop, node);
+                } catch (Exception e) {
+                    // Catching Exception instead of RuntimeException, since
+                    // Kotlin-implemented DynamicProperty would throw checked exceptions
+                    logger.warn("Failed to set value updated from CentralDogma for {}", definition.name(), e);
+                }
+            });
             try {
+                JsonNode node = child.initialValueFuture().join().value(); //doesn't fail since it's a child watcher
                 setValue(prop, node);
             } catch (Exception e) {
                 // Catching Exception instead of RuntimeException, since
                 // Kotlin-implemented DynamicProperty would throw checked exceptions
-                logger.warn("Failed to set value updated from CentralDogma for {}", definition.name(), e);
+                logger.warn("Failed to set initial value from CentralDogma for {}", definition.name(), e);
             }
+
+            return prop;
         });
-        try {
-            JsonNode node = child.initialValueFuture().join().value(); //doesn't fail since it's a child watcher
-            setValue(prop, node);
-        } catch (Exception e) {
-            // Catching Exception instead of RuntimeException, since
-            // Kotlin-implemented DynamicProperty would throw checked exceptions
-            logger.warn("Failed to set initial value from CentralDogma for {}", definition.name(), e);
-        }
 
-        return Optional.of(prop);
+        if (cachedProp.definition().runtimeType() != definition.runtimeType()) {
+            throw new IllegalStateException("Several different properties have the same name: " + definition.name());
+        }
+        //noinspection unchecked
+        return Optional.of((Property<T>) cachedProp);
     }
 
     @Override
     public void close() {
         rootWatcher.close();
+        cachedProperties.clear();
     }
 
     /**

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

@@ -31,6 +31,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
 
 import org.junit.Rule;
 import org.junit.Test;
@@ -119,6 +120,11 @@ public void testCDIntegration() throws InterruptedException {
 
         latch.await();
         assertEquals(20, prop.value().intValue());
+
+        IntStream.range(0, 10000)
+                 .mapToObj(i -> CONFIG_PARTITION_CONCURRENCY)
+                 .map(supplier::getProperty);
+        assertEquals(1, supplier.getCachedProperties().size());
     }
 
     @Test