[mysql] add custom jdbc properties to debezium mysql connection (#674)

[ruanhang1993]

This PR fix the bug that the mysql jdbc connection does not contain the jdbc properties set by users.
The problem is that the debezium mysql connection uses the fixed url pattern.

Changes:

1. add new mysql connection class(MySqlConnectionWithJdbcProperties) and new inner mysql connection configuration class
2. change the usage in mysql cdc from MySqlConnection to MySqlConnectionWithJdbcProperties

[leonardBang]

Thanks @ruanhang1993 for the contribution, I left some comments

[leonardBang]

Suggested change

	DebeziumUtils.createMySqlConnection(
	from(dbzProperties), sourceConfig.getJdbcProperties());
	DebeziumUtils.createMySqlConnection(sourceConfig);

[ruanhang1993]

This is used by the legacy source whose sourceConfig is null. I will revert this change to use new Properties().

[foxmorg]

Hi there! Trying to update com.ververica:flink-connector-mysql-cdc from 2.2.1 to 2.3.0 version and getting such Exception:

Caused by: java.lang.NoSuchMethodError: 'void io.debezium.connector.mysql.MySqlConnection$MySqlConnectionConfiguration.<init>(io.debezium.config.Configuration, java.util.Properties)'

    at com.ververica.cdc.connectors.mysql.debezium.DebeziumUtils.createMySqlConnection(DebeziumUtils.java:78)

    at com.ververica.cdc.connectors.mysql.debezium.DebeziumUtils.createMySqlConnection(DebeziumUtils.java:71)

    at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.checkSplitOrStartNext(MySqlSplitReader.java:138)

    at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.fetch(MySqlSplitReader.java:71)

    at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)

    at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:142)

    at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:105)

    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)

    at java.base/java.util.concurrent.FutureTask.run(Unknown Source)

    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)

    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)

    ... 1 more

As I understand, the reason because I have two dependencies as result:
debezium-connector-mysql-1.6.4.Final.jar
flink-connector-mysql-cdc-2.3.0.jar

with the same class io.debezium.connector.mysql.MySqlConnection$MySqlConnectionConfiguration but first one with only one constructor. So the question is, why was this class renamed in this commit 5415bcf and how may I fix this issue?

[foxmorg]

With 2.4.0 the same issue