Realtime Compute for Apache Flink：KMS 憑據管理整合（公測）

建立 RAM 角色

1. 登入 RAM 控制台 - 角色頁面，單擊建立角色。保持預設選項，並單擊確定。

2. 填寫一個有辨識度的角色名稱，例如 FlinkRoleForKMSRead，單擊確定。

添加信任策略

角色建立成功後，在角色詳情頁的信任策略頁簽下，單擊編輯信任策略，在 Principal 中添加 "Service": ["stream.aliyuncs.com"]，信任 Flink 扮演該角色。更多資訊，請參見修改 RAM 角色的信任策略。

{
  "Statement": [
    {
      "Action": "sts:AssumeRole",
      "Effect": "Allow",
      "Principal": {
        "Service": [
          "stream.aliyuncs.com"
        ]
      }
    }
  ],
  "Version": "1"
}

為 RAM 角色授予 KMS 存取權限

扮演該角色後，Flink 將以角色身份調用 KMS，需提前在角色權限原則中授予 KMS 相關許可權。

在角色詳情頁的許可權管理頁簽下，單擊新增授權，添加 AliyunKMSSecretUserAccess（擷取 KMS 中憑據的許可權）權限原則。若憑據使用了使用者指定的密鑰（CMK）加密，還需有密鑰的 kms:Decrypt 許可權，可通過添加 AliyunKMSCryptoUserAccess 權限原則來授權。

若期望細粒度控制到密鑰及憑據粒度，可自訂權限原則，授予 GetSecretValue、Decrypt 許可權，並將資源限制為指定資源。詳情請參見 KMS 自訂權限原則參考、建立自訂權限原則。

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "kms:GetSecretValue",
      "Resource": "acs:kms:<region>:<account-id>:secret/<secret-name>"
    },
    {
      "Effect": "Allow",
      "Action": "kms:Decrypt",
      "Resource": "acs:kms:<region>:<account-id>:key/<key-id>"
    }
  ]
}

建立權限原則

登入 RAM 控制台 - 權限原則頁面，單擊建立權限原則。切換至指令碼編輯，編輯權限原則內容。

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "ram:ListRoles"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "ram:PassRole",
      "Resource": "acs:ram::<account-id>:role/<role-name>",
      "Condition": {
        "StringEquals": {
          "acs:Service": "stream.aliyuncs.com"
        }
      }
    }
  ]
}

單擊確定，將權限原則儲存，例如命名為 FlinkPassRolePolicy。更多資訊，請參見建立自訂權限原則。

為 RAM 使用者授權

1. 登入 RAM 控制台 - 使用者頁面，找到需要授權的 RAM 使用者，單擊操作列的新增授權。

2. 在新增授權面板中，搜尋並選中步驟三建立的權限原則（例如 FlinkPassRolePolicy），單擊確認新增授權。更多資訊，請參見管理RAM使用者的許可權。

Postgres CDC Connector 樣本

SQL 作業：以下樣本以 Postgres CDC 源表為例，使用 KMS 憑據管理的資料庫帳號密碼。KMS 中已建立資料庫憑據 my-db-secret，其 SecretData 為 {"AccountName":"alice","AccountPassword":"xxx"}。

CREATE TEMPORARY TABLE postgrescdc_source (
    id INT NOT NULL,
    name STRING,
    description STRING,
    weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = 'secret://kms.my-db-secret.AccountName',
  'password' = 'secret://kms.my-db-secret.AccountPassword',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>',
  'slot.name' = '<slot name>',
  'decoding.plugin.name' = 'pgoutput',
  'kms.endpoint' = '<YOUR-KMS-INSTANCE-VPC-ENDPOINT>',
  -- KMS 專屬網關的 CA 憑證路徑
  'kms.ca' = 'oss://flink-fullymanaged-<工作空間ID>/artifacts/namespaces/<專案空間名稱>/<檔案名稱>',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

Postgres CDC 連接器的參數說明、前提條件與類型映射，請參見Postgres CDC 連接器。

Kafka Connector 樣本

Kafka 的憑據寫在 SASL JAAS 配置中。將 properties.sasl.jaas.config 內的 username 和 password 替換為 secret:// 引用，其餘 Kafka 參數保持不變。KMS 中已建立憑據 my-kafka-secret，其 SecretData 為 {"AccountName":"<SASL 使用者名稱>","AccountPassword":"<SASL 密碼>"}。

SQL 作業：以下樣本以 Kafka 源表為例，結果表的配置方式相同。

CREATE TEMPORARY TABLE kafka_source (
  order_id BIGINT,
  user_id BIGINT,
  amount INT,
  payload STRING
) WITH (
  'connector' = 'kafka',
  'topic' = '<topic name>',
  'properties.bootstrap.servers' = '<broker 1>,<broker 2>,<broker 3>',
  'properties.group.id' = '<group id>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv',
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  /* 使用者名稱與密碼引用 KMS 憑據 */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";',
  'kms.endpoint' = 'kms.cn-beijing.aliyuncs.com',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

資料攝入 YAML 作業：源端與目標端分別配置 KMS 參數。樣本中源端使用共用網關，目標端使用專屬網關，一次展示兩種配置形態。

source:
  type: kafka
  name: Kafka Source
  topic: <topic name>
  value.format: json
  scan.startup.mode: earliest-offset
  properties.bootstrap.servers: <broker 1>,<broker 2>,<broker 3>
  properties.group.id: <group id>
  properties.security.protocol: SASL_PLAINTEXT
  properties.sasl.mechanism: PLAIN
  properties.sasl.jaas.config: org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";
  # 共用網關：Endpoint 按地區固定，無需配置 kms.ca
  kms.endpoint: kms.cn-beijing.aliyuncs.com
  kms.akless.assume-role.role-name: FlinkRoleForKMSRead

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: <broker 1>,<broker 2>,<broker 3>
  # 阿里雲 Kafka 不支援等冪寫入，VVR 8.0.0 及以上版本需顯式關閉
  properties.enable.idempotence: false
  properties.security.protocol: SASL_PLAINTEXT
  properties.sasl.mechanism: PLAIN
  properties.sasl.jaas.config: org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";
  # 專屬網關：需同時配置 kms.ca
  kms.endpoint: <KMS 執行個體的 VPC 端點>
  kms.ca: oss://flink-fullymanaged-<工作空間ID>/artifacts/namespaces/<專案空間名稱>/PrivateKmsCA_kst-******.pem
  kms.akless.assume-role.role-name: FlinkRoleForKMSRead

Kafka 連接器的完整參數與安全認證說明，請參見Kafka SQL 連接器和Kafka 資料攝入連接器。

參數說明