Getting started with Amazon MSK

Amazon MSK is a fully managed service that enables you to build and run applications that use Apache Kafka to process streaming data. ## Create MSK Cluster ### Via Console Please check out [here](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html) ### Via Amazon CDK The source code is available [here](https://github.com/wingkwong/aws-playground/tree/master/msk/cdk) Using CDK, we will create a VPC for Your MSK Cluster with a Single Public Subnet where you can modify ``lib/msk-stack.ts`` to meet your requirements. ``` const vpc = new Vpc(this, 'AWSKafkaVPC', { cidr: '10.0.0.0/16', maxAzs: 3, subnetConfiguration: [ { cidrMask: 24, name: 'kafka', subnetType: SubnetType.PUBLIC } ] }); ``` and then CDK will provision a MSK Cluster which is also defined in ``lib/msk-stack.ts``. ``` const cluster = new CfnCluster(this, 'mskCluster', { clusterName: 'AWSKafkaCluster', kafkaVersion: KafkaVersion.VERSION_2_3_1, encryptionInfo: { encryptionInTransit: { inCluster: true, clientBroker: 'TLS' } }, numberOfBrokerNodes: 2, brokerNodeGroupInfo: { clientSubnets: [ vpc.publicSubnets[0].subnetId, vpc.publicSubnets[1].subnetId, vpc.publicSubnets[2].subnetId ], brokerAzDistribution: 'DEFAULT', instanceType: KafkaInstanceType.T3_SMALL, storageInfo: { ebsStorageInfo: { volumeSize: 10 } } } }); ``` Copy .env.sample and paste as .env and update the environment varibles. ``` CDK_DEFAULT_REGION=XXXXXXXXXXXXXXXXXXXX CDK_DEFAULT_ACCOUNT=XXXXXXXXXXXXXXXXXXX ``` Run ``npm run build`` to compile typescript to js Run ``cdk deploy`` to deploy this stack to your default AWS account/region ## Create a bastion machine You need a machine to create a topic that produces and consumes data. Let's create a t2.xlarge instance of Amazon Linux 2 AMI (HVM), SSD Volume Type with Public IP enabled. ## Create Kafka Topic on the bastion machine ``` #!/bin/sh zookeeperConnectString="" # retrieved from "View Client Information" in Amazon MSK Console kafka_topic="" replication_factor=1 partitions=1 # Change directory to Kafka bin cd ~/kafka_2.13-2.6.0/bin/ # Execute kafka-topics.sh ./kafka-topics.sh --create --zookeeper $zookeeperConnectString --replication-factor $replication_factor --partitions $partitions --topic $kafka_topic ``` ## Produce Data Here's a sample producer ``` from time import sleep from json import dumps from kafka import KafkaProducer # Define Amazon MSK Brokers brokers=[':9092', ':9092'] # Define Kafka topic to be produced to kafka_topic='' # A Kafka client that publishes records to the Kafka cluster producer = KafkaProducer(bootstrap_servers=brokers, value_serializer=lambda x: dumps(x).encode('utf-8')) # To produce 1000 numbers from 0 to 999 for num in range(1000): data = {'number' : num} producer.send(kafka_topic, value=data) sleep(1) ``` ## Consume Data Here's a sample consumer ``` from kafka import KafkaConsumer from json import loads # Define Amazon MSK Brokers brokers=[':9092', ':9092'] # Define Kafka topic to be consumed from kafka_topic='' # A Kafka client that consumes records from a Kafka cluster consumer = KafkaConsumer( kafka_topic, bootstrap_servers=brokers, auto_offset_reset='earliest', enable_auto_commit=True, group_id='my-group', value_deserializer=lambda x: loads(x.decode('utf-8'))) for message in consumer: message = message.value print('{}'.format(message)) ```

Writing middlewares using Middy for AWS Lambda

[Middy](https://github.com/middyjs/middy) is the stylish Node.js middleware engine for AWS Lambda. It allows us to focus on the strict business logic of your Lambda and attach common modular elements such as authentication, authorization, validation etc. To install middy via NPM ``` npm install --save @middy/core ``` Middy supports multiple middlewares, we can create a new folder called ``middlewares`` and define the middlewares there. Let's create ``middlewares/index.js``. In this file, it includes other middlewares. You can treat it as an entry point of all middlewares. ```js const middy = require('@middy/core') const middleware1 = require("./middleware1"); const middleware2 = require("./middleware2"); const middleware3 = require("./middleware3"); const middlewares = [ middleware1(), middleware2(), middleware3() ]; module.exports = { middlewares }; ``` For example, if you wanna update your handler header for each Lambda function, you can create a middleware like ```js const NEW_RESPONSE_HEADER = { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Credentials": true, "Access-Control-Allow-Methods": "POST,GET", "Access-Control-Allow-Headers": "*", "Strict-Transport-Security": "max-age= 63072000", "X-Content-Type-Options": "nosniff", "X-Frame-Options": "DENY", "X-XSS-Protection": "1; mode=block", "Referrer-Policy": "same-origin", }; const updateHandlerHeaders = (handler) => { if (!handler.response) { handler.response = {}; } handler.response.headers = { ...RESPONSE_SECURITY_HEADER, ...handler.response.headers, }; }; const headers = () => ({ before: async (handler) => updateHandlerHeaders(handler), after: async (handler) => updateHandlerHeaders(handler), onError: async (handler) => updateHandlerHeaders(handler), }); module.exports = headers; ``` In your ``handle.js``, import core and your middlewares ```js const middy = require('@middy/core') const { middlewares }= require("../middleware"); ``` Then define your business logic ```js const func = (event, context, callback) => { const { data } = event.body // business logic goes here return callback(null, { result: 'success', message: 'hello world'}) } const handler = middy(func).use(middlewares) ``` Middlewares have two phases - ``before`` and ``after``. Therefore, the order does matter. In the previous example, there are three middlewares, the expected order of execution is ``` middleware1 (before) middleware2 (before) middleware3 (before) handler middleware3 (after) middleware2 (after) middleware1 (after) ``` By using Middy, we can address some common concerns like setting CORS headers and keep our business logic clean. For more details, please check out [here](https://github.com/middyjs/middy).