DEV Community: Chanaka Supun

How I Upgraded RDS PostgreSQL Version From 13.20 to 17.6

Chanaka Supun — Tue, 04 Nov 2025 04:52:25 +0000

i had to upgrade PostgreSQL version last month because it will be end of life in next month.

Our organization uses postrgeSQL as the main database engine. We have staging and production environments both running version 13.20. According to announcements from AWS -> Amazon RDS PostgreSQL 13.x end of standard support is February 28, 2026. With community end of life being November 2025. So we had to prepare for the upgrade. Below describe the process i used and steps taken during the upgrade process.

Steps Followed:

Check Changelog

First i take a look at the release notes from 13.2 version to 17.6 version and note down the breaking/major changes that are there for my applications. Identify If any i need to pay attention to. Here are few critical changes that are there.
Release Notes
Summary from RN

## Critical Breaking Changes (PostgreSQL 17)

### 1. Search Path Changes in Maintenance Operations ⚠️ CRITICAL

**Issue:** Functions used by expression indexes and materialized views must now specify search_path explicitly.

**Impact:** Maintenance operations (ANALYZE, CLUSTER, CREATE INDEX, CREATE MATERIALIZED VIEW, REFRESH MATERIALIZED VIEW, REINDEX, VACUUM) now use a safe search_path.

**Action Required:**
sql
-- Check for functions used in indexes/materialized views without explicit search_path
SELECT n.nspname, p.proname, pg_get_functiondef(p.oid)
FROM pg_proc p
JOIN pg_namespace n ON p.pronamespace = n.oid
WHERE p.oid IN (
  SELECT indexrelid::regclass::oid 
  FROM pg_index 
  WHERE indexprs IS NOT NULL
)
AND pg_get_functiondef(p.oid) NOT LIKE '%SET search_path%';

-- Fix: Add search_path to function definition
ALTER FUNCTION schema_name.function_name() SET search_path = schema_name, pg_catalog;

### 2. Removed: old_snapshot_threshold Parameter

**Issue:** Server variable `old_snapshot_threshold` has been removed.

**Action Required:**
sql
-- Check if parameter is set
SHOW old_snapshot_threshold;

-- If set, remove from postgresql.conf before upgrade
-- This feature may be re-added in future versions


### 3. Removed: db_user_namespace Feature

**Issue:** Per-database user simulation feature removed.

**Action Required:**
sql
-- Check if enabled
SHOW db_user_namespace;

-- If 'on', must be disabled before upgrade
-- Migrate to standard user management


### 4. Removed: adminpack Extension

**Issue:** adminpack contrib extension removed (was used by pgAdmin III).

**Action Required:**
sql
-- Check if installed
SELECT * FROM pg_extension WHERE extname = 'adminpack';

-- Drop before upgrade
DROP EXTENSION IF EXISTS adminpack;

### 5. System Catalog Column Renames

**Issue:** Several system catalog columns renamed in PostgreSQL 17.

**Action Required:**
sql
-- Check for queries/views using old column names:

-- pg_collation.colliculocale → colllocale
-- pg_database.daticulocale → datlocale
-- pg_attribute.attstattarget (now NULL for default instead of -1)
-- pg_stat_progress_vacuum columns renamed
-- pg_stat_slru columns renamed
-- pg_stat_statements: blk_read_time → shared_blk_read_time
-- pg_stat_statements: blk_write_time → shared_blk_write_time

-- Search for usage in views/functions
SELECT schemaname, viewname, definition
FROM pg_views
WHERE definition LIKE '%colliculocale%'
   OR definition LIKE '%daticulocale%'
   OR definition LIKE '%blk_read_time%'
   OR definition LIKE '%blk_write_time%';

### 6. Removed: pg_stat_bgwriter Columns

**Issue:** `buffers_backend` and `buffers_backend_fsync` removed from pg_stat_bgwriter.

**Action Required:**
sql
-- Check for queries using these columns
SELECT schemaname, viewname, definition
FROM pg_views
WHERE definition LIKE '%buffers_backend%'
   OR definition LIKE '%buffers_backend_fsync%';

-- Migrate to pg_stat_io view instead
SELECT * FROM pg_stat_io;

