🎉 Destination MySQL: Add jdbc_url_params support for optional JDBC parameters

airbyte-integrations/connectors/destination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java

@@ -6,6 +6,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap.Builder;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.db.Databases;
 import io.airbyte.db.jdbc.JdbcDatabase;
@@ -18,6 +19,8 @@
 import io.airbyte.protocol.models.AirbyteConnectionStatus.Status;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,6 +32,8 @@ public class MySQLDestination extends AbstractJdbcDestination implements Destina
 
   public static final String DRIVER_CLASS = "com.mysql.cj.jdbc.Driver";
 
+  public static final String JDBC_URL_PARAMS_KEY = "jdbc_url_params";
+
   public static Destination sshWrappedDestination() {
     return new SshWrappedDestination(new MySQLDestination(), HOST_KEY, PORT_KEY);
   }
@@ -78,13 +83,7 @@ protected JdbcDatabase getDatabase(final JsonNode config) {
 
   @Override
   public JsonNode toJdbcConfig(final JsonNode config) {
-    final List<String> additionalParameters = new ArrayList<>();
-
-    if (!config.has("ssl") || config.get("ssl").asBoolean()) {
-      additionalParameters.add("useSSL=true");
-      additionalParameters.add("requireSSL=true");
-      additionalParameters.add("verifyServerCertificate=false");
-    }
+    final List<String> additionalParameters = getAdditionalParameters(config);
 
     final StringBuilder jdbcUrl = new StringBuilder(String.format("jdbc:mysql://%s:%s/%s",
         config.get("host").asText(),
@@ -95,6 +94,7 @@ public JsonNode toJdbcConfig(final JsonNode config) {
     // and can't
     // remove zero date values.
     // since zero dates are placeholders, we convert them to null by default
+    //FIXME(girarda): do we also want to prevent people from overriding zeroDateTimeBehavior?
     jdbcUrl.append("?zeroDateTimeBehavior=convertToNull");
     if (!additionalParameters.isEmpty()) {
       jdbcUrl.append("&");
@@ -112,6 +112,61 @@ public JsonNode toJdbcConfig(final JsonNode config) {
     return Jsons.jsonNode(configBuilder.build());
   }
 
+  private List<String> getAdditionalParameters(final JsonNode config) {
+    final List<String> additionalParameters = new ArrayList<>();
+    addJDBCUrlParameters(config, additionalParameters);
+    if (useSSL(config)) {
+      addSSLJDBCParameters(additionalParameters);
+    }
+    return additionalParameters;
+  }
+
+  private void addJDBCUrlParameters(final JsonNode config, final List<String> additionalParameters) {
+    if (config.has(JDBC_URL_PARAMS_KEY)) {
+      final String additionalParams = config.get(JDBC_URL_PARAMS_KEY).asText();
+      if (additionalParams.isBlank()) {
+        return;
+      }
+      if (useSSL(config)) {
+        for (final String p : getSSLParameters().keySet()) {
+          if (containsParameterKey(additionalParams, p)) {
+            throw new RuntimeException("Cannot overwrite JDBC parameter " + p);
+          }
+        }
+      }
+      additionalParameters.add(additionalParams);
+    }
+  }
+
+  private boolean containsParameterKey(final String additionalParametersString, final String key) {
+    final String s = String.format("%s=", key);
+    return additionalParametersString.contains(s);
+  }
+
+  private void addSSLJDBCParameters(final List<String> additionalParameters) {
+    for (final Entry<String, String> sslParameter : getSSLParameters().entrySet()) {
+      final String param = formatParameter(sslParameter.getKey(), sslParameter.getValue());
+      additionalParameters.add(param);
+    }
+  }
+
+  private boolean useSSL(final JsonNode config) {
+    return !config.has("ssl") || config.get("ssl").asBoolean();
+  }
+
+  static Map<String, String> getSSLParameters() {
+    final Builder<String, String> builder = ImmutableMap.builder();
+    return builder
+        .put("useSSL", "true")
+        .put("requireSSL", "true")
+        .put("verifyServerCertificate", "false")
+        .build();
+  }
+
+  static String formatParameter(final String key, final String value) {
+    return String.format("%s=%s", key, value);
+  }
+
   public static void main(final String[] args) throws Exception {
     final Destination destination = MySQLDestination.sshWrappedDestination();
     LOGGER.info("starting destination: {}", MySQLDestination.class);

airbyte-integrations/connectors/destination-mysql/src/main/resources/spec.json

@@ -3,12 +3,20 @@
   "supportsIncremental": true,
   "supportsNormalization": true,
   "supportsDBT": true,
-  "supported_destination_sync_modes": ["overwrite", "append"],
+  "supported_destination_sync_modes": [
+    "overwrite",
+    "append"
+  ],
   "connectionSpecification": {
     "$schema": "http://json-schema.org/draft-07/schema#",
     "title": "MySQL Destination Spec",
     "type": "object",
-    "required": ["host", "port", "username", "database"],
+    "required": [
+      "host",
+      "port",
+      "username",
+      "database"
+    ],
     "additionalProperties": true,
     "properties": {
       "host": {
@@ -24,7 +32,9 @@
         "minimum": 0,
         "maximum": 65536,
         "default": 3306,
-        "examples": ["3306"],
+        "examples": [
+          "3306"
+        ],
         "order": 1
       },
       "database": {
@@ -52,6 +62,12 @@
         "type": "boolean",
         "default": true,
         "order": 5
+      },
+      "jdbc_url_params": {
+        "description": "Additional properties to pass to the JDBC URL string when connecting to the database formatted as 'key=value' pairs separated by the symbol '&'. (example: key1=value1&key2=value2&key3=value3).",
+        "title": "JDBC URL Params",
+        "type": "string",
+        "order": 6
       }
     }
   }

airbyte-integrations/connectors/destination-mysql/src/test/java/io/airbyte/integrations/destination/mysql/MySQLDestinationTest.java

@@ -0,0 +1,98 @@
+package io.airbyte.integrations.destination.mysql;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.spy;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import java.util.Map.Entry;
+import org.junit.jupiter.api.Test;
+
+public class MySQLDestinationTest {
+
+  private MySQLDestination getDestination() {
+    final MySQLDestination result = spy(MySQLDestination.class);
+    return result;
+  }
+
+  private JsonNode buildConfigNoJDBCParameters() {
+    final JsonNode config = Jsons.jsonNode(ImmutableMap.of(
+        "host", "localhost",
+        "port", 1337,
+        "username", "user",
+        "database", "db"
+    ));
+    return config;
+  }
+
+  private JsonNode buildConfigWithExtraJDBCParameters(final String extraParam) {
+    final JsonNode config = Jsons.jsonNode(ImmutableMap.of(
+        "host", "localhost",
+        "port", 1337,
+        "username", "user",
+        "database", "db",
+        "jdbc_url_params", extraParam
+    ));
+    return config;
+  }
+
+  private JsonNode buildConfigWithExtraJDBCParametersNoSSL(final String extraParam) {
+    final JsonNode config = Jsons.jsonNode(ImmutableMap.of(
+        "host", "localhost",
+        "port", 1337,
+        "username", "user",
+        "database", "db",
+        "ssl", false,
+        "jdbc_url_params", extraParam
+    ));
+    return config;
+  }
+
+  @Test
+  void testNoExtraParams() {
+    final JsonNode jdbcConfig = getDestination().toJdbcConfig(buildConfigNoJDBCParameters());
+    final String url = jdbcConfig.get("jdbc_url").asText();
+    assertEquals("jdbc:mysql://localhost:1337/db?zeroDateTimeBehavior=convertToNull&useSSL=true&requireSSL=true&verifyServerCertificate=false&", url);
+  }
+
+  @Test
+  void testEmptyExtraParams() {
+    final JsonNode jdbcConfig = getDestination().toJdbcConfig(buildConfigWithExtraJDBCParameters(""));
+    final String url = jdbcConfig.get("jdbc_url").asText();
+    assertEquals("jdbc:mysql://localhost:1337/db?zeroDateTimeBehavior=convertToNull&useSSL=true&requireSSL=true&verifyServerCertificate=false&", url);
+  }
+
+  @Test
+  void testExtraParams() {
+    final String extraParam = "key1=value1&key2=value2&key3=value3";
+    final JsonNode jdbcConfig = getDestination().toJdbcConfig(buildConfigWithExtraJDBCParameters(extraParam));
+    final String url = jdbcConfig.get("jdbc_url").asText();
+    assertEquals(
+        "jdbc:mysql://localhost:1337/db?zeroDateTimeBehavior=convertToNull&key1=value1&key2=value2&key3=value3&useSSL=true&requireSSL=true&verifyServerCertificate=false&",
+        url);
+  }
+
+  @Test
+  void testExtraParamsWithSSLParameter() {
+    for (final Entry<String, String> entry : MySQLDestination.getSSLParameters().entrySet()) {
+      final String extraParam = MySQLDestination.formatParameter(entry.getKey(), entry.getValue());
+      assertThrows(RuntimeException.class, () ->
+          getDestination().toJdbcConfig(buildConfigWithExtraJDBCParameters(extraParam))
+      );
+    }
+  }
+
+  @Test
+  void testExtraParamsWithSSLParameterNoSSL() {
+    for (final Entry<String, String> entry : MySQLDestination.getSSLParameters().entrySet()) {
+      final String extraParam = MySQLDestination.formatParameter(entry.getKey(), entry.getValue());
+      final JsonNode jdbcConfig = getDestination().toJdbcConfig(buildConfigWithExtraJDBCParametersNoSSL(extraParam));
+      final String url = jdbcConfig.get("jdbc_url").asText();
+      final String expected = String.format("jdbc:mysql://localhost:1337/db?zeroDateTimeBehavior=convertToNull&%s&", extraParam);
+      assertEquals(expected,
+          url);
+    }
+  }
+}