[FLINK-24679][conf] Hide secret values from parser error messages

flink-core/src/main/java/org/apache/flink/configuration/Configuration.java

@@ -717,9 +717,11 @@ public <T> Optional<T> getOptional(ConfigOption<T> option) {
             }
         } catch (Exception e) {
             throw new IllegalArgumentException(
-                    String.format(
-                            "Could not parse value '%s' for key '%s'.",
-                            rawValue.map(Object::toString).orElse(""), option.key()),
+                    GlobalConfiguration.isSensitive(option.key())
+                            ? String.format("Could not parse value for key '%s'.", option.key())
+                            : String.format(
+                                    "Could not parse value '%s' for key '%s'.",
+                                    rawValue.map(Object::toString).orElse(""), option.key()),
                     e);
         }
     }

flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java

@@ -363,7 +363,7 @@ static Map<String, String> convertToProperties(Object o) {
                             pair -> {
                                 if (pair.size() != 2) {
                                     throw new IllegalArgumentException(
-                                            "Could not parse pair in the map " + pair);
+                                            "Map item is not a key-value pair (missing ':'?)");
                                 }
                             })
                     .collect(Collectors.toMap(a -> a.get(0), a -> a.get(1)));

flink-core/src/main/java/org/apache/flink/configuration/GlobalConfiguration.java

@@ -201,9 +201,7 @@ private static Configuration loadYAMLResource(File file) {
                                         + file
                                         + ":"
                                         + lineNo
-                                        + ": \""
-                                        + line
-                                        + "\"");
+                                        + ": Line is not a key-value pair (missing space after ':'?)");
                         continue;
                     }
 
@@ -217,9 +215,7 @@ private static Configuration loadYAMLResource(File file) {
                                         + file
                                         + ":"
                                         + lineNo
-                                        + ": \""
-                                        + line
-                                        + "\"");
+                                        + ": Key or value was empty");
                         continue;
                     }
 

flink-core/src/test/java/org/apache/flink/configuration/ConfigurationTest.java

@@ -18,9 +18,11 @@
 
 package org.apache.flink.configuration;
 
+import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.InstantiationUtil;
 import org.apache.flink.util.TestLogger;
 
+import org.assertj.core.api.Assertions;
 import org.junit.Test;
 
 import java.time.Duration;
@@ -30,6 +32,7 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertArrayEquals;
@@ -455,6 +458,44 @@ public void testMapRemovePrefix() {
         assertFalse(cfg.containsKey(MAP_PROPERTY_2));
     }
 
+    @Test
+    public void testListParserErrorDoesNotLeakSensitiveData() {
+        ConfigOption<List<String>> secret =
+                ConfigOptions.key("secret").stringType().asList().noDefaultValue();
+
+        Assertions.assertThat(GlobalConfiguration.isSensitive(secret.key())).isTrue();
+
+        final Configuration cfg = new Configuration();
+        // missing closing quote
+        cfg.setString(secret.key(), "'secret_value");
+
+        assertThatThrownBy(() -> cfg.get(secret))
+                .isInstanceOf(IllegalArgumentException.class)
+                .satisfies(
+                        e ->
+                                Assertions.assertThat(ExceptionUtils.stringifyException(e))
+                                        .doesNotContain("secret_value"));
+    }
+
+    @Test
+    public void testMapParserErrorDoesNotLeakSensitiveData() {
+        ConfigOption<Map<String, String>> secret =
+                ConfigOptions.key("secret").mapType().noDefaultValue();
+
+        Assertions.assertThat(GlobalConfiguration.isSensitive(secret.key())).isTrue();
+
+        final Configuration cfg = new Configuration();
+        // malformed map representation
+        cfg.setString(secret.key(), "secret_value");
+
+        assertThatThrownBy(() -> cfg.get(secret))
+                .isInstanceOf(IllegalArgumentException.class)
+                .satisfies(
+                        e ->
+                                Assertions.assertThat(ExceptionUtils.stringifyException(e))
+                                        .doesNotContain("secret_value"));
+    }
+
     // --------------------------------------------------------------------------------------------
     // Test classes
     // --------------------------------------------------------------------------------------------