Add transaction manager config

conf/database.properties

@@ -14,6 +14,9 @@ scalar.db.password=cassandra
 # Namespace prefix. The default is empty.
 #scalar.db.namespace_prefix=
 
+# The type of the transaction manager. "consensus-commit" or "jdbc" or "grpc" can be set. The default is "consensus-commit"
+#scalar.db.transaction_manager=consensus-commit
+
 # Default isolation level. Either SNAPSHOT or SERIALIZABLE can be specified. SNAPSHOT is used by default.
 #scalar.db.isolation_level=
 

core/src/main/java/com/scalar/db/config/DatabaseConfig.java

@@ -6,6 +6,7 @@
 import com.google.common.base.Strings;
 import com.scalar.db.api.DistributedStorage;
 import com.scalar.db.api.DistributedStorageAdmin;
+import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.api.Isolation;
 import com.scalar.db.storage.cassandra.Cassandra;
 import com.scalar.db.storage.cassandra.CassandraAdmin;
@@ -19,7 +20,10 @@
 import com.scalar.db.storage.multistorage.MultiStorageAdmin;
 import com.scalar.db.storage.rpc.GrpcAdmin;
 import com.scalar.db.storage.rpc.GrpcStorage;
+import com.scalar.db.transaction.consensuscommit.ConsensusCommitManager;
 import com.scalar.db.transaction.consensuscommit.SerializableStrategy;
