Client does not support authentication protocol requested by server; consider upgrading MySQL client

[nickevin]

经过调整 MySQL 版本，5.7 正常运行，8.* 有问题。

下面是测试数据：

1. zeppelin-0.9.0-preview2
2. flink-1.11.1-bin-scala_2.11

FLINK_HOME/lib/flink-sql-connector-mysql-cdc-1.1.0.jar
FLINK_HOME/lib/flink-sql-connector-postgres-cdc-1.1.0.jar

version: '3'
services:
  mysql:
    image: mysql:8.0.21
    ports:
      - 3306:3306

  postgres:
    image: postgres:9.6.19
    ports:
      - 5432:5432

%flink.ssql

DROP TABLE IF EXISTS products;

CREATE TABLE products (
  id INT,
  name STRING,
  description STRING
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'root',
  'password' = 'password',
  'database-name' = 'test',
  'table-name' = 'products'
);

DROP TABLE IF EXISTS orders;

CREATE TABLE orders (
  order_id INT,
  order_date TIMESTAMP(0),
  customer_name STRING,
  price DECIMAL(10, 5),
  product_id INT,
  order_status BOOLEAN
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'root',
  'password' = 'password',
  'database-name' = 'test',
  'table-name' = 'orders'
);

DROP TABLE IF EXISTS shipments;

CREATE TABLE shipments (
  shipment_id INT,
  order_id INT,
  origin STRING,
  destination STRING,
  is_arrived BOOLEAN
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = 'localhost',
  'port' = '5432',
  'username' = 'root',
  'password' = 'password',
  'database-name' = 'test',
  'schema-name' = 'public',
  'table-name' = 'shipments'
);

%flink.ssql(type=update)

SELECT o.* , p.name, p.description
FROM orders AS o
LEFT JOIN products AS p ON o.product_id = p.id

Caused by: org.apache.kafka.connect.errors.ConnectException: Failed to authenticate to the MySQL database at localhost:3306 with user 'root'
	at io.debezium.connector.mysql.BinlogReader.doStart(BinlogReader.java:441)
	at io.debezium.connector.mysql.AbstractReader.start(AbstractReader.java:116)
	at io.debezium.connector.mysql.ChainedReader.startNextReader(ChainedReader.java:206)
	at io.debezium.connector.mysql.ChainedReader.readerCompletedPolling(ChainedReader.java:158)
	at io.debezium.connector.mysql.AbstractReader.cleanupResources(AbstractReader.java:309)
	at io.debezium.connector.mysql.AbstractReader.poll(AbstractReader.java:288)
	at io.debezium.connector.mysql.ChainedReader.poll(ChainedReader.java:146)
	at io.debezium.connector.mysql.MySqlConnectorTask.doPoll(MySqlConnectorTask.java:443)
	at io.debezium.connector.common.BaseSourceTask.poll(BaseSourceTask.java:131)
	at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:779)
	at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:170)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: com.github.shyiko.mysql.binlog.network.AuthenticationException: Client does not support authentication protocol requested by server; consider upgrading MySQL client
	at com.github.shyiko.mysql.binlog.BinaryLogClient.authenticate(BinaryLogClient.java:728)
	at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:515)
	at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:860)
	... 1 more