### 6. Postgres Extention version upgrade - check for breaking changes
pg_partman extention version needed to be upgraded from version "4.5.1" to version "5.2.4" . you can see the breaking changes here 
[changelog](https://github.com/pgpartman/pg_partman/blob/development/CHANGELOG.md)

## Breaking Changes (PostgreSQL 16)

### 1. PL/pgSQL Bound Cursor Variable Changes

**Issue:** String value of bound cursor variables no longer matches variable name during assignment.

**Action Required:**
sql
-- Review PL/pgSQL functions with bound cursors
SELECT n.nspname, p.proname, pg_get_functiondef(p.oid)
FROM pg_proc p
JOIN pg_namespace n ON p.pronamespace = n.oid
WHERE p.prolang = (SELECT oid FROM pg_language WHERE lanname = 'plpgsql')
  AND pg_get_functiondef(p.oid) LIKE '%CURSOR%FOR%';

-- If cursor variable name is used, assign explicitly before OPEN
-- Old behavior: cursor_var := 'cursor_name' (automatic)
-- New behavior: Must explicitly assign before OPEN

### 2. Removed: Postfix Operators

**Issue:** Support for postfix operators removed.

**Action Required:**
sql
-- Check for postfix operators
SELECT oprname, oprleft, oprright
FROM pg_operator
WHERE oprright = 0;

-- pg_dump and pg_upgrade will warn about these
-- Convert to prefix or infix operators before upgrade

## Breaking Changes (PostgreSQL 15)

### 1. PUBLIC Schema Permission Changes ⚠️ CRITICAL

**Issue:** PUBLIC creation permission removed from public schema by default.

**Impact:** Users can no longer create objects in public schema without explicit GRANT.

**Action Required:**
sql
-- Check current permissions
SELECT nspname, nspacl FROM pg_namespace WHERE nspname = 'public';

-- If needed, restore old behavior (not recommended for security)
GRANT CREATE ON SCHEMA public TO PUBLIC;

-- Better: Grant to specific roles
GRANT CREATE ON SCHEMA public TO app_role;

### 2. Removed: Exclusive Backup Mode

**Issue:** pg_start_backup()/pg_stop_backup() exclusive mode removed.

**Action Required:**sql
-- Check for usage of exclusive backup mode
-- Search application code for:
-- pg_start_backup(label, true)  -- Second parameter 'true' = exclusive mode

-- Migrate to non-exclusive mode:
-- pg_backup_start(label, false)  -- Note: function renamed
-- pg_backup_stop(false)

### 3. Removed: Python 2.x Support

**Issue:** plpython2u and plpythonu (Python 2) removed.

**Action Required:**
sql
-- Check for Python 2 functions
SELECT n.nspname, p.proname, l.lanname
FROM pg_proc p
JOIN pg_namespace n ON p.pronamespace = n.oid
JOIN pg_language l ON p.prolang = l.oid
WHERE l.lanname IN ('plpython2u', 'plpythonu');

-- Migrate to plpython3u
-- Rewrite functions for Python 3 compatibility

### 4. array_to_tsvector() Empty String Error

**Issue:** Now generates error for empty-string array elements.

**Action Required:**sql
-- Check for empty lexemes in tsvector columns
SELECT tablename, attname
FROM pg_stats
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
  AND atttypid = 'tsvector'::regtype;

-- Verify no empty lexemes exist
SELECT * FROM your_table WHERE to_tsvector('') = ANY(your_tsvector_column);

-- Clean up empty lexemes before upgrade

## Breaking Changes (PostgreSQL 14)

### 1. Array Function Signature Changes ⚠️ CRITICAL

**Issue:** Array functions changed from `anyarray` to `anycompatiblearray`.

**Affected Functions:**
- array_append()
- array_prepend()
- array_cat()
- array_position()
- array_positions()
- array_remove()
- array_replace()
- width_bucket()

**Action Required:**
sql
-- Find user-defined objects referencing these functions
SELECT n.nspname, p.proname, pg_get_functiondef(p.oid)
FROM pg_proc p
JOIN pg_namespace n ON p.pronamespace = n.oid
WHERE pg_get_functiondef(p.oid) ~ 'array_(append|prepend|cat|position|positions|remove|replace)|width_bucket'
  AND n.nspname NOT IN ('pg_catalog', 'information_schema');

-- Find aggregates using these functions
SELECT n.nspname, a.aggfnoid::regproc
FROM pg_aggregate a
JOIN pg_proc p ON a.aggfnoid = p.oid
JOIN pg_namespace n ON p.pronamespace = n.oid
WHERE n.nspname NOT IN ('pg_catalog', 'information_schema');

-- Drop and recreate after upgrade

### 2. Removed: Containment Operators @ and ~

**Issue:** Deprecated operators @ and ~ removed for geometric types and contrib modules.

**Affected:** geometric data types, cube, hstore, intarray, seg

**Action Required:**
sql
-- Search for usage of @ and ~ operators
SELECT schemaname, viewname, definition
FROM pg_views
WHERE definition ~ '[@~]'
  AND schemaname NOT IN ('pg_catalog', 'information_schema');

-- Replace with <@ and @> operators
-- Old: geometry1 @ geometry2
-- New: geometry1 <@ geometry2

### 3. to_tsquery() and websearch_to_tsquery() Parsing Changes

**Issue:** Discarded tokens now properly parsed.

**Action Required:**
sql
-- Review queries using these functions with underscores or other discarded tokens
-- Test query results after upgrade
-- Example: websearch_to_tsquery('"pg_class pg"')
-- Old output: ( 'pg' & 'class' ) <-> 'pg'
-- New output: 'pg' <-> 'class' <-> 'pg'

### 4. Regular Expression \D and \W Changes

**Issue:** \D and \W now match newlines in newline-sensitive mode.

**Action Required:**
sql
-- Review regex patterns using \D or \W
SELECT schemaname, viewname, definition
FROM pg_views
WHERE definition ~ '\\[DW]'
  AND schemaname NOT IN ('pg_catalog', 'information_schema');

-- Use [^[:digit:]] or [^[:word:]] for old behavior

### 5. Removed: pg_standby Utility

**Issue:** contrib program pg_standby removed.

**Action Required:**
- Check if pg_standby is used in recovery.conf or restore_command
- Migrate to restore_command with alternative tools

In my case non of the above changes were affected my applications so no changes were required from my end. Typically that would be the case here unless you are using anything special to postgreSQL.

Check Official AWS upgrade Guide

Then using the AWS documentation i generated a set of steps need to do in order to perform major version upgrade. You can check them from below link.
RDS Major Upgrade Guide

check target version

After opening the doc i needed to decide my upgrade target 13.2-> 17.6 is supported. It was supported.

Check database instance compatibility

Afterward, I needed to checked if my database instance type was compatible with the target version. In my case db.m5.large. It was supported.
InstanceClass.Support

Decide on Upgrade Path

I considered below upgrade paths

Blue/Green Deployment (Recommended) : we had some pre-requisites that don’t match our database to support blue/green.
blue-green-deployments
Snapshot & In-Place Upgrade
pg_dump/restore : pg_dump/restore would incur longer downtime (rebuilding indexes is slow for restore) .

After doing some feasibility study identified the most feasible path for us is in-place upgrade with thorough testing.

Create parameter group

I created custom parameter group for postgres17. In my currently database i am using pg_cron extention. So needed to update shared_libraries parameter to include pg_cron in new param group as well. Just like that you can compare the parameter groups and do required changes if you changed any defaults values.

Upgrading extentions

A PostgreSQL upgrade doesn’t upgrade any PostgreSQL extensions. To upgrade extensions, see
PostgreSQL.ExtensionUpgrades

I am using pg_partman and pg_cron extentions. In order to check supported version for postgreSQL 17 see below doc.
postgresql-extensions-support

In my case upgrade requirement

pg_partman : 4.5.1 to 5.2.4
pg_cron 1.3 to 1.6.

I checked the changes in partman and pg_cron release notes identified the changes that need to happen.

Other checks needed..

There are some other checks in this doc that need to be fulfilled in order to do a successful upgrade. you can check below by following the official aws doc mentioned previously.

1.Check for unsupported usage
2.Check for invalid databases
3.Handle logical replication slots 
4.Handle read replicas 
5.Handle zero-ETL integrations

In my case 1–3 all are zero and i don’t have replicas or zero-etl configured. So now i am all set for the upgrade.

Perform an upgrade dry run
Good thing about using AWS is you can quickly spin up new servers with click of a button. So in order to test my upgrade procedure working i created a new db instance using a snapshot of my existing db. Then perform pre-checks above mentioned on it and after verification i performed an in place upgrade to the new instance. You can do this my modifying the instance and update postgreSQL version to 17.6 and update parameter group to the new one created. Then applied the changes immediately. You can use the process to time your process and identify the downtime needed during the real upgrade.

In my case database upgrade took around 12mins so during this time database was unavailable. AWS takes a backup of instance before and after the upgrade. So after the upgrade it is recomended to run the ANALYZE on all databases to update pg_statistic table. This took around 30mins. Also i am using pg_partman, pg_cron which then i upgraded to supported versions. Then connected to database locally and performed validations. Then Documented all steps performed during dry-run.

Upgrade Staging/Production Database
After successful dry-run then its was time to perform staging upgrade. Can follow same steps for prod as well. Inform the stake holders beforehand and acquired a required downtime. Better to perform during an off-peak time interval. Then perform same steps as the dry-run.

Check logs

After upgrading, I checked the logs. There should be two new logs.

pg_upgrade_internal.log: Log of RDS upgrade (pg_upgrade)
pg_upgrade_server.log: Log of stopping/starting RDS

Checked any Error or Fatal log in those two logs and confirmed that “Upgrade Complete” is printed at the bottom of internal.log.

After the upgrade connected to the database and perform some validations on content.

I had an issue where some of the pods in eks were crashloopbackoff state after the upgrade. After checking the logs identified it was failing to connect database due to SSL error. Identified this was due to in postgreSQL17 param group rds.force_ssl is 1 by default and in postgreSQL13 it was 0. So after updating that pods were running without errors. Noted this down for production upgrade.

Also note after the upgrade some SQL queries were taking longer time to execute hence increasing CPU load for rds and causing performance degradations.

By using performance insights in AWS identified this was mainly due to a single slow query that was happening.

After analysing query identified it was due to a VIEW with anti join. Had to rewrite the query for it and after that performance was back to normal.

That’s all! Database was upgraded successfully.

How I Orchestrated 400K+ Nested API Calls with AWS Step Functions and Lambda

Chanaka Supun — Tue, 19 Aug 2025 18:26:22 +0000

When working with large-scale data pipelines, a common challenge is efficiently orchestrating high-volume API calls that have multiple layers of dependency. Recently, I encountered such a challenge where I needed to generate 409,600 JSON files from deeply nested API responses.

The Challenge
The requirements looked straightforward at first but quickly revealed significant complexity:

Fetch 100 latest content items via an API call.
For each content item, fetch the 16 latest comments.
For each comment, fetch the 16 latest replies.
For each reply, fetch the 16 latest nested replies.
This results in a large combinatorial expansion:

100 (contents) × 16 (comments) × 16 (replies) × 16 (nested replies) = 409,600 JSON files
Each generated JSON file represents the fully expanded content tree for one leaf node.

Why AWS Step Functions
To manage this orchestration at scale, I chose AWS Step Functions. Specifically, I leveraged the Map state, which allows parallel execution over dynamic lists without needing to manually code distributed concurrency handling.

Key benefits of using Step Functions here:

Native parallelism → Handles thousands of executions concurrently.
Error handling & retries → Simplifies recovery from transient API failures.
Visibility → Provides execution history, making it easier to debug complex workflows.
Integration → Works seamlessly with AWS Lambda for API calls and file generation.

High-Level Workflow
|-Initial API Call (Lambda): Get the 100 latest content items.

— —|-Step Functions Map State (Level 1): Iterate over the 100 content items.

— — — — -Inside this map:

— — — — — — — — — — Call API to fetch 16 comments.

— — — — — — — — — — |-Nested Map State (Level 2): Iterate over comments.

— — — — — — — — — — — — —Fetch 16 replies per comment.

— — — — — — — — — — — — —|-Nested Map State (Level 3): Iterate over replies.

— — — — — — — — — — — — — — — — -Fetch 16 nested replies per reply.

This recursive orchestration fans out into 409,600 parallel tasks at the deepest level.

Scalability Considerations
Handling such scale required careful planning:

Map concurrency limits: Step Functions supports up to 40,000 concurrent executions per account (with quotas adjustable via AWS Support).
API throttling: Implemented exponential backoff and batching logic in Lambda functions to avoid rate-limit errors.
File storage: Each JSON file is stored in Amazon S3 with a structured key hierarchy for easy retrieval.
Cost control: Since Step Functions are billed per state transition, I optimized workflows to reduce unnecessary states.
Payload Size: AWS Step Functions has a payload size limit of 256 KB for both input and output data passed between states in a workflow execution. Instead we can save response to s3 bucket after combining results from each map iteration and pass object name as input to next stage.

Diagram

Outcomes
Using Step Functions significantly reduced the operational complexity of managing this pipeline. Instead of building custom queueing, concurrency, and retry logic, I relied on AWS-native orchestration. The final system was scalable, fault-tolerant, and easy to monitor.

Here is the stepfunction ASL
(you can use this to create your own workflow)

{
    "Comment": "Content, Comments, Replies, and Reply Comments Processing Workflow",
    "StartAt": "GenerateFeedJSON",
    "States": {
        "GenerateAnonymousFeedJSON": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "Parameters": {
                "FunctionName": "${GenerateFeedLambdaArn}"
            },
            "Retry": [
                {
                    "ErrorEquals": [
                        "Lambda.ServiceException",
                        "Lambda.AWSLambdaException",
                        "Lambda.SdkClientException",
                        "Lambda.TooManyRequestsException"
                    ],
                    "IntervalSeconds": 1,
                    "MaxAttempts": 3,
                    "BackoffRate": 2,
                    "JitterStrategy": "FULL"
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "WorkflowFailed",
                    "ResultPath": "$.error"
                }
            ],
            "TimeoutSeconds": 300,
            "Next": "Pass"
        },
        "WorkflowFailed": {
            "Type": "Fail"
        },
        "Pass": {
            "Type": "Pass",
            "Next": "ProcessIndividualContent",
            "Parameters": {
                "input.$": "States.StringToJson($.Payload.body)"
            },
            "OutputPath": "$.input",
            "Assign": {
                "key.$": "$.input.key"
            }
        },
        "ProcessIndividualContent": {
            "Type": "Map",
            "ItemProcessor": {
                "ProcessorConfig": {
                    "Mode": "DISTRIBUTED",
                    "ExecutionType": "STANDARD"
                },
                "StartAt": "GenerateIndividualContentJSON",
                "States": {
                    "GenerateIndividualContentJSON": {
                        "Type": "Task",
                        "Resource": "arn:aws:states:::lambda:invoke",
                        "Parameters": {
                            "FunctionName": "${GenerateIndivcontentJsonsLambdaArn}",
                            "Payload.$": "$"
                        },
                        "Retry": [
                            {
                                "ErrorEquals": [
                                    "Lambda.ServiceException",
                                    "Lambda.AWSLambdaException",
                                    "Lambda.SdkClientException",
                                    "Lambda.TooManyRequestsException"
                                ],
                                "IntervalSeconds": 1,
                                "MaxAttempts": 3,
                                "BackoffRate": 2,
                                "JitterStrategy": "FULL"
                            }
                        ],
                        "TimeoutSeconds": 300,
                        "End": true
                    }
                }
            },
            "ItemsPath": "$.detail",
            "MaxConcurrency": 1000,
            "Next": "ProcessComments",
            "ResultPath": "$.processedContent",
            "ItemBatcher": {
                "MaxItemsPerBatch": 10
            },
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "WorkflowFailed",
                    "ResultPath": "$.error"
                }
            ]
        },
        "ProcessComments": {
            "Type": "Map",
            "ItemProcessor": {
                "ProcessorConfig": {
                    "Mode": "DISTRIBUTED",
                    "ExecutionType": "STANDARD"
                },
                "StartAt": "GenerateComments",
                "States": {
                    "GenerateComments": {
                        "Type": "Task",
                        "Resource": "arn:aws:states:::lambda:invoke",
                        "Parameters": {
                            "FunctionName": "${GenerateContentCommentJsonsLambdaArn}",
                            "Payload.$": "$"
                        },
                        "Retry": [
                            {
                                "ErrorEquals": [
                                    "Lambda.ServiceException",
                                    "Lambda.AWSLambdaException",
                                    "Lambda.SdkClientException",
                                    "Lambda.TooManyRequestsException"
                                ],
                                "IntervalSeconds": 1,
                                "MaxAttempts": 3,
                                "BackoffRate": 2,
                                "JitterStrategy": "FULL"
                            }
                        ],
                        "TimeoutSeconds": 300,
                        "End": true,
                        "OutputPath": "$.Payload.body"
                    }
                }
            },
            "ItemsPath": "$.detail",
            "MaxConcurrency": 1000,
            "ResultPath": "$",
            "Next": "IsReplyLevelCountReached",
            "ItemBatcher": {
                "MaxItemsPerBatch": 10
            },
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "WorkflowFailed",
                    "ResultPath": "$.error"
                }
            ],
            "ResultWriter": {
                "Resource": "arn:aws:states:::s3:putObject",
                "Parameters": {
                    "Bucket": "${content_sfn_artifact_bucket}",
                    "Prefix": "ProcessComments/"
                },
                "WriterConfig": {
                    "OutputType": "JSON",
                    "Transformation": "FLATTEN"
                }
            },
            "ResultSelector": {
                "output": {
                    "detail": {
                        "fileKey.$": "States.Format('ProcessComments/{}/SUCCEEDED_0.json', States.ArrayGetItem(States.StringSplit($.ResultWriterDetails.Key,'/'), 1))"
                    }
                },
                "counter": 0
            }
        },
        "IsReplyLevelCountReached": {
            "Type": "Choice",
            "Choices": [
                {
                    "Next": "consolidateoutputs",
                    "Variable": "$.counter",
                    "NumericLessThan": 2
                }
            ],
            "Default": "CopyToLatestJson"
        },
        "consolidateoutputs": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "Parameters": {
                "FunctionName": "${ConsolidateOutputsLambdaArn}",
                "Payload.$": "$"
            },
            "Retry": [
                {
                    "ErrorEquals": [
                        "Lambda.ServiceException",
                        "Lambda.AWSLambdaException",
                        "Lambda.SdkClientException",
                        "Lambda.TooManyRequestsException"
                    ],
                    "IntervalSeconds": 1,
                    "MaxAttempts": 3,
                    "BackoffRate": 2,
                    "JitterStrategy": "FULL"
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "WorkflowFailed",
                    "ResultPath": "$.error"
                }
            ],
            "TimeoutSeconds": 300,
            "Next": "ProcessReplies",
            "ResultSelector": {
                "input.$": "States.StringToJson($.Payload.body)"
            },
            "OutputPath": "$.input"
        },
        "ProcessReplies": {
            "Type": "Map",
            "ItemProcessor": {
                "ProcessorConfig": {
                    "Mode": "DISTRIBUTED",
                    "ExecutionType": "EXPRESS"
                },
                "StartAt": "GenerateReplies",
                "States": {
                    "GenerateReplies": {
                        "Type": "Task",
                        "Resource": "arn:aws:states:::lambda:invoke",
                        "Parameters": {
                            "FunctionName": "${GeneratCommentRepliesJsonsLambdaArn}",
                            "Payload.$": "$"
                        },
                        "Retry": [
                            {
                                "ErrorEquals": [
                                    "Lambda.ServiceException",
                                    "Lambda.AWSLambdaException",
                                    "Lambda.SdkClientException",
                                    "Lambda.TooManyRequestsException"
                                ],
                                "IntervalSeconds": 1,
                                "MaxAttempts": 3,
                                "BackoffRate": 2,
                                "JitterStrategy": "FULL"
                            }
                        ],
                        "TimeoutSeconds": 300,
                        "End": true,
                        "OutputPath": "$.Payload.body"
                    }
                }
            },
            "MaxConcurrency": 500,
            "ResultPath": "$.output",
            "Next": "IsReplyLevelCountReached",
            "ItemReader": {
                "Resource": "arn:aws:states:::s3:getObject",
                "ReaderConfig": {
                    "InputType": "JSON"
                },
                "Parameters": {
                    "Bucket.$": "$.bucketName",
                    "Key.$": "$.fileName"
                }
            },
            "ItemBatcher": {
                "MaxItemsPerBatch": 10
            },
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "WorkflowFailed",
                    "ResultPath": "$.error"
                }
            ],
            "ResultWriter": {
                "Resource": "arn:aws:states:::s3:putObject",
                "Parameters": {
                    "Bucket": "${content_sfn_artifact_bucket}",
                    "Prefix": "ProcessReplies/"
                },
                "WriterConfig": {
                    "OutputType": "JSON",
                    "Transformation": "FLATTEN"
                }
            },
            "ResultSelector": {
                "detail": {
                    "fileKey.$": "States.Format('ProcessReplies/{}/SUCCEEDED_0.json', States.ArrayGetItem(States.StringSplit($.ResultWriterDetails.Key,'/'), 1))"
                }
            }
        },
        "CopyToLatestJson": {
            "Type": "Task",
            "Parameters": {
                "Bucket": "${content_s3_bucket}",
                "CopySource.$": "States.Format('${content_s3_bucket}/{}', $key)",
                "Key": "feed/anonymous/latest.json"
            },
            "Resource": "arn:aws:states:::aws-sdk:s3:copyObject",
            "Retry": [
                {
                    "ErrorEquals": [
                        "S3.ServiceException",
                        "S3.AWSServiceException",
                        "S3.SdkClientException"
                    ],
                    "IntervalSeconds": 1,
                    "MaxAttempts": 3,
                    "BackoffRate": 2,
                    "JitterStrategy": "FULL"
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "WorkflowFailed",
                    "ResultPath": "$.error"
                }
            ],
            "Next": "ListObjectVersions"
        },
        "ListObjectVersions": {
            "Type": "Task",
            "Parameters": {
                "Bucket": "${content_s3_bucket}",
                "Prefix": "feed/anonymous/latest.json",
                "MaxKeys": 2
            },
            "Resource": "arn:aws:states:::aws-sdk:s3:listObjectVersions",
            "ResultSelector": {
                "OldVersionId.$": "$.Versions[1].VersionId",
                "LatestVersionId.$": "$.Versions[0].VersionId"
            },
            "Next": "CheckIfPreviousVersionExists",
            "Retry": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "BackoffRate": 2,
                    "IntervalSeconds": 1,
                    "MaxAttempts": 3,
                    "JitterStrategy": "FULL"
                }
            ],
            "TimeoutSeconds": 30,
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "ResultPath": "$.error",
                    "Next": "Fail"
                }
            ]
        },
        "Fail": {
            "Type": "Fail"
        },
        "CheckIfPreviousVersionExists": {
            "Type": "Choice",
            "Choices": [
                {
                    "Next": "CopyOldVersion",
                    "Variable": "$.OldVersionId",
                    "IsPresent": true
                }
            ],
            "Default": "Success"
        },
        "CopyOldVersion": {
            "Type": "Task",
            "Parameters": {
                "Bucket": "${content_s3_bucket}",
                "CopySource.$": "States.Format('${content_s3_bucket}/feed/anonymous/latest.json?versionId={}', $.OldVersionId)",
                "Key": "feed/anonymous/latest-previous.json",
                "MetadataDirective": "COPY",
                "TaggingDirective": "COPY"
            },
            "Resource": "arn:aws:states:::aws-sdk:s3:copyObject",
            "Next": "HeadObject",
            "ResultSelector": {
                "ArchivedVersionId.$": "$.VersionId"
            },
            "Retry": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "BackoffRate": 2,
                    "IntervalSeconds": 1,
                    "MaxAttempts": 3,
                    "JitterStrategy": "FULL"
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "Fail",
                    "ResultPath": "$.error"
                }
            ],
            "TimeoutSeconds": 30,
            "ResultPath": "$.ArchivedVersion"
        },
        "HeadObject": {
            "Type": "Task",
            "Parameters": {
                "Bucket": "${content_s3_bucket}",
                "Key": "feed/anonymous/latest-previous.json",
                "VersionId.$": "$.ArchivedVersion.ArchivedVersionId"
            },
            "Resource": "arn:aws:states:::aws-sdk:s3:headObject",
            "Next": "DeleteObject",
            "Retry": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "BackoffRate": 2,
                    "IntervalSeconds": 1,
                    "MaxAttempts": 3,
                    "JitterStrategy": "FULL"
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "Fail",
                    "ResultPath": "$.error"
                }
            ],
            "TimeoutSeconds": 30,
            "ResultPath": "$.verificationresults"
        },
        "DeleteObject": {
            "Type": "Task",
            "Parameters": {
                "Bucket": "${content_s3_bucket}",
                "Key": "feed/anonymous/latest.json",
                "VersionId.$": "$.OldVersionId"
            },
            "Resource": "arn:aws:states:::aws-sdk:s3:deleteObject",
            "Retry": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "BackoffRate": 2,
                    "IntervalSeconds": 1,
                    "MaxAttempts": 3,
                    "JitterStrategy": "FULL"
                }
            ],
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "Fail",
                    "ResultPath": "$.error"
                }
            ],
            "TimeoutSeconds": 30,
            "Next": "ValidateAnonFeed"
        },
        "ValidateAnonFeed": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "Parameters": {
                "FunctionName": "${validate_anonymous_feed}"
            },
            "Retry": [
                {
                    "ErrorEquals": [
                        "Lambda.ServiceException",
                        "Lambda.AWSLambdaException",
                        "Lambda.SdkClientException",
                        "Lambda.TooManyRequestsException"
                    ],
                    "IntervalSeconds": 1,
                    "MaxAttempts": 3,
                    "BackoffRate": 2,
                    "JitterStrategy": "FULL"
                }
            ],
            "End": true
        },
        "Success": {
            "Type": "Succeed"
        }
    }
}

