[FLINK-35354] Add doc for host mapping feature

Author: Mrart

docs/content.zh/docs/connectors/flink-sources/tidb-cdc.md

@@ -124,6 +124,13 @@ Connector Options
       <td style="word-wrap: break-word;">(none)</td>
       <td>String</td>
       <td>TiKV cluster's PD address.</td>
+    </tr>
+     <tr>
+      <td>host-mapping</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">(none)</td>
+      <td>String</td>
+      <td>TiKV cluster's host-mapping used to configure public IP and intranet IP mapping. When the TiKV cluster is running on the intranet, you can map a set of intranet IPs to public IPs for an outside Flink cluster to access. The format is {Intranet IP1}:{Public IP1};{Intranet IP2}:{Public IP2}, e.g. 192.168.0.2:8.8.8.8;192.168.0.3:9.9.9.9.</td>
     </tr>
     <tr>
       <td>tikv.grpc.timeout_in_ms</td>

docs/content/docs/connectors/flink-sources/tidb-cdc.md

@@ -125,6 +125,13 @@ Connector Options
       <td>String</td>
       <td>TiKV cluster's PD address.</td>
     </tr>
+    <tr>
+      <td>host-mapping</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">(none)</td>
+      <td>String</td>
+      <td>TiKV cluster's host-mapping used to configure public IP and intranet IP mapping. When the TiKV cluster is running on the intranet, you can map a set of intranet IPs to public IPs for an outside Flink cluster to access. The format is {Intranet IP1}:{Public IP1};{Intranet IP2}:{Public IP2}, e.g. 192.168.0.2:8.8.8.8;192.168.0.3:9.9.9.9.</td>
+    </tr>
     <tr>
       <td>tikv.grpc.timeout_in_ms</td>
       <td>optional</td>

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/TDBSourceOptions.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.connectors.tidb;
 
+import org.apache.flink.cdc.common.utils.Preconditions;
 import org.apache.flink.cdc.connectors.tidb.table.utils.UriHostMapping;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
@@ -27,6 +28,7 @@
 import org.tikv.common.TiConfiguration;
 
 import java.util.Map;
+import java.util.Optional;
 
 /** Configurations for {@link TiDBSource}. */
 public class TDBSourceOptions {
@@ -64,7 +66,7 @@ private TDBSourceOptions() {}
                     .stringType()
                     .noDefaultValue()
                     .withDescription(
-                            "This is route map used to configure public IP and intranet IP mapping. When the TiDB cluster is running on the intranet, you can map a set of intranet IPs to public IPs for an outside Flink cluster to access. The format is {Intranet IP1}:{Public IP1};{Intranet IP2}:{Public IP2}, e.g. 192.168.0.2:8.8.8.8;192.168.0.3:9.9.9.9.");
+                            "TiKV cluster's host-mapping used to configure public IP and intranet IP mapping. When the TiKV cluster is running on the intranet, you can map a set of intranet IPs to public IPs for an outside Flink cluster to access. The format is {Intranet IP1}:{Public IP1};{Intranet IP2}:{Public IP2}, e.g. 192.168.0.2:8.8.8.8;192.168.0.3:9.9.9.9.");
     public static final ConfigOption<Long> TIKV_GRPC_TIMEOUT =
             ConfigOptions.key(ConfigUtils.TIKV_GRPC_TIMEOUT)
                     .longType()
@@ -94,9 +96,7 @@ public static TiConfiguration getTiConfiguration(
         final Configuration configuration = Configuration.fromMap(options);
 
         final TiConfiguration tiConf = TiConfiguration.createDefault(pdAddrsStr);
-        if (StringUtils.isNotBlank(hostMapping)) {
-            tiConf.setHostMapping(new UriHostMapping(hostMapping));
-        }
+        Optional.of(new UriHostMapping(hostMapping)).ifPresent(tiConf::setHostMapping);
         configuration.getOptional(TIKV_GRPC_TIMEOUT).ifPresent(tiConf::setTimeout);
         configuration.getOptional(TIKV_GRPC_SCAN_TIMEOUT).ifPresent(tiConf::setScanTimeout);
         configuration

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-tidb-cdc/src/main/java/org/apache/flink/cdc/connectors/tidb/table/TiDBTableSource.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.connectors.tidb.table;
 
+import com.sun.istack.internal.Nullable;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.cdc.connectors.tidb.TDBSourceOptions;
 import org.apache.flink.cdc.connectors.tidb.TiDBSource;
@@ -52,7 +53,7 @@ public class TiDBTableSource implements ScanTableSource, SupportsReadingMetadata
     private final String database;
     private final String tableName;
     private final String pdAddresses;
-
+    @Nullable
     private final String hostMapping;
     private final StartupOptions startupOptions;
     private final Map<String, String> options;