+import com.scalar.db.transaction.jdbc.JdbcTransactionManager;
+import com.scalar.db.transaction.rpc.GrpcTransactionManager;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
@@ -40,6 +44,7 @@ public class DatabaseConfig {
   private Class<? extends DistributedStorage> storageClass;
   private Class<? extends DistributedStorageAdmin> adminClass;
   private Optional<String> namespacePrefix;
+  private Class<? extends DistributedTransactionManager> transactionManagerClass;
   private Isolation isolation = Isolation.SNAPSHOT;
   private SerializableStrategy strategy = SerializableStrategy.EXTRA_READ;
   public static final String PREFIX = "scalar.db.";
@@ -49,6 +54,7 @@ public class DatabaseConfig {
   public static final String PASSWORD = PREFIX + "password";
   public static final String STORAGE = PREFIX + "storage";
   public static final String NAMESPACE_PREFIX = PREFIX + "namespace_prefix";
+  public static final String TRANSACTION_MANAGER = PREFIX + "transaction_manager";
   public static final String ISOLATION_LEVEL = PREFIX + "isolation_level";
   public static final String CONSENSUS_COMMIT_PREFIX = PREFIX + "consensus_commit.";
   public static final String SERIALIZABLE_STRATEGY =
@@ -74,10 +80,9 @@ public Properties getProperties() {
   }
 
   protected void load() {
-    if (props.getProperty(STORAGE) == null) {
-      storageClass = Cassandra.class;
-      adminClass = CassandraAdmin.class;
-    } else {
+    storageClass = Cassandra.class;
+    adminClass = CassandraAdmin.class;
+    if (!Strings.isNullOrEmpty(props.getProperty(STORAGE))) {
       switch (props.getProperty(STORAGE).toLowerCase()) {
         case "cassandra":
           storageClass = Cassandra.class;
@@ -104,15 +109,16 @@ protected void load() {
           adminClass = GrpcAdmin.class;
           break;
         default:
-          throw new IllegalArgumentException(props.getProperty(STORAGE) + " isn't supported");
+          throw new IllegalArgumentException(
+              "storage '" + props.getProperty(STORAGE) + "' isn't supported");
       }
     }
 
     if (storageClass != MultiStorage.class) {
       checkNotNull(props.getProperty(CONTACT_POINTS));
 
       contactPoints = Arrays.asList(props.getProperty(CONTACT_POINTS).split(","));
-      if (props.getProperty(CONTACT_PORT) == null) {
+      if (Strings.isNullOrEmpty(props.getProperty(CONTACT_PORT))) {
         contactPort = 0;
       } else {
         contactPort = Integer.parseInt(props.getProperty(CONTACT_PORT));
@@ -132,6 +138,34 @@ protected void load() {
       namespacePrefix = Optional.empty();
     }
 
+    transactionManagerClass = ConsensusCommitManager.class;
+    if (!Strings.isNullOrEmpty(props.getProperty(TRANSACTION_MANAGER))) {
+      switch (props.getProperty(TRANSACTION_MANAGER).toLowerCase()) {
+        case "consensus-commit":
+          transactionManagerClass = ConsensusCommitManager.class;
+          break;
+        case "jdbc":
+          if (storageClass != JdbcDatabase.class) {
+            throw new IllegalArgumentException(
+                "'jdbc' transaction manager ("
+                    + TRANSACTION_MANAGER
+                    + ") is supported only for 'jdbc' storage ("
+                    + STORAGE
+                    + ")");
+          }
+          transactionManagerClass = JdbcTransactionManager.class;
+          break;
+        case "grpc":
+          transactionManagerClass = GrpcTransactionManager.class;
+          break;
+        default:
+          throw new IllegalArgumentException(
+              "transaction manager '"
+                  + props.getProperty(TRANSACTION_MANAGER)
+                  + "' isn't supported");
+      }
+    }
+
     if (!Strings.isNullOrEmpty(props.getProperty(ISOLATION_LEVEL))) {
       isolation = Isolation.valueOf(props.getProperty(ISOLATION_LEVEL).toUpperCase());
     }
@@ -170,6 +204,10 @@ public Optional<String> getNamespacePrefix() {
     return namespacePrefix;
   }
 
+  public Class<? extends DistributedTransactionManager> getTransactionManagerClass() {
+    return transactionManagerClass;
+  }
+
   public Isolation getIsolation() {
     return isolation;
   }

core/src/main/java/com/scalar/db/service/TransactionModule.java

@@ -7,11 +7,8 @@
 import com.scalar.db.api.DistributedTransactionManager;
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.dynamo.DynamoDatabaseConfig;
-import com.scalar.db.storage.jdbc.JdbcDatabase;
 import com.scalar.db.storage.jdbc.JdbcDatabaseConfig;
 import com.scalar.db.storage.multistorage.MultiStorageConfig;
-import com.scalar.db.transaction.consensuscommit.ConsensusCommitManager;
-import com.scalar.db.transaction.jdbc.JdbcTransactionManager;
 
 public class TransactionModule extends AbstractModule {
   private final DatabaseConfig config;
@@ -23,25 +20,9 @@ public TransactionModule(DatabaseConfig config) {
   @Override
   protected void configure() {
     bind(DistributedStorage.class).to(config.getStorageClass()).in(Singleton.class);
-
-    if (useJdbcTransaction()) {
-      bind(DistributedTransactionManager.class)
-          .to(JdbcTransactionManager.class)
-          .in(Singleton.class);
-      return;
-    }
-
-    bind(DistributedTransactionManager.class).to(ConsensusCommitManager.class).in(Singleton.class);
-  }
-
-  private boolean useJdbcTransaction() {
-    if (config.getStorageClass() == JdbcDatabase.class) {
-      JdbcDatabaseConfig jdbcDatabaseConfig = provideJdbcDatabaseConfig();
-      return jdbcDatabaseConfig
-          .getTransactionManagerType()
-          .equals(JdbcDatabaseConfig.TRANSACTION_MANAGER_TYPE_JDBC);
-    }
-    return false;
+    bind(DistributedTransactionManager.class)
+        .to(config.getTransactionManagerClass())
+        .in(Singleton.class);
   }
 
   @Singleton

core/src/main/java/com/scalar/db/storage/jdbc/JdbcDatabaseConfig.java

@@ -23,24 +23,17 @@ public class JdbcDatabaseConfig extends DatabaseConfig {
   public static final String PREPARED_STATEMENTS_POOL_MAX_OPEN =
       PREFIX + "prepared_statements_pool.max_open";
 
-  public static final String TRANSACTION_MANAGER_TYPE = PREFIX + "transaction_manager_type";
-  public static final String TRANSACTION_MANAGER_TYPE_CONSENSUS_COMMIT = "consensus-commit";
-  public static final String TRANSACTION_MANAGER_TYPE_JDBC = "jdbc";
-
   public static final int DEFAULT_CONNECTION_POOL_MIN_IDLE = 5;
   public static final int DEFAULT_CONNECTION_POOL_MAX_IDLE = 10;
   public static final int DEFAULT_CONNECTION_POOL_MAX_TOTAL = 25;
   public static final boolean DEFAULT_PREPARED_STATEMENTS_POOL_ENABLED = false;
   public static final int DEFAULT_PREPARED_STATEMENTS_POOL_MAX_OPEN = -1;
-  public static final String DEFAULT_TRANSACTION_MANAGER_TYPE =
-      TRANSACTION_MANAGER_TYPE_CONSENSUS_COMMIT;
 
   private int connectionPoolMinIdle;
   private int connectionPoolMaxIdle;
   private int connectionPoolMaxTotal;
   private boolean preparedStatementsPoolEnabled;
   private int preparedStatementsPoolMaxOpen;
-  private String transactionManagerType;
 
   public JdbcDatabaseConfig(File propertiesFile) throws IOException {
     super(propertiesFile);
@@ -70,19 +63,6 @@ protected void load() {
         getBoolean(PREPARED_STATEMENTS_POOL_ENABLED, DEFAULT_PREPARED_STATEMENTS_POOL_ENABLED);
     preparedStatementsPoolMaxOpen =
         getInt(PREPARED_STATEMENTS_POOL_MAX_OPEN, DEFAULT_PREPARED_STATEMENTS_POOL_MAX_OPEN);
-
-    transactionManagerType =
-        getProperties().getProperty(TRANSACTION_MANAGER_TYPE, DEFAULT_TRANSACTION_MANAGER_TYPE);
-    if (!transactionManagerType.equals(TRANSACTION_MANAGER_TYPE_CONSENSUS_COMMIT)
-        && !transactionManagerType.equals(TRANSACTION_MANAGER_TYPE_JDBC)) {
-      if (!transactionManagerType.isEmpty()) {
-        LOGGER.warn(
-            "the specified value of '{}' is invalid. using the default value: {}",
-            TRANSACTION_MANAGER_TYPE,
-            DEFAULT_TRANSACTION_MANAGER_TYPE);
-      }
-      transactionManagerType = DEFAULT_TRANSACTION_MANAGER_TYPE;
-    }
   }
 
   private int getInt(String name, int defaultValue) {
@@ -128,8 +108,4 @@ public boolean isPreparedStatementsPoolEnabled() {
   public int getPreparedStatementsPoolMaxOpen() {
     return preparedStatementsPoolMaxOpen;
   }
-
-  public String getTransactionManagerType() {
-    return transactionManagerType;
-  }
 }