Lambda Functions in the Workflow
The workflow relies on multiple AWS Lambda functions, each designed with a single responsibility to keep the architecture modular and maintainable.

GenerateFeedLambda
Purpose: Calls the API to retrieve the latest 100 content items.
Input: Triggered at the start of the workflow.
Output: A list of content IDs passed into the first Map state.
GenerateIndivcontentJsonsLambda

Purpose: Generate json for each of 100 items and save in s3 bucket. Output latest 100 content items to next stage.
Input: Single content ID.
Output: A list of content IDs passed into the first Map state.

GenerateContentCommentJsonsLambda

Purpose: Given a content ID, calls the API to fetch the 16 latest comments for that content and save each in s3.
Input: Single content ID.
Output: List of comment IDs saved in s3. pass object name for the next processing stage.

Replies Fetcher Lambda (Level 1/Level 2)

Purpose: Given a comment ID, fetches the 16 latest replies and save each in s3. In our case same API can be used to generate replies and nested replies. So same lambda is used in step function.
Input: Single comment ID.
Output: List of reply IDs saved in s3. pass object name for the next nested Map state.

Consolidateoutput Lambda

Purpose: output from previous stage saved in s3 is not a list. we need a list type object to pass to map stage. this lambda does that conversion.
Input: s3 object name (output from previous stage saved in s3)
Output: JSON object written to S3 with a structured

