FlinkStreaming(Scala) + FlinkCDC2.0

def main(args: Array[String]): Unit = {

    val debeziumProperties = new Properties
    debeziumProperties.put("snapshot.locking.mode", "none") // do not use lock
    import org.apache.flink.streaming.api.scala._
    val sourceFunction = MySqlSource
      .builder()
      .hostname("cm1")
      .port(3306)
      .databaseList("db")
      .tableList("db.dim_tbl_config_info")
      .username("root")
      .password("123456")
      .deserializer(new JsonDebeziumDeserializationSchema)
      .debeziumProperties(debeziumProperties)
      .build()
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.enableCheckpointing(3000)
    env.setParallelism(1)
    env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "test").print()
    
    env.execute()

  }