Added redis.insecure option to disable verification of TLS certificates for Redis connections. Resolves #10

Author: jruaux

src/main/java/com/redislabs/kafka/connect/common/RedisEnterpriseConfig.java

@@ -17,6 +17,7 @@
 
 import com.github.jcustenborder.kafka.connect.utils.config.ConfigKeyBuilder;
 import com.github.jcustenborder.kafka.connect.utils.config.validators.Validators;
+import io.lettuce.core.RedisURI;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 
@@ -28,30 +29,34 @@ public class RedisEnterpriseConfig extends AbstractConfig {
     private static final String REDIS_URI_DEFAULT = "redis://localhost:6379";
     private static final String REDIS_URI_DOC = "URI of the Redis Enterprise database to connect to, e.g. redis://redis-12000.redislabs.com:12000";
 
-    private final String redisUri;
+    public static final String INSECURE = "redis.insecure";
+    public static final String INSECURE_DEFAULT = "false";
+    public static final String INSECURE_DOC = "Allow insecure connections (e.g. invalid certificates) to Redis Enterprise when using SSL.";
 
     public RedisEnterpriseConfig(ConfigDef config, Map<?, ?> originals) {
         super(config, originals);
-        redisUri = getString(REDIS_URI);
     }
 
-    public String getRedisUri() {
-        return redisUri;
+    public RedisURI getRedisURI() {
+        RedisURI uri = RedisURI.create(getString(REDIS_URI));
+        uri.setVerifyPeer(!getBoolean(INSECURE));
+        return uri;
     }
 
     protected static class RedisEnterpriseConfigDef extends ConfigDef {
 
         protected RedisEnterpriseConfigDef() {
-            defineRedisURI();
+            defineConfigs();
         }
 
         protected RedisEnterpriseConfigDef(ConfigDef base) {
             super(base);
-            defineRedisURI();
+            defineConfigs();
         }
 
-        private void defineRedisURI() {
+        private void defineConfigs() {
             define(ConfigKeyBuilder.of(REDIS_URI, ConfigDef.Type.STRING).documentation(REDIS_URI_DOC).defaultValue(REDIS_URI_DEFAULT).importance(ConfigDef.Importance.HIGH).validator(Validators.validURI("redis", "rediss")).build());
+            define(ConfigKeyBuilder.of(INSECURE, ConfigDef.Type.BOOLEAN).documentation(INSECURE_DOC).defaultValue(INSECURE_DEFAULT).importance(ConfigDef.Importance.MEDIUM).build());
         }
 
     }

src/main/java/com/redislabs/kafka/connect/sink/RedisEnterpriseSinkTask.java

@@ -78,7 +78,7 @@ public String version() {
     @Override
     public void start(final Map<String, String> props) {
         config = new RedisEnterpriseSinkConfig(props);
-        client = RedisClient.create(config.getRedisUri());
+        client = RedisClient.create(config.getRedisURI());
         connection = client.connect();
         charset = config.getCharset();
         writer = OperationItemWriter.operation(operation()).codec(new ByteArrayCodec()).client(client).transactional(Boolean.TRUE.equals(config.isMultiexec())).build();

src/main/java/com/redislabs/kafka/connect/source/AbstractSourceRecordReader.java

@@ -24,7 +24,7 @@ protected AbstractSourceRecordReader(RedisEnterpriseSourceConfig sourceConfig) {
 
     @Override
     public void open() {
-        this.client = RedisModulesClient.create(sourceConfig.getRedisUri());
+        this.client = RedisModulesClient.create(sourceConfig.getRedisURI());
         open(client);
     }