Performance metrics
⚡ By distributing work across nested Map states, the pipeline scaled horizontally without requiring manual queue or concurrency management. (duration ~30s to generate all)

Step Functions vs Manual Orchestration
Manual Orchestration (Queues + Custom Logic)

Must design and manage worker pools manually
Requires custom retry/backoff logic
Logs spread across services, harder to trace
Pay for compute + queue infrastructure
Higher engineering effort, more boilerplate

AWS Step Functions (Map State)

Automatically scales with Map state parallelism
Built-in retries, catchers, error paths
Centralized execution history and visualization
Pay per state transition + Lambda usage
Faster implementation with declarative workflow
Simplified workflow as state machine JSON

This demonstrated the power of combining AWS Step Functions with Lambda to handle large-scale, highly nested workloads. By leveraging the Map state effectively, I was able to orchestrate large amount of dependent API calls and generate 409,600 JSON files — all while keeping the system resilient, observable, and cost-efficient.

Transferring Data Between Amazon S3 Buckets Across AWS Accounts with AWS DataSync

Chanaka Supun — Thu, 27 Mar 2025 17:33:00 +0000

Introduction

In multi-account AWS environments, teams often need to transfer data across Amazon S3 buckets residing in different AWS accounts. While traditional methods like S3 cross-account replication or AWS CLI-based transfers exist, AWS DataSync provides a more robust, managed solution that enables automated, secure, and high-performance data transfers with monitoring and scheduling capabilities.

