Realtime Compute for Apache Flink:KMS credential management integration (Beta)

1. Create a RAM role

1. Log on to the RAM console - Roles page and click Create Role. Keep the default options and click OK.

2. Enter a recognizable role name, for example, FlinkRoleForKMSRead, and click OK.

2. Add a trust policy

After the role is created, on the role details page, click the Trust Policy tab and click Edit Trust Policy. Add "Service": ["stream.aliyuncs.com"] to Principal to trust Flink to assume the role. For more information, see Modify the trust policy of a RAM role.

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

3. Grant KMS access permissions to the RAM role

After Flink assumes the role, it calls KMS as the role. Grant the required KMS permissions to the role's permission policy in advance.

On the role details page, click the Permissions tab and click Grant Permission. Add the AliyunKMSSecretUserAccess policy (which grants permission to retrieve credentials in KMS). If credentials are encrypted with a customer master key (CMK), also grant the kms:Decrypt permission by adding the AliyunKMSCryptoUserAccess policy.

To control permissions at the key and credential level, create a custom policy that grants GetSecretValue and Decrypt permissions and restricts the resource to specified resources. For details, see KMS custom policy reference and Create a custom policy.

{
  "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>"
    }
  ]
}

Create a permission policy

Log on to the RAM console - Policies page and click Create Policy. Switch to Script mode and edit the policy content.

{
  "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"
        }
      }
    }
  ]
}

Click OK to save the policy, for example, as FlinkPassRolePolicy. For more information, see Create a custom policy.

Grant permissions to the RAM user

1. Log on to the RAM console - Users page. Find the RAM user to authorize and click Add Permissions in the Actions column.

2. In the Add Permissions panel, search for and select the policy created in Step 3 (for example, FlinkPassRolePolicy), then click Confirm. For more information, see Grant permissions to a RAM user.

Postgres CDC Connector example

SQL job: The following example uses a Postgres CDC source table and hosts the database username and password with a KMS credential. A database credential named my-db-secret has been created in KMS, with 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' = 'oss://<bucket>/kms-ca.pem',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

For the parameter descriptions, prerequisites, and type mapping of the Postgres CDC connector, see Postgres CDC connector.

Kafka Connector example

Kafka credentials are referenced in the SASL JAAS configuration. Replace the username and password values in properties.sasl.jaas.config with secret:// references. Other Kafka parameters stay the same. A credential named my-kafka-secret has been created in KMS, with SecretData {"AccountName":"<SASL username>","AccountPassword":"<SASL password>"}.

SQL job: The following example uses a Kafka source table. Configure a result table in the same way.

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',
  /* Reference KMS credentials in username and password */
  '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'
);

Data ingestion YAML job: The source and the sink specify KMS parameters separately. The source uses a shared gateway and the sink uses a dedicated gateway, so this example covers both.

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";
  # Shared gateway: the endpoint is fixed per region, and kms.ca is not required
  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>
  # ApsaraMQ for Kafka does not support idempotent writes. Disable idempotence on VVR 8.0.0 and later
  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";
  # Dedicated gateway: kms.ca is also required
  kms.endpoint: <your-kms-instance-vpc-endpoint>
  kms.ca: oss://flink-fullymanaged-<workspace-id>/artifacts/namespaces/<namespace>/PrivateKmsCA_kst-******.pem
  kms.akless.assume-role.role-name: FlinkRoleForKMSRead

For the complete parameters and security settings of the Kafka connector, see Kafka SQL connector and Kafka YAML connector.

Parameter description