Add MySQL ssl option

embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/AbstractJdbcOutputPlugin.java

@@ -387,6 +387,7 @@ private PluginTask begin(final PluginTask task,
                 public void run() throws SQLException
                 {
                     JdbcOutputConnection con = newConnection(task, true, false);
+                    con.showDriverVersion();
                     try {
                         doBegin(con, task, schema, taskCount);
                     } finally {

embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/JdbcOutputConnection.java

@@ -9,6 +9,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Locale;
 
 import org.slf4j.Logger;
 
@@ -19,7 +20,7 @@
 public class JdbcOutputConnection
         implements AutoCloseable
 {
-    private final Logger logger = Exec.getLogger(JdbcOutputConnection.class);
+    protected final Logger logger = Exec.getLogger(JdbcOutputConnection.class);
     protected final Connection connection;
     protected final String schemaName;
     protected final DatabaseMetaData databaseMetaData;
@@ -593,4 +594,9 @@ protected SQLException safeRollback(Connection con, SQLException cause)
             return ex;
         }
     }
+    public void showDriverVersion() throws SQLException {
+        DatabaseMetaData meta = connection.getMetaData();
+        logger.info(String.format(Locale.ENGLISH,"Using JDBC Driver %s",meta.getDriverVersion()));
+    }
+
 }

embulk-output-redshift/src/main/java/org/embulk/output/redshift/Ssl.java → embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/Ssl.java

@@ -1,4 +1,4 @@
-package org.embulk.output.redshift;
+package org.embulk.output.jdbc;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonValue;

embulk-output-mysql/README.md

@@ -14,6 +14,7 @@ MySQL output plugin for Embulk loads records to MySQL.
 - **port**: database port number (integer, default: 3306)
 - **user**: database login user name (string, required)
 - **password**: database login password (string, default: "")
+- **ssl**: use SSL to connect to the database (string, default: `disable`. `enable` uses SSL without server-side validation and `verify` checks the certificate. For compatibility reasons, `true` behaves as `enable` and `false` behaves as `disable`.)
 - **database**: destination database name (string, required)
 - **temp_database**: database name for intermediate tables. by default, intermediate tables will be created in the database specified by `database`. (string, optional)
 - **table**: destination table name (string, required)

embulk-output-mysql/src/main/java/org/embulk/output/MySQLOutputPlugin.java

@@ -9,6 +9,7 @@
 import org.embulk.config.Config;
 import org.embulk.config.ConfigDefault;
 import org.embulk.output.jdbc.AbstractJdbcOutputPlugin;
+import org.embulk.output.jdbc.Ssl;
 import org.embulk.output.jdbc.BatchInsert;
 import org.embulk.output.jdbc.JdbcOutputConnection;
 import org.embulk.output.jdbc.MergeConfig;
@@ -44,6 +45,11 @@ public interface MySQLPluginTask
         @Config("temp_database")
         @ConfigDefault("null")
         public Optional<String> getTempDatabase();
+
+        @Config("ssl")
+        @ConfigDefault("\"disable\"") // backward compatibility
+        public Ssl getSsl();
+
     }
 
     @Override
@@ -80,21 +86,21 @@ protected MySQLOutputConnector getConnector(PluginTask task, boolean retryableMe
         // Socket options TCP_KEEPCNT, TCP_KEEPIDLE, and TCP_KEEPINTVL are not configurable.
         props.setProperty("tcpKeepAlive", "true");
 
-        // TODO
-        //switch t.getSssl() {
-        //when "disable":
-        //    break;
-        //when "enable":
-        //    props.setProperty("useSSL", "true");
-        //    props.setProperty("requireSSL", "false");
-        //    props.setProperty("verifyServerCertificate", "false");
-        //    break;
-        //when "verify":
-        //    props.setProperty("useSSL", "true");
-        //    props.setProperty("requireSSL", "true");
-        //    props.setProperty("verifyServerCertificate", "true");
-        //    break;
-        //}
+        switch (t.getSsl()) {
+            case DISABLE:
+                props.setProperty("useSSL", "false");
+                break;
+            case ENABLE:
+                props.setProperty("useSSL", "true");
+                props.setProperty("requireSSL", "true");
+                props.setProperty("verifyServerCertificate", "false");
+                break;
+            case VERIFY:
+                props.setProperty("useSSL", "true");
+                props.setProperty("requireSSL", "true");
+                props.setProperty("verifyServerCertificate", "true");
+                break;
+        }
 
         if (!retryableMetadataOperation) {
             // non-retryable batch operation uses longer timeout

embulk-output-mysql/src/main/java/org/embulk/output/mysql/MySQLOutputConnection.java

@@ -119,4 +119,20 @@ public void compareTimeZone() throws SQLException
         timeZoneComparison.compareTimeZone();
     }
 
+    //
+    //
+    // The MySQL Connector/J 5.1.35 introduce new option `Current MySQL Connect`.
+    // It has incompatibility behavior current version and 5.1.35.
+    //
+    // This method announces users about this change before the update driver version.
+    //
+    @Override
+    public void showDriverVersion() throws SQLException {
+        super.showDriverVersion();
+        logger.warn("This plugin will update MySQL Connector/J version in the near future release.");
+        logger.warn("It has some incompatibility changes.");
+        logger.warn("For example, the 5.1.35 introduced `noTimezoneConversionForDateType` and `cacheDefaultTimezone` options.");
+        logger.warn("Please read a document and make sure configuration carefully before updating the plugin.");
+    }
+
 }

embulk-output-redshift/src/main/java/org/embulk/output/RedshiftOutputPlugin.java

@@ -18,9 +18,9 @@
 import org.embulk.output.jdbc.JdbcOutputConnection;
 import org.embulk.output.jdbc.MergeConfig;
 import org.embulk.output.jdbc.TableIdentifier;
+import org.embulk.output.jdbc.Ssl;
 import org.embulk.output.redshift.RedshiftOutputConnector;
 import org.embulk.output.redshift.RedshiftCopyBatchInsert;
-import org.embulk.output.redshift.Ssl;
 
 public class RedshiftOutputPlugin
         extends AbstractJdbcOutputPlugin