Part 3: Orchestrate, Monitor, and Handle Failures in Informatica IDMC Taskflow Runs with AWS Step Functions

Learn how to orchestrate, monitor, and handle Informatica IDMC taskflow runs using AWS Step Functions Standard, AWS Lambda, Amazon S3, EventBridge, DynamoDB, SNS, and CloudWatch. This guide demonstrates durable polling, idempotency, failure handling, timeout management, execution tracking, notifications, monitoring, testing, and production deployment using AWS SAM for a scalable and resilient IDMC integration architecture.

Shireesha Mangannagari

Published Oct 1, 2026

Last modified Oct 1, 2026

From Start-and-forget to a pipeline you can trust

Part 3 closes the loop: every taskflow the S3 trigger starts is now tracked to SUCCESS or FAILED, retried when it is safe, and reported to your team when it is not.

In Part 1 we built a reusable Python SDK for the Informatica IDMC REST API and shipped it as an AWS Lambda layer. In Part 2 we used that layer in a Lambda function that starts an IDMC taskflow whenever a file lands in Amazon S3.

Part 2 has one gap. The Lambda gets back a run ID and exits. Nobody knows if the taskflow finished, failed at 2 a.m., or hung for six hours. Lambda cannot wait for it either, because a single invocation stops at 15 minutes and many taskflows run longer.

In this part we add:

AWS Step Functions to start the taskflow, poll its status on a timer, and branch on the result
Idempotency, so a duplicate S3 event never starts the same load twice
Retries with exponential backoff for transient IDMC and API errors, with session re-authentication handled by the shared SDK layer
A run ledger in Amazon DynamoDB that records every file, run ID, status, IDMC start and end times, duration, and failure reason
Amazon SNS alerts and CloudWatch metrics for failures and timeouts
An AWS SAM template that deploys the whole thing in one command

The code reuses the idmc_sdk layer from Part 1. Where your Part 1 method names differ from the ones below, swap them in; the pattern stays the same.

Architecture: one Step Functions execution per file

Each file that lands in S3 starts its own Step Functions execution, which owns that IDMC run from start to final status.

Part 3 architecture: S3 to EventBridge to Step Functions to IDMC. Every outcome is written to the DynamoDB run ledger; failures and timeouts also go to SNS.

The wait-and-check loop repeats until IDMC reports a final state or polling times out. Every outcome is written to the ledger by one Lambda; failures and timeouts also go to SNS.

Step 1: Teach the SDK layer to read taskflow status

Add one method to the Part 1 client: get_taskflow_status(run_id), which calls IDMC's taskflow status resource and maps the answer to four states the state machine understands.

IDMC exposes taskflow status at GET <serverUrl>/active-bpel/services/tf/status/<runId>, authenticated with the INFA-SESSION-ID header from the v3 login (Informatica docs ). A running taskflow comes back with endTime set to null. Pass subtaskDetails=No when you only need the top-level status, which keeps the response small.

Add this to idmc_sdk/taskflows.py in the layer:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# idmc_sdk/taskflows.py  (additions for Part 3)

from enum import Enum



class RunState(str, Enum):

    RUNNING = "RUNNING"

    SUCCEEDED = "SUCCEEDED"

    FAILED = "FAILED"

    UNKNOWN = "UNKNOWN"



# IDMC returns mixed-case values; normalise before mapping.

_STATUS_MAP = {

    "SUCCESS": RunState.SUCCEEDED,

    "RUNNING": RunState.RUNNING,

    "QUEUED": RunState.RUNNING,

    "FAILED": RunState.FAILED,

    "SUSPENDED": RunState.FAILED,   # treat as needing a human

    "STOPPED": RunState.FAILED,

}



class TaskflowMixin:

    def get_taskflow_status(self, run_id: str, subtask_details: bool = False) -> dict:

        """Return {'state': RunState, 'raw': <IDMC payload>} for a taskflow run."""

        url = f"{self.server_url}/active-bpel/services/tf/status/{run_id}"

        params = {"subtaskDetails": "Yes" if subtask_details else "No"}

        resp = self._request("GET", url, params=params)   # Part 1 helper: adds

                                                          # INFA-SESSION-ID, retries,

                                                          # re-login on 401

        raw = resp.json()

        status = str(raw.get("status", "")).upper()

        if raw.get("status") == "No status available.":

            state = RunState.UNKNOWN          # run not visible yet

        else:

            state = _STATUS_MAP.get(status, RunState.UNKNOWN)

        return {"state": state, "raw": raw}

