3.7 —Configure Security Group Inbound rules for your Kafka cluster.

a. Open the AWS Cloud9 console, and select the View details tab for your created environment.

e. On the Security Groups page, choose the Inbound rules tab, and choose Edit inbound rules.

f. On the Edit inbound rules page, add the copied security group ID as an Inbound rule for the Kafka Security group using Port 9094 and 2181.

3.8 —Create a new topic AWSKafkaTutorialTopic in the Kafka cluster. This topic will be used by producer and consumer to publish and subscribe messages. In the following commands  (if applicable) change the name of the topic that was set as variable in section 3.6.c.

This will create a new topic AWSKafkaTutorialTopic. Verify that the commands returns the following message: Created topic AWSKafkaTutorialTopic.

Note: If you receive a Client Session timed out error appears, check that your cluster Security Group Inbound Rules were configured correctly.

4.1 — In your AWS Cloud9 terminal, run the following commands to create a client folder to store certificate files, and copy Java KeyStore to this folder.

4.7 — Run the following command to use AWS CLI to fetch the certificate using the received Certificate ARN in the previous step. The received response is in JSON format, which is parsed by jq, and saved into a file signed-certificate-from-acm.

4.8 —  Run the following command to import the signed certificate to KeyStore.

a. Enter Yes when prompted in the terminal.  

b. The client file should have 4 files.

• client-cert-sign-request
• kafka.client.keystore.jks
• kafka.client.truststore.jks
• signed-certificate-from-acm
  

4.9 —The MSK cluster that was previously configured requires all client communication to be TLS secured. The file kafka.client.keystore.jks has to be uploaded to Secrets Manager as a SecretBinary to authenticate IoT core rule.

Run the following command to create a Secret named Kafka_Keystore in HAQM Secrets Manager.

You can verify the newly created Secret in the Secrets Manager console.

4.10 —Set up a Kafka consumer in a new Terminal while keeping the producer session active.

Note: A new Terminal can be opened by choosing the + symbol or alt+T in AWS Cloud9 console.

In the new Terminal session,  run the following commands to set the initial variables, and start a console producer.

4.12 —Run the following commands to start a Kafka consumer session.

You should now have a Kafka producer session and consumer session initiated.

4.13 —Type any string and press enter into the producer terminal to publish to the Kafka stream.

5.1 — Create an AWS IAM role with AWS Secrets Manager permissions to allow an AWS IoT Core rule to access the Kafka KeyStore stored in Secrets Manager.

a. Open the IAM console, choose Roles from the left navigation pane, and choose Create role.

b. For Select type of trusted entity, choose AWS service. For Choose a use case, choose EC2, then choose Next: Permissions.

5.2 —On the Create role page, in the search box, type SecretsManagerReadWrite, select the SecretsManagerReadWrite policy, and choose Next:Tags.

5.3 — On the Create role page, in the Add tags (optional) section, add any necessary tags. Then, choose Next: Review.

5.4 —On the Create role page, in the Review section, for Role name, type kafkaRole and optionally add a description. Then, choose Create role.

5.5 —Once the Role is created, choose the newly created kafkaRole. On the Summary page for the role, choose the Trust relationships tab, choose Edit trust relationship, and copy and paste the following details.

5.6 —Then, choose Update Trust Policy.

5.7 —Open the AWS IoT Core console, and in the left navigation pane, choose ACT. Then, choose Rules. Choose Create a rule.

5.8 —On the Create a rule page, type a Name and Description. In the Role query statement section, choose the SQL version, and enter a Rule query statement. Then, choose Add action.

5.9 — On the Select an action page, choose Send a message to an Apache Kafka cluster, then, choose Configure action.

5.10 —On the Configure action page, choose Create a VPC destination.

5.11 — On the Create a VPC destination page, choose the same VPC ID where you created your Kafka cluster.

5.12 — Choose the Subnet IDs, and choose the Security group.

Note: Make sure you select a Security Group with access to Kafka cluster Security Group.

Select the security group settings of the EC2 instance you created previously and/ or the security group of Kafka cluster itself.

5.13 — Choose Create Role, on the Create a new role window, name the role AWSKafkaTutorialTopic. Then, choose Create role.

5.14 — Choose Create Destination.

Note: It takes 5-10 mins for the Destination to be Enabled. Once the status is Enabled, continue back with the Rule creation page.

5.15 — In your web browser, navigate to the previously opened AWS IoT - Rules- Configuration action page. Then, choose the newly created VPC Destination from the drop down, and enter AWSKafkaTutorialTopic for the Kafka topic. Leave Key and Partition empty.

5.16 —On the Configure action page, in the Client properties section, enter the following details.

• Bootstrap.server = The TLS bootstrap string for Kafka cluster
• security.protocol = SSL
• ssl.truststore = Leave EMPTY
  
• ss.truststore.password = Leave EMPTY
• ssl.keystore = ${get_secret('Kafka_Keystore','SecretBinary','arn:aws:iam::[YOUR ROLE ARN FOR IAM]:role/kafkaRole')}
• ssl.keystore.password = myPassword
  
  Note: The password is the variable you entered for myPassword, which was configured in the previous steps.
  
  

5.18 — On the Create a rule page, choose Create rule.

5.19 — Use the following procedure to test the MQTT Stream.

a. In the left navigation pane of AWS IoT Core, choose Test, and choose MQTT test client.

b. On the MQTT test client page, choose the publish to a topic tab.

c. Under Topic name, enter a name. Under Message payload, enter a message.

d. In the active AWS Cloud9 consumer session (created/opened in the previous step), you should see data published from the MQTT topic and streamed to Kafka consumer.

6.1 — Open the AWS IoT console, in the left navigation pane, choose Settings. Then, in the Logs section, choose Manage Logs.

a. In the Log level section, choose Debug (most verbosity) from the drop down, and choose Update. 

6.2 —To create an IoT log level, open the IoT console, and choose Rules. Then, choose the kafkaRole.

6.3 — In the Error action section, choose Add action. Then, choose Republish message to an AWS IoT topic, and choose Configure action.

You can now view your IoT Logs in HAQM Cloudwatch, Log groups, and AWS IoT Logs.

7.10 — Open the Secrets Manager console and in the left navigation pane, choose Secrets.

7.11 — Choose the kafka_keystore secret.

7.12 — In the Secret details section, choose Delete secret.

7.13 — Open the ACM Private CA console.

7.14 — Choose Private CAs.

7.15 — Choose your private CA from the list.

7.16 — If your CA is in the ACTIVE state, you must disable it. On the Actions menu, choose Disable.

7.17 — On the Actions menu, choose Delete.

7.18 — If your CA is in the PENDING_CERTIFICATE, EXPIRED, or DISABLED state, specify a restoration period of seven to 30 days. Then choose Delete.

Note: If your private CA is not in one of these states, it can not be restored later.

7.19 — If you are certain that you want to delete the private CA, choose Permanently delete when prompted. The status of the private CA changes to DELETED.

7.20 — Open the AWS Cloud9 console.

7.21 — In the top navigation bar, choose the AWS Region where the environment is located.

7.22 — In the list of environments, for the environment you want to delete, choose the title of the card for the environment.

7.23 —Choose Delete.

7.24 — In the Delete dialog box, type Delete, and then choose Delete.