Validate config values for 'not-null' (#544)

Author: namelessvoid

src/Confluent.Kafka/Consumer.cs

@@ -819,15 +819,21 @@ public Consumer(IEnumerable<KeyValuePair<string, object>> config)
             config
                 .Where(prop => prop.Key != "default.topic.config")
                 .ToList()
-                .ForEach((kvp) => { configHandle.Set(kvp.Key, kvp.Value.ToString()); });
+                .ForEach((kvp) => {
+                    if (kvp.Value == null) throw new ArgumentException($"'{kvp.Key}' configuration parameter must not be null.");
+                    configHandle.Set(kvp.Key, kvp.Value.ToString());
+                });
 
             // Note: Setting default topic configuration properties via default.topic.config is depreciated 
             // and this functionality will be removed in a future version of the library.
             var defaultTopicConfig = (IEnumerable<KeyValuePair<string, object>>)config.FirstOrDefault(prop => prop.Key == "default.topic.config").Value;
             if (defaultTopicConfig != null)
             {
                 defaultTopicConfig.ToList().ForEach(
-                    (kvp) => { configHandle.Set(kvp.Key, kvp.Value.ToString()); }
+                    (kvp) => {
+                        if (kvp.Value == null) throw new ArgumentException($"'{kvp.Key}' configuration parameter in 'default.topic.config' must not be null.");
+                        configHandle.Set(kvp.Key, kvp.Value.ToString());
+                    }
                 );
             }
 

src/Confluent.Kafka/Producer.cs

@@ -279,7 +279,10 @@ public Producer(IEnumerable<KeyValuePair<string, object>> config, bool manualPol
                 modifiedConfig = modifiedConfig.Concat(new KeyValuePair<string, object>[] { new KeyValuePair<string, object>("produce.offset.report", "true") });
             }
 
-            modifiedConfig.ToList().ForEach((kvp) => { configHandle.Set(kvp.Key, kvp.Value.ToString()); });
+            modifiedConfig.ToList().ForEach((kvp) => {
+                if (kvp.Value == null) throw new ArgumentException($"'{kvp.Key}' configuration parameter must not be null.");
+                configHandle.Set(kvp.Key, kvp.Value.ToString());
+            });
 
             IntPtr configPtr = configHandle.DangerousGetHandle();
 

test/Confluent.Kafka.UnitTests/Consumer.cs

@@ -25,34 +25,52 @@ namespace Confluent.Kafka.UnitTests
 {
     public class ConsumerTests
     {
-        /// <summary>
-        ///     Test that the Consumer constructor throws an exception if
-        ///     the group.id configuration parameter is not set and that
-        ///     the message of the exception mentions group.id (i.e. is
-        ///     not some unrelated exception).
-        /// </summary>
         [Fact]
         public void Constuctor()
         {
+            // Throw exception if 'group.id' is not set in config and ensure that exception
+            // mentions 'group.id'.
             var config = new Dictionary<string, object>();
             var e = Assert.Throws<ArgumentException>(() => { var c = new Consumer(config); });
             Assert.True(e.Message.Contains("group.id"));
             e = Assert.Throws<ArgumentException>(() => { var c = new Consumer<Null, string>(config, null, new StringDeserializer(Encoding.UTF8)); });
             Assert.True(e.Message.Contains("group.id"));
 
+            // Throw exception if a config value is null and ensure that exception mentions the
+            // respective config key.
+            var configWithNullValue = CreateValidConfiguration();
+            configWithNullValue["sasl.password"] = null;
+            e = Assert.Throws<ArgumentException>(() => { var c = new Consumer<byte[], byte[]>(configWithNullValue, new ByteArrayDeserializer(), new ByteArrayDeserializer()); });
+            Assert.Contains("sasl.password", e.Message);
+
+            // Throw exception if a config value within default.topic.config is null and
+            // ensure that exception mentions the respective config key.
+            var configWithDefaultTopicNullValue = CreateValidConfiguration();
+            configWithDefaultTopicNullValue["default.topic.config"] = new Dictionary<string, object>() { { "auto.offset.reset", null } };
+            e = Assert.Throws<ArgumentException>(() => { var c = new Consumer<byte[], byte[]>(configWithDefaultTopicNullValue, new ByteArrayDeserializer(), new ByteArrayDeserializer()); });
+            Assert.Contains("default.topic.config", e.Message);
+            Assert.Contains("auto.offset.reset", e.Message);
+
+            // Throw exception when serializer and deserializer are equal and ensure that exception
+            // message indicates the issue.
             e = Assert.Throws<ArgumentException>(() => 
             {
-                var validConfig = new Dictionary<string, object>
-                {
-                    { "bootstrap.servers", "localhost:9092" },
-                    { "group.id", "my-group" }
-                };
+                var validConfig = CreateValidConfiguration();
                 var deserializer = new StringDeserializer(Encoding.UTF8);
                 var c = new Consumer<string, string>(validConfig, deserializer, deserializer); 
             });
             Assert.True(e.Message.Contains("must not be the same object"));
 
             // positve case covered by integration tests. here, avoiding creating a rd_kafka_t instance.
         }
+
+        private static Dictionary<string, object> CreateValidConfiguration()
+        {
+            return new Dictionary<string, object>
+            {
+                { "bootstrap.servers", "localhost:9092" },
+                { "group.id", "my-group" }
+            };
+        }
     }
 }

test/Confluent.Kafka.UnitTests/Producer.cs

@@ -0,0 +1,42 @@
+// Copyright 2016-2017 Confluent Inc.
+//
+// Licensed 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.
+//
+// Refer to LICENSE for more information.
+
+using Xunit;
+using System;
+using System.Text;
+using System.Collections.Generic;
+using Confluent.Kafka.Serialization;
+
+
+namespace Confluent.Kafka.UnitTests
+{
+    public class ProducerTests
+    {
+        [Fact]
+        public void Constuctor()
+        {
+            // Throw exception if a config value is null and ensure that exception mentions the
+            // respective config key.
+            var configWithNullValue = new Dictionary<string, object>
+            {
+                { "sasl.password", null }
+            };
+            configWithNullValue["sasl.password"] = null;
+            var e = Assert.Throws<ArgumentException>(() => { var c = new Producer<byte[], byte[]>(configWithNullValue, new ByteArraySerializer(), new ByteArraySerializer()); });
+            Assert.Contains("sasl.password", e.Message);
+        }
+    }
+}