Three design notes:

UNKNOWN is not FAILED. Right after a start call, IDMC can briefly report no status. The state machine keeps polling on UNKNOWN for a few cycles before giving up.
SUSPENDED is treated as a failure. A suspended taskflow is waiting on an operator, so it should page someone rather than poll forever.
Session handling stays in _request. IDMC sessions expire after a period of inactivity, so the Part 1 helper should re-authenticate once on a 401 and retry. Every Lambda in this part gets that for free.

Rebuild and publish a new layer version, then note the new version ARN for the SAM template:

1
2
3
4
5
6
7
8
9
cd layer && rm -rf python && mkdir python

pip install -r requirements.txt -t python

cp -r ../idmc_sdk python/

zip -r idmc-sdk-layer.zip python

aws lambda publish-layer-version \

  --layer-name idmc-sdk \

  --zip-file fileb://idmc-sdk-layer.zip \

  --compatible-runtimes python3.12 \

  --compatible-architectures arm64

Step 2: Split the Part 2 Lambda into three small functions

The Part 2 function becomes three: start_taskflow claims the file and starts the run, check_status reports where the run is, and record_outcome writes the final result to the ledger. None of them waits; Step Functions does the waiting.

Shared helpers

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# functions/common.py

import os, boto3

from datetime import datetime, timezone

from idmc_sdk import IDMCClient           # Part 1 layer

from idmc_sdk.errors import IDMCAuthError, IDMCHTTPError



ddb = boto3.resource("dynamodb").Table(os.environ["LEDGER_TABLE"])

_client = None



class TransientIdmcError(Exception):

    """Safe to retry: 429, 5xx, timeouts."""



class PermanentIdmcError(Exception):

    """Do not retry: 400, 403, 404, bad taskflow name."""



def client() -> IDMCClient:

    # Reuse across warm invocations; credentials come from Secrets Manager (Part 1).

    global _client

    if _client is None:

        _client = IDMCClient.from_secret(os.environ["IDMC_SECRET_ARN"])

    return _client



def classify(err: Exception) -> Exception:

    if isinstance(err, IDMCHTTPError):

        if err.status_code in (429, 500, 502, 503, 504):

            return TransientIdmcError(str(err))

        return PermanentIdmcError(str(err))

    if isinstance(err, IDMCAuthError):

        return PermanentIdmcError(f"Auth failed: {err}")

    return TransientIdmcError(str(err))    # network blips, timeouts



def now() -> str:

    return datetime.now(timezone.utc).isoformat(timespec="seconds")

How the error names reach Step Functions. When a Python Lambda function raises an unhandled exception, the runtime reports the exception's class name as errorType. Step Functions uses that value as the error name it matches in Retry and Catch, so TransientIdmcError raised here arrives as TransientIdmcError in the state machine. Two rules keep this reliable: raise these classes directly rather than wrapped in another exception, and confirm the names in the execution history during testing (Step 6 shows the command). Duplicate files do not use an exception at all; start_taskflow returns a flag instead, so that branch does not depend on error naming.

start_taskflow

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# functions/start_taskflow.py

import os

from botocore.exceptions import ClientError

from common import ddb, client, classify, now



TASKFLOW = os.environ["TASKFLOW_NAME"]



def object_identity(bucket: str, obj: dict) -> str:

    # Prefer the S3 version ID (versioned buckets); fall back to the ETag.

    if obj.get("version-id"):

        return f"s3://{bucket}/{obj['key']}?versionId={obj['version-id']}"

    return f"s3://{bucket}/{obj['key']}#{obj.get('etag', 'na')}"



