Merge pull request #2 from Rathan8/main

removed $ char & adjusted timeouts for init conn

Author: nwheeler81

README.MD

@@ -25,27 +25,27 @@ to run workloads with variable traffic patterns rather than traditional approach
 
 Manually set the AWS region to use create the stack
 ```
-$export AWS_REGION=us-east-1
+export AWS_REGION=us-east-1
 ```
 or use the configured region, to check current region
 ```
-$export AWS_REGION=$(aws configure get region)
+export AWS_REGION=$(aws configure get region)
 ```
 Add environment variables for your AWS Region and the account ID from your AWS configuration
 ```
-$export AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
+export AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
 ```
 First use aws-cli to create s3 bucket to store the cfn template, custom plugins which makes the post deployment cleanup easier
 ```
-$aws s3api create-bucket --bucket blog-eks-msk-aks-$AWS_ACCOUNT_ID
+aws s3api create-bucket --bucket blog-eks-msk-aks-$AWS_ACCOUNT_ID
 ```
 Switch to `Templates` folder deploy the cfn-eks-msk-aks.yml, the template creates the AWS resources. 
 Deploying the CFN does requires you to pass parameters for SSH key (pass the key pair that already exists in the aws account else need to create a new one) 
 and IP address range to access the Kafka client instance.
 
 ```
-$cd Templates
-$aws cloudformation deploy --template-file cfn-eks-msk-aks.yml --stack-name eks-msk-aks-sink-stack --parameter-overrides KeyName= < SSH-KEY-Name >   SSHLocation=< ip-address > --tags aws-blog=eks-msk-aks-sink --s3-bucket blog-eks-msk-aks-$AWS_ACCOUNT_ID  --capabilities CAPABILITY_NAMED_IAM  --on-failure ROLLBACK
+cd Templates
+aws cloudformation deploy --template-file cfn-eks-msk-aks.yml --stack-name eks-msk-aks-sink-stack --parameter-overrides KeyName= < SSH-KEY-Name >   SSHLocation=< ip-address > --tags aws-blog=eks-msk-aks-sink --s3-bucket blog-eks-msk-aks-$AWS_ACCOUNT_ID  --capabilities CAPABILITY_NAMED_IAM  --on-failure ROLLBACK
 ```
 KeyName is ssh key name to be used for the kafkaclient instance. SSHLocation is ip address range which allow ssh access (default is 0.0.0.0/0).
 
@@ -60,51 +60,51 @@ The CloudFormation stack will create the following resources:
 
 Run the following command to get the output of the stack created and save it in stack_resources_output file
 ```
-$aws cloudformation describe-stacks --stack-name eks-msk-aks-sink-stack --query "Stacks[0].Outputs[*].[OutputKey,OutputValue]" --output text > stack_resources_output
+aws cloudformation describe-stacks --stack-name eks-msk-aks-sink-stack --query "Stacks[0].Outputs[*].[OutputKey,OutputValue]" --output text > stack_resources_output
 ```
 
 Based on stack_resources_output the helper _update-param_ script updates the parameters in template files used for pod and connector deployment
 ```
-$sh update-param.sh
+sh update-param.sh
 ```
 So far you have deployed the stack using CFN template. Used the output of stack to update parameters for pod and connector deployment templates
 ## Build and Deploy Event source application on EKS
 
 Run the following command to log in in the ECR registry
 ```
-$cd ..
-$aws ecr get-login-password --region $AWS_REGION | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com
+cd ..
+aws ecr get-login-password --region $AWS_REGION | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com
 ```
 
 Run the following command to build docker container image, add a tag, and push the container image to the ECR registry
 ```
-$mvn package
-$docker-compose build
-$docker tag amazon-keyspaces-with-apache-kafka:latest  $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/amazon-keyspaces-with-apache-kafka:latest
-$docker push $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/amazon-keyspaces-with-apache-kafka:latest
+mvn package
+docker-compose build
+docker tag amazon-keyspaces-with-apache-kafka:latest  $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/amazon-keyspaces-with-apache-kafka:latest
+docker push $AWS_ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com/amazon-keyspaces-with-apache-kafka:latest
 ```
 
 aws eks update-kubeconfig command updates the default kubeconfig file to use eks-twitter-cluster as the current context
 ```
-$aws eks update-kubeconfig --name eks-twitter-cluster 
+aws eks update-kubeconfig --name eks-twitter-cluster 
 ```
 Run the following kubectl commands to create namespace, verify if namespace is created
 ```
-$kubectl create namespace eks-msk-twitter
-$kubectl get namespaces
+kubectl create namespace eks-msk-twitter
+kubectl get namespaces
 ```
 
 Switch to `Templates` folder
 ```
-$cd Templates
+cd Templates
 ```
 You need to replace value of bearer token in the _twitter-cfn-app-deployment.yaml_ template for twitter access, deploy the pods to eks-msk-twitter namespace
 
 _<< Bearer Token >>_, Run the following commands to deploy the pods and verify the pod status
 
 ```
-$kubectl apply -f twitter-cfn-app-deployment.yaml
-$kubectl get pods -n=eks-msk-twitter -o wide
+kubectl apply -f twitter-cfn-app-deployment.yaml
+kubectl get pods -n=eks-msk-twitter -o wide
 ```
 
 Now the application is deployed, and it starts reading sample stream data, parse tweet data and write it to the kafka topic, next step would be to create the msk connector to sink data from the kafka topic to Amazon keyspaces
