[oceanbase] Introduce 'table-list' option to support capture list of tables (#1369)

This closes #1369.

Author: whhe

docs/content/connectors/oceanbase-cdc.md

@@ -94,6 +94,12 @@ Flink SQL> SELECT * FROM orders;
 Connector Options
 ----------------
 
+The OceanBase CDC Connector contains some options for both sql and stream api as the following sheet. 
+
+*Note*: The connector supports two ways to specify the table list to listen to, and will get the union of the results when both way are used at the same time.
+1. Use `database-name` and `table-name` to match database and table names in regex. As the `obcdc` (former `liboblog`) only supports `fnmatch` now, we can't use regex directly to filter change events, so these two options can only be used in `initial` startup mode.
+2. Use `table-list` to match the exact value of database and table names.
+
 <div class="highlight">
     <table class="colwidths-auto docutils">
         <thead>
@@ -148,21 +154,28 @@ Connector Options
                 <td>required</td>
                 <td style="word-wrap: break-word;">(none)</td>
                 <td>String</td>
-                <td>Tenant name of OceanBase to monitor.</td>
+                <td>Tenant name of OceanBase to monitor, should be exact value.</td>
             </tr>
             <tr>
                 <td>database-name</td>
-                <td>required</td>
+                <td>optional</td>
                 <td style="word-wrap: break-word;">(none)</td>
                 <td>String</td>
-                <td>Database name of OceanBase to monitor.</td>
+                <td>Database name of OceanBase to monitor, should be regular expression. Only can be used with 'initial' mode.</td>
             </tr>
             <tr>
                 <td>table-name</td>
-                <td>required</td>
+                <td>optional</td>
+                <td style="word-wrap: break-word;">(none)</td>
+                <td>String</td>
+                <td>Table name of OceanBase to monitor, should be regular expression. Only can be used with 'initial' mode.</td>
+            </tr>
+            <tr>
+                <td>table-list</td>
+                <td>optional</td>
                 <td style="word-wrap: break-word;">(none)</td>
                 <td>String</td>
-                <td>Table name of OceanBase to monitor.</td>
+                <td>List of full names of tables, separated by commas, e.g. "db1.table1, db2.table2".</td>
             </tr>
             <tr>
                 <td>hostname</td>
@@ -211,7 +224,7 @@ Connector Options
                 <td>optional</td>
                 <td style="word-wrap: break-word;">By rule.</td>
                 <td>String</td>
-                <td>Id of a log proxy client connection, will be in format {flink_ip}_{process_id}_{timestamp}_{tenant}.{db}.{table} by default.</td>
+                <td>Id of a log proxy client connection, will be in format {flink_ip}_{process_id}_{timestamp}_{thread_id}_{tenant} by default.</td>
             </tr>
             <tr>
                 <td>rootserver-list</td>

docs/content/quickstart/oceanbase-tutorial.md

@@ -133,8 +133,8 @@ Flink SQL> CREATE TABLE orders (
     'username' = 'root',
     'password' = 'pswd',
     'tenant-name' = 'sys',
-    'database-name' = 'ob',
-    'table-name' = 'orders',
+    'database-name' = '^ob$',
+    'table-name' = '^orders$',
     'hostname' = 'localhost',
     'port' = '2881',
     'rootserver-list' = '127.0.0.1:2882:2881',
@@ -153,8 +153,8 @@ Flink SQL> CREATE TABLE products (
     'username' = 'root',
     'password' = 'pswd',
     'tenant-name' = 'sys',
-    'database-name' = 'ob',
-    'table-name' = 'products',
+    'database-name' = '^ob$',
+    'table-name' = '^products$',
     'hostname' = 'localhost',
     'port' = '2881',
     'rootserver-list' = '127.0.0.1:2882:2881',

docs/content/快速上手/oceanbase-tutorial-zh.md

@@ -131,8 +131,8 @@ Flink SQL> CREATE TABLE orders (
     'username' = 'root',
     'password' = 'pswd',
     'tenant-name' = 'sys',
-    'database-name' = 'ob',
-    'table-name' = 'orders',
+    'database-name' = '^ob$',
+    'table-name' = '^orders$',
     'hostname' = 'localhost',
     'port' = '2881',
     'rootserver-list' = '127.0.0.1:2882:2881',
@@ -151,8 +151,8 @@ Flink SQL> CREATE TABLE products (
     'username' = 'root',
     'password' = 'pswd',
     'tenant-name' = 'sys',
-    'database-name' = 'ob',
-    'table-name' = 'products',
+    'database-name' = '^ob$',
+    'table-name' = '^products$',
     'hostname' = 'localhost',
     'port' = '2881',
     'rootserver-list' = '127.0.0.1:2882:2881',

flink-connector-oceanbase-cdc/src/main/java/com/ververica/cdc/connectors/oceanbase/OceanBaseSource.java

@@ -54,6 +54,7 @@ public static class Builder<T> {
         private String tenantName;
         private String databaseName;
         private String tableName;
+        private String tableList;
         private String serverTimeZone;
         private Duration connectTimeout;
 
@@ -102,6 +103,11 @@ public Builder<T> tableName(String tableName) {
             return this;
         }
 
+        public Builder<T> tableList(String tableList) {
+            this.tableList = tableList;
+            return this;
+        }
+
         public Builder<T> serverTimeZone(String serverTimeZone) {
             this.serverTimeZone = serverTimeZone;
             return this;
@@ -182,6 +188,17 @@ public SourceFunction<T> build() {
                             startupMode + " mode is not supported.");
             }
 
+            if (StringUtils.isNotEmpty(databaseName) || StringUtils.isNotEmpty(tableName)) {
+                if (StringUtils.isEmpty(databaseName) || StringUtils.isEmpty(tableName)) {
+                    throw new IllegalArgumentException(
+                            "'database-name' and 'table-name' should be configured at the same time");
+                }
+            } else {
+                checkNotNull(
+                        tableList,
+                        "'database-name', 'table-name' or 'table-list' should be configured");
+            }
+
             if (serverTimeZone == null) {
                 serverTimeZone = "UTC";
             }
@@ -191,15 +208,13 @@ public SourceFunction<T> build() {
                 connectTimeout = Duration.ofSeconds(30);
             }
 
-            String tableWhiteList =
-                    String.format(
-                            "%s.%s.%s",
-                            checkNotNull(tenantName),
-                            checkNotNull(databaseName),
-                            checkNotNull(tableName));
-
             if (logProxyClientId == null) {
-                logProxyClientId = ClientIdGenerator.generate() + "_" + tableWhiteList;
+                logProxyClientId =
+                        String.format(
+                                "%s_%s_%s",
+                                ClientIdGenerator.generate(),
+                                Thread.currentThread().getId(),
+                                checkNotNull(tenantName));
             }
             ClientConf clientConf =
                     ClientConf.builder()
@@ -219,7 +234,6 @@ public SourceFunction<T> build() {
             }
             obReaderConfig.setUsername(username);
             obReaderConfig.setPassword(password);
-            obReaderConfig.setTableWhiteList(tableWhiteList);
             obReaderConfig.setStartTimestamp(startupTimestamp);
             obReaderConfig.setTimezone(zoneOffset.getId());
 
@@ -230,6 +244,7 @@ public SourceFunction<T> build() {
                     tenantName,
                     databaseName,
                     tableName,
+                    tableList,
                     zoneOffset,
                     connectTimeout,
                     hostname,