This guide walks you through setting up AWS DataSync to transfer data from a source S3 bucket in one AWS account to a destination S3 bucket in another AWS account.

Step 1: Create IAM Roles for AWS DataSync
AWS DataSync requires IAM roles to access the source and destination S3 buckets securely.

1.1 Create IAM Role in the Source Account
Role: datasync-source-role

Go to IAM → Roles → Create Role
Select AWS Service → Choose DataSync as the trusted entity
Attach the following trust policy to allow DataSync to assume this role:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "datasync.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

Attach the following inline policy to allow access to the source S3 bucket:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "Statement1",
            "Effect": "Allow",
            "Action": [
                "s3:GetBucketLocation",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads"
            ],
            "Resource": [
                "arn:aws:s3:::source-bucket"
            ]
        },
        {
            "Sid": "Statement2",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:GetObjectTagging",
                "s3:ListMultipartUploadParts"
            ],
            "Resource": [
                "arn:aws:s3:::source-bucket/*"
            ]
        }
    ]
}

Role: datasync-destination-role

Go to IAM → Roles → Create Role
Select AWS Service → Choose DataSync
Attach the following trust policy to allow DataSync to assume this role:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "datasync.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

Attach the following inline policy to allow DataSync to write to the destination S3 bucket:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "Statement1",
            "Effect": "Allow",
            "Action": [
                "s3:GetBucketLocation",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads"
            ],
            "Resource": [
                "arn:aws:s3:::destination-bucket"
            ]
        },
        {
            "Sid": "Statement2",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:GetObjectTagging",
                "s3:ListMultipartUploadParts",
                "s3:PutObject",
                "s3:PutObjectTagging",
                "s3:AbortMultipartUpload",
                "s3:DeleteObject"
            ],
            "Resource": [
                "arn:aws:s3:::destination-bucket/*"
            ]
        }
    ]
}

