Use the Kafka connector for Kerberos authentication - Realtime Compute for Apache Flink

Kerberos is a computer-network authentication protocol that is used for identity authentication to ensure the security of communication. Your Realtime Compute for Apache Flink deployment can access a Kafka cluster for which Kerberos authentication is enabled only if you configure valid Kafka and Kerberos information in the development console of Realtime Compute for Apache Flink. This topic describes how to enable Kerberos authentication for the Kafka client.

Limits

ApsaraMQ for Kafka does not support Kerberos authentication.

Step 1: Make preparations

Before you enable Kerberos authentication, perform the following operations:

Obtain the Kerberos configuration files.
- krb5.conf is a configuration file in the Kerberos authentication environment. The krb5.conf configuration file is used to specify the cluster for which you want to enable Kerberos authentication and the parameters that are used for connection between Kerberos clients and servers. For more information about the configuration items, see MIT Kerberos Documentation.
- keytab is a configuration file that contains identity authentication credentials of users, which are used to identify users during authentication on the Kerberos server.
Configure domain name resolution.
Kerberos authentication relies on domain name resolution. You must configure the domain names and IP addresses of the Key Distribution Center (KDC) server and Kafka in the development console of Realtime Compute for Apache Flink.
- KDC server: The domain name and IP address of the KDC server are defined in the krb5.conf configuration file.
- Kafka broker: The domain names and IP addresses of Kafka brokers are defined in the server.properties configuration file.
  Important
  The Kafka client must use the domain name of a Kafka broker that is registered in Kerberos to connect to the Kafka broker. Otherwise, the Kafka client fails Kerberos authentication and cannot connect to the Kafka broker. You can check the principal to determine whether the domain name of the Kafka broker is registered in Kerberos. For more information, see Basic operations on Kerberos. You must configure domain name resolution for all Kafka brokers in your Kafka cluster.
  For example, if the principal of the Kafka broker named broker1 in Kerberos is kafka/broker1.example.com@EXAMPLE.com, you must configure domain name resolution for broker1.example.com.
Create an access rule.
You must create an access rule to allow Realtime Compute for Apache Flink to access the KDC server over port 88 of the KDC server.

Step 2: Upload the Kerberos configuration files

Log on to the Realtime Compute for Apache Flink console.
Find the workspace that you want to manage and click Console in the Actions column.
In the left-side navigation pane, click Artifacts.
In the upper-left corner of the Artifacts page, click Upload Artifact. Select the krb5.conf and keytab configuration files that you want to upload.
If the files that you upload are named krb5.conf and keytab when you configure a deployment, you can use the paths /flink/usrlib/krb5.conf and /flink/usrlib/keytab to reference the files.
Note
You can specify different Kerberos principals for different Kafka clients in the development console of Realtime Compute for Apache Flink. If your deployment involves multiple Kerberos principals, upload the keytab files of all principals.
Click Open to upload the files.

Step 3: Enable Kerberos authentication

Method 1: Enable Kerberos authentication in an SQL deployment

Create a Kafka table.

Develop an SQL draft or open an existing SQL draft. For more information, see Develop an SQL draft.

Add the following configurations to the WITH clause of the Kafka table to enable Kerberos authentication for the Kafka client:

CREATE TEMPORARY TABLE kafka_kerberos (
     ...
)WITH (
    'connector' = 'kafka',
    // Specify the domain name of a Kafka broker that is registered in Kerberos.
    'properties.bootstrap.servers' = 'broker1.example.com:9092',
    // Use SASL_PLAINTEXT in this example.
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'GSSAPI',
    // Modify the username and realm in the principal in the configuration based on your business requirements.
    'properties.sasl.jaas.config' = 'com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/flink/usrlib/keytab" principal="user@EXAMPLE.COM";',
    // Specify the service name of the principal that is registered in Kerberos.
    'properties.sasl.kerberos.service.name' = 'kafka',
    // Add other configurations.
    ...
)

For more information about the parameters, see CONFIGURATION.

Deploy the draft. For more information, see Create a deployment.

Configure runtime parameters for the deployment.
1. On the O&M > Deployments page, click the name of the desired deployment.
2. In the Parameters section of the Configuration tab, add the following configuration to the Other Configuration field to specify the Kerberos configuration file for the Java virtual machine (JVM):
```
env.java.opts: '-Djava.security.krb5.conf=/flink/usrlib/krb5.conf'
```
3. In the upper-right corner of the Parameters section, click Save.
In the upper-right corner of the Deployments page, click Start.

Method 2: Enable Kerberos authentication in a DataStream deployment

Create a Kafka source and Kafka sink.

Develop a DataStream draft. For more information, see Develop a JAR draft.

Add the following configurations to the code to enable Kerberos authentication for the Kafka client:

KafkaSource.builder()
    ...
    // Specify the domain name of a Kafka broker that is registered in Kerberos.
    .setBootstrapServers("broker1.example.com:9092")
    // Use SASL_PLAINTEXT in this example.
    .setProperty("security.protocol", "SASL_PLAINTEXT")
    .setProperty("sasl.mechanism", "GSSAPI")
    // Modify the username and realm in the principal in the configuration based on the information registered in Kerberos.
    .setProperty("sasl.jaas.config", "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab=\"/flink/usrlib/keytab\" principal=\"user@EXAMPLE.COM\";")
     // Specify the service name of the principal that is registered in Kerberos.
    .setProperty("sasl.kerberos.service.name", "kafka")

KafkaSink.builder()
    ...
    // Specify the domain name of a Kafka broker that is registered in Kerberos.
    .setBootstrapServers("broker1.example.com:9092")
    // Use SASL_PLAINTEXT in this example.
    .setProperty("security.protocol", "SASL_PLAINTEXT")
    .setProperty("sasl.mechanism", "GSSAPI")
    // Modify the username and realm in the principal in the configuration based on the information registered in Kerberos.
    .setProperty("sasl.jaas.config", "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab=\"/flink/usrlib/keytab\" principal=\"user@EXAMPLE.COM\";")
    // Specify the service name of the principal that is registered in Kerberos.
    .setProperty("sasl.kerberos.service.name", "kafka")

For more information about the parameters, see CONFIGURATION.

Configure runtime parameters for the deployment.
1. After you develop a DataStream draft, package the program.
2. Upload and deploy the generated JAR file to the development console of Realtime Compute for Apache Flink. For more information, see Create a JAR deployment.
3. In the Parameters section of the Configuration tab on the Deployments page, add the following configuration to the Other Configuration field to specify the Kerberos configuration file for the JVM:
```
env.java.opts: '-Djava.security.krb5.conf=/flink/usrlib/krb5.conf'
```
4. In the upper-right corner of the Parameters section, click Save.
In the upper-right corner of the Deployments page, click Start.