Merge branch '5.0.x' into 5.1.x

Author: ConfluentJenkins

src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfig.java

@@ -608,8 +608,13 @@ private Map<String, String> parseMapConfig(List<String> values) {
   private static class UrlListValidator implements Validator {
 
     @Override
+    @SuppressWarnings("unchecked")
     public void ensureValid(String name, Object value) {
-      @SuppressWarnings("unchecked")
+      if (value == null) {
+        throw new ConfigException(
+            name, value, "The provided url list is null."
+        );
+      }
       List<String> urls = (List<String>) value;
       for (String url : urls) {
         try {

src/test/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfigTest.java

@@ -7,6 +7,7 @@
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG;
 import static org.junit.Assert.assertEquals;
 
+import org.apache.kafka.common.config.ConfigException;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -40,4 +41,10 @@ public void testSetHttpTimeoutsConfig() {
     assertEquals(config.readTimeoutMs(), 10000);
     assertEquals(config.connectionTimeoutMs(), 15000);
   }
+
+  @Test(expected = ConfigException.class)
+  public void shouldNotAllowNullUrlList(){
+    props.put(CONNECTION_URL_CONFIG, null);
+    new ElasticsearchSinkConnectorConfig(props);
+  }
 }