Use new config options hosts, user, password instead of uri

Author: Jiabao-Sun

docs/content/connectors/mongodb-cdc.md

@@ -81,7 +81,9 @@ CREATE TABLE products (
   PRIMARY KEY(_id) NOT ENFORCED
 ) WITH (
   'connector' = 'mongodb-cdc',
-  'uri' = 'mongodb://flinkuser:flinkpw@localhost:27017',
+  'hosts' = 'localhost:27017,localhost:27018,localhost:27019',
+  'user' = 'flinkuser',
+  'password' = 'flinkpw',
   'database' = 'inventory',
   'collection' = 'products'
 );
@@ -119,11 +121,31 @@ Connector Options
       <td>Specify what connector to use, here should be <code>'mongodb-cdc'</code>.</td>
     </tr>
     <tr>
-      <td>uri</td>
+      <td>hosts</td>
       <td>required</td>
       <td style="word-wrap: break-word;">(none)</td>
       <td>String</td>
-      <td>A MongoDB connection URI string.</td>
+      <td>The comma-separated list of hostname and port pairs of the MongoDB servers.<br>
+          eg. localhost:27017,localhost:27018
+      </td>
+    </tr>
+    <tr>
+      <td>user</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">(none)</td>
+      <td>String</td>
+      <td>Name of the database user to be used when connecting to MongoDB.<br>
+          This is required only when MongoDB is configured to use authentication.
+      </td>
+    </tr>
+    <tr>
+      <td>password</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">(none)</td>
+      <td>String</td>
+      <td>Password to be used when connecting to MongoDB.<br>
+          This is required only when MongoDB is configured to use authentication.
+      </td>
     </tr>
     <tr>
       <td>database</td>
@@ -139,6 +161,60 @@ Connector Options
       <td>String</td>
       <td>Name of the collection in the database to watch for changes.</td>
     </tr>
+    <tr>
+      <td>mongodb.replicaset</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">(none)</td>
+      <td>String</td>
+      <td>Specifies the name of the replica set. <br>
+          It is not necessary, but can speed up your connection times to 
+          explicitly state the servers are in a replica set in the connection.
+      </td>
+    </tr>
+    <tr>
+      <td>mongodb.authsource</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">admin</td>
+      <td>String</td>
+      <td>Database (authentication source) containing MongoDB credentials. <br>
+          This is required only when MongoDB is configured to use authentication 
+          with another authentication database than admin.
+      </td>
+    </tr>
+    <tr>
+      <td>mongodb.connect.timeout.ms</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">10000</td>
+      <td>String</td>
+      <td>The time in milliseconds to attempt a connection before timing out. <br>
+          Defaults to 10 seconds.
+      </td>
+    </tr>
+    <tr>
+      <td>mongodb.socket.timeout.ms</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">0</td>
+      <td>String</td>
+      <td>The time in milliseconds to attempt a send or receive on a socket before the attempt times out. <br>
+          A value of 0 disables this behavior.
+      </td>
+    </tr>
+    <tr>
+      <td>mongodb.ssl.enabled</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">false</td>
+      <td>Boolean</td>
+      <td>Connector will use SSL to connect to MongoDB instances.</td>
+    </tr>
+    <tr>
+      <td>mongodb.ssl.invalid.hostname.allowed</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">false</td>
+      <td>Boolean</td>
+      <td>When SSL is enabled this setting controls 
+          whether strict hostname checking is disabled during connection phase.
+      </td>
+    </tr>
     <tr>
       <td>errors.tolerance</td>
       <td>optional</td>
@@ -208,15 +284,8 @@ Connector Options
       <td>Integer</td>
       <td>The length of time in milliseconds between sending heartbeat messages. Use 0 to disable.</td>
     </tr>
-    <tr>
-      <td>local-time-zone</td>
-      <td>optional</td>
-      <td style="word-wrap: break-word;">UTC</td>
-      <td>String</td>
-      <td>The local time zone.</td>
-    </tr>
     </tbody>
-</table>    
+</table>
 </div>
 
 Note: `heartbeat.interval.ms` is highly recommended to set a proper value larger than 0 if the collection changes slowly.
@@ -229,6 +298,10 @@ Features
 
 The MongoDB CDC connector is a Flink Source connector which will read database snapshot first and then continues to read change stream events with **exactly-once processing** even failures happen. 
 
