fix: handling of explicit configuration properties with @registryproperties

Test `io.apicurio.registry.utils.tests.KafkaSqlStartupVerificationIT#testBadSnapshotsTopicConfig2Override` in extra tests is failing because `apicurio.kafkasql.topic-configuration-verification-override-enabled` property has the prefix `apicurio.kafkasql.topic` which propagates it into topic configuration when creating the topic. The same issue is with `apicurio.kafkasql.topic.auto-create`, but we did not notice it because nobody has passed `-Dapicurio.kafkasql.topic.auto-create=true` to Registry before, since it's the default. This fixes the issue.

Author: jsenko

app/src/main/java/io/apicurio/registry/rules/RulesConfiguration.java

@@ -12,7 +12,7 @@ public class RulesConfiguration {
     @Produces
     @ApplicationScoped
     public RulesProperties rulesProperties(
-            @RegistryProperties(value = { "apicurio.rules.global" }) Properties properties) {
+            @RegistryProperties(prefixes = { "apicurio.rules.global" }) Properties properties) {
         return new RulesPropertiesImpl(properties);
     }
 

app/src/main/java/io/apicurio/registry/storage/impl/kafkasql/KafkaSqlConfiguration.java

@@ -76,7 +76,7 @@ public Duration getResponseTimeout() {
     String topic;
 
     @Inject
-    @RegistryProperties(value = "apicurio.kafkasql.topic")
+    @RegistryProperties(prefixes = "apicurio.kafkasql.topic", excluded = {"auto-create", "configuration-verification-override-enabled"})
     @Info(category = CATEGORY_STORAGE, description = """
             Kafka sql storage topic properties. \
             There are two optional Registry-specific configuration properties: 'partitions' and 'replication.factor'.""")
@@ -105,7 +105,7 @@ public Map<String, String> getTopicProperties() {
     String snapshotsTopic;
 
     @Inject
-    @RegistryProperties(value = "apicurio.kafkasql.snapshots.topic")
+    @RegistryProperties(prefixes = "apicurio.kafkasql.snapshots.topic")
     @Info(category = CATEGORY_STORAGE, description = """
             Kafka sql snapshots topic properties. \
             There are two optional Registry-specific configuration properties: 'partitions' and 'replication.factor'. \
@@ -148,7 +148,7 @@ public Map<String, String> getSnapshotTopicProperties() {
     String eventsTopic;
 
     @Inject
-    @RegistryProperties(value = "apicurio.events.kafka.topic")
+    @RegistryProperties(prefixes = "apicurio.events.kafka.topic")
     @Info(category = CATEGORY_STORAGE, description = """
             Kafka sql events topic properties. \
             There is an optional Registry-specific configuration property: 'replication.factor'. \
@@ -189,7 +189,7 @@ public Duration getPollTimeout() {
     String groupPrefix;
 
     @Inject
-    @RegistryProperties(value = {"apicurio.kafka.common", "apicurio.kafkasql.consumer"}, empties = {"ssl.endpoint.identification.algorithm="})
+    @RegistryProperties(prefixes = {"apicurio.kafka.common", "apicurio.kafkasql.consumer"}, defaults = {"ssl.endpoint.identification.algorithm="})
     Properties consumerProperties;
 
     public Map<String, String> getConsumerProperties() {
@@ -213,7 +213,7 @@ public Map<String, String> getConsumerProperties() {
     // === Producer configurations ===
 
     @Inject
-    @RegistryProperties(value = {"apicurio.kafka.common", "apicurio.kafkasql.producer"}, empties = {"ssl.endpoint.identification.algorithm="})
+    @RegistryProperties(prefixes = {"apicurio.kafka.common", "apicurio.kafkasql.producer"}, defaults = {"ssl.endpoint.identification.algorithm="})
     Properties producerProperties;
 
     public Map<String, String> getProducerProperties() {
@@ -236,7 +236,7 @@ public Map<String, String> getProducerProperties() {
     // === Admin client configurations ===
 
     @Inject
-    @RegistryProperties(value = {"apicurio.kafka.common", "apicurio.kafkasql.admin"}, empties = {"ssl.endpoint.identification.algorithm="})
+    @RegistryProperties(prefixes = {"apicurio.kafka.common", "apicurio.kafkasql.admin"}, defaults = {"ssl.endpoint.identification.algorithm="})
     Properties adminProperties;
 
     public Map<String, String> getAdminProperties() {

common/src/main/java/io/apicurio/registry/utils/PropertiesUtil.java

@@ -8,7 +8,7 @@
 
 import java.util.Arrays;
 import java.util.HashMap;
-import java.util.Map;
+import java.util.HashSet;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.stream.Collectors;
@@ -27,52 +27,68 @@ public class PropertiesUtil {
      * @return filtered/stripped properties
      */
     public static Properties properties(InjectionPoint ip) {
+
         RegistryProperties cp = ip.getAnnotated().getAnnotation(RegistryProperties.class);
         if (cp == null) {
-            throw new IllegalArgumentException(ip.getMember() + " is not annotated with @RegistryProperties");
+            throw new IllegalArgumentException(ip.getMember() + " must be annotated with @RegistryProperties.");
         }
-        String[] prefixes = Stream.of(cp.value())
+
+        var prefixes = Stream.of(cp.prefixes())
                 .map(pfx -> pfx.isEmpty() || pfx.endsWith(".") ? pfx : pfx + ".").distinct()
-                .toArray(String[]::new);
-        if (prefixes.length == 0) {
+                .toList();
+
+        if (prefixes.isEmpty()) {
             throw new IllegalArgumentException("Annotation @RegistryProperties on " + ip.getMember()
-                    + " is missing non-empty 'value' attribute");
+                    + " must have a non-empty 'prefixes' attribute.");
         }
 
         Properties properties = new Properties();
         Config config = ConfigProviderResolver.instance().getConfig();
 
         if (debug && log.isDebugEnabled()) {
-            String dump = StreamSupport.stream(config.getPropertyNames().spliterator(), false).sorted()
+            String dump = StreamSupport.stream(config.getPropertyNames().spliterator(), false)
+                    .sorted()
                     .map(key -> key + "=" + config.getOptionalValue(key, String.class).orElse(""))
                     .collect(Collectors.joining("\n  ", "  ", "\n"));
             log.debug("Injecting config properties with prefixes {} into {} from the following...\n{}",
-                    Arrays.toString(prefixes), ip.getMember(), dump);
+                    prefixes, ip.getMember(), dump);
         }
 
-        // some security properties take empty value as config
-        Map<String, String> defaults = new HashMap<>();
-        if (cp != null) {
-            String[] empties = cp.empties();
-            for (String e : empties) {
-                int p = e.indexOf("=");
-                defaults.put(e.substring(0, p), e.substring(p + 1));
-            }
+        var defaults = new HashMap<String, String>();
+        for (String d : cp.defaults()) {
+            int p = d.indexOf("=");
+            defaults.put(d.substring(0, p), d.substring(p + 1));
         }
 
-        // collect properties with specified prefixes in order of prefixes (later prefix overrides earlier)
+        var excludedList = Arrays.asList(cp.excluded());
+        var excluded = new HashSet<>();
+        // When the property is passed as an env. variable, dashes are interpreted as dots,
+        // so we need to exclude both forms.
+        excluded.addAll(excludedList);
+        excluded.addAll(excludedList.stream().map(s -> s.replace('-', '.')).toList());
+
+        // Collect properties with specified prefixes in order of prefixes (later prefix overrides earlier)
         for (String prefix : prefixes) {
             for (String key : config.getPropertyNames()) {
                 if (key.startsWith(prefix)) {
-                    // property can exist with key, but no value ...
-                    Optional<String> value = config.getOptionalValue(key, String.class);
-                    if (value.isPresent()) {
-                        properties.put(key.substring(prefix.length()), value.get());
-                    } else if (defaults.size() > 0) {
-                        String sKey = key.substring(prefix.length());
-                        String defaultValue = defaults.get(sKey);
-                        if (defaultValue != null) {
-                            properties.put(sKey, defaultValue);
+                    var suffix = key.substring(prefix.length());
+                    if (!suffix.isEmpty()) {
+                        log.debug("Processing property '{}'", key);
+                        if (!excluded.contains(suffix)) {
+                            // Property can exist with a key but no value...
+                            Optional<String> value = config.getOptionalValue(key, String.class);
+                            if (value.isPresent()) {
+                                properties.put(suffix, value.get());
+                            } else {
+                                String defaultValue = defaults.get(suffix);
+                                if (defaultValue != null) {
+                                    properties.put(suffix, defaultValue);
+                                } else {
+                                    log.debug("Property '{}' has no value and no default, skipping.", key);
+                                }
+                            }
+                        } else {
+                            log.debug("Property '{}' is excluded, skipping.", key);
                         }
                     }
                 }

common/src/main/java/io/apicurio/registry/utils/RegistryProperties.java

@@ -7,9 +7,27 @@
 import static java.lang.annotation.RetentionPolicy.RUNTIME;
 
 @Retention(RUNTIME)
-@Target({ ElementType.METHOD, ElementType.FIELD, ElementType.PARAMETER })
+@Target({ElementType.METHOD, ElementType.FIELD, ElementType.PARAMETER})
 public @interface RegistryProperties {
-    String[] value();
 
-    String[] empties() default {};
+    /**
+     * The prefixes of keys of configuration properties that should be included.
+     * Keys with a later prefix in the array override the earlier ones.
+     */
+    String[] prefixes();
+
+    /**
+     * If a configuration property (excluding the prefix) key is found, but has no value,
+     * the default value from this array is used.
+     * This can be used to set empty string values for properties that require it.
+     * Each entry must be of the form "key-prefix=value".
+     */
+    String[] defaults() default {};
+
+    /**
+     * Keys of configuration properties (excluding the prefix) that should be ignored.
+     * This is useful, for example, if we have defined standard configuration property
+     * in addition to a prefix-based set, and we don't want it to be included in the resulting Properties object.
+     */
+    String[] excluded() default {};
 }

docs/config-generator/src/main/java/io/apicurio/registry/docs/GenerateAllConfigPartial.java

@@ -269,7 +269,7 @@ public static Map<String, Option> extractConfigurations(String jarFile, Map<Stri
                             .orElse(""));
 
             // Get the property prefixes from @RegistryProperties annotation
-            var values = annotation.value("value").asStringArray();
+            var values = annotation.value("prefixes").asStringArray();
 
             for (String prefix : values) {
                 prefix = prefix.replace("app.authn.", "apicurio.auth.");