Step 2: Update the S3 Bucket Policy in the Destination Account
To allow AWS DataSync to write data to the destination S3 bucket, add the following bucket policy to destination-bucket:

Navigate to S3 → destination-bucket → Permissions → Bucket Policy
Add the following policy, replacing SOURCE_ACCOUNT_ID with the actual AWS Account ID:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "Allowdatasync",
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::SOURCE_ACCOUNT_ID:role/datasync-destination-role"
            },
            "Action": [
                "s3:List*",
                "s3:Get*",
                "s3:AbortMultipartUpload",
                "s3:DeleteObject",
                "s3:PutObject",
                "s3:PutObjectTagging"
            ],
            "Resource": [
                "arn:aws:s3:::destination-bucket/*",
                "arn:aws:s3:::destination-bucket"
            ]
        }
    ]
}

Step 3: Configure AWS DataSync in the Source Account
3.1 Create the Source Location in AWS DataSync

Open the AWS DataSync console in the source account.
Click Create *Location *→ Select Amazon S3.
Choose the source-bucket as the location.
Select datasync-source-role as the IAM role.
Click Create Location.
3.2 Create the Destination Location (via AWS CloudShell)
Since the destination S3 bucket belongs to a different AWS account, use AWS CLI (or AWS CloudShell) to create the destination location.
Run the following AWS CLI command in the source account:

aws datasync create-location-s3 \
    --s3-bucket-arn arn:aws:s3:::destination-bucket \
    --s3-config '{ "BucketAccessRoleArn": "arn:aws:iam::SOURCE_ACCOUNT_ID:role/datasync-destination-role" }' \
    --region ap-southeast-2

Replace ap-southeast-2 with the region where your bucket exists.

you will get a response back as follows.

{
    "LocationArn": "arn:aws:datasync:ap-southeast-2:SOURCE_ACCOUNT_ID:location/loc-xxxxxxxx"
}

Then refresh the page will be able to see two locations now created.

Step 4: Create and Start the DataSync Task
4.1 Create the DataSync Task

Open AWS DataSync in the source account.
Click Create Task.
Select Source Location (previously created).

Next Select Destination Location (created via CLI).
Configure the following settings:
Task Mode: Ensure Enhanced.
keep rest of the settings as default.
Click Create Task.

4.2 Start the DataSync Task

Once the task is created, start the transfer using the AWS CLI or console.

Start via Console:
Navigate to AWS DataSync → Tasks.
Select your newly created task.
Click Start Task with defaults.

Step 5: Monitor the Data Transfer

Go to AWS DataSync → Task Executions to view progress.
Check CloudWatch logs for errors if the task fails.
Validate that files appear in the destination S3 bucket after completion.