@@ -116,7 +116,7 @@ Make changes to connector configuration in kafka-keyspaces-connector.json file
 Replace the username and password to connect to Amazon Keyspaces in _kafka-keyspaces-connector.json_ file  "auth.username": "< keyspaces-user-at > ", "auth.password": “< password >",
 Run the following command to deploy the connector
 ```
-$aws kafkaconnect create-connector --cli-input-json file://kafka-keyspaces-connector.json
+aws kafkaconnect create-connector --cli-input-json file://kafka-keyspaces-connector.json
 ```
 After deploy is completed the sink connector uses interface VPC endpoints to connect and write messages from the kafka topic to Amazon Keyspaces tables
 Now the data pipeline is complete, you can use the cqlsh-expansion library which extends the existing cqlsh library with additional helpers, or the Amazon Keyspaces CQL console the read the data from Amazon Keyspaces tables
@@ -125,14 +125,14 @@ Use the following command to connect to Amazon Keyspaces using the cqlsh-expansi
 
 To install and confgure the cqlsh-expansion utility
 ```
-$pip install --user cqlsh-expansion
-$cqlsh-expansion.init
+pip install --user cqlsh-expansion
+cqlsh-expansion.init
 ```
 
 To connect to Amazon Keyspaces to check the data 
 
 ```
-$cqlsh-expansion cassandra.$AWS_REGION.amazonaws.com 9142 --ssl --auth-provider "SigV4AuthProvider"
+cqlsh-expansion cassandra.$AWS_REGION.amazonaws.com 9142 --ssl --auth-provider "SigV4AuthProvider"
 ```
 ### Sample CQL Queries to get data from Amazon Keyspaces 
 CQL Query to get sample data from tweet_by_tweet_id and tweet_by_user tables
@@ -158,7 +158,7 @@ Run delete-stack bash script to delete all the resources, created as part of thi
 
 Run the following command to delete the stack
 ```
-$./delete-stack.sh
+./delete-stack.sh
 
 ```
 ## Summary

Templates/kafka-keyspaces-connector.json

@@ -45,9 +45,8 @@
         "datastax-java-driver.basic.load-balancing-policy.local-datacenter":"keyspaces_dc",
         "datastax-java-driver.advanced.retry-policy.class": "DefaultRetryPolicy",
         "datastax-java-driver.basic.default-idempotence": "true",
-        "datastax-java-driver.advanced.connection.init-query-timeout": "10",
-        "datastax-java-driver.advanced.control-connection.timeout": "10",
-        "datastax-java-driver.basic.consistency": "LOCAL_QUORUM",
+        "datastax-java-driver.advanced.connection.init-query-timeout": "500",
+        "datastax-java-driver.advanced.control-connection.timeout": "500",
         "auth.provider":"PLAIN",
         "auth.username": "< keyspaces-user-at >",
         "auth.password": "< password >",
@@ -58,11 +57,13 @@
         "connectionPoolLocalSize":"3",
         "maxNumberOfRecordsInBatch": "1",
         "ssl.hostnameValidation": "false",
-        "queryExecutionTimeout":"10",
+        "queryExecutionTimeout":"100",
         "topic.twitter_input.aws_blog.tweet_by_tweet_id.deletesEnabled":"false",
+        "topic.twitter_input.aws_blog.tweet_by_tweet_id.consistency":"false",
         "topic.twitter_input.aws_blog.tweet_by_tweet_id.query":"INSERT INTO aws_blog.tweet_by_tweet_id(hashtag,tweet_id,tweet,lang,username,tweet_time) VALUES (:tag,:id,:text,:lang,:username,:timestamp) USING TTL 259200;",
         "topic.twitter_input.aws_blog.tweet_by_tweet_id.mapping": "tag=value.hashtag, text=value.text, id=value.id, timestamp=value.createdAt, username=value.username, lang=value.lang",
         "topic.twitter_input.aws_blog.tweet_by_user.deletesEnabled":"false",
+        "topic.twitter_input.aws_blog.tweet_by_user.consistency":"false",
         "topic.twitter_input.aws_blog.tweet_by_user.query":"INSERT INTO aws_blog.tweet_by_user(hashtag,tweet_id,tweet,lang,username,tweet_time) VALUES (:tag,:id,:text,:lang,:username,:timestamp) USING TTL 259200;",
         "topic.twitter_input.aws_blog.tweet_by_user.mapping": "tag=value.hashtag, text=value.text, id=value.id, timestamp=value.createdAt, username=value.username, lang=value.lang"   
     }