Add TRANSACTION_ISOLATION_LEVEL config in PostgreSQL plugins

Author: psainics

database-commons/src/main/java/io/cdap/plugin/db/ConnectionConfig.java

@@ -45,6 +45,8 @@ public abstract class ConnectionConfig extends PluginConfig implements DatabaseC
   public static final String CONNECTION_ARGUMENTS = "connectionArguments";
   public static final String JDBC_PLUGIN_NAME = "jdbcPluginName";
   public static final String JDBC_PLUGIN_TYPE = "jdbc";
+  public static final String TRANSACTION_ISOLATION_LEVEL = "transactionIsolationLevel";
+
 
   @Name(JDBC_PLUGIN_NAME)
   @Description("Name of the JDBC driver to use. This is the value of the 'jdbcPluginName' key defined in the JSON " +

database-commons/src/main/java/io/cdap/plugin/db/config/AbstractDBSpecificSourceConfig.java

@@ -28,6 +28,8 @@
 import io.cdap.plugin.db.TransactionIsolationLevel;
 import io.cdap.plugin.db.connector.AbstractDBConnectorConfig;
 import io.cdap.plugin.db.source.AbstractDBSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -49,6 +51,7 @@ public abstract class AbstractDBSpecificSourceConfig extends PluginConfig implem
   public static final String DATABASE = "database";
   public static final String FETCH_SIZE = "fetchSize";
   public static final String DEFAULT_FETCH_SIZE = "1000";
+  public static final Logger LOG = LoggerFactory.getLogger(AbstractDBSpecificSourceConfig.class);
 
   @Name(Constants.Reference.REFERENCE_NAME)
   @Description(Constants.Reference.REFERENCE_NAME_DESCRIPTION)

database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBConnectorConfig.java

@@ -25,6 +25,8 @@
 import io.cdap.plugin.common.KeyValueListParser;
 import io.cdap.plugin.common.db.DBConnectorProperties;
 import io.cdap.plugin.db.ConnectionConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -37,6 +39,8 @@
  */
 public abstract class AbstractDBConnectorConfig extends PluginConfig implements DBConnectorProperties {
 
+  private static final Logger LOG = LoggerFactory.getLogger(AbstractDBConnectorConfig.class);
+
   @Name(ConnectionConfig.JDBC_PLUGIN_NAME)
   @Description("Name of the JDBC driver to use. This is the value of the 'jdbcPluginName' key defined in the JSON " +
     "file for the JDBC plugin.")

database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnector.java

@@ -37,6 +37,8 @@
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
 import org.apache.hadoop.mapreduce.lib.db.DBWritable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.sql.Connection;
@@ -56,6 +58,7 @@ public abstract class AbstractDBSpecificConnector<T extends DBWritable> extends
   implements BatchConnector<LongWritable, T> {
 
   private final AbstractDBConnectorConfig config;
+  private static final Logger LOG = LoggerFactory.getLogger(AbstractDBSpecificConnector.class);
 
   protected AbstractDBSpecificConnector(AbstractDBConnectorConfig config) {
     super(config);
@@ -99,6 +102,7 @@ public InputFormatProvider getInputFormatProvider(ConnectorContext context, Samp
       tableQuery, null, false);
     connectionConfigAccessor.setConnectionArguments(Maps.fromProperties(config.getConnectionArgumentsProperties()));
     connectionConfigAccessor.getConfiguration().setInt(MRJobConfig.NUM_MAPS, 1);
+    LOG.debug("Moving inside AbstractDBConnectorConfig");
     Map<String, String> additionalArguments = config.getAdditionalArguments();
     for (Map.Entry<String, String> argument : additionalArguments.entrySet()) {
       connectionConfigAccessor.getConfiguration().set(argument.getKey(), argument.getValue());

database-commons/src/main/java/io/cdap/plugin/db/connector/AbstractDBSpecificConnectorConfig.java

@@ -20,8 +20,12 @@
 import io.cdap.cdap.api.annotation.Macro;
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.plugin.db.ConnectionConfig;
+import io.cdap.plugin.db.TransactionIsolationLevel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import javax.annotation.Nullable;
 
@@ -30,9 +34,11 @@
  */
 public abstract class AbstractDBSpecificConnectorConfig extends AbstractDBConnectorConfig {
 
+  private static final Logger LOG = LoggerFactory.getLogger(AbstractDBSpecificConnectorConfig.class);
   @Name(ConnectionConfig.HOST)
   @Description("Database host")
   @Macro
+
   @Nullable
   protected String host;
 
@@ -42,6 +48,12 @@ public abstract class AbstractDBSpecificConnectorConfig extends AbstractDBConnec
   @Nullable
   protected Integer port;
 
+  @Name(ConnectionConfig.TRANSACTION_ISOLATION_LEVEL)
+  @Description("The transaction isolation level for the database session.")
+  @Macro
+  @Nullable
+  protected String transactionIsolationLevel;
+
   public String getHost() {
     return host;
   }
@@ -55,4 +67,22 @@ public int getPort() {
   public boolean canConnect() {
     return super.canConnect() && !containsMacro(ConnectionConfig.HOST) && !containsMacro(ConnectionConfig.PORT);
   }
+
+  @Override
+  public Map<String, String> getAdditionalArguments() {
+    Map<String, String> additonalArguments = new HashMap<>();
+    LOG.debug("inside get AdditionalArguemnts of AbstractDBSpecificConnectorConfig");
+    if (getTransactionIsolationLevel() != null) {
+      additonalArguments.put(TransactionIsolationLevel.CONF_KEY, getTransactionIsolationLevel());
+    }
+    return additonalArguments;
+  }
+
+  public String getTransactionIsolationLevel() {
+    if (transactionIsolationLevel == null) {
+      return null;
+    }
+    return TransactionIsolationLevel.Level.valueOf(transactionIsolationLevel).name();
+  }
 }
+

mysql-plugin/docs/MySQL-connector.md

@@ -22,6 +22,12 @@ authentication. Optional for databases that do not require authentication.
 
 **Password:** Password to use to connect to the specified database.
 
+**Transaction Isolation Level** The transaction isolation level of the databse connection
+- TRANSACTION_READ_COMMITTED: No dirty reads. Non-repeatable reads and phantom reads are possible.
+- TRANSACTION_SERIALIZABLE: No dirty reads. Non-repeatable and phantom reads are prevented.
+- TRANSACTION_REPEATABLE_READ: No dirty reads. Prevents non-repeatable reads, but phantom reads are still possible.
+- TRANSACTION_READ_UNCOMMITTED: Allows dirty reads (reading uncommitted changes from other transactions). Non-repeatable reads and phantom reads are possible.
+
 **Connection Arguments:** A list of arbitrary string tag/value pairs as connection arguments. These arguments
 will be passed to the JDBC driver, as connection arguments, for JDBC drivers that may need additional configurations.
 This is a semicolon-separated list of key-value pairs, where each pair is separated by a equals '=' and specifies

mysql-plugin/docs/Mysql-batchsink.md

@@ -39,6 +39,12 @@ You also can use the macro function ${conn(connection-name)}.
 
 **Password:** Password to use to connect to the specified database.
 
+**Transaction Isolation Level** The transaction isolation level of the databse connection
+- TRANSACTION_READ_COMMITTED: No dirty reads. Non-repeatable reads and phantom reads are possible.
+- TRANSACTION_SERIALIZABLE: No dirty reads. Non-repeatable and phantom reads are prevented.
+- TRANSACTION_REPEATABLE_READ: No dirty reads. Prevents non-repeatable reads, but phantom reads are still possible.
+- TRANSACTION_READ_UNCOMMITTED: Allows dirty reads (reading uncommitted changes from other transactions). Non-repeatable reads and phantom reads are possible.
+
 **Connection Arguments:** A list of arbitrary string key/value pairs as connection arguments. These arguments
 will be passed to the JDBC driver as connection arguments for JDBC drivers that may need additional configurations.
 

mysql-plugin/docs/Mysql-batchsource.md

@@ -49,6 +49,12 @@ For example, 'SELECT MIN(id),MAX(id) FROM table'. Not required if numSplits is s
 
 **Password:** Password to use to connect to the specified database.
 
+**Transaction Isolation Level** The transaction isolation level of the databse connection
+- TRANSACTION_READ_COMMITTED: No dirty reads. Non-repeatable reads and phantom reads are possible.
+- TRANSACTION_SERIALIZABLE: No dirty reads. Non-repeatable and phantom reads are prevented.
+- TRANSACTION_REPEATABLE_READ: No dirty reads. Prevents non-repeatable reads, but phantom reads are still possible.
+- TRANSACTION_READ_UNCOMMITTED: Allows dirty reads (reading uncommitted changes from other transactions). Non-repeatable reads and phantom reads are possible.
+
 **Connection Arguments:** A list of arbitrary string key/value pairs as connection arguments. These arguments
 will be passed to the JDBC driver as connection arguments for JDBC drivers that may need additional configurations.
 

mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSource.java

@@ -197,9 +197,15 @@ public MysqlConnectorConfig getConnection() {
       return connection;
     }
 
+    @Override
+    public String getTransactionIsolationLevel() {
+      return connection.getTransactionIsolationLevel();
+    }
+
     @Override
     public void validate(FailureCollector collector) {
       ConfigUtil.validateConnection(this, useConnection, connection, collector);
+      connection.getAdditionalArguments();
       super.validate(collector);
     }
 

mysql-plugin/widgets/MySQL-connector.json

@@ -30,6 +30,20 @@
           "widget-attributes": {
             "default": "3306"
           }
+        },
+        {
+          "widget-type": "select",
+          "label": "Transaction Isolation Level",
+          "name": "transactionIsolationLevel",
+          "widget-attributes": {
+            "values": [
+              "TRANSACTION_READ_UNCOMMITTED",
+              "TRANSACTION_READ_COMMITTED",
+              "TRANSACTION_REPEATABLE_READ",
+              "TRANSACTION_SERIALIZABLE"
+            ],
+            "default": "TRANSACTION_SERIALIZABLE"
+          }
         }
       ]
     },