Conclusion
By following this approach, you can securely and efficiently transfer data across AWS accounts using AWS DataSync. This solution offers:

Automated Data Movement: No need for manual copying.
Incremental Transfers: Only modified files are transferred.
Monitoring & Logs: AWS CloudWatch integration for tracking.
Scalability: Can handle large-scale transfers.
This method is ideal for one-time migrations as well as ongoing cross-account synchronization.

Public http API Gateway to Private API Gateway : Cross-Account Integration

Chanaka Supun — Tue, 25 Mar 2025 18:33:00 +0000

When architecting services across multiple AWS accounts—especially in a bulkhead architecture—enabling secure service-to-service communication is crucial. A Private API Gateway is often used to keep internal APIs accessible only within a VPC, but what if you need external access?

A Regional API Gateway with an authorizer (e.g., Lambda Authorizer) can act as a single-entry point for multiple private endpoints. This article demonstrates how to securely connect a Public Regional API Gateway in one AWS account (Account A) to a Private API Gateway in another AWS account (Account B), using:

✅ VPC Link to securely route traffic to internal services
✅ Network Load Balancer (NLB) to distribute traffic efficiently
✅ VPC Endpoint to expose the Private API Gateway securely

🏗 Prerequisites
Before starting, ensure you have:

Two AWS accounts (Account A & Account B)
VPC networking in both accounts with Transit Gateway (TGW) connectivity

📌 Architecture Overview
The diagram below illustrates the integration:

Key Components:
1️⃣ Private API Gateway (Account B) – Hosts internal APIs, accessible via a VPC Endpoint
2️⃣ Network Load Balancer (NLB) (Account A) – Routes traffic from the Public API Gateway to the Private API Gateway
3️⃣ VPC Link (Account A) – Connects the Public API Gateway to the NLB
4️⃣ Regional API Gateway (Account A) – Exposes the Private API securely to external users

Lets Get started !

🏗 Step 1: Create a Private API Gateway & VPC Endpoint (Account B)

1️⃣ Log into Account B and navigate to VPC Endpoints

2️⃣ Create a new VPC Endpoint:

VPC: Select the VPC where the Private API Gateway resides
Security Group: Allow inbound traffic from Account A

Policy: Set to Full Access for now

Then click create. Will take few minutes to status to become available.

3️⃣ Once created, note down the IP addresses of the VPC Endpoint

Step 2: Create a Private API Gateway (Account B)

1️⃣ Log into Account B and navigate to API Gateway

Select rest api private.

3️⃣ Create a resource and a method (e.g., GET /test)
4️⃣ Configure a mock integration for testing:

I'm using mock endpoint here. It will just return.

{
 "response" : "integration-successful"
}

Select method and go to test.

5️⃣ Set up a Resource Policy to allow traffic only from VPC Endpoint in Account B:

For resource include arn of private endpoints and sourcevpce should be id of vpc endpoint created above. This will make sure traffic will accepted only from vpc endpoint.

6️⃣ Deploy the API

Step 3 : Create a Regional API Gateway (Account A)

Type: REST API (Regional)

Name: e.g., http-gateway

Create the http gateway.

Step 3 : Create a Network Load Balancer (Account A)

1️⃣ Navigate to EC2 > Load Balancers

2️⃣ Create a new Network Load Balancer

Scheme: Internal
VPC: Must have connectivity to the Private API Gateway's VPC
Subnets: Private subnets in Account A

3️⃣ Create a Target Group:

Target Type: IP
Port: 443 (for HTTPS traffic to Private API Gateway)
VPC: that has connectivity with Private API Gateway

Target IPs: Enter VPC Endpoint IPs noted earlier

For IP addresses enter the Ip's of VPV endpoint previously note down in step 01 and keep port as 443.

4️⃣ Attach the Target Group to the Load Balancer
5️⃣ Create and activate the NLB

Step 4 : Create a VPC Link (Account A)

1️⃣ Navigate to API Gateway > VPC Links

2️⃣ Create a new VPC Link

Type: Network Load Balancer
NLB ARN: Select the NLB created in Account A

3️⃣ Wait for the VPC Link to become available

step 5 : Integrate http API Gateway with VPC Link (Account A)

1️⃣ Go back to the http API Gateway

2️⃣ Create a new resource & method

Method Type: ANY
PATH (use greedy path to pass all requests to private API: {proxy+}
Integration Type: VPC Link
VPC Link: Select the one created earlier
Endpoint URL: Use the Private API Gateway Invoke URL from Step 1.

Select the load balancer created previously

Select the vpc link created above.

in the advance section for server name include invoke domain of private api gateway. Otherwise will get a certificate validation error.

Step 6 : Test

1️⃣ test the integration ✅
When sending the request to the regional api gateway invoke url need to include the header x-apigw-api-id with value as private api gateway ID.