[improve][admin] Validate dynamic configuration updates against FieldContext constraints

pulsar-broker-common/src/main/java/org/apache/pulsar/common/configuration/FieldContextValidator.java

@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.configuration;
+
+import java.lang.reflect.Field;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.common.util.FieldParser;
+
+/**
+ * Validates {@link FieldContext} value constraints ({@code required}, numeric bounds,
+ * {@code maxCharLength}, and boolean format for string inputs) on reflected configuration fields.
+ * <p>
+ * Used when validating configuration completeness at startup
+ * ({@link PulsarConfigurationLoader#isComplete}) and when validating dynamic configuration
+ * updates via the admin API.
+ */
+public final class FieldContextValidator {
+
+    private FieldContextValidator() {
+    }
+
+    /**
+     * Validates a string configuration value for a reflected field.
+     * <p>
+     * Parses {@code valueStr} with {@link FieldParser}, then checks {@code required},
+     * numeric bounds, {@code maxCharLength}, and boolean format. Blank {@code valueStr}
+     * is valid for optional fields; for dynamic configuration, blank means clearing the
+     * override.
+     *
+     * @param field    configuration field
+     * @param valueStr raw string value from the admin API
+     * @return error message if validation fails, or empty if the field has no
+     *         {@link FieldContext} or the value passes
+     */
+    public static Optional<String> validateString(Field field, String valueStr) {
+        if (field == null || !field.isAnnotationPresent(FieldContext.class)) {
+            return Optional.empty();
+        }
+        FieldContext fieldContext = field.getAnnotation(FieldContext.class);
+        if (StringUtils.isBlank(valueStr)) {
+            if (fieldContext.required()) {
+                return Optional.of(String.format("Required %s is null", field.getName()));
+            }
+            return Optional.empty();
+        }
+        if (isBooleanField(field) && !isValidBooleanString(valueStr)) {
+            return Optional.of(String.format("%s must be 'true' or 'false'", field.getName()));
+        }
+        final Object parsedValue;
+        try {
+            parsedValue = FieldParser.value(valueStr, field);
+        } catch (Exception e) {
+            String message = e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName();
+            return Optional.of(String.format("Failed to parse %s: %s", field.getName(), message));
+        }
+        return validateParsedValue(field, parsedValue);
+    }
+
+    /**
+     * Validates a configuration value that is already parsed to the field's Java type.
+     * <p>
+     * Checks {@code required}, numeric {@code minValue}/{@code maxValue}, and
+     * {@code maxCharLength} from {@link FieldContext}. Does not parse strings or validate
+     * boolean string format.
+     *
+     * @param field configuration field
+     * @param value value in the field's Java type
+     * @return error message if validation fails, or empty if the field has no
+     *         {@link FieldContext} or the value passes
+     */
+    public static Optional<String> validateParsedValue(Field field, Object value) {
+        if (field == null || !field.isAnnotationPresent(FieldContext.class)) {
+            return Optional.empty();
+        }
+        FieldContext fieldContext = field.getAnnotation(FieldContext.class);
+        if (fieldContext.required() && isEmpty(value)) {
+            return Optional.of(String.format("Required %s is null", field.getName()));
+        }
+        if (value != null && Number.class.isAssignableFrom(value.getClass())) {
+            long fieldVal = ((Number) value).longValue();
+            long minValue = fieldContext.minValue();
+            long maxValue = fieldContext.maxValue();
+            if (fieldVal < minValue || fieldVal > maxValue) {
+                return Optional.of(String.format("%s value %d doesn't fit in given range (%d, %d)",
+                        field.getName(), fieldVal, minValue, maxValue));
+            }
+        }
+        if (value instanceof String stringValue && stringValue.length() > fieldContext.maxCharLength()) {
+            return Optional.of(String.format("%s exceeds maxCharLength %d",
+                    field.getName(), fieldContext.maxCharLength()));
+        }
+        return Optional.empty();
+    }
+
+    private static boolean isBooleanField(Field field) {
+        Class<?> type = field.getType();
+        return type == boolean.class || type == Boolean.class;
+    }
+
+    private static boolean isValidBooleanString(String valueStr) {
+        return "true".equalsIgnoreCase(valueStr) || "false".equalsIgnoreCase(valueStr);
+    }
+
+    private static boolean isEmpty(Object value) {
+        if (value == null) {
+            return true;
+        }
+        if (value instanceof String stringValue) {
+            return StringUtils.isBlank(stringValue);
+        }
+        return false;
+    }
+}

