Merge branch 'develop' into PLUGIN-72

# Conflicts:
#	README.md
#	docker-compose/db-plugins-env/docker-compose.yml
#	pom.xml

Author: akhalymon

README.md

@@ -13,10 +13,12 @@ mvn clean test \
 ```
 Notice that you must change properties for Aurora MySQL and Aurora Postgresql to real before running tests.
 ## Setup Local Environment
-MySQL, Postgresql, MSSQL, DB2, SAP HANA are using prebuild images.
+MySQL, Postgresql, MSSQL, DB2, MemSQL, SAP HANA are using prebuild images.
 
 Oracle DB image should be build separately.
 
+MemSQL image should be configure after start.
+
 Note that you should login into docker account to pull SAP HANA image. 
 Account can be created [here](https://hub.docker.com/signup)
 Also, please note SAP HANA is sensitive to some CPU instructions.
@@ -30,12 +32,37 @@ Netezza requires VMware Player for running Netezza emulator.
 * [Install Docker Compose](https://docs.docker.com/compose/install/)
 * Build local docker images
   * [Build Oracle DB docker image version 12.1.0.2-ee](https://github.com/oracle/docker-images/tree/master/OracleDatabase/SingleInstance)
-* Start docker environment by running commands:
+* Enter the folder with docker-compose file:
+```bash
+cd docker-compose/db-plugins-env/
+```
+* Export your license key for MemSQL to environment variable:
+```bash
+export MEMSQL_LICENSE_KEY=YOUR_LICENSE_KEY
+```
+* Initialize Memsql container:
+```bash
+docker-compose up memsql
+```
+* Start SAP HANA password service
 ```bash
 bash saphana-password-server.sh &
