Field Notes: Deliver Messages Using an IoT Rule Action to Amazon Managed Streaming for Apache Kafka
AWS Architecture BlogField Notes: Deliver Messages Using an IoT Rule Action to Amazon Managed Streaming for Apache Kafkaby Siddhesh Keluskar | on29 JUL 2021| inAmazon Managed Streaming for Apache Kafka (Amazon MSK),Architecture,Technical How-to|Permalink|ShareWith IoT devices scaling up rapidly, real-time data integration and data processing has become a major challenge. This is why customers often chooseMessage Queuing Telemetry Transport(MQTT) for message ingestion, and Apache Kafka to build a real-time streaming data pipeline.AWS IoT Corenow supports a new IoT rule action to deliver messages from your devices directly to yourAmazon MSKor self-managed Apache Kafka clusters for data analysis and visualization, without you having to write a single line of code.In this post, you learn how to set up a real-time streaming data pipeline for IoT data using AWS IoT Core rule and Amazon Managed Streaming for Apache Kafka. The audience for this post is architects and developers creating solutions to ingest sensor data, and high-volume high-frequency streaming data, and process it using a Kafka cluster. Also, this blog describes the SASL_SSL (using user name and password) method to access your Kafka cluster.Overview of solutionFigure 1 represents an IoT data ingestion pipeline where multiple IoT devices connect to AWS IoT Core. These devices can send messages to AWS IoT Core over MQTT or HTTPS protocol. AWS IoT Core rule for Kafka is configured to intercept messages from the desired topic and route them to the Apache Kafka cluster. These messages can then be received by multiple consumers connected to the Kafka cluster. In this post, we will use AWS Python SDK to represent IoT devices and publish messages.Figure 1 – Architecture representing an IoT ingestion pipelinePrerequisitesFamiliarity with MQTT protocol andKafka command line interface (CLI)AWS accountwith console accessWalkthroughI will show you how to stream AWS IoT data on an Amazon MSK cluster using AWS IoT Core rules and SASL_SSL SCRAM-SHA-512 mechanism of authentication. Following are the steps for this walkthrough:Create an Apache Kafka cluster usingAmazon MSK.Configure an Apache Kafka cluster for SASL_SSL authentication.Set up a Kafka producer and consumer on AWS Cloud9 to test the setup.Configure an IoT Rule action to send a message to Kafka.1. Create an Apache Kafka cluster using Amazon MSKThe first step is to create an Apache Kafka cluster. Open the service page for Amazon MSK by signing in to your AWS account.ChooseCreate Cluster, and selectCustom Create. AWS IoT Core supports SSL and SASL_SSL based authentication for Amazon MSK. We are using custom settings to configure these authentication methods.Figure 2 – Screenshot showing how to create an MSK cluster.Assign a cluster name, and select Apache Kafka (version of your choice), for this walkthrough, we are using 2.6.1.Keep the configuration asAmazon MSK default configuration. Choose yourNetworking components: VPC, number of Availability Zones (a minimum of two is required for high availability), and subnets.ChooseSASL/SCRAM authentication(default selection isNone).Use the encryption settings as shown in the following screenshot:Figure 3 – Screenshot showing Encryption SettingsKeep the monitoring settings asBasic Monitoring, and ChooseCreate Cluster.It takes approximately 15–20 minutes for the cluster to be created.2. Configure an Apache Kafka cluster for SASL_SSL authenticationWhen the Apache Kafka cluster is available, we must then configure authentication for producers and consumers.OpenAWS Secrets Manager, chooseStore a new secret, and then chooseOther type of secrets.Enter user name and password as two keys, and assign the user name and password values of your choice.Figure 4 – Screenshot showing how to store a new secretNext, selectAdd new key link.Note:Do not selectDefaultEncryptionKey! A secret created with the default key cannot be used with an Amazon MSK cluster. Only aCustomer managed keycan be used as an encryption key for an Amazon MSK–compatible secret.To add a new key, selectCreate key, selectSymmetrickey, and chooseNext.Type anAlias, and chooseNext.Select appropriate users as Key administrators, and choose Next.Review the configuration, and selectFinish.Figure 5 – Select the newly-created Customer Managed Key as the encryption keyFigure 6 – Specify the key value pair to be stored in this secretSelect the newly-created Customer Managed Key as theencryption key, and chooseNext.Provide aSecret name(Secret name must start with AmazonMSK_ for Amazon MSK cluster to recognize it), for example, AmazonMSK_SECRET_NAME.ChooseNexttwice, and then chooseStore.Figure 7 – Storing a new secretOpen theAmazon MSKservice page, and select your Amazon MSK cluster. ChooseAssociate Secrets, and then selectChoose secrets(this will only be available after the cluster is created and in Active Status).Choose the secret we created in the previous step, and chooseAssociate secrets. Only the secret name starting withAmazonMSK_will be visible.3. Set up Kafka producer and consumer on AWS Cloud9 to test the setupTo test if the cluster and authentication is correctly setup, we use Kafka SDK onAWS Cloud9IDE.ChooseCreate environment, and follow the console to create a new AWS Cloud9 environment. You can use an existing AWS Cloud9 environment, in addition to an environment with Kafka consumer and producer already configured.This blog requires Java 8 or earlier.Verify your version of Java with the command:java -version. Next, add your AWS Cloud9 instanceSecurity Groupto inbound rules of your Kafka cluster.Open the Amazon MSK page and select your cluster, then chooseSecurity groups applied.Figure 8 – Selecting Security Groups AppliedNext, chooseInbound rules, and then chooseEdit inbound rules.ChooseAdd rule, and addCustom TCP ports2181 and 9096 with Security Group of your AWS Cloud9 instance.Figure 9 – Screenshot showing rules appliedThe Security Group for your AWS Cloud9 can be found in the Environment details section of your AWS Cloud9 instance.Figure 10 – Screenshot showing Edit Inbound Rules addedUsePort rangevalues as per the client information section of your Bootstrap server and Zookeeper connection.Figure 11 – Screenshot showing where to access ‘View client information’Figure 12 – Screenshot showing client integration informationInvoke the following commands on AWS Cloud9 console to download and extract Kafka CLI tools:wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz
tar -xzf kafka_2.12-2.2.1.tgz
cd kafka_2.12-2.2.1/
mkdir client && cd clientNext, create a file users_jass.conf, and add the user name and password that you added in Secrets Manage:sudo nano users_jaas.confPaste the following configuration andsave. Verify the user name and passwords are the same as saved in Secrets Manager.KafkaClient {
   org.apache.kafka.common.security.scram.ScramLoginModule required
   username="hello"
   password="world";
};Invoke the following commands:export KAFKA_OPTS=-Djava.security.auth.login.config=$PWD/users_jaas.confCreate a new file with nameclient_sasl.properties.sudo nano client_sasl.propertiesCopy the following content to file:security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
ssl.truststore.location=<path-to-keystore-file>/kafka.client.truststore.jks<path-to-keystore-file>can be retrieved by running following command:cd ~/environment/kafka_2.12-2.2.1/client
echo $PWDNext, copy the cacerts file from your Javalibfolder toclientfolder. The path of Java lib folder might be different based on your version of Java.cd ~/environment/kafka_2.12-2.2.1/client
cp /usr/lib/jvm/java-11-amazon-corretto.x86_64/lib/security/cacerts kafka.client.truststore.jksFigure 13 – Screenshot showing client integration informationSave the previous endpoints asBOOTSTRAP_SERVERandZOOKEEPER_STRING.export BOOTSTRAP_SERVER=b-2.iot-demo-cluster.slu5to.c13.kafka.us-east-1.amazonaws.com:9096,b-1.iot-demo-cluster.slu5to.c13.kafka.us-east-1.amazonaws.com:9096
export ZOOKEEPER_STRING=z-1.iot-demo-cluster.slu5to.c13.kafka.us-east-1.amazonaws.com:2181,z-3.iot-demo-cluster.slu5to.c13.kafka.us-east-1.amazonaws.com:2181,z-2.iot-demo-cluster.slu5to.c13.kafka.us-east-1.amazonaws.com:2181Save theTopic namein an environment variable.TOPIC="AWSKafkaTutorialTopic"Next, create a new Topic using the Zookeeper String.cd ~/environment/kafka_2.12-2.2.1
bin/kafka-topics.sh --create --zookeeper $ZOOKEEPER_STRING --replication-factor 2 --partitions 1 --topic $TOPICConfirm that you receive the message:Created topic AWSKafkaTutorialTopic.Start Kafka producer by running this command in your Kafka folder:cd ~/environment/kafka_2.12-2.2.1