def handler(event, context):

    # Step Functions passes the EventBridge "Object Created" event through.

    bucket = event["detail"]["bucket"]["name"]

    obj = event["detail"]["object"]

    key = obj["key"]

    file_id = object_identity(bucket, obj)

    result = {"file_id": file_id, "run_id": None, "taskflow": TASKFLOW,

              "bucket": bucket, "key": key, "duplicate": False}



    # 1. Claim the file: a new object, or one whose last attempt failed to start.

    try:

        ddb.put_item(

            Item={"file_id": file_id, "status": "STARTING",

                  "taskflow": TASKFLOW, "started_at": now(),

                  "execution": event.get("execution_name", "")},

            ConditionExpression="attribute_not_exists(file_id) OR #s = :sf",

            ExpressionAttributeNames={"#s": "status"},

            ExpressionAttributeValues={":sf": "START_FAILED"},

        )

    except ClientError as e:

        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":

            return {**result, "duplicate": True}

        raise



    # 2. Start the taskflow, passing the file location as input parameters.

    try:

        run_id = client().run_taskflow(TASKFLOW, inputs={

            "in_bucket": bucket, "in_key": key})

    except Exception as err:

        # Keep an audit trail; a retry can re-claim a START_FAILED item.

        ddb.update_item(

            Key={"file_id": file_id},

            UpdateExpression="SET #s = :s, error_message = :m, failed_at = :t",

            ExpressionAttributeNames={"#s": "status"},

            ExpressionAttributeValues={":s": "START_FAILED",

                                       ":m": str(err)[:1000], ":t": now()},

        )

        raise classify(err) from err



    ddb.update_item(

        Key={"file_id": file_id},

        UpdateExpression="SET #s = :s, run_id = :r",

        ExpressionAttributeNames={"#s": "status"},

        ExpressionAttributeValues={":s": "RUNNING", ":r": str(run_id)},

    )

    return {**result, "run_id": str(run_id)}

The conditional put_item is the idempotency guard. S3 can deliver the same event more than once, and an operator may re-upload a file by mistake. When the bucket has versioning turned on, the version ID identifies one specific upload, which is the strongest key available.

Without versioning, the function falls back to the ETag as a practical duplicate-detection key. An ETag is not a universal content hash: multipart uploads and SSE-KMS encryption produce ETags that are not an MD5 of the bytes, so identical files can carry different ETags. Treat an ETag match as "the same upload, delivered twice", not "the same content".

A failed start is no longer deleted. The item stays as START_FAILED with the error message, which gives you an audit trail, and the claim condition still lets a Step Functions retry or a corrected upload start the run.

check_status

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# functions/check_status.py

from datetime import datetime

from common import client, classify



def seconds_between(start, end):

    if not start or not end:

        return None

    try:

        s = datetime.fromisoformat(str(start).replace("Z", "+00:00"))

        e = datetime.fromisoformat(str(end).replace("Z", "+00:00"))

        return int((e - s).total_seconds())

    except ValueError:

        return None



def handler(event, context):

    try:

        result = client().get_taskflow_status(event["run_id"])

    except Exception as err:

        raise classify(err) from err



    raw = result["raw"]

    error = raw.get("errorMessage") or raw.get("details")

    return {   # every key always present, so ResultSelector never hits a missing path

        "state": result["state"].value,          # RUNNING | SUCCEEDED | FAILED | UNKNOWN

        "idmc_status": raw.get("status"),

        "start_time": raw.get("startTime"),

        "end_time": raw.get("endTime"),

        "duration_seconds": seconds_between(raw.get("startTime"), raw.get("endTime")),

        "error": str(error)[:1000] if error else None,

    }

record_outcome

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# functions/record_outcome.py

from common import ddb, now



def handler(event, context):

    run, outcome = event["run"], event["outcome"]

    status = event.get("status") or {}       # last CheckStatus result, if any

    step_error = event.get("error") or {}    # Step Functions Catch output, if any

    err = status.get("error") or step_error.get("Cause")



    fields = {

        "status": outcome["status"],                 # SUCCEEDED | FAILED | TIMED_OUT

        "failure_type": outcome.get("failure_type"),

        "ended_at": now(),

        "polls": (event.get("polls") or {}).get("count"),

        "idmc_status": status.get("idmc_status"),

        "idmc_start_time": status.get("start_time"),

        "idmc_end_time": status.get("end_time"),

        "duration_seconds": status.get("duration_seconds"),

        "error_message": str(err)[:1000] if err else None,

    }

    fields = {k: v for k, v in fields.items() if v is not None}   # skip empty values



    ddb.update_item(

        Key={"file_id": run["file_id"]},

        UpdateExpression="SET " + ", ".join(f"#{k} = :{k}" for k in fields),

        ExpressionAttributeNames={f"#{k}": k for k in fields},

        ExpressionAttributeValues={f":{k}": v for k, v in fields.items()},

    )

    return {"recorded": outcome["status"]}

Recording through one small Lambda, rather than direct DynamoDB integrations, keeps null handling and field names in code you can unit-test. The ledger now keeps the IDMC status, the IDMC start and end times, the computed duration, the failure type, and the error message.

Each function runs for well under a second per call, so cost is driven by the number of polls, not by how long the taskflow runs.

Step 3: Define the state machine

