Accept com.mysql.cj.util.TimeUtil in addition to com.mysql.jdbc.TimeUtil

embulk-input-mysql/src/main/java/org/embulk/input/mysql/MySQLInputPlugin.java

@@ -8,6 +8,8 @@
 import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicReference;
@@ -159,20 +161,20 @@ private void loadTimeZoneMappings()
         // Here implements a workaround as as workaround.
         Field f = null;
         try {
-            Class<?> timeUtilClass = Class.forName("com.mysql.jdbc.TimeUtil");
+            Class<?> timeUtilClass = getTimeUtilClass();
             f = timeUtilClass.getDeclaredField("timeZoneMappings");
             f.setAccessible(true);
 
             Properties timeZoneMappings = (Properties) f.get(null);
             if (timeZoneMappings == null) {
                 timeZoneMappings = new Properties();
                 synchronized (timeUtilClass) {
-                    timeZoneMappings.load(this.getClass().getResourceAsStream("/com/mysql/jdbc/TimeZoneMapping.properties"));
+                    loadTimezoneMappings(timeZoneMappings);
                 }
                 f.set(null, timeZoneMappings);
             }
         }
-        catch (ClassNotFoundException | IllegalAccessException | NoSuchFieldException | IOException e) {
+        catch (IllegalAccessException | NoSuchFieldException e) {
             throw new RuntimeException(e);
         }
         finally {
@@ -252,7 +254,40 @@ private static Class<? extends java.sql.Driver> loadJdbcDriverClassForName(final
     {
         return (Class<? extends java.sql.Driver>) Class.forName(className);
     }
+    private Class<?> getTimeUtilClass() {
+        Class<?> timeUtilClass;
+        List<String> timeUtilClassNames = Arrays.asList(
+                "com.mysql.jdbc.TimeUtil",   // Connector/J v5.X
+                "com.mysql.cj.util.TimeUtil" // Connector/J v8.X
+        );
+
+        for (String name : timeUtilClassNames) {
+            try {
+                timeUtilClass = Class.forName(name);
+                return timeUtilClass;
+            } catch (ClassNotFoundException e) {
+                // do nothing.
+            }
+        }
+        throw new ConfigException("Can't find TimeUtil Class");
+    }
+
+    private void loadTimezoneMappings(Properties timeZoneMappings) {
+        List<String> timeZoneResources = Arrays.asList(
+                "/com/mysql/jdbc/TimeZoneMapping.properties",   // Connector/J v5.X
+                "/com/mysql/cj/util/TimeZoneMapping.properties" // Connector/J v8.X
+        );
 
+        for (String resource : timeZoneResources) {
+            try {
+                timeZoneMappings.load(this.getClass().getResourceAsStream(resource));
+                return;
+            } catch (IOException | NullPointerException e) {
+                // do nothing.
+            }
+        }
+        throw new ConfigException("Can't find Timezone mapping property file.");
+    }
     private static final AtomicReference<Class<? extends java.sql.Driver>> mysqlJdbcDriver = new AtomicReference<>();
 
     private static final Logger logger = LoggerFactory.getLogger(MySQLInputPlugin.class);