Preventing Duplicate Executions in Step Function

Описание: Sometime we might have a hard requirement to prevent executing this state machine twice with the same input.

We can use the default feature of start_execution API for Step Fucntion for preventing dupe executions.

Start Execution Request Syntax
---------------------------------------------------

{
"input": "string",
"name": "string",
"stateMachineArn": "string"
}

Cases:
-------------
Invoke Step Functions with a name 'A': the execution will be successfully invoked and is assigned that name.
Invoke Step Functions with the same name 'A': If the execution is running with same name, the Step Functions API will return the same response as the original request.
Invoke Step Functions with the same name 'A': If the execution with name 'A' has completed (failed/successful) or teh Step Function is running with some other name , the Step Functions API returns an error --
Execution Already Exists.

Configure name:
-------------------------
1)In Order Processing Workflow , to avoid processing same order twice , we can use order id as name
2)Apply hashing (e.g. md5) on your json payload and use the hash value as the execution name.


V.V.I.
--------
Execution names can be used after 90 days , so there can be a possibility of reprocessing of same input after 90 days..
In that case , we need a custom solution using a datastore to keep track of execution.

Step 1:
--------------
Create an SNS Topic & create an email subscriber

Step 2:
--------------
Create a Step Function to publish the message in the SNS Topic after some wait time

Step 3:
--------------
Lambda Code:
----------------------------

import json
import boto3
import hashlib

client = boto3.client('stepfunctions')

def lambda_handler(event, context):
TODO implement
print(event)

country_name=json.loads(event['body'])['country']

encoded_country_name = hashlib.md5(country_name.encode()).hexdigest()

input_json=json.dumps({"body":'I love my {}'.format(country_name)})

try:
response = client.start_execution(
stateMachineArn='',
name=encoded_country_name,
input=input_json
)
print(response)

send_back={"execution_arn":response['executionArn']}

return {
'body': send_back
}
except client.exceptions.ExecutionAlreadyExists as e:
return 'Already executed'


Step 4:
----------------------

Create API using API Gateway to trigger the Lambda Funciton

Step Function Json:
-----------------------------------------
{
"Comment": "A description of my state machine",
"StartAt": "Wait",
"States": {
"Wait": {
"Type": "Wait",
"Seconds": 30,
"Next": "SNS Publish"
},
"SNS Publish": {
"Type": "Task",
"Resource": "arn:aws:states:::sns:publish",
"Parameters": {
"Message.$": "$.body",
"TopicArn": "{SNS ARN}"
},
"End": true
}
}
}

Prerequisite:
------------------------
Creating a POST Api | AWS API Gateway Passing Data to AWS Lambda
   • Creating a POST Api | AWS API Gateway Pass...  

To know more about the start_execution API , you can refer this link--
https://docs.aws.amazon.com/step-func...

Boto3 documentation reference--
https://boto3.amazonaws.com/v1/docume...

Learn AWS Step Fucniton from Scratch:
   • AWS Step Functions Simplified  

Check this playlist for more AWS Projects in Big Data domain:
   • Demystifying Data Engineering with Cloud C...  

🙏🙏🙏🙏🙏🙏🙏🙏
YOU JUST NEED TO DO
3 THINGS to support my channel
LIKE
SHARE
&
SUBSCRIBE
TO MY YOUTUBE CHANNEL