From cac4bec0357e4bb5a115bbf68c118ab1a7eb266f Mon Sep 17 00:00:00 2001
From: Ewen Cheslack-Postava <me@ewencp.org>
Date: Fri, 10 Oct 2014 23:57:04 -0700
Subject: [PATCH] KAFKA-1698 Validate parsed ConfigDef values in addition to
 the default values.

---
 .../org/apache/kafka/common/config/ConfigDef.java  |  2 ++
 .../apache/kafka/common/config/ConfigDefTest.java  | 41 ++++++++++++++++++++++
 2 files changed, 43 insertions(+)

diff --git a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
index 227309e..c4cea2c 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
@@ -124,6 +124,8 @@ public class ConfigDef {
                 throw new ConfigException("Missing required configuration \"" + key.name + "\" which has no default value.");
             else
                 value = key.defaultValue;
+            if (key.validator != null)
+                key.validator.ensureValid(key.name, value);
             values.put(key.name, value);
         }
         return values;
diff --git a/clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java b/clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java
index 09a82fe..7869e46 100644
--- a/clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java
@@ -16,12 +16,15 @@ import static java.util.Arrays.asList;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 
 import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Validator;
 import org.apache.kafka.common.config.ConfigDef.Range;
+import org.apache.kafka.common.config.ConfigDef.ValidString;
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.junit.Test;
 
@@ -97,4 +100,42 @@ public class ConfigDefTest {
             }
         }
     }
+
+    @Test(expected = ConfigException.class)
+    public void testInvalidDefaultRange() {
+        ConfigDef def = new ConfigDef().define("name", Type.INT, -1, Range.between(0,10), Importance.HIGH, "docs");
+    }
+
+    @Test(expected = ConfigException.class)
+    public void testInvalidDefaultString() {
+        ConfigDef def = new ConfigDef().define("name", Type.STRING, "bad", ValidString.in(Arrays.asList("valid", "values")), Importance.HIGH, "docs");
+    }
+
+    @Test
+    public void testValidators() {
+        testValidators(Type.INT, Range.between(0,10), 5, new Object[]{1, 5, 9}, new Object[]{-1, 11});
+        testValidators(Type.STRING, ValidString.in(Arrays.asList("good", "values", "default")), "default",
+                new Object[]{"good", "values", "default"}, new Object[]{"bad", "inputs"});
+    }
+
+    private void testValidators(Type type, Validator validator, Object defaultVal, Object[] okValues, Object[] badValues) {
+        ConfigDef def = new ConfigDef().define("name", type, defaultVal, validator, Importance.HIGH, "docs");
+
+        for (Object value : okValues) {
+            Map<String, Object> m = new HashMap<String, Object>();
+            m.put("name", value);
+            def.parse(m);
+        }
+
+        for (Object value : badValues) {
+            Map<String, Object> m = new HashMap<String, Object>();
+            m.put("name", value);
+            try {
+                def.parse(m);
+                fail("Expected a config exception due to invalid value " + value);
+            } catch (ConfigException e) {
+                // this is good
+            }
+        }
+    }
 }
-- 
2.1.2