How to use the flink-dts-connector - Data Transmission Service

After you configure a subscription instance, you can use the flink-dts-connector to consume data from the instance with a Flink client. This topic describes how to use the flink-dts-connector.

Usage notes

This connector supports Flink clients that use the DataStream API, Table API, or SQL.
If your Flink client uses the Table API or SQL, you can consume data from only a single table per configuration. To consume data from multiple tables, you must configure and run multiple independent tasks.

Procedure

This topic uses IntelliJ IDEA Community Edition 2020.1 for Windows to demonstrate how to use the flink-dts-connector to consume data from a subscription instance.

Create a change tracking task. For more information, see the relevant topics in Overview of change tracking scenarios.
Create one or more consumer groups. For more information, see Add a consumer group.
Download and decompress the flink-dts-connector.
Open IntelliJ IDEA and click Open or Import.
In the dialog box that appears, navigate to the decompressed flink-dts-connector directory, expand the folders, and then find the Project Object Model (pom.xml) file.
In the dialog box that appears, select Open as Project.

Add the following dependency to the pom.xml file:

<dependency>
      <groupId>com.alibaba.flink</groupId>
      <artifactId>flink-dts-connector</artifactId>
      <version>1.1.1-SNAPSHOT</version>
      <classifier>jar-with-dependencies</classifier>
</dependency>

In IntelliJ IDEA, expand the folders and select the Java file that corresponds to the program type of your Flink connector.
- If your Flink client uses the DataStream API, double-click the flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\datastream\DtsExample.java file and perform the following steps:
  1. At the top of the IntelliJ IDEA window, click the Run icon.
  2. In the menu that appears, click DtsExample > Edit.
  3. In the Program arguments field, enter the parameters and their values based on the following example. Then, click Run to start the flink-dts-connector.
    
    Note
    For more information about the parameters and how to obtain their values, see Parameters.
```
--broker-url dts-cn-******.******.***:****** --topic cn_hangzhou_rm_**********_dtstest_version2 --sid dts****** --user dtstest --password Test123456 --checkpoint 1624440043
```
  4. The output confirms that the client can subscribe to data changes from the source database.
    
    Note
    To query specific data change records, log on to the Task Manager UI of your Flink client.
- If your Flink client uses the Table API or SQL, double-click the flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\sql\DtsTableISelectTCaseTest.java file and perform the following steps:
  
  Note
  Each DtsTableISelectTCaseTest.java file can be configured to consume data from only one table. To configure and consume data from multiple tables, you must repeat the configuration and run multiple independent tasks.
  1. Add two forward slashes (//) to comment out the specified line of code.
  2. Configure the information for the single table from which you want to consume data. SQL statements are supported.
  3. Configure the parameters for the subscription instance. For more information about the parameters and how to obtain their values, see Parameters.
  4. At the top of the IntelliJ IDEA window, click Run'DtsTableISelectTCaseTest' to start the flink-dts-connector.
  5. The output confirms that the client can subscribe to data changes from the source database.
    
    Note
    To query specific data change records, log on to the Task Manager UI of your Flink client.

Parameters

DtsExample.java	DtsTableISelectTCaseTest.java	Description	Query method
`broker-url`	`dts.server`	The network endpoint and port of the subscription instance. Note If the ECS instance where your Flink client is deployed and the subscription instance are in the same classic network or Virtual Private Cloud (VPC), we recommend that you use an internal endpoint for data subscription to minimize network latency. We do not recommend using a public endpoint because of potential network instability.	In the DTS console, click the ID of the target subscription instance. On the Basic Information page, you can obtain the Topic and Network information.
`topic`	`topic`	The topic of the subscription instance.
`sid`	`dts.sid`	The consumer group ID.	In the DTS console, click the ID of the target subscription instance. In the left-side navigation pane, click Consume Data. You can obtain the Consumer Group ID/Name and the Account of the consumer group. Note The password for the consumer group account is specified when you create the consumer group.
`user`	`dts.user`	The account of the consumer group. Warning If you are not using the flink-dts-connector provided in this topic, you must set the username in the `<consumer_group_account>-<consumer_group_ID>` format, such as `dtstest-dtsae******bpv`. Otherwise, the connection fails.
`password`	`dts.password`	The password of the account.
`checkpoint`	`dts.checkpoint`	The consumer offset. This parameter specifies the Unix timestamp of the first data record to consume. Example: 1624440043. Note You can use the consumer offset to: Resume consumption from the last offset after an application interruption to prevent data loss. Specify an offset to start consuming data from a specific point in time.	The start time for consumption must be within the data range of the subscription instance and must be converted to a Unix timestamp. Note You can view the data range of the subscription instance in the Data Range column of the subscription task list. You can use a search engine to find a Unix timestamp conversion tool.
N/A	`dts-cdc.table.name`	The subscribed object. You can specify only a single table. The format requirements are as follows: If the Database Type is MySQL, PolarDB for MySQL, PolarDB-X 1.0, or PolarDB-X 2.0, use the format `<database_name>.<table_name>`. For other database types, use the format `<schema_name>.<table_name>`.	In the DTS console, click the ID of the target subscription instance. At the top of the Basic Information or Task Management page, click View Objects to find the subscribed database and table.

FAQ

Error message	Possible cause	Solution
`Cluster changed from * to *, consumer require restart.`	A switchover in the DStore module, which Data Transmission Service (DTS) uses to read incremental data, causes the Flink client's consumer offset to be lost.	Do not restart the client. Instead, query the client's consumer offset and pass the value again to the `checkpoint` or `dts.checkpoint` parameter in the DtsExample.java or DtsTableISelectTCaseTest.java file to resume data consumption.