bin/kafka-console-producer.sh --broker-list $BOOTSTRAP_SERVER --topic $TOPIC --producer.config client/client_sasl.propertiesNext, open a new Terminal by pressing the + button, and initiate the following commands to configure the environment variables:export BOOTSTRAP_SERVER=b-2.iot-demo-cluster.slu5to.c13.kafka.us-east-1.amazonaws.com:9096,b-1.iot-demo-cluster.slu5to.c13.kafka.us-east-1.amazonaws.com:9096
TOPIC="AWSKafkaTutorialTopic"

cd ~/environment/kafka_2.12-2.2.1/client
export KAFKA_OPTS=-Djava.security.auth.login.config=$PWD/users_jaas.conf

cd ~/environment/kafka_2.12-2.2.1/
bin/kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC --from-beginning --consumer.config client/client_sasl.properties --from-beginningNow that you have a Kafka consumer and producer opened side-by-side, you can type in producer terminal and verify it from the consumer terminal.Figure 14 – Screenshot showing Kafka consumer and producer opened side-by-side4. Configure an IoT Rule action to send a message to KafkaCreate anAWS Identity and Access Management (IAM)rolewith SecretsManager permissions to allow IoT rule to access Kafka KeyStore in AWS Secrets Manager.Sign in to IAM, selectPoliciesfrom the left-side panel, chooseCreate policy.SelectChoose a service, and search forAWS KMS.In Actions, chooseAll AWS KMS actions.SelectAll resourcesin the Resources section, and chooseNext.Name the policyKMSfullAccess, and chooseCreate policy.Select Roles from the left-side panel, chooseCreate Role, then selectEC2from Choose a use case, and chooseNext:Permissions.Assign the policySecretsManagerReadWrite.Note:if you do not select EC2, SecretsManager Policy will be unavailable.Search for and selectSecretsManagerReadWriteandKMSfullAccess Policy.Add tags, type Role name askafkaSASLRole, and chooseCreate Role.After the Role is created, search the newly-created Role name to view the Summary of the role.Choose theTrust relationshipstab, and chooseEdit trust relationship.Enter the following trust relationship:{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "iot.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}ChooseUpdate Trust Policy.Next, create a new AWS IoT Core rule by signing in to theAWS IoT Coreservice. ChooseActfrom the left side-menu, and selectRules.ChooseCreate. Insert details for Name, Description, and Rule query statement, and then chooseAdd action. The following query is used for this post:SELECT * from ‘iot/topic’SelectSend a message to an Apache Kafka cluster. Next, chooseConfigure action.Figure 15 – Screenshot to create a ruleFigure 16 – How to Create a VPC destinationCreate aVPC destination(if you do not already have one).Select the VPC ID of your Kafka cluster. Select a Security Group with access to Kafka cluster Security Group.Choose security group settings of the EC2 instance we created, or the security group of Kafka cluster.ChooseCreate Role, and then selectCreate Destination. It takes approximately 5–10 minutes for the Destination to be Enabled. After the status is Enabled, navigate back to the Rule creation page and select the VPC Destination.EnterAWSKafkaTutorialTopicas Kafka topic (confirm there is no extra space after the topic name, or you will get an error). Do not update Key and Partition boxes.Figure 17 – Screenshot showing how to enter the AWSKafkaTutorialTopicVerify the Security Group of your VPC destination is added to the inbound list for your Kafka cluster.Figure 18 – Showing Security Group for Kafka ClusterFigure 19 -Screenshot showing Inbound Inbound rulesThe first two Custom TCP entries are for AWS Cloud9 security group. The last two entries are for VPC endpoint.Set the Client properties as follows:Bootstrap.server = The TLS bootstrap string for Kafka clustersecurity.protocol = SASL_SSLssl.truststore = EMPTY for Amazon MSK, enter SecretBinary template for self-managed Kafkass.truststore.password = EMPTY for Amazon MSK, enter truststore password for self-managed Kafkasasl.mechanism = SCRAM-SHA-512Replace the secret name with your stored secret name starting withAmazonMSK_, replace the IAM role ARN with your IAM role ARN.The secret and IAM role are created in previous steps of this post. Enter the following template in thesasl.scram.usernamefield to retrieve username from Secrets Manager.${get_secret('AmazonMSK_cluster_secret','SecretString','username','arn:aws:iam::318219976534:role/kafkaSASLRole')}Perform a similar step for sasl.scram.password field:${get_secret('AmazonMSK_cluster_secret','SecretString','password','arn:aws:iam::318219976534:role/kafkaSASLRole')}ChooseAdd action.ChooseCreate rule.Testing the data pipelineOpen MQTT test client from AWS IoT Core page.Publish the message to the MQTT topic that you configured while creating the rule.Keep the consumer session active (created in earlier step). You will see data published on the MQTT topic being streamed to Kafka consumer.Figure 20 – Screenshot showing testing the data pipelineCommon troubleshooting checksConfirm that your:AWS Cloud9 Security Group is added to Amazon MSK Security Group Inbound ruleVPC endpoint Security Group is added to Amazon MSK Security Group Inbound ruleTopic is created in the Kafka clusterIAM role has Secrets Manager and KMS permissionsEnvironment variables are correctly configured in terminalFolder paths have been correctly followedCleaning upTo avoid incurring future changes, delete the following resources:Amazon MSK clusterAWS IoT Core ruleIAM roleSecrets Manager SecretAWS Cloud9 instanceConclusionIn this post, I showed you how to configure an IoT Rule Action to deliver messages to Apache Kafka cluster using AWS IoT Core and Amazon MSK. You can now build a real-time streaming data pipeline by securely delivering MQTT messages to a highly-scalable, durable, and reliable system using Apache Kafka.Field Notes provides hands-on technical guidance from AWS Solutions Architects, consultants, and technical account managers, based on their experiences in the field solving real-world business problems for customers.TAGS:Field NotesSiddhesh KeluskarSiddhesh is a Solutions Architect at AWS, with extensive experience in Internet of Things, Analytics, and Machine Learning. He helps customers build secure, performant, and scalable solutions on AWS.
With IoT devices scaling up rapidly, real-time data integration and data processing has become a major challenge. This is why customers often choose Message Queuing Telemetry Transport (MQTT) for message ingestion, and Apache Kafka to build a real-time streaming data pipeline. AWS IoT Core now supports a new IoT rule action to deliver messages from […]
Siddhesh Keluskar
Amazon Managed Streaming for Apache Kafka (Amazon MSK) Architecture Technical How-to