实时计算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 数据摄入连接器。

参数说明