src/main/java/org/embulk/input/MySQLInputPlugin.java in embulk-input-mysql-0.8.4 vs src/main/java/org/embulk/input/MySQLInputPlugin.java in embulk-input-mysql-0.8.5

@@ -2,15 +2,16 @@
 
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.Properties;
 import java.sql.Connection;
-import java.sql.Driver;
+import java.sql.DriverManager;
 import java.sql.SQLException;
 
+import com.google.common.base.Optional;
 import com.google.common.base.Throwables;
-import com.mysql.jdbc.TimeUtil;
+
 import org.embulk.config.Config;
 import org.embulk.config.ConfigDefault;
 import org.embulk.input.jdbc.AbstractJdbcInputPlugin;
 import org.embulk.input.jdbc.Ssl;
 import org.embulk.input.jdbc.JdbcInputConnection;
@@ -25,10 +26,14 @@
         extends AbstractJdbcInputPlugin
 {
     public interface MySQLPluginTask
             extends PluginTask
     {
+        @Config("driver_path")
+        @ConfigDefault("null")
+        public Optional<String> getDriverPath();
+
         @Config("host")
         public String getHost();
 
         @Config("port")
         @ConfigDefault("3306")
@@ -58,18 +63,20 @@
     @Override
     protected MySQLInputConnection newConnection(PluginTask task) throws SQLException
     {
         MySQLPluginTask t = (MySQLPluginTask) task;
 
+        loadDriver("com.mysql.jdbc.Driver", t.getDriverPath());
+
         String url = String.format("jdbc:mysql://%s:%d/%s",
                 t.getHost(), t.getPort(), t.getDatabase());
 
         Properties props = new Properties();
         props.setProperty("user", t.getUser());
         props.setProperty("password", t.getPassword());
 
-        // convert 0000-00-00 to NULL to avoid this exceptoin:
+        // convert 0000-00-00 to NULL to avoid this exception:
         //   java.sql.SQLException: Value '0000-00-00' can not be represented as java.sql.Date
         props.setProperty("zeroDateTimeBehavior", "convertToNull");
 
         props.setProperty("useCompression", "true");
 
@@ -108,19 +115,12 @@
         props.putAll(t.getOptions());
 
         // load timezone mappings
         loadTimeZoneMappings();
 
-        Driver driver;
+        Connection con = DriverManager.getConnection(url, props);
         try {
-            driver = new com.mysql.jdbc.Driver();  // new com.mysql.jdbc.Driver throws SQLException
-        } catch (SQLException ex) {
-            throw new RuntimeException(ex);
-        }
-
-        Connection con = driver.connect(url, props);
-        try {
             MySQLInputConnection c = new MySQLInputConnection(con);
             con = null;
             return c;
         } finally {
             if (con != null) {
@@ -147,22 +147,23 @@
         // that loaded com.mysql.jdbc.TimeUtil class rather than system class loader to read the
         // property file because the file should be in the same classpath with the class.
         // Here implements a workaround as as workaround.
         Field f = null;
         try {
-            f = TimeUtil.class.getDeclaredField("timeZoneMappings");
+            Class<?> timeUtilClass = Class.forName("com.mysql.jdbc.TimeUtil");
+            f = timeUtilClass.getDeclaredField("timeZoneMappings");
             f.setAccessible(true);
 
             Properties timeZoneMappings = (Properties) f.get(null);
             if (timeZoneMappings == null) {
                 timeZoneMappings = new Properties();
-                synchronized (TimeUtil.class) {
+                synchronized (timeUtilClass) {
                     timeZoneMappings.load(this.getClass().getResourceAsStream("/com/mysql/jdbc/TimeZoneMapping.properties"));
                 }
                 f.set(null, timeZoneMappings);
             }
         }
-        catch (IllegalAccessException | NoSuchFieldException | IOException e) {
+        catch (ClassNotFoundException | IllegalAccessException | NoSuchFieldException | IOException e) {
             throw Throwables.propagate(e);
         }
         finally {
             if (f != null) {
                 f.setAccessible(false);