Example on how to use Kafka config providers in Flink Kafka connectors, using Table API & SQL(Java), with mTLS authentication. (#72)

Authored-by: “Ravtej” <“sravtej”@amazon.co.uk”>

Author: ravtejsingh

java/KafkaConfigProviders/Kafka-SASL_SSL-ConfigProviders/README.md

@@ -162,7 +162,7 @@ is controlled via networking (SecurityGroups, NACL) and by the SASL credentials
 
 When running on Amazon Managed Service for Apache Flink the runtime configuration is read from *Runtime Properties*.
 
-When running locally, the configuration is read from the [`resources/flink-application-properties-dev.json`](resources/flink-application-properties-dev.json) file located in the resources folder.
+When running locally, the configuration is read from the [`resources/flink-application-properties-dev.json`](src/main/resources/flink-application-properties-dev.json) file located in the resources folder.
 
 Runtime parameters:
 
@@ -186,4 +186,4 @@ All parameters are case-sensitive.
 
 You can run this example directly in IntelliJ, without any local Flink cluster or local Flink installation.
 
-See [Running examples locally](../running-examples-locally.md) for details.
+See [Running examples locally](../../running-examples-locally.md) for details.

java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/.gitignore

@@ -1 +1 @@
-!*.jar
+local-repo/**/*.jar

java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/README.md

@@ -1,16 +1,16 @@
 ## Sample illustrating how to use MSK config providers in Flink Kafka connectors, for mTLS authentication
 
-* Flink version: 1.19
+* Flink version: 1.20
 * Flink API: DataStream API
 * Language: Java (11)
 * Flink connectors: Kafka (mTLS authentication)
 
 This sample illustrates how to configure the Flink Kafka connectors (KafkaSource and KafkaSink) 
-retrieving custom KeyStore and TrustStore at runtime, using Config Providers.
+retrieving custom KeyStore at runtime, using Config Providers.
 More details on the MSK Config Providers in [this repo](https://github.com/aws-samples/msk-config-providers).
 
-* KeyStore and TrustStore are fetched from S3, when the job starts.
-* The passwords to open both KeyStore and TrustStore are also fetched when the job starts, from AWS Secret Manager.
+* KeyStore is fetched from S3, when the job starts.
+* The password to open KeyStore is also fetched when the job starts, from AWS Secret Manager.
 * No secret is packaged with the application.
 
 ### High level approach
@@ -45,16 +45,13 @@ builder.setProperty("config.providers.s3import.class", "com.amazonaws.kafka.conf
 String region = appProperties.get("S3BucketRegion");
 String keystoreS3Bucket = appProperties.get("KeystoreS3Bucket");
 String keystoreS3Path = appProperties.get("KeystoreS3Path");
-String truststoreS3Bucket = appProperties.get("TruststoreS3Bucket");
-String truststoreS3Path = appProperties.get("TruststoreS3Path");
 String keystorePassSecret = appProperties.get("KeystorePassSecret");
 String keystorePassSecretField = appProperties.get("KeystorePassSecretField");
 
 // region, etc..
 builder.setProperty("config.providers.s3import.param.region", region);
 
 // properties
-builder.setProperty("ssl.truststore.location", "${s3import:" + region + ":" + truststoreS3Bucket + "/" + truststoreS3Path + "}");
 builder.setProperty("ssl.keystore.type", "PKCS12");
 builder.setProperty("ssl.keystore.location", "${s3import:" + region + ":" + keystoreS3Bucket + "/" + keystoreS3Path + "}");
 builder.setProperty("ssl.keystore.password", "${secretsmanager:" + keystorePassSecret + ":" + keystorePassSecretField + "}");
@@ -73,7 +70,7 @@ Access Policy/Role associated with the application that is running a config prov
 
 When running on Amazon Managed Service for Apache Flink the runtime configuration is read from *Runtime Properties*.
 
-When running locally, the configuration is read from the [`resources/flink-application-properties-dev.json`](resources/flink-application-properties-dev.json) file located in the resources folder.
+When running locally, the configuration is read from the [`resources/flink-application-properties-dev.json`](src/main/resources/flink-application-properties-dev.json) file located in the resources folder.
 
 Runtime parameters:
 
@@ -82,11 +79,9 @@ Runtime parameters:
 | `Input0` | `bootstrap.servers`     |             | kafka cluster boostrap servers                                     |
 | `Input0` | `topic`                 | `source`    | source topic name                                                  |
 | `Input0` | `group.id`              | `flink-app` | kafka consumer group id                                            |
-| `Input0` | `bucket.region`         |             | region of the S3 bucket(s) containing the keystore and truststore  |
+| `Input0` | `bucket.region`         |             | region of the S3 bucket(s) containing the keystore  |
 | `Input0` | `keystore.bucket`       |             | name of the S3 bucket containing the keystore                      |
 | `Input0` | `keystore.path`         |             | path to the keystore object, omitting any trailing `/`             |
-| `Input0` | `truststore.bucket`     |             | name of the S3 bucket containing the truststore                    |
-| `Input0` | `truststore.path`       |             | path to the truststore object, omitting any trailing `/`           |
 | `Input0` | `keystore.secret`       |             | SecretManager secret ID  containing the password of the keystore   |
 | `Input0` | `keystore.secret.field` |             | SecretManager secret key containing the password of the keystore   |
 
@@ -99,4 +94,4 @@ All parameters are case-sensitive.
 
 You can run this example directly in IntelliJ, without any local Flink cluster or local Flink installation.
 
-See [Running examples locally](../running-examples-locally.md) for details.
+See [Running examples locally](../../running-examples-locally.md) for details.

java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/docs/step-by-step.md

@@ -22,9 +22,9 @@ Also add a self-reference inbound rule to the Security Group for port 9094, that
 
 ### 2. Setup client bastion
 
-We can use the AWS Cloud9 environment and `KafkaClientEC2Instance` setup as described in the [MSK Labs Workshop](https://catalog.workshops.aws/msk-labs/en-US/overview/setup) for testing.
+We can setup the Kafka client EC2 instance using the steps described in the [MSK Labs Workshop](https://catalog.workshops.aws/msk-labs/en-US/overview/prerequisites) for testing.
 
-SSH into `KafkaClientEC2Instance` from Cloud9 terminal.
+Connect to the Kafka client EC2 instance via SSM.
 
 Export the brokers and zookeeper connection strings for the Amazon MSK cluster-
 ```
@@ -109,13 +109,12 @@ bin/kafka-acls.sh --authorizer-properties zookeeper.connect=$zkeeper --add --all
 
 ### 6. Setup Amazon S3 Bucket to be used by Amazon Managed Service for Apache Flink
 
-Create an S3 bucket in the required AWS region e.g. us-east-1. We will use this bucket e.g. `kafkaclientstore` to get the keystore/truststore and the Flink application jar in subsequent steps.
+Create an S3 bucket in the required AWS region e.g. us-east-1. We will use this bucket e.g. `kafkaclientstore` to get the keystore and Flink application jar in subsequent steps.
 
-Now copy the keystore and truststore from `/tmp` to this S3 bucket
+Now copy the keystore from `/tmp` to this S3 bucket
 ```
 cd /tmp
 aws s3 cp kafka.client.keystore.jks s3://kafkaclientstore
-aws s3 cp kafka.client.truststore.jks s3://kafkaclientstore
 aws s3 cp private_key.pem s3://kafkaclientstore
 aws s3 cp client_cert.pem s3://kafkaclientstore
 ```
@@ -165,11 +164,9 @@ Under `Runtime properties`, add the following properties:
 | `Input0` | `bootstrap.servers`     |             | kafka cluster boostrap servers                                     |
 | `Input0` | `topic`                 | `source`    | source topic name                                                  |
 | `Input0` | `group.id`              | `flink-app` | kafka consumer group id                                            |
-| `Input0` | `bucket.region`         |             | region of the S3 bucket(s) containing the keystore and truststore  |
+| `Input0` | `bucket.region`         |             | region of the S3 bucket(s) containing the keystore  |
 | `Input0` | `keystore.bucket`       |             | name of the S3 bucket containing the keystore                      |
 | `Input0` | `keystore.path`         |             | path to the keystore object, omitting any trailing `/`             |
-| `Input0` | `truststore.bucket`     |             | name of the S3 bucket containing the truststore                    |
-| `Input0` | `truststore.path`       |             | path to the truststore object, omitting any trailing `/`           |
 | `Input0` | `keystore.secret`       |             | SecretManager secret ID  containing the password of the keystore   |
 | `Input0` | `keystore.secret.field` |             | SecretManager secret key containing the password of the keystore   |
 

java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/docs/troubleshoot-guide.md

@@ -14,8 +14,6 @@ Few important points to consider that will help troubleshoot or prevent any pote
 
 2. While configuring Managed Apache Flink application, please ensure that the VPC connectivity, Subnets and Security Groups (under ‘Networking’ section) are correctly selected and are allowing access to the required resources, e.g. Kafka cluster and brokers. Depending on the setup e.g. for mTLS you may need to add a self-reference inbound rule to the security group for port 9094. Also, check if there are any Kafka ACLs set on the respective topic(s) for authorization and if the required operation(s) have the `ALLOW` permissionType.  
 
-3. While configuring the Runtime properties for the Apache Flink application, please ensure that the values for `keystore.bucket` and `truststore.bucket` do not contain the prefix `s3://` . This is different from `Application code location` section where the specified Amazon S3 bucket needs to have the format `s3://bucket`.Also, the path to S3 object(s) e.g. `keystore.path` doesn't need a trailing slash. For e.g., if the keystore `kafka.client.keystore.jks` is stored within the S3 bucket `my-bucket`, then the runtime properties can be as follows:
-
-    ![Runtime Properties](../images/flink-keystore-s3-example.png)
+3. While configuring the Runtime properties for the Apache Flink application, please ensure that the value for `keystore.bucket` does not contain the prefix `s3://` . This is different from `Application code location` section where the specified Amazon S3 bucket needs to have the format `s3://bucket`.Also, the path to S3 object(s) e.g. `keystore.path` doesn't need a trailing slash.
 
 4. When running the Apache Flink application, if you are getting a `SecretsManagerException` with Status Code 400 (e.g. not authorized to perform: `secretsmanager:GetSecretValue` on resource: SSL_KEYSTORE_PASS because no identity-based policy allows the `secretsmanager:GetSecretValue` action), please make sure that the IAM Role for the application has the necessary permission policy for SecretsManager.

java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/images/flink-keystore-s3-example.png

@@ -16,7 +16,7 @@
         <target.java.version>11</target.java.version>
         <maven.compiler.source>${target.java.version}</maven.compiler.source>
         <maven.compiler.target>${target.java.version}</maven.compiler.target>
-        <flink.version>1.19.1</flink.version>
+        <flink.version>1.20.0</flink.version>
         <kafka.connector.version>3.2.0-1.19</kafka.connector.version>
         <kda.runtime.version>1.2.0</kda.runtime.version>
         <jackson.databind.version>2.13.4.2</jackson.databind.version>

java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/pom.xml

@@ -26,16 +26,14 @@ public class StreamingJob {
     private static final String DEFAULT_SOURCE_TOPIC = "source";
     private static final String DEFAULT_CONSUMER_GROUP = "flink-app";
 
-    public static final String KAFKA_SOURCE_TOPIC_KEY = "topic";
-    public static final String MSKBOOTSTRAP_SERVERS_KEY = "bootstrap.servers";
-    public static final String KAFKA_CONSUMER_GROUP_ID_KEY = "group.id";
-    public static final String S3_BUCKET_REGION_KEY = "bucket.region";
-    public static final String KEYSTORE_S3_BUCKET_KEY = "keystore.bucket";
-    public static final String KEYSTORE_S3_PATH_KEY = "keystore.path";
-    public static final String TRUSTSTORE_S3_BUCKET_KEY = "truststore.bucket";
-    public static final String TRUSTSTORE_S3_PATH_KEY = "truststore.path";
-    public static final String KEYSTORE_PASS_SECRET_KEY = "keystore.secret";
-    public static final String KEYSTORE_PASS_SECRET_FIELD_KEY = "keystore.secret.field";
+    private static final String KAFKA_SOURCE_TOPIC_KEY = "topic";
+    private static final String MSKBOOTSTRAP_SERVERS_KEY = "bootstrap.servers";
+    private static final String KAFKA_CONSUMER_GROUP_ID_KEY = "group.id";
+    private static final String S3_BUCKET_REGION_KEY = "bucket.region";
+    private static final String KEYSTORE_S3_BUCKET_KEY = "keystore.bucket";
+    private static final String KEYSTORE_S3_PATH_KEY = "keystore.path";
+    private static final String KEYSTORE_PASS_SECRET_KEY = "keystore.secret";
+    private static final String KEYSTORE_PASS_SECRET_FIELD_KEY = "keystore.secret.field";
 
     private static final String LOCAL_APPLICATION_PROPERTIES_RESOURCE = "flink-application-properties-dev.json";
 
@@ -90,16 +88,13 @@ private static void configureConnectorPropsWithConfigProviders(KafkaSourceBuilde
         String region = appProperties.getProperty(S3_BUCKET_REGION_KEY);
         String keystoreS3Bucket = appProperties.getProperty(KEYSTORE_S3_BUCKET_KEY);
         String keystoreS3Path = appProperties.getProperty(KEYSTORE_S3_PATH_KEY);
-        String truststoreS3Bucket = appProperties.getProperty(TRUSTSTORE_S3_BUCKET_KEY);
-        String truststoreS3Path = appProperties.getProperty(TRUSTSTORE_S3_PATH_KEY);
         String keystorePassSecret = appProperties.getProperty(KEYSTORE_PASS_SECRET_KEY);
         String keystorePassSecretField = appProperties.getProperty(KEYSTORE_PASS_SECRET_FIELD_KEY);
 
         // region, etc..
         builder.setProperty("config.providers.s3import.param.region", region);
 
         // properties
-        builder.setProperty("ssl.truststore.location", "${s3import:" + region + ":" + truststoreS3Bucket + "/" + truststoreS3Path + "}");
         builder.setProperty("ssl.keystore.type", "PKCS12");
         builder.setProperty("ssl.keystore.location", "${s3import:" + region + ":" + keystoreS3Bucket + "/" + keystoreS3Path + "}");
         builder.setProperty("ssl.keystore.password", "${secretsmanager:" + keystorePassSecret + ":" + keystorePassSecretField + "}");

java/KafkaConfigProviders/Kafka-mTLS-Keystore-ConfigProviders/src/main/java/com/amazonaws/services/msf/StreamingJob.java

@@ -0,0 +1 @@
+local-repo/**/*.jar