Errorhandling and ssl support (#10)

patitapaban19 · web-flow · commit e177741a9033 · 2020-03-11T13:45:59.000-07:00
behavior.on.error : Error handling behavior setting. Must be configured to one of the following:

scylladb.ssl.keystore.path : Path to the Java Keystore.

scylladb.ssl.keystore.password : Password to open the Java Keystore with.

scylladb.ssl.cipherSuites : The cipher suites to enable. Defaults to none, resulting in a minimal quality of service according to JDK documentation.

scylladb.ssl.openssl.keyCertChain : Path to the SSL certificate file, when using OpenSSL.

ssl.openssl.privateKey : Path to the private key file, when using OpenSSL.
diff --git a/src/main/java/io/connect/scylladb/ScyllaDbSchemaBuilder.java b/src/main/java/io/connect/scylladb/ScyllaDbSchemaBuilder.java
@@ -124,7 +124,7 @@ DataType dataType(Schema schema) {
           dataType = DataType.varchar();
           break;
         default:
-          throw new UnsupportedOperationException(
+          throw new DataException(
               String.format("Unsupported type %s", schema.type())
           );
       }
diff --git a/src/main/java/io/connect/scylladb/ScyllaDbSessionFactory.java b/src/main/java/io/connect/scylladb/ScyllaDbSessionFactory.java
@@ -15,14 +15,19 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLException;
 import javax.net.ssl.TrustManagerFactory;
+import java.io.BufferedInputStream;
+import java.io.File;
 import java.io.FileInputStream;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.security.KeyStore;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
 import java.security.cert.CertificateException;
 
 public class ScyllaDbSessionFactory {
@@ -59,25 +64,13 @@ public ScyllaDbSession newSession(ScyllaDbSinkConnectorConfig config) {
 
       if (null != config.trustStorePath) {
         log.info("Configuring SSLContext to use Truststore {}", config.trustStorePath);
-        final KeyStore keyStore;
-        try {
-          keyStore = KeyStore.getInstance("JKS");
-          try (InputStream inputStream = new FileInputStream(config.trustStorePath)) {
-            keyStore.load(inputStream, config.trustStorePassword);
-          } catch (IOException e) {
-            throw new ConnectException("Exception while reading keystore", e);
-          } catch (CertificateException | NoSuchAlgorithmException e) {
-            throw new ConnectException("Exception while loading keystore", e);
-          }
-        } catch (KeyStoreException e) {
-          throw new ConnectException("Exception while creating keystore", e);
-        }
+        final KeyStore trustKeyStore = createKeyStore(config.trustStorePath, config.trustStorePassword);
 
         final TrustManagerFactory trustManagerFactory;
         try {
           trustManagerFactory =
               TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
-          trustManagerFactory.init(keyStore);
+          trustManagerFactory.init(trustKeyStore);
         } catch (NoSuchAlgorithmException e) {
           throw new ConnectException("Exception while creating TrustManagerFactory", e);
         } catch (KeyStoreException e) {
@@ -86,6 +79,42 @@ public ScyllaDbSession newSession(ScyllaDbSinkConnectorConfig config) {
         sslContextBuilder.trustManager(trustManagerFactory);
       }
 
+      if (null != config.keyStorePath) {
+        log.info("Configuring SSLContext to use Keystore {}", config.keyStorePath);
+        final KeyStore keyStore = createKeyStore(config.keyStorePath, config.keyStorePassword);
+
+        final KeyManagerFactory keyManagerFactory;
+        try {
+          keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+          keyManagerFactory.init(keyStore, config.keyStorePassword);
+        } catch (NoSuchAlgorithmException e) {
+          throw new ConnectException("Exception while creating KeyManagerFactory", e);
+        } catch (UnrecoverableKeyException | KeyStoreException e) {
+          throw new ConnectException("Exception while calling KeyManagerFactory.init()", e);
+        }
+        sslContextBuilder.keyManager(keyManagerFactory);
+      }
+
+      if (config.cipherSuites.size() > 0) {
+        sslContextBuilder.ciphers(config.cipherSuites);
+      }
+
+      if (config.certFilePath != null && config.privateKeyPath != null) {
+        try {
+          sslContextBuilder.keyManager(new BufferedInputStream(new FileInputStream(config.certFilePath)),
+                  new BufferedInputStream(new FileInputStream(config.privateKeyPath)));
+        }
+        catch (IllegalArgumentException e) {
+          throw new ConnectException(String.format("Invalid certificate or private key: %s", e.getMessage()));
+        } catch (FileNotFoundException e) {
+          throw new ConnectException("Invalid certificate or private key file path", e);
+        }
+      } else if (config.certFilePath == null != (config.privateKeyPath == null)) {
+        throw new ConnectException(String.format("%s cannot be set without %s and vice-versa: %s is not set",
+                "scylladb.ssl.openssl.keyCertChain", "scylladb.ssl.openssl.privateKey",
+                (config.certFilePath == null) ? "scylladb.ssl.openssl.keyCertChain" : "scylladb.ssl.openssl.privateKey"));
+      }
+
       final SslContext context;
       try {
         context = sslContextBuilder.build();
@@ -101,4 +130,21 @@ public ScyllaDbSession newSession(ScyllaDbSinkConnectorConfig config) {
     final Session session = cluster.newSession();
     return new ScyllaDbSessionImpl(config, cluster, session);
   }
+
+  private KeyStore createKeyStore(File path, char[] password) {
+    KeyStore keyStore;
+    try {
+      keyStore = KeyStore.getInstance("JKS");
+      try (InputStream inputStream = new FileInputStream(path)) {
+        keyStore.load(inputStream, password);
+      } catch (IOException e) {
+        throw new ConnectException("Exception while reading keystore", e);
+      } catch (CertificateException | NoSuchAlgorithmException e) {
+        throw new ConnectException("Exception while loading keystore", e);
+      }
+    } catch (KeyStoreException e) {
+      throw new ConnectException("Exception while creating keystore", e);
+    }
+    return keyStore;
+  }
 }
diff --git a/src/main/java/io/connect/scylladb/ScyllaDbSinkConnectorConfig.java b/src/main/java/io/connect/scylladb/ScyllaDbSinkConnectorConfig.java
@@ -1,6 +1,7 @@
 package io.connect.scylladb;
 
 import java.io.File;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -47,18 +48,23 @@ public class ScyllaDbSinkConnectorConfig extends AbstractConfig {
   public final TableOptions.CompressionOptions tableCompressionAlgorithm;
   public final char[] trustStorePassword;
   public final File trustStorePath;
+  public final char[] keyStorePassword;
+  public final File keyStorePath;
   public final String offsetStorageTable;
   public final long statementTimeoutMs;
   public final int maxBatchSizeKb;
   public final String loadBalancingLocalDc;
   public final long timestampResolutionMs;
   public final Map<String, TopicConfigs> topicWiseConfigs;
   public final Integer ttl;
+  public final BehaviorOnError behaviourOnError;
+  public final List<String> cipherSuites;
+  public final File certFilePath;
+  public final File privateKeyPath;
 
   private static final Pattern TOPIC_KS_TABLE_SETTING_PATTERN =
           Pattern.compile("topic\\.([a-zA-Z0-9._-]+)\\.([^.]+|\"[\"]+\")\\.([^.]+|\"[\"]+\")\\.(mapping|consistencyLevel|ttlSeconds|deletesEnabled)$");
 
-
   static final Map<String, ProtocolOptions.Compression> CLIENT_COMPRESSION = 
       ImmutableMap.of(
           "NONE", ProtocolOptions.Compression.NONE,
@@ -95,6 +101,18 @@ public ScyllaDbSinkConnectorConfig(Map<?, ?> originals) {
     this.trustStorePassword =
             this.getPassword(SSL_TRUSTSTORE_PASSWORD_CONFIG).value().toCharArray();
 
+    final String keyStorePath = this.getString(SSL_KEYSTORE_PATH_CONFIG);
+    this.keyStorePath = Strings.isNullOrEmpty(keyStorePath) ? null : new File(keyStorePath);
+    this.keyStorePassword =
+            this.getPassword(SSL_KEYSTORE_PASSWORD_CONFIG).value().toCharArray();
+
+    this.cipherSuites = getList(SSL_CIPHER_SUITES_CONFIG);
+
+    final String certFilePath = this.getString(SSL_OPENSLL_KEYCERTCHAIN_CONFIG);
+    this.certFilePath = Strings.isNullOrEmpty(certFilePath) ? null : new File(certFilePath);
+    final String privateKeyPath = this.getString(SSL_OPENSLL_PRIVATEKEY_CONFIG);
+    this.privateKeyPath = Strings.isNullOrEmpty(privateKeyPath) ? null : new File(privateKeyPath);
+
     final String compression = getString(COMPRESSION_CONFIG);
     this.compression = CLIENT_COMPRESSION.get(compression);
     this.sslProvider = ConfigUtils.getEnum(SslProvider.class, this, SSL_PROVIDER_CONFIG);
@@ -127,6 +145,8 @@ public ScyllaDbSinkConnectorConfig(Map<?, ?> originals) {
     this.maxBatchSizeKb = getInt(MAX_BATCH_SIZE_CONFIG);
     this.loadBalancingLocalDc = getString(LOAD_BALANCING_LOCAL_DC_CONFIG);
     this.timestampResolutionMs = getLong(TIMESTAMP_RESOLUTION_MS_CONF);
+    this.behaviourOnError = BehaviorOnError.valueOf(getString(BEHAVIOR_ON_ERROR_CONFIG).toUpperCase());
+
     Map<String, Map<String, String>> topicWiseConfigsMap = new HashMap<>();
     for (final Map.Entry<String, String> entry : ((Map<String, String>) originals).entrySet()) {
       final String name2 = entry.getKey();
@@ -236,6 +256,22 @@ public ScyllaDbSinkConnectorConfig(Map<?, ?> originals) {
   public static final String SSL_TRUSTSTORE_PASSWORD_CONFIG = "scylladb.ssl.truststore.password";
   private static final String SSL_TRUSTSTORE_PASSWORD_DOC = "Password to open the Java Truststore with.";
 
+  public static final String SSL_KEYSTORE_PATH_CONFIG = "scylladb.ssl.keystore.path";
+  private static final String SSL_KEYSTORE_PATH_DOC = "Path to the Java Keystore";
+
+  public static final String SSL_KEYSTORE_PASSWORD_CONFIG = "scylladb.ssl.keystore.password";
+  private static final String SSL_KEYSTORE_PASSWORD_DOC = "Password to open the Java Keystore with.";
+
+  public static final String SSL_CIPHER_SUITES_CONFIG = "scylladb.ssl.cipherSuites";
+  private static final String SSL_CIPHER_SUITES_DOC = "The cipher suites to enable. "
+          + "Defaults to none, resulting in a ``minimal quality of service`` according to JDK documentation.";
+
+  public static final String SSL_OPENSLL_KEYCERTCHAIN_CONFIG = "scylladb.ssl.openssl.keyCertChain";
+  private static final String SSL_OPENSLL_KEYCERTCHAIN_DOC = "Path to the SSL certificate file, when using OpenSSL.";
+
+  public static final String SSL_OPENSLL_PRIVATEKEY_CONFIG = "ssl.openssl.privateKey";
+  private static final String SSL_OPENSLL_PRIVATEKEY_DOC = "Path to the private key file, when using OpenSSL.";
+
   public static final String TTL_CONFIG = "scylladb.ttl";
   /*If TTL value is not specified then skip setting ttl value while making insert query*/
   public static final String TTL_DEFAULT = null;
@@ -263,6 +299,23 @@ public ScyllaDbSinkConnectorConfig(Map<?, ?> originals) {
           + "local to the machine on which the connector is running. It is a recommended config if "
           + "we have more than one DC.";
 
+  public static final String BEHAVIOR_ON_ERROR_CONFIG = "behavior.on.error";
+  public static final String BEHAVIOR_ON_ERROR_DEFAULT = BehaviorOnError.FAIL.name();
+  private static final String BEHAVIOR_ON_ERROR_DISPLAY = "Behavior On Error";
+  private static final String BEHAVIOR_ON_ERROR_DOC = "Error handling behavior setting. "
+          + "Must be configured to one of the following:\n"
+          + "``fail``\n"
+          + "The Connector throws ConnectException and stops processing records "
+          + "when an error occurs while processing or inserting records into ScyllDB.\n"
+          + "``ignore``\n"
+          + "Continues to process next set of records "
+          + "when error occurs while processing or inserting records into ScyllDB.\n"
+          + "``log``\n"
+          + "Logs the error via connect-reporter when an error occurs while processing or "
+          + "inserting records into ScyllDB and continues to process next set of records, "
+          + "available in the kafka topics.";
+
+  public static final String SCYLLADB_GROUP = "ScyllaDB";
   public static final String CONNECTION_GROUP = "Connection";
   public static final String SSL_GROUP = "SSL";
   public static final String KEYSPACE_GROUP = "Keyspace";
@@ -389,6 +442,56 @@ public static ConfigDef config() {
                     "SSL Truststore Password")
             //TODO .validator(Validators.blankOr(ValidFile.of()))
             //TODO .recommender(Recommenders.visibleIf(SSL_ENABLED_CONFIG, true))
+            .define(
+                    SSL_KEYSTORE_PATH_CONFIG,
+                    ConfigDef.Type.STRING,
+                    "",
+                    ConfigDef.Importance.MEDIUM,
+                    SSL_KEYSTORE_PATH_DOC,
+                    SSL_GROUP,
+                    2,
+                    ConfigDef.Width.SHORT,
+                    "SSL Keystore Path")
+            .define(
+                    SSL_KEYSTORE_PASSWORD_CONFIG,
+                    ConfigDef.Type.PASSWORD,
+                    "password123",
+                    ConfigDef.Importance.MEDIUM,
+                    SSL_KEYSTORE_PASSWORD_DOC,
+                    SSL_GROUP,
+                    3,
+                    ConfigDef.Width.SHORT,
+                    "SSL Keystore Password")
+            .define(
+                    SSL_CIPHER_SUITES_CONFIG,
+                    ConfigDef.Type.LIST,
+                    (Object) Collections.EMPTY_LIST,
+                    ConfigDef.Importance.HIGH,
+                    SSL_CIPHER_SUITES_DOC,
+                    SSL_GROUP,
+                    4,
+                    ConfigDef.Width.LONG,
+                    "The cipher suites to enable")
+            .define(
+                    SSL_OPENSLL_KEYCERTCHAIN_CONFIG,
+                    ConfigDef.Type.STRING,
+                    "",
+                    ConfigDef.Importance.HIGH,
+                    SSL_OPENSLL_KEYCERTCHAIN_DOC,
+                    SSL_GROUP,
+                    5,
+                    ConfigDef.Width.SHORT,
+                    "The path to the certificate chain file")
+            .define(
+                    SSL_OPENSLL_PRIVATEKEY_CONFIG,
+                    ConfigDef.Type.STRING,
+                    "",
+                    ConfigDef.Importance.HIGH,
+                    SSL_OPENSLL_PRIVATEKEY_DOC,
+                    SSL_GROUP,
+                    6,
+                    ConfigDef.Width.SHORT,
+                    "The path to the private key file")
             .define(
                     CONSISTENCY_LEVEL_CONFIG,
                     ConfigDef.Type.STRING,
@@ -524,7 +627,21 @@ public static ConfigDef config() {
                     WRITE_GROUP,
                     6,
                     ConfigDef.Width.SHORT,
-                    "Timestamp Threshold in MS");
+                    "Timestamp Threshold in MS")
+            .define(
+                    BEHAVIOR_ON_ERROR_CONFIG,
+                    ConfigDef.Type.STRING,
+                    BEHAVIOR_ON_ERROR_DEFAULT,
+                    ConfigDef.ValidString.in(BehaviorOnError.FAIL.name(),
+                            BehaviorOnError.LOG.name(), BehaviorOnError.IGNORE.name()),
+                    ConfigDef.Importance.MEDIUM,
+                    BEHAVIOR_ON_ERROR_DOC,
+                    SCYLLADB_GROUP,
+                    0,
+                    ConfigDef.Width.NONE,
+                    BEHAVIOR_ON_ERROR_DISPLAY
+                    //Recommenders.enumValues(BehaviorOnError.class)
+            );
   }
 
   private String tryMatchTopicName(final String name) {
@@ -535,6 +652,15 @@ private String tryMatchTopicName(final String name) {
     throw new IllegalArgumentException("The setting: " + name + " does not match topic.keyspace.table nor topic.codec regular expression pattern");
   }
 
+  /**
+   * Enums for behavior on error.
+   */
+  public enum BehaviorOnError {
+    IGNORE,
+    LOG,
+    FAIL
+  }
+
   public boolean isOffsetEnabledInScyllaDb() {
     return getBoolean(ENABLE_OFFSET_STORAGE_TABLE);
   }
diff --git a/src/main/java/io/connect/scylladb/ScyllaDbSinkTask.java b/src/main/java/io/connect/scylladb/ScyllaDbSinkTask.java

Original file line number	Diff line number	Diff line change
`@@ -124,7 +124,7 @@ DataType dataType(Schema schema) {`
`124`	`124`	`dataType = DataType.varchar();`
`125`	`125`	`break;`
`126`	`126`	`default:`
`127`		`- throw new UnsupportedOperationException(`
	`127`	`+ throw new DataException(`
`128`	`128`	`String.format("Unsupported type %s", schema.type())`
`129`	`129`	`);`
`130`	`130`	`}`