apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: flink
  name: flink-deployment-manager
rules:
- apiGroups: 
  - flink.apache.org
  resources: 
  - flinkdeployments
  verbs: 
  - 'get'
  - 'list'
  - 'create'
  - 'update'
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: flink-deployment-manager-binding
  namespace: flink
subjects:
- kind: User
  name: "277293711358271379"  
  apiGroup: rbac.authorization.k8s.io
roleRef:
  kind: Role
  name: flink-deployment-manager
  apiGroup: rbac.authorization.k8s.io

apiVersion: "flink.apache.org/v1beta1"
kind: "FlinkDeployment"
metadata:
  name: "financial-job"
spec:
  image: "cr.registry.res.cloud.wuxi-yqgcy.cn/mirror/financial-topic:1.5-oss"
  flinkVersion: "v1_17"
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "8"
    fs.oss.endpoint: http://ay-test.oss-cn-jswx-xuelang-d01-a.ops.cloud.wuxi-yqgcy.cn/
    fs.oss.accessKeyId: 4gqOVOfQqCsCUwaC
    fs.oss.accessKeySecret: xxx
  ingress:
    template: "flink.k8s.io/{{namespace}}/{{name}}(/|$)(.*)"
    className: "nginx"
    annotations:
      cert-manager.io/cluster-issuer: "self-signed-ca-issuer"
      nginx.ingress.kubernetes.io/rewrite-target: "/$2"
  serviceAccount: "flink"
  podTemplate:
    apiVersion: "v1"
    kind: "Pod"
    metadata:
      name: "financial-job"
    spec:
      containers:
        - name: "flink-main-container"
          env:
            - name: ENABLE_BUILT_IN_PLUGINS
              value: flink-oss-fs-hadoop-1.17.2.jar
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  job:
    jarURI: "local:///app/application.jar"
    parallelism: 1
    upgradeMode: "stateless"

apiVersion: "flink.apache.org/v1beta1"
kind: "FlinkDeployment"
metadata:
  name: "financial-job"
spec:
  image: "cr.registry.res.cloud.wuxi-yqgcy.cn/mirror/financial-topic:1.5"
  flinkVersion: "v1_17"
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "8"
    s3a.endpoint: http://172.27.253.89:9000
    s3a.access-key: minioadmin
    s3a.secret-key: minioadmin
  ingress:
    template: "flink.k8s.io/{{namespace}}/{{name}}(/|$)(.*)"
    className: "nginx"
    annotations:
      cert-manager.io/cluster-issuer: "self-signed-ca-issuer"
      nginx.ingress.kubernetes.io/rewrite-target: "/$2"
  serviceAccount: "flink"
  podTemplate:
    apiVersion: "v1"
    kind: "Pod"
    metadata:
      name: "financial-job"
    spec:
      containers:
        - name: "flink-main-container"
          env:
            - name: ENABLE_BUILT_IN_PLUGINS
              value: flink-s3-fs-hadoop-1.17.2.jar
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  job:
    jarURI: "local:///app/application.jar"
    parallelism: 1
    upgradeMode: "stateless"

MySqlSource<String> mySqlSource =
    MySqlSource.<String>builder()
        .hostname("192.168.56.107")
        .port(3306)
        .databaseList("test") // set captured database
        .tableList("test.table_a") // set captured table
        .username("root")
        .password("mysql")
        .deserializer(
            new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
        .serverTimeZone("UTC")
        .build();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// enable checkpoint
env.enableCheckpointing(3000);

env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
    // set 4 parallel source tasks
    .setParallelism(4)
    .print()
    .setParallelism(1); // use parallelism 1 for sink to keep message ordering

env.execute("Print MySQL Snapshot + Binlog");