+### Whether Snapshot When Startup
+
+The config option `copy.existing` specifies whether do snapshot when MongoDB CDC consumer startup. Defaults to `true`.
+
 ### Change Streams
 
 We integrates the [MongoDB's official Kafka Connector](https://docs.mongodb.com/kafka-connector/current/kafka-source/) to read snapshot or change events from MongoDB and drive it by Debezium's `EmbeddedEngine`.
@@ -264,7 +337,9 @@ import com.ververica.cdc.connectors.mongodb.MongoDBSource;
 public class MongoDBSourceExample {
     public static void main(String[] args) throws Exception {
         SourceFunction<String> sourceFunction = MongoDBSource.<String>builder()
-                .connectionUri("mongodb://localhost:27017")
+                .hosts("localhost:27017")
+                .user("flink")
+                .password("flinkpw")
                 .database("inventory")
                 .collection("products")
                 .deserializer(new StringDebeziumDeserializationSchema())

docs/content/tutorials/mongodb-tutorial-zh.md

@@ -229,7 +229,9 @@ Flink SQL> CREATE TABLE shipments (
    PRIMARY KEY (_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mongodb-cdc',
-   'uri' = 'mongodb://mongouser:mongopw@localhost:27017',
+   'hosts' = 'localhost:27017',
+   'user' = 'mongodb',
+   'password' = 'mongopw',
    'database' = 'mgdb',
    'collection' = 'customers'
  );
@@ -293,50 +295,3 @@ db.customers.updateOne(
 --MySQL
 DELETE FROM orders WHERE order_id = 10004;
 ```
-
-7. Kafka changelog json format
-
-在 SQL CLI 中：
-
-```sql
---Flink SQL
-Flink SQL> CREATE TABLE kafka_gmv (
-   day_str STRING,
-   gmv DECIMAL(10, 5)
- ) WITH (
-     'connector' = 'kafka',
-     'topic' = 'kafka_gmv',
-     'scan.startup.mode' = 'earliest-offset',
-     'properties.bootstrap.servers' = 'localhost:9092',
-     'format' = 'changelog-json'
- );
-
-Flink SQL> INSERT INTO kafka_gmv
- SELECT DATE_FORMAT(order_date, 'yyyy-MM-dd') as day_str, SUM(price) as gmv
- FROM orders
- WHERE order_status = true
- GROUP BY DATE_FORMAT(order_date, 'yyyy-MM-dd');
-
--- 读取 Kafka 的 changelog 数据，观察 materialize 后的结果
-Flink SQL> SELECT * FROM kafka_gmv;
-```
-观察 kafka 的输出：
-
-```
-docker-compose exec kafka bash -c 'kafka-console-consumer.sh --topic kafka_gmv --bootstrap-server kafka:9094 --from-beginning'
-```
-
-更新 orders 数据，观察SQL CLI 和 kafka console 的输出
-```sql
--- MySQL
-UPDATE orders SET order_status = true WHERE order_id = 10001;
-UPDATE orders SET order_status = true WHERE order_id = 10002;
-UPDATE orders SET order_status = true WHERE order_id = 10003;
-
-INSERT INTO orders
-VALUES (default, '2020-07-30 17:33:00', 1001, 50.00, 104, true);
-
-UPDATE orders SET price = 40.00 WHERE order_id = 10005;
-
-DELETE FROM orders WHERE order_id = 10005;
-```

docs/content/tutorials/mongodb-tutorial.md

@@ -229,7 +229,9 @@ Flink SQL> CREATE TABLE shipments (
    PRIMARY KEY (_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mongodb-cdc',
-   'uri' = 'mongodb://mongouser:mongopw@localhost:27017',
+   'hosts' = 'localhost:27017',
+   'user' = 'mongodb',
+   'password' = 'mongopw',
    'database' = 'mgdb',
    'collection' = 'customers'
  );
@@ -293,50 +295,3 @@ db.customers.updateOne(
 --MySQL
 DELETE FROM orders WHERE order_id = 10004;
 ```
-
-8. Kafka Changelog JSON format
-
-Execute following SQL in Flink SQL CLI: 
-
-```sql
--- Flink SQL
-Flink SQL> CREATE TABLE kafka_gmv (
-   day_str STRING,
-   gmv DECIMAL(10, 5)
- ) WITH (
-     'connector' = 'kafka',
-     'topic' = 'kafka_gmv',
-     'scan.startup.mode' = 'earliest-offset',
-     'properties.bootstrap.servers' = 'localhost:9092',
-     'format' = 'changelog-json'
- );
-
-Flink SQL> INSERT INTO kafka_gmv
- SELECT DATE_FORMAT(order_date, 'yyyy-MM-dd') as day_str, SUM(price) as gmv
- FROM orders
- WHERE order_status = true
- GROUP BY DATE_FORMAT(order_date, 'yyyy-MM-dd');
-
--- Consumer changelog data from Kafka, and check the result of materialized view: 
-Flink SQL> SELECT * FROM kafka_gmv;
-```
-To consumer records in Kafka using `kafka-console-consumer`:
-
-```
-docker-compose exec kafka bash -c 'kafka-console-consumer.sh --topic kafka_gmv --bootstrap-server kafka:9094 --from-beginning'
-```
-
-Update orders data and check the output in Flink SQL CLI and Kafka console consumer: 
-```sql
--- MySQL
-UPDATE orders SET order_status = true WHERE order_id = 10001;
-UPDATE orders SET order_status = true WHERE order_id = 10002;
-UPDATE orders SET order_status = true WHERE order_id = 10003;
-
-INSERT INTO orders
-VALUES (default, '2020-07-30 17:33:00', 1001, 50.00, 104, true);
-
-UPDATE orders SET price = 40.00 WHERE order_id = 10005;
-
-DELETE FROM orders WHERE order_id = 10005;
-```