The state machine starts the run, waits 60 seconds, checks status, and loops until IDMC reports a final state. Each ending is labeled in a Pass state, recorded by record_outcome, and failures and timeouts are sent to SNS.

Save this as statemachine/idmc_taskflow.asl.json. The ${...} placeholders are filled by SAM at deploy time.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
{

  "Comment": "Start an IDMC taskflow for an S3 file, poll to completion, record and alert",

  "StartAt": "StartTaskflow",

  "States": {

    "StartTaskflow": {

      "Type": "Task",

      "Resource": "arn:aws:states:::lambda:invoke",

      "Parameters": {

        "FunctionName": "${StartTaskflowFnArn}",

        "Payload": { "detail.$": "$.detail", "execution_name.$": "$$.Execution.Name" }

      },

      "ResultSelector": { "file_id.$": "$.Payload.file_id", "run_id.$": "$.Payload.run_id",

                          "taskflow.$": "$.Payload.taskflow", "key.$": "$.Payload.key",

                          "duplicate.$": "$.Payload.duplicate" },

      "ResultPath": "$.run",

      "Retry": [

        { "ErrorEquals": ["TransientIdmcError", "Lambda.ServiceException",

                          "Lambda.TooManyRequestsException", "Lambda.SdkClientException"],

          "IntervalSeconds": 10, "BackoffRate": 2.0, "MaxAttempts": 4, "JitterStrategy": "FULL" }

      ],

      "Catch": [

        { "ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "NotifyStartFailure" }

      ],

      "Next": "IsDuplicate"

    },



    "IsDuplicate": {

      "Type": "Choice",

      "Choices": [ { "Variable": "$.run.duplicate", "BooleanEquals": true, "Next": "AlreadyProcessed" } ],

      "Default": "InitPollCount"

    },



    "InitPollCount": {

      "Type": "Pass", "Result": { "count": 0 }, "ResultPath": "$.polls", "Next": "WaitForRun"

    },



    "WaitForRun": { "Type": "Wait", "Seconds": 60, "Next": "CheckStatus" },



    "CheckStatus": {

      "Type": "Task",

      "Resource": "arn:aws:states:::lambda:invoke",

      "Parameters": {

        "FunctionName": "${CheckStatusFnArn}",

        "Payload": { "run_id.$": "$.run.run_id" }

      },

      "ResultSelector": {

        "state.$": "$.Payload.state", "idmc_status.$": "$.Payload.idmc_status",

        "start_time.$": "$.Payload.start_time", "end_time.$": "$.Payload.end_time",

        "duration_seconds.$": "$.Payload.duration_seconds", "error.$": "$.Payload.error"

      },

      "ResultPath": "$.status",

      "Retry": [

        { "ErrorEquals": ["TransientIdmcError", "Lambda.ServiceException",

                          "Lambda.TooManyRequestsException"],

          "IntervalSeconds": 15, "BackoffRate": 2.0, "MaxAttempts": 5, "JitterStrategy": "FULL" }

      ],

      "Catch": [

        { "ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "LabelStatusCheckError" }

      ],

      "Next": "CountPoll"

    },



    "CountPoll": {

      "Type": "Pass",

      "Parameters": { "count.$": "States.MathAdd($.polls.count, 1)" },

      "ResultPath": "$.polls",

      "Next": "EvaluateStatus"

    },



    "EvaluateStatus": {

      "Type": "Choice",

      "Choices": [

        { "Variable": "$.status.state", "StringEquals": "SUCCEEDED", "Next": "LabelSucceeded" },

        { "Variable": "$.status.state", "StringEquals": "FAILED",    "Next": "LabelFailed" },

        { "And": [

            { "Variable": "$.status.state", "StringEquals": "UNKNOWN" },

            { "Variable": "$.polls.count", "NumericGreaterThan": 10 } ],

          "Next": "LabelNotRegistered" },

        { "Variable": "$.polls.count", "NumericGreaterThan": 360, "Next": "LabelTimedOut" }

      ],

      "Default": "WaitForRun"

    },



    "LabelSucceeded": {

      "Type": "Pass", "ResultPath": "$.outcome", "Next": "RecordOutcome",

      "Parameters": { "status": "SUCCEEDED", "failure_type": null, "note": "Run completed." }

    },

    "LabelFailed": {

      "Type": "Pass", "ResultPath": "$.outcome", "Next": "RecordOutcome",

      "Parameters": { "status": "FAILED", "failure_type": "IDMC_TASKFLOW",

                      "note": "IDMC reported the taskflow as failed or suspended. Check IDMC Monitor." }

    },

    "LabelNotRegistered": {

      "Type": "Pass", "ResultPath": "$.outcome", "Next": "RecordOutcome",

      "Parameters": { "status": "FAILED", "failure_type": "RUN_NOT_REGISTERED",

                      "note": "IDMC returned no status for this run ID after 10 polls." }

    },

    "LabelStatusCheckError": {

      "Type": "Pass", "ResultPath": "$.outcome", "Next": "RecordOutcome",

      "Parameters": { "status": "FAILED", "failure_type": "STATUS_CHECK_ERROR",

                      "note": "The IDMC status API kept failing after retries. The run may still be active." }

    },

    "LabelTimedOut": {

      "Type": "Pass", "ResultPath": "$.outcome", "Next": "RecordOutcome",

      "Parameters": { "status": "TIMED_OUT", "failure_type": "ORCHESTRATION_TIMEOUT",

                      "note": "Polling stopped after 6 hours. The IDMC run may still be active; check or stop it in IDMC Monitor." }

    },



    "RecordOutcome": {

      "Type": "Task",

      "Resource": "arn:aws:states:::lambda:invoke",

      "Parameters": { "FunctionName": "${RecordOutcomeFnArn}", "Payload.$": "$" },

      "ResultPath": null,

      "Retry": [

        { "ErrorEquals": ["Lambda.ServiceException", "Lambda.TooManyRequestsException",

                          "Lambda.SdkClientException"],

          "IntervalSeconds": 5, "BackoffRate": 2.0, "MaxAttempts": 3 }

      ],

      "Catch": [

        { "ErrorEquals": ["States.ALL"], "ResultPath": "$.record_error", "Next": "NotifyRecordFailure" }

      ],

      "Next": "AfterRecord"

    },



    "AfterRecord": {

      "Type": "Choice",

      "Choices": [ { "Variable": "$.outcome.status", "StringEquals": "SUCCEEDED", "Next": "Done" } ],

      "Default": "NotifyOutcome"

    },



    "NotifyOutcome": {

      "Type": "Task",

      "Resource": "arn:aws:states:::sns:publish",

      "Parameters": {

        "TopicArn": "${AlertTopicArn}",

        "Subject": "IDMC taskflow needs attention",

        "Message.$": "States.Format('Taskflow {} ended as {} ({}) for {}. {} Run ID: {}. Execution: {}', $.run.taskflow, $.outcome.status, $.outcome.failure_type, $.run.key, $.outcome.note, $.run.run_id, $$.Execution.Id)"

      },

      "ResultPath": null,

      "Next": "RunFailed"

    },



    "NotifyRecordFailure": {

      "Type": "Task",

      "Resource": "arn:aws:states:::sns:publish",

      "Parameters": {

        "TopicArn": "${AlertTopicArn}",

        "Subject": "IDMC ledger write failed",

        "Message.$": "States.Format('Ledger write failed for {} (outcome {}). Run ID: {}. Error: {}', $.run.key, $.outcome.status, $.run.run_id, $.record_error.Cause)"

      },

      "ResultPath": null,

      "Next": "RunFailed"

    },



    "NotifyStartFailure": {

      "Type": "Task",

      "Resource": "arn:aws:states:::sns:publish",

      "Parameters": {

        "TopicArn": "${AlertTopicArn}",

        "Subject": "IDMC taskflow could not start",

        "Message.$": "States.Format('Could not start taskflow for s3://{}/{}. Ledger status: START_FAILED. Error: {}', $.detail.bucket.name, $.detail.object.key, $.error.Cause)"

      },

      "ResultPath": null,

      "Next": "RunFailed"

    },



    "AlreadyProcessed": { "Type": "Succeed", "Comment": "Duplicate event for an object already claimed" },

    "Done":             { "Type": "Succeed" },

    "RunFailed":        { "Type": "Fail", "Error": "IdmcTaskflowFailed" }

  }

}

How the key choices play out:

Poll interval of 60 seconds. Standard workflows bill per state transition, not per second of waiting, so a long Wait costs nothing. Shorten it for short taskflows; lengthen it for overnight loads.
Every ending is labeled. Each terminal path sets status, failure_type and note in a Pass state, so the ledger and the alert say exactly why the run ended: IDMC_TASKFLOW, RUN_NOT_REGISTERED, STATUS_CHECK_ERROR or ORCHESTRATION_TIMEOUT. If the ledger write itself fails after its retries, NotifyRecordFailure still sends an alert, so no outcome goes unreported.
A timeout stops the polling, not the IDMC run. After 360 polls (6 hours) the execution records TIMED_OUT and alerts, but the taskflow may still be running in IDMC, and the alert says so. If your process needs automatic cancellation, add a stop step before LabelTimedOut that calls your org's stop or cancel operation. Check the IDMC REST API reference for your release; this article does not assume one exists.
Duplicates are a normal result, not an error. IsDuplicate reads a flag from the Lambda response, so that branch never depends on exception names.
RunFailed is a real Fail state. The execution shows as failed in the console and in CloudWatch, which drives the alarm in Step 5.
Standard, not Express. Standard executions can run for up to a year; Express workflows stop at 5 minutes, so they cannot follow a long taskflow.

Step 4: Route S3 events through EventBridge and deploy with SAM

In Part 2, S3 invoked the Lambda directly. Here S3 sends events to Amazon EventBridge, and an EventBridge rule starts one Step Functions execution per file.

EventBridge gives you prefix and suffix filters in one place, a built-in retry policy toward Step Functions, and the option to fan the same event out to other targets later.

Turn on EventBridge for the bucket

1
2
3
aws s3api put-bucket-notification-configuration \

  --bucket my-idmc-landing-bucket \

  --notification-configuration '{"EventBridgeConfiguration": {}}'

Warning: this call replaces the bucket's whole notification configuration. That removes the Part 2 direct Lambda trigger, which is what you want; otherwise both paths would start the taskflow. If the bucket has other notifications you need to keep, include them in the same JSON.

template.yaml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
AWSTemplateFormatVersion: "2010-09-09"

Transform: AWS::Serverless-2016-10-31

Description: Part 3 - Orchestrate and monitor Informatica IDMC taskflows



Parameters:

  LandingBucket:   { Type: String, Default: my-idmc-landing-bucket }

  InboundPrefix:   { Type: String, Default: inbound/ }

  TaskflowName:    { Type: String, Default: tf_load_customer_files }

  IdmcSecretArn:   { Type: String }

  IdmcSdkLayerArn: { Type: String }

  AlertEmail:      { Type: String }



Globals:

  Function:

    Runtime: python3.12

    Architectures: [arm64]

    Timeout: 60

    MemorySize: 256

    Layers: [!Ref IdmcSdkLayerArn]

    Environment:

      Variables:

        IDMC_SECRET_ARN: !Ref IdmcSecretArn

        LEDGER_TABLE: !Ref RunLedger

        TASKFLOW_NAME: !Ref TaskflowName



Resources:

  RunLedger:

    Type: AWS::DynamoDB::Table

    Properties:

      BillingMode: PAY_PER_REQUEST

      AttributeDefinitions:

        - { AttributeName: file_id, AttributeType: S }

      KeySchema:

        - { AttributeName: file_id, KeyType: HASH }

      PointInTimeRecoverySpecification: { PointInTimeRecoveryEnabled: true }



  AlertTopic:            # detailed technical notices for the data team

    Type: AWS::SNS::Topic

    Properties:

      Subscription:

        - { Protocol: email, Endpoint: !Ref AlertEmail }



  OnCallTopic:           # escalation; subscribe your paging or chat integration

    Type: AWS::SNS::Topic



  FailedRunsAlarm:       # failed or timed-out executions -> on-call

    Type: AWS::CloudWatch::Alarm

    Properties:

      AlarmDescription: IDMC taskflow orchestration failed or timed out

      Namespace: AWS/States

      MetricName: ExecutionsFailed

      Dimensions:

        - Name: StateMachineArn

          Value: !Ref TaskflowStateMachine

      Statistic: Sum

      Period: 300

      EvaluationPeriods: 1

      Threshold: 1

      ComparisonOperator: GreaterThanOrEqualToThreshold

      TreatMissingData: notBreaching

      AlarmActions:

        - !Ref OnCallTopic



  StartTaskflowFn:

    Type: AWS::Serverless::Function

    Properties:

      CodeUri: functions/

      Handler: start_taskflow.handler

      Policies:

        - DynamoDBCrudPolicy: { TableName: !Ref RunLedger }

        - AWSSecretsManagerGetSecretValuePolicy: { SecretArn: !Ref IdmcSecretArn }



  CheckStatusFn:

    Type: AWS::Serverless::Function

    Properties:

      CodeUri: functions/

      Handler: check_status.handler

      Timeout: 30

      Policies:

        - AWSSecretsManagerGetSecretValuePolicy: { SecretArn: !Ref IdmcSecretArn }



  RecordOutcomeFn:

    Type: AWS::Serverless::Function

    Properties:

      CodeUri: functions/

      Handler: record_outcome.handler

      Timeout: 15

      Policies:

        - DynamoDBCrudPolicy: { TableName: !Ref RunLedger }



  TaskflowStateMachine:

    Type: AWS::Serverless::StateMachine

    Properties:

      Type: STANDARD

      DefinitionUri: statemachine/idmc_taskflow.asl.json

      DefinitionSubstitutions:

        StartTaskflowFnArn: !GetAtt StartTaskflowFn.Arn

        CheckStatusFnArn: !GetAtt CheckStatusFn.Arn

        RecordOutcomeFnArn: !GetAtt RecordOutcomeFn.Arn

        AlertTopicArn: !Ref AlertTopic

      Tracing: { Enabled: true }

      Policies:

        - LambdaInvokePolicy: { FunctionName: !Ref StartTaskflowFn }

        - LambdaInvokePolicy: { FunctionName: !Ref CheckStatusFn }

        - LambdaInvokePolicy: { FunctionName: !Ref RecordOutcomeFn }

        - SNSPublishMessagePolicy: { TopicName: !GetAtt AlertTopic.TopicName }

      Events:

        FileLanded:

          Type: EventBridgeRule

          Properties:

            Pattern:

              source: [aws.s3]

              detail-type: ["Object Created"]

              detail:

                bucket: { name: [!Ref LandingBucket] }

                object:

                  key: [{ prefix: !Ref InboundPrefix }]

            RetryPolicy:

              MaximumRetryAttempts: 5

              MaximumEventAgeInSeconds: 3600



Outputs:

  StateMachineArn: { Value: !Ref TaskflowStateMachine }

  LedgerTableName: { Value: !Ref RunLedger }

  OnCallTopicArn:  { Value: !Ref OnCallTopic }

Deploy

1
2
3
4
5
6
sam build

sam deploy --guided \

  --parameter-overrides \

    IdmcSecretArn=arn:aws:secretsmanager:us-east-1:123456789012:secret:idmc/prod \

    IdmcSdkLayerArn=arn:aws:lambda:us-east-1:123456789012:layer:idmc-sdk:3 \

    AlertEmail=data-oncall@example.com

Confirm the SNS subscription email after the first deploy, or alerts will not arrive.

Step 5: See every run and get paged on the right ones

You now have three views of the pipeline: the DynamoDB ledger for "what happened to this file", Step Functions history for "where did it break", and a CloudWatch alarm for "wake someone up".

The run ledger

Each item in the ledger tells the story of one file:

started_at and ended_at are orchestration times. The idmc_ times are what IDMC reports for the run itself, and duration_seconds is computed from them, so polling delay does not inflate it.

To answer "did yesterday's customer file load?", look up its file_id. The execution attribute takes you to the visual run history in the Step Functions console, and the run_id takes you to the same run in IDMC Monitor.

Alerts: two topics, two jobs

A failure produces two signals by design, so send them to different places:

AlertTopic gets the state machine's message for every failed file: taskflow, file, run ID, failure type and note. This is the detailed technical notice for the data team.
OnCallTopic gets the CloudWatch alarm. It fires once per 5-minute window however many files fail, so it is the right signal for paging or a chat channel.

If one team receives both, subscribe them to only one topic to avoid duplicate notifications. The FailedRunsAlarm resource is already included in the Step 4 template, so sam deploy creates it; it watches the ExecutionsFailed metric for this state machine, which counts both failed and timed-out runs because both end in the RunFailed state.

Add a second alarm on ExecutionThrottled if you expect bursts of hundreds of files at once.

Logs and traces

Lambda logs in CloudWatch Logs carry the IDMC error text for every failed API call. Log the run_id and file_id on every line so you can filter by run.
Time the IDMC calls yourself. Wrap _request with time.perf_counter() and log the duration with the run ID. That is the reliable source for IDMC API latency.
X-Ray tracing, enabled on the state machine in the template, correlates Step Functions and Lambda timing and shows where the orchestration path is slow. It does not trace inside IDMC; the outbound HTTP call shows only as client-side time in the Lambda segment.
Never log the session ID or credentials. The Part 1 SDK should already mask the INFA-SESSION-ID header; check before you raise the log level to DEBUG.

Step 6: Test it, price it, and ship it

Six tests prove the pipeline: a clean run, a duplicate event, a failing taskflow, bad credentials, a transient IDMC outage, and a polling timeout. At roughly a third of a cent per 30-minute run, the orchestration costs far less than the data it protects.

Test locally first

Save a sample EventBridge event as events/object_created.json:

1
2
3
4
5
6
7
8
9
{

  "source": "aws.s3",

  "detail-type": "Object Created",

  "detail": {

    "bucket": { "name": "my-idmc-landing-bucket" },

    "object": { "key": "inbound/customers_20261001.csv",

                "size": 18234, "etag": "9b2cf535f27731c974343645a3985328" }

  }

}

Then invoke the start function against your dev IDMC org:

1
2
sam local invoke StartTaskflowFn -e events/object_created.json \

  --env-vars env.dev.json

Then run these six tests end to end

In a versioned bucket, re-uploading the same file creates a new version and runs again by design, which is why the duplicate test replays the event instead.

After the failure tests, confirm the error names Step Functions actually received:

1
2
3
aws stepfunctions get-execution-history \

  --execution-arn <execution-arn> \

  --query "events[?type=='TaskFailed'].taskFailedEventDetails.error"

You should see TransientIdmcError or PermanentIdmcError. Any other name means the retry rules are not matching, so fix that before going live.

What it costs

Step Functions Standard workflows are billed per state transition, about $0.025 per 1,000 in us-east-1 (check current AWS Step Functions pricing for your Region). Each poll cycle is 4 transitions, so a 30-minute taskflow polled every 60 seconds uses about 130 transitions, or roughly $0.003 per run. At 500 files a day, Step Functions Standard costs about $48.75 a month in us-east-1 (130 transitions x 500 files x 30 days at $0.000025 each), before the monthly free tier of 4,000 transitions and excluding extra transitions from retries. Lambda, DynamoDB, SNS, CloudWatch and X-Ray are billed separately.

Production checklist

Remove or disable the Part 2 S3 to Lambda trigger in the stack that owns it before turning on the Part 3 EventBridge trigger; a manual bucket notification change can be overwritten by that stack's next deployment
Use a dedicated IDMC integration user with only the roles needed to run and monitor the taskflow
Store IDMC credentials in Secrets Manager and turn on rotation
Tune Seconds in WaitForRun and the 360-poll limit to each taskflow's normal runtime
Confirm the SNS email subscription and connect OnCallTopic to your on-call tool
Add a DynamoDB TTL attribute if you do not need ledger history beyond a set period
Deploy through a pipeline with separate dev and prod stacks, each pointing at its own IDMC org
Turn on versioning for the landing bucket so the version ID identifies each upload

Production caveats

Polling is a deliberate choice. This design polls because IDMC exposes run status through a status resource. Tune the interval to each taskflow's runtime and to your org's API limits.
The 6-hour limit governs the orchestration, not IDMC. Timing out does not end the IDMC run. Add a stop or cancel step if your process requires it.
ETag deduplication has limits. It is a practical duplicate check, not a content hash. Turn on versioning for stronger object identity.
Error names are a contract. Renaming an exception class silently changes retry behavior, so keep the execution-history check in your test suite.

Wrapping up

Across the three parts, a raw REST API became a reusable SDK layer, then an event-driven trigger, and now a monitored, idempotent pipeline that records every outcome. Every file that lands in S3 is now accounted for from upload to load or to a clearly labeled failure.

A natural next step is routing different file patterns to different taskflows from a configuration table, and running dependent taskflows in parallel with a Step Functions Map state.

If you build on this pattern, share what you changed in the comments. I would like to hear how others handle long-running IDMC jobs.

# python # etl # data-automation-api # amazon-s3 # data-integration

Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.

Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article

Builder Center

Part 3: Orchestrate, Monitor, and Handle Failures in Informatica IDMC Taskflow Runs with AWS Step Functions

Shireesha Mangannagari

From Start-and-forget to a pipeline you can trust

Architecture: one Step Functions execution per file

Step 1: Teach the SDK layer to read taskflow status

Step 2: Split the Part 2 Lambda into three small functions

Shared helpers

start_taskflow

check_status

record_outcome

Step 3: Define the state machine

Step 4: Route S3 events through EventBridge and deploy with SAM

Turn on EventBridge for the bucket

template.yaml

Deploy

Step 5: See every run and get paged on the right ones

The run ledger

Alerts: two topics, two jobs

Logs and traces

Step 6: Test it, price it, and ship it

Test locally first

Then run these six tests end to end

What it costs

Production checklist

Production caveats

Wrapping up