+```
+* Start docker environment by running commands:
+```bash
 cd docker-compose/db-plugins-env/
 docker-compose up -d
 ```
+* Connect to MemSQL Studio at [http://localhost:8888](http://localhost:8888) 
+The default Username is root and Password should be left blank.
+* Create `mydb` database in MemSQL Studio
+```sql
+create database mydb
+```
+* Set password for `root` user in MemSQL Studio
+```sql
+grant all on *.* to 'root'@'%' identified by 'root' with grant option;
+```
 * [Install and start Netezza emulator](http://dwgeek.com/install-vmware-player-netezza-emulator.html/)
 * Create database `mydb` in Netezza emulator
 
@@ -77,6 +104,12 @@ docker-compose up -d
 * **netezza.database** - Server namespace for test databases. Default: mydb.
 * **netezza.username** - Server username. Default: admin.
 * **netezza.password** - Server password. Default: password.
+#### MemSQL
+* **memsql.host** - Server host. Default: localhost.
+* **memsql.port** - Server port. Default: 3308.
+* **memsql.database** - Server namespace for test databases. Default: mydb.
+* **memsql.username** - Server username. Default: root.
+* **memsql.password** - Server password. Default: root.
 #### Aurora MySQL
 * **auroraMysql.clusterEndpoint** - Cluster endpoint.
 * **auroraMysql.port** - Server port.

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

@@ -17,6 +17,10 @@
 package io.cdap.plugin.db.batch;
 
 
+import io.cdap.cdap.etl.api.FailureCollector;
+import io.cdap.plugin.db.DBConfig;
+import io.cdap.plugin.db.batch.sink.AbstractDBSink;
+
 import java.sql.Connection;
 import java.util.Arrays;
 import javax.annotation.Nullable;
@@ -69,17 +73,20 @@ public static int getLevel(@Nullable String level) {
    * Validates that the given level is either null or one of the possible transaction isolation levels.
    *
    * @param level the level to check
+   * @param collector failure collector
    */
-  public static void validate(@Nullable String level) {
+  public static void validate(@Nullable String level, FailureCollector collector) {
     if (level == null) {
       return;
     }
     try {
       Level.valueOf(level.toUpperCase());
     } catch (IllegalArgumentException e) {
-      throw new IllegalArgumentException(String.format(
-        "Transaction isolation level must be one of the following values: %s, but got: %s.",
-        Arrays.toString(Level.values()), level));
+      collector.addFailure(
+        "Unsupported Transaction Isolation Level.",
+        String.format("Transaction Isolation Level must be one of the following values: %s",
+                      Arrays.toString(Level.values())))
+        .withConfigProperty(AbstractDBSink.DBSinkConfig.TRANSACTION_ISOLATION_LEVEL);
     }
   }
 }

database-commons/src/main/java/io/cdap/plugin/db/batch/action/AbstractQueryAction.java

@@ -16,6 +16,7 @@
 
 package io.cdap.plugin.db.batch.action;
 
+import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchActionContext;
 import io.cdap.cdap.etl.api.batch.PostAction;
@@ -40,7 +41,9 @@ public AbstractQueryAction(QueryActionConfig config, Boolean enableAutoCommit) {
 
   @Override
   public void run(BatchActionContext batchContext) throws Exception {
-    config.validate();
+    FailureCollector collector = batchContext.getFailureCollector();
+    config.validate(collector);
+    collector.getOrThrowException();
 
     if (!config.shouldRun(batchContext)) {
       return;
@@ -53,7 +56,8 @@ public void run(BatchActionContext batchContext) throws Exception {
 
   @Override
   public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException {
-    config.validate();
+    FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector();
+    config.validate(collector);
     DBUtils.validateJDBCPluginPipeline(pipelineConfigurer, config, JDBC_PLUGIN_ID);
   }
 }

database-commons/src/main/java/io/cdap/plugin/db/batch/action/QueryActionConfig.java

@@ -18,6 +18,7 @@
 
 import io.cdap.cdap.api.annotation.Description;
 import io.cdap.cdap.api.annotation.Macro;
+import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.batch.BatchActionContext;
 import io.cdap.plugin.common.batch.action.Condition;
 import io.cdap.plugin.common.batch.action.ConditionConfig;
@@ -43,10 +44,14 @@ public QueryActionConfig() {
     runCondition = Condition.SUCCESS.name();
   }
 
-  public void validate() {
+  public void validate(FailureCollector collector) {
     // have to delegate instead of inherit, since we can't extend both ConditionConfig and ConnectionConfig.
-    if (!containsMacro("runCondition")) {
-      new ConditionConfig(runCondition).validate();
+    if (!containsMacro(RUN_CONDITION)) {
+      try {
+        new ConditionConfig(runCondition).validate();
+      } catch (IllegalArgumentException e) {
+        collector.addFailure(e.getMessage(), null).withConfigProperty(RUN_CONDITION);
+      }
     }
   }
 

database-commons/src/main/java/io/cdap/plugin/db/batch/sink/AbstractDBSink.java

@@ -28,7 +28,9 @@
 import io.cdap.cdap.api.dataset.lib.KeyValue;
 import io.cdap.cdap.api.plugin.PluginConfig;
 import io.cdap.cdap.etl.api.Emitter;
+import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
+import io.cdap.cdap.etl.api.StageConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
 import io.cdap.cdap.etl.api.batch.BatchSinkContext;
 import io.cdap.cdap.etl.api.validation.InvalidStageException;
@@ -64,6 +66,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 /**
@@ -91,12 +94,16 @@ private String getJDBCPluginId() {
   @Override
   public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
     super.configurePipeline(pipelineConfigurer);
+    StageConfigurer configurer = pipelineConfigurer.getStageConfigurer();
     DBUtils.validateJDBCPluginPipeline(pipelineConfigurer, dbSinkConfig, getJDBCPluginId());
-    Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema();
+    Schema inputSchema = configurer.getInputSchema();
     if (Objects.nonNull(inputSchema)) {
       Class<? extends Driver> driverClass = DBUtils.getDriverClass(
         pipelineConfigurer, dbSinkConfig, ConnectionConfig.JDBC_PLUGIN_TYPE);
-      validateSchema(driverClass, dbSinkConfig.tableName, inputSchema);
+      if (driverClass != null) {
+        FailureCollector collector = configurer.getFailureCollector();
+        validateSchema(collector, driverClass, dbSinkConfig.tableName, inputSchema);
+      }
     }
   }
 
@@ -117,7 +124,9 @@ public void prepareRun(BatchSinkContext context) {
     // make sure that the destination table exists and column types are correct
     try {
       if (Objects.nonNull(outputSchema)) {
-        validateSchema(driverClass, dbSinkConfig.tableName, outputSchema);
+        FailureCollector collector = context.getFailureCollector();
+        validateSchema(collector, driverClass, dbSinkConfig.tableName, outputSchema);
+        collector.getOrThrowException();
       } else {
         outputSchema = inferSchema(driverClass);
       }
@@ -269,15 +278,17 @@ static List<ColumnType> getMatchedColumnTypeList(ResultSetMetaData resultSetMeta
     return columnTypes;
   }
 
-  private void validateSchema(Class<? extends Driver> jdbcDriverClass, String tableName, Schema inputSchema) {
+  private void validateSchema(FailureCollector collector, Class<? extends Driver> jdbcDriverClass, String tableName,
+                              Schema inputSchema) {
     String connectionString = dbSinkConfig.getConnectionString();
 
     try {
       DBUtils.ensureJDBCDriverIsAvailable(jdbcDriverClass, connectionString, dbSinkConfig.jdbcPluginName);
     } catch (IllegalAccessException | InstantiationException | SQLException e) {
-      throw new InvalidStageException(String.format("Unable to load or register JDBC driver '%s' while checking for " +
-                                                      "the existence of the database table '%s'.",
-                                                    jdbcDriverClass, tableName), e);
+      collector.addFailure(String.format("Unable to load or register JDBC driver '%s' while checking for " +
+                                           "the existence of the database table '%s'.",
+                                         jdbcDriverClass, tableName), null).withStacktrace(e.getStackTrace());
+      throw collector.getOrThrowException();
     }
 
     Properties connectionProperties = new Properties();
@@ -286,23 +297,26 @@ private void validateSchema(Class<? extends Driver> jdbcDriverClass, String tabl
       executeInitQueries(connection, dbSinkConfig.getInitQueries());
       try (ResultSet tables = connection.getMetaData().getTables(null, null, tableName, null)) {
         if (!tables.next()) {
-          throw new InvalidStageException("Table " + tableName + " does not exist. " +
-                                            "Please check that the 'tableName' property has been set correctly, " +
-                                            "and that the connection string  " + connectionString +
-                                            "points to a valid database.");
+          collector.addFailure(
+            String.format("Table '%s' does not exist.", tableName),
+            String.format("Ensure table '%s' is set correctly and that the connection string '%s' points " +
+                            "to a valid database.", tableName, connectionString))
+            .withConfigProperty(DBSinkConfig.TABLE_NAME);
+          return;
         }
       }
 
       try (PreparedStatement pStmt = connection.prepareStatement("SELECT * FROM " + dbSinkConfig.getEscapedTableName()
                                                                    + " WHERE 1 = 0");
            ResultSet rs = pStmt.executeQuery()) {
-        getFieldsValidator().validateFields(inputSchema, rs);
+        getFieldsValidator().validateFields(inputSchema, rs, collector);
       }
-
     } catch (SQLException e) {
       LOG.error("Exception while trying to validate schema of database table {} for connection {}.",
                 tableName, connectionString, e);
-      throw Throwables.propagate(e);
+      collector.addFailure(
+        String.format("Exception while trying to validate schema of database table '%s' for connection '%s'.",
+                      tableName, connectionString), null).withStacktrace(e.getStackTrace());
     }
   }
 

database-commons/src/main/java/io/cdap/plugin/db/batch/sink/CommonFieldsValidator.java

@@ -18,15 +18,14 @@
 
 import com.google.common.base.Preconditions;
 import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.etl.api.FailureCollector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Types;
-import java.util.HashSet;
-import java.util.Set;
 
 /**
  * Common fields validator.
@@ -35,37 +34,29 @@ public class CommonFieldsValidator implements FieldsValidator {
   protected static final Logger LOG = LoggerFactory.getLogger(CommonFieldsValidator.class);
 
   @Override
-  public void validateFields(Schema inputSchema, ResultSet resultSet) throws SQLException {
+  public void validateFields(Schema inputSchema, ResultSet resultSet, FailureCollector collector) throws SQLException {
     ResultSetMetaData rsMetaData = resultSet.getMetaData();
-
     Preconditions.checkNotNull(inputSchema.getFields());
-    Set<String> invalidFields = new HashSet<>();
     for (Schema.Field field : inputSchema.getFields()) {
       int columnIndex = resultSet.findColumn(field.getName());
       boolean isColumnNullable = (ResultSetMetaData.columnNullable == rsMetaData.isNullable(columnIndex));
       boolean isNotNullAssignable = !isColumnNullable && field.getSchema().isNullable();
+      String name = field.getName();
       if (isNotNullAssignable) {
-        LOG.error("Field '{}' was given as nullable but the database column is not nullable", field.getName());
-        invalidFields.add(field.getName());
+        collector.addFailure(
+          String.format("Field '%s' was given as nullable but database column is not nullable.", name),
+          "Ensure that the field is not nullable.").withInputSchemaField(name);
       }
 
       if (!isFieldCompatible(field, rsMetaData, columnIndex)) {
         String sqlTypeName = rsMetaData.getColumnTypeName(columnIndex);
         Schema fieldSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema();
-        Schema.Type fieldType = fieldSchema.getType();
-        Schema.LogicalType fieldLogicalType = fieldSchema.getLogicalType();
-        LOG.error("Field '{}' was given as type '{}' but the database column is actually of type '{}'.",
-                  field.getName(),
-                  fieldLogicalType != null ? fieldLogicalType.getToken() : fieldType,
-                  sqlTypeName
-        );
-        invalidFields.add(field.getName());
+        collector.addFailure(
+          String.format("Field '%s' was given as type '%s' but the database column is actually of type '%s'.",
+                        name, fieldSchema.getDisplayName(), sqlTypeName),
+          "Ensure that the field is not nullable.").withInputSchemaField(name);
       }
     }
-
-    Preconditions.checkArgument(invalidFields.isEmpty(),
-                                "Couldn't find matching database column(s) for input field(s) '%s'.",
-                                String.join(",", invalidFields));
   }
 
   /**

database-commons/src/main/java/io/cdap/plugin/db/batch/sink/FieldsValidator.java

@@ -17,10 +17,12 @@
 package io.cdap.plugin.db.batch.sink;
 
 import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.etl.api.FailureCollector;
 
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
+import java.util.Set;
 
 /**
  * Main Interface to validate db fields.
@@ -33,7 +35,7 @@ public interface FieldsValidator {
    * @param inputSchema input schema.
    * @param resultSet   resultSet with database fields.
    */
-  void validateFields(Schema inputSchema, ResultSet resultSet) throws SQLException;
+  void validateFields(Schema inputSchema, ResultSet resultSet, FailureCollector collector) throws SQLException;
 
   /**
    * Checks if field is compatible to be written into database column of the given sql index.