pulsar-broker-common/src/main/java/org/apache/pulsar/common/configuration/PulsarConfigurationLoader.java

@@ -29,10 +29,10 @@
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.TreeMap;
 import lombok.CustomLog;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.ServiceConfiguration;
 
 /**
@@ -135,21 +135,8 @@ public static boolean isComplete(Object obj) throws IllegalArgumentException {
                         .attr("field", field.getName())
                         .attr("value", value)
                         .log("Validating configuration field");
-                boolean isRequired = field.getAnnotation(FieldContext.class).required();
-                long minValue = field.getAnnotation(FieldContext.class).minValue();
-                long maxValue = field.getAnnotation(FieldContext.class).maxValue();
-                if (isRequired && isEmpty(value)) {
-                    error.append(String.format("Required %s is null,", field.getName()));
-                }
-
-                if (value != null && Number.class.isAssignableFrom(value.getClass())) {
-                    long fieldVal = ((Number) value).longValue();
-                    boolean valid = fieldVal >= minValue && fieldVal <= maxValue;
-                    if (!valid) {
-                        error.append(String.format("%s value %d doesn't fit in given range (%d, %d),", field.getName(),
-                                fieldVal, minValue, maxValue));
-                    }
-                }
+                Optional<String> validationError = FieldContextValidator.validateParsedValue(field, value);
+                validationError.ifPresent(err -> error.append(err).append(","));
             }
         }
         if (error.length() > 0) {
@@ -158,16 +145,6 @@ public static boolean isComplete(Object obj) throws IllegalArgumentException {
         return true;
     }
 
-    private static boolean isEmpty(Object obj) {
-        if (obj == null) {
-            return true;
-        } else if (obj instanceof String) {
-            return StringUtils.isBlank((String) obj);
-        } else {
-            return false;
-        }
-    }
-
     /**
      * Converts a PulsarConfiguration object to a ServiceConfiguration object.
      *

pulsar-broker-common/src/test/java/org/apache/pulsar/common/configuration/FieldContextValidatorTest.java

@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.configuration;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import java.lang.reflect.Field;
+import java.util.Optional;
+import org.testng.annotations.Test;
+
+public class FieldContextValidatorTest {
+
+    @Test
+    public void testValidateStringSkipsWhenFieldHasNoAnnotation() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("noAnnotation");
+        assertFalse(FieldContextValidator.validateString(field, "value").isPresent());
+    }
+
+    @Test
+    public void testValidateStringAllowsBlankForOptionalField() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("optionalString");
+        assertFalse(FieldContextValidator.validateString(field, "").isPresent());
+        assertFalse(FieldContextValidator.validateString(field, "   ").isPresent());
+        assertFalse(FieldContextValidator.validateString(field, null).isPresent());
+    }
+
+    @Test
+    public void testValidateStringRejectsBlankForRequiredField() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("requiredString");
+        Optional<String> error = FieldContextValidator.validateString(field, "");
+        assertTrue(error.isPresent());
+        assertEquals(error.get(), "Required requiredString is null");
+    }
+
+    @Test
+    public void testValidateStringAcceptsValidBooleanValues() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("enabled");
+        assertFalse(FieldContextValidator.validateString(field, "true").isPresent());
+        assertFalse(FieldContextValidator.validateString(field, "false").isPresent());
+        assertFalse(FieldContextValidator.validateString(field, "TRUE").isPresent());
+    }
+
+    @Test
+    public void testValidateStringRejectsInvalidBooleanValues() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("enabled");
+        Optional<String> error = FieldContextValidator.validateString(field, "abc");
+        assertTrue(error.isPresent());
+        assertEquals(error.get(), "enabled must be 'true' or 'false'");
+    }
+
+    @Test
+    public void testValidateStringRejectsUnparseableNumericValue() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("boundedInt");
+        Optional<String> error = FieldContextValidator.validateString(field, "abc");
+        assertTrue(error.isPresent());
+        assertTrue(error.get().startsWith("Failed to parse boundedInt:"));
+    }
+
+    @Test
+    public void testValidateStringRejectsOutOfRangeNumericValue() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("boundedInt");
+        Optional<String> error = FieldContextValidator.validateString(field, "0");
+        assertTrue(error.isPresent());
+        assertEquals(error.get(), "boundedInt value 0 doesn't fit in given range (1, 3)");
+    }
+
+    @Test
+    public void testValidateStringAcceptsValidNumericValue() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("boundedInt");
+        assertFalse(FieldContextValidator.validateString(field, "2").isPresent());
+    }
+
+    @Test
+    public void testValidateStringRejectsStringExceedingMaxCharLength() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("shortString");
+        Optional<String> error = FieldContextValidator.validateString(field, "abcd");
+        assertTrue(error.isPresent());
+        assertEquals(error.get(), "shortString exceeds maxCharLength 3");
+    }
+
+    @Test
+    public void testValidateParsedValueSkipsWhenFieldHasNoAnnotation() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("noAnnotation");
+        assertFalse(FieldContextValidator.validateParsedValue(field, "value").isPresent());
+    }
+
+    @Test
+    public void testValidateParsedValueRejectsRequiredNull() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("requiredString");
+        Optional<String> error = FieldContextValidator.validateParsedValue(field, null);
+        assertTrue(error.isPresent());
+        assertEquals(error.get(), "Required requiredString is null");
+    }
+
+    @Test
+    public void testValidateParsedValueRejectsRequiredBlankString() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("requiredString");
+        Optional<String> error = FieldContextValidator.validateParsedValue(field, "  ");
+        assertTrue(error.isPresent());
+        assertEquals(error.get(), "Required requiredString is null");
+    }
+
+    @Test
+    public void testValidateParsedValueRejectsOutOfRangeNumber() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("boundedInt");
+        Optional<String> error = FieldContextValidator.validateParsedValue(field, 4);
+        assertTrue(error.isPresent());
+        assertEquals(error.get(), "boundedInt value 4 doesn't fit in given range (1, 3)");
+    }
+
+    @Test
+    public void testValidateParsedValueAcceptsInRangeNumber() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("boundedInt");
+        assertFalse(FieldContextValidator.validateParsedValue(field, 2).isPresent());
+    }
+
+    @Test
+    public void testValidateParsedValueRejectsStringExceedingMaxCharLength() throws Exception {
+        Field field = TestConfig.class.getDeclaredField("shortString");
+        Optional<String> error = FieldContextValidator.validateParsedValue(field, "abcd");
+        assertTrue(error.isPresent());
+        assertEquals(error.get(), "shortString exceeds maxCharLength 3");
+    }
+
+    static class TestConfig {
+        String noAnnotation;
+
+        @FieldContext(required = true)
+        String requiredString;
+
+        @FieldContext
+        String optionalString;
+
+        @FieldContext(dynamic = true)
+        boolean enabled;
+
+        @FieldContext(minValue = 1, maxValue = 3)
+        int boundedInt;
+
+        @FieldContext(maxCharLength = 3)
+        String shortString;
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java

@@ -44,6 +44,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
@@ -338,9 +339,10 @@ public void getRuntimeConfiguration(@Suspended AsyncResponse asyncResponse) {
      */
     private synchronized CompletableFuture<Void> persistDynamicConfigurationAsync(
             String configName, String configValue) {
-        if (!pulsar().getBrokerService().validateDynamicConfiguration(configName, configValue)) {
-            return FutureUtil
-                    .failedFuture(new RestException(Status.PRECONDITION_FAILED, " Invalid dynamic-config value"));
+        Optional<String> validationError = pulsar().getBrokerService()
+                .getDynamicConfigurationValidationError(configName, configValue);
+        if (validationError.isPresent()) {
+            return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, validationError.get()));
         }
         if (pulsar().getBrokerService().isDynamicConfiguration(configName)) {
             return dynamicConfigurationResources().setDynamicConfigurationWithCreateAsync(old -> {