Step-by-Step CDC Recovery Guide: Oracle to PostgreSQL Migration

When dealing with AWS DMS CDC failures, you have options for restoring CDC from specific points in time, but there are important constraints:

Recovery Point Options for CDC

1. Using CdcStartPosition

   • You can modify a replication task to start from a specific transaction log position

   • For Oracle, this is an SCN (System Change Number)

   • For PostgreSQL, it's an LSN (Log Sequence Number)

   • For MySQL, it's a binary log position

2. Using CdcStartTime

   • Instead of a specific log position, you can set a timestamp

   • DMS will attempt to find the corresponding position in the transaction logs

   • Example (AWS CLI):

   Copyaws dms modify-replication-task \
     --replication-task-arn <task-arn> \
     --cdc-start-time "2025-01-15T12:00:00Z"

Critical Limitations

1. Transaction Log Retention

   • You can only go back as far as your source database retains transaction logs

   • For Oracle: redo logs and archived logs

   • For PostgreSQL: WAL segments

   • For MySQL: binary logs

2. Log Availability

   • If logs have been purged or are unavailable, you cannot restart CDC from that point

   • Many database systems automatically purge older logs

3. Database-Specific Constraints

   • Oracle: Requires ARCHIVELOG mode and appropriate retention

   • PostgreSQL: Requires logical replication slots to be maintained

   • MySQL: Binary log expiration policy limits recovery points

Best Practices for CDC Recovery

1. Configure adequate log retention

   • Extend transaction log retention periods on source databases

   • For Oracle, ensure ARCHIVELOG mode is enabled with adequate retention

   • For PostgreSQL, monitor and maintain logical replication slots

   • For MySQL, set binlog_expire_logs_seconds appropriately

2. Create recovery checkpoints

   • Periodically capture and store the current log position

   • Consider creating DMS task bookmarks for critical points

   • Document SCN/LSN positions after major data loads or changes

3. Implement monitoring for log retention

   • Monitor source database log space and retention

   • Set up alerts for log purging events

   • Track CDC lag metrics in CloudWatch

If logs for your desired recovery point have been purged, your only option would be to perform a new full load and restart CDC from that point. This is why preventive measures like extended log retention are crucial for CDC recovery strategies.

Step-by-step Gide

This guide explains how to set up, maintain, and recover CDC (Change Data Capture) during an Oracle to PostgreSQL migration using AWS DMS.

Part 1: Prerequisites and Configuration

Step 1: Configure Oracle Source for Optimal CDC Recovery

1. Enable ARCHIVELOG Mode (if not already enabled):

   -- Connect as SYSDBA
   SHUTDOWN IMMEDIATE;
   STARTUP MOUNT;
   ALTER DATABASE ARCHIVELOG;
   ALTER DATABASE OPEN;
   
   -- Verify ARCHIVELOG mode is enabled
   SELECT log_mode FROM v$database;

2. Configure Supplemental Logging:

   -- Enable database-level supplemental logging
   ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
   
   -- Enable primary key supplemental logging for all tables
   ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;
   
   -- For tables without primary keys, add all columns
   ALTER TABLE schema.table_without_pk ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

3. Configure Adequate Redo Log Retention:

   -- Check current retention policy
   SELECT name, value FROM v$parameter WHERE name = 'db_recovery_file_dest_size';
   
   -- Increase the retention space (adjust size as needed)
   ALTER SYSTEM SET db_recovery_file_dest_size = 100G SCOPE=BOTH;
   
   -- Set archive log retention hours (e.g., 168 hours = 7 days)
   ALTER SYSTEM SET db_flashback_retention_target = 168 SCOPE=BOTH;

4. Create a Regular SCN Checkpoint Process:

   -- Create table to store SCN checkpoints
   CREATE TABLE admin.dms_scn_checkpoints (
     checkpoint_id NUMBER GENERATED ALWAYS AS IDENTITY,
     checkpoint_time TIMESTAMP DEFAULT SYSTIMESTAMP,
     checkpoint_name VARCHAR2(100),
     scn NUMBER,
     notes VARCHAR2(4000),
     PRIMARY KEY (checkpoint_id)
   );
   
   -- Procedure to capture current SCN
   CREATE OR REPLACE PROCEDURE admin.capture_scn_checkpoint(
     p_checkpoint_name IN VARCHAR2,
     p_notes IN VARCHAR2 DEFAULT NULL
   )
   AS
     v_current_scn NUMBER;
   BEGIN
     SELECT CURRENT_SCN INTO v_current_scn FROM V$DATABASE;
     
     INSERT INTO admin.dms_scn_checkpoints (
       checkpoint_name, scn, notes
     ) VALUES (
       p_checkpoint_name, v_current_scn, p_notes
     );
     
     COMMIT;
   END;
   /
   
   -- Schedule to run every 6 hours
   BEGIN
     DBMS_SCHEDULER.CREATE_JOB (
       job_name        => 'SCN_CHECKPOINT_JOB',
       job_type        => 'STORED_PROCEDURE',
       job_action      => 'admin.capture_scn_checkpoint',
       start_date      => SYSTIMESTAMP,
       repeat_interval => 'FREQ=HOURLY;INTERVAL=6',
       enabled         => TRUE,
       comments        => 'Job to capture SCN checkpoints every 6 hours',
       auto_drop       => FALSE,
       job_class       => 'DEFAULT_JOB_CLASS'
     );
     
     -- Pass parameters for scheduled job
     DBMS_SCHEDULER.SET_JOB_ARGUMENT_VALUE (
       job_name          => 'SCN_CHECKPOINT_JOB',
       argument_position => 1,
       argument_value    => 'SCHEDULED_CHECKPOINT'
     );
   END;
   /

5. Monitor Oracle Archive Log Generation:

   -- Create monitoring view
   CREATE OR REPLACE VIEW admin.archive_log_stats AS
   SELECT
     TRUNC(FIRST_TIME) AS log_date,
     COUNT(*) AS logs_generated,
     ROUND(SUM(BLOCKS * BLOCK_SIZE) / 1024 / 1024 / 1024, 2) AS size_gb
   FROM
     V$ARCHIVED_LOG
   WHERE
     FIRST_TIME > SYSDATE - 7
   GROUP BY
     TRUNC(FIRST_TIME)
   ORDER BY
     log_date;

Step 2: Configure PostgreSQL Target for CDC

1. Set Up Target PostgreSQL with Appropriate Settings:

   -- Connect to PostgreSQL and create a CDC status tracking table
   CREATE SCHEMA IF NOT EXISTS admin;
   
   CREATE TABLE admin.dms_cdc_checkpoints (
     checkpoint_id SERIAL PRIMARY KEY,
     checkpoint_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     checkpoint_name VARCHAR(100),
     task_arn TEXT,
     oracle_scn NUMERIC,
     postgres_lsn TEXT,
     notes TEXT
   );

2. Configure PostgreSQL for Performance During CDC:

   -- These settings may need to be adjusted in postgresql.conf
   
   -- Increase WAL level for logical replication compatibility
   ALTER SYSTEM SET wal_level = logical;
   
   -- Other helpful settings for DMS performance
   ALTER SYSTEM SET max_wal_senders = 10;
   ALTER SYSTEM SET max_replication_slots = 10;
   
   -- Apply changes
   SELECT pg_reload_conf();

Step 3: Configure AWS DMS for Recoverable CDC

1. Create Initial DMS Task with CDC in Mind:

   # Using AWS CLI to create a task with both full load and CDC
   aws dms create-replication-task \
     --replication-task-identifier "oracle-to-pg-migration" \
     --source-endpoint-arn "arn:aws:dms:region:account:endpoint:source-oracle-endpoint" \
     --target-endpoint-arn "arn:aws:dms:region:account:endpoint:target-pg-endpoint" \
     --replication-instance-arn "arn:aws:dms:region:account:rep:instance-name" \
     --migration-type "full-load-and-cdc" \
     --table-mappings file://table-mappings.json \
     --replication-task-settings file://task-settings.json

2. Configure task-settings.json with Logging and Oracle Settings:

   {
     "TargetMetadata": {
       "TargetSchema": "",
       "SupportLobs": true,
       "FullLobMode": false,
       "LimitedSizeLobMode": true,
       "LobChunkSize": 64,
       "LobMaxSize": 32,
       "BatchApplyEnabled": true
     },
     "FullLoadSettings": {
       "TargetTablePrepMode": "DO_NOTHING",
       "CreatePkAfterFullLoad": false,
       "StopTaskCachedChangesApplied": false,
       "StopTaskCachedChangesNotApplied": false,
       "MaxFullLoadSubTasks": 8,
       "TransactionConsistencyTimeout": 600,
       "CommitRate": 10000
     },
     "Logging": {
       "EnableLogging": true,
       "LogComponents": [
         {
           "Id": "TRANSFORMATION",
           "Severity": "LOGGER_SEVERITY_DEFAULT"
         },
         {
           "Id": "SOURCE_UNLOAD",
           "Severity": "LOGGER_SEVERITY_DEFAULT"
         },
         {
           "Id": "IO",
           "Severity": "LOGGER_SEVERITY_DEFAULT"
         },
         {
           "Id": "TARGET_LOAD",
           "Severity": "LOGGER_SEVERITY_DEFAULT"
         },
         {
           "Id": "PERFORMANCE",
           "Severity": "LOGGER_SEVERITY_DEFAULT"
         }
       ]
     },
     "OracleSettings": {
       "ReadTableSpaceName": false,
       "EnableHomogenousTablespace": false,
       "StandbyDelayTime": 0,
       "ArchivedLogsOnly": false,
       "ArchivedLogDestId": 0,
       "UsePathPrefix": "",
       "UseLogminerReader": true,
       "SecurityDbEncryption": false,
       "SecurityDbEncryptionName": "",
       "DirectPathNoLog": false,
       "AllowSelectNestedTables": true,
       "ConvertTimestampWithZoneToUTC": true,
       "AddSupplementalLogging": true
     },
     "ChangeProcessingTuning": {
       "BatchApplyPreserveTransaction": true,
       "BatchApplyTimeoutMin": 1,
       "BatchApplyTimeoutMax": 30,
       "BatchApplyMemoryLimit": 500,
       "BatchSplitSize": 0,
       "MinTransactionSize": 1000,
       "CommitTimeout": 1,
       "MemoryLimitTotal": 1024,
       "MemoryKeepTime": 60,
       "StatementCacheSize": 50
     }
   }

3. Create a Checkpoint Recording Script:

   #!/bin/bash
   # dms-record-checkpoints.sh
   
   # Configuration
   TASK_ARN="arn:aws:dms:region:account:task:your-task-arn"
   PG_HOST="your-postgres-host"
   PG_PORT="5432"
   PG_DB="your-postgres-db"
   PG_USER="your-postgres-user"
   PG_PASSWORD="your-postgres-password"
   
   # Get DMS task status
   TASK_INFO=$(aws dms describe-replication-tasks \
     --filters Name=replication-task-arn,Values=$TASK_ARN \
     --query 'ReplicationTasks[0]')
   
   TASK_STATUS=$(echo $TASK_INFO | jq -r '.Status')
   
   # Only record if task is running
   if [ "$TASK_STATUS" != "running" ]; then
     echo "Task is not running (status: $TASK_STATUS). Skipping checkpoint."
     exit 0
   fi
   
   # Create checkpoint name based on date/time
   CHECKPOINT_NAME="AUTO_$(date +%Y%m%d_%H%M%S)"
   NOTES="Automatic checkpoint created by monitoring script"
   
   # Query Oracle SCN (requires connection to Oracle)
   ORACLE_SCN=$(sqlplus -s $ORACLE_USER/$ORACLE_PASSWORD@$ORACLE_CONNECTION << EOF
   SET HEADING OFF FEEDBACK OFF VERIFY OFF
   SELECT CURRENT_SCN FROM V\$DATABASE;
   EXIT;
   EOF
   )
   ORACLE_SCN=$(echo $ORACLE_SCN | tr -d ' ')
   
   # Determine last processed SCN from DMS if available
   CACHED_CHANGES=$(echo $TASK_INFO | jq -r '.ReplicationTaskStats.CdcLatencySource')
   
   # Record checkpoint in PostgreSQL
   PGPASSWORD=$PG_PASSWORD psql -h $PG_HOST -p $PG_PORT -d $PG_DB -U $PG_USER << EOF
   INSERT INTO admin.dms_cdc_checkpoints 
   (checkpoint_name, task_arn, oracle_scn, notes)
   VALUES 
   ('$CHECKPOINT_NAME', '$TASK_ARN', $ORACLE_SCN, '$NOTES - CDC Latency: $CACHED_CHANGES seconds');
   EOF
   
   echo "Recorded checkpoint $CHECKPOINT_NAME with SCN $ORACLE_SCN"

4. Schedule Checkpoint Script:

   # Add to crontab to run every hour
   crontab -e
   
   # Add this line to run every hour
   0 * * * * /path/to/dms-record-checkpoints.sh >> /var/log/dms-checkpoints.log 2>&1

Part 2: CDC Recovery Procedures

Step 1: Identifying Available Recovery Points

1. Check Available Oracle SCN Checkpoints:

   -- Connect to Oracle and run:
   SELECT
     checkpoint_id,
     checkpoint_time,
     checkpoint_name,
     scn,
     notes
   FROM admin.dms_scn_checkpoints
   ORDER BY checkpoint_time DESC;

2. Check Available Archive Logs:

   -- Connect to Oracle and run:
   SELECT
     sequence#,
     first_change#,
     first_time,
     next_change#,
     next_time,
     archived,
     status
   FROM v$archived_log
   WHERE first_time > SYSDATE - 7
   ORDER BY sequence# DESC;

3. Check Minimum Available SCN:

   -- Connect to Oracle and run:
   SELECT
     MIN(first_change#) AS min_available_scn,
     TO_CHAR(MIN(first_time), 'YYYY-MM-DD HH24:MI:SS') AS min_available_time
   FROM v$archived_log
   WHERE status = 'A';

4. Check Recorded DMS Checkpoints in PostgreSQL:

   -- Connect to PostgreSQL and run:
   SELECT
     checkpoint_id,
     checkpoint_time,
     checkpoint_name,
     task_arn,
     oracle_scn,
     notes
   FROM admin.dms_cdc_checkpoints
   ORDER BY checkpoint_time DESC;

Step 2: Recover CDC from a Specific SCN

1. Stop the Current DMS Task:

   aws dms stop-replication-task \
     --replication-task-arn "arn:aws:dms:region:account:task:your-task-arn"

2. Modify Task to Start from Specific SCN:

   aws dms modify-replication-task \
     --replication-task-arn "arn:aws:dms:region:account:task:your-task-arn" \
     --cdc-start-position "Oracle SCN:6916533"

3. Start Task with New SCN Position:

   aws dms start-replication-task \
     --replication-task-arn "arn:aws:dms:region:account:task:your-task-arn" \
     --start-replication-task-type start-replication

4. Monitor CDC Recovery Progress:

   # Check task status
   aws dms describe-replication-tasks \
     --filters Name=replication-task-arn,Values="arn:aws:dms:region:account:task:your-task-arn" \
     --query 'ReplicationTasks[0].{Status:Status,CDCLatencySource:ReplicationTaskStats.CdcLatencySource,CDCLatencyTarget:ReplicationTaskStats.CdcLatencyTarget}'

Step 3: Recover CDC from a Specific Timestamp

1. Stop the Current DMS Task:

   aws dms stop-replication-task \
     --replication-task-arn "arn:aws:dms:region:account:task:your-task-arn"

2. Find the Nearest SCN for a Timestamp:

   -- Connect to Oracle and run (replace timestamp):
   SELECT
     scn
   FROM admin.dms_scn_checkpoints
   WHERE checkpoint_time <= TO_TIMESTAMP('2025-02-15 14:30:00', 'YYYY-MM-DD HH24:MI:SS')
   ORDER BY checkpoint_time DESC
   FETCH FIRST 1 ROW ONLY;
   
   -- Alternative using Oracle timestamp:
   SELECT timestamp_to_scn(TO_TIMESTAMP('2025-02-15 14:30:00', 'YYYY-MM-DD HH24:MI:SS')) FROM dual;

3. Modify Task to Start from a Specific Time:

   # Using timestamp directly
   aws dms modify-replication-task \
     --replication-task-arn "arn:aws:dms:region:account:task:your-task-arn" \
     --cdc-start-time "2025-02-15T14:30:00Z"
   
   # Alternative using SCN from previous step
   aws dms modify-replication-task \
     --replication-task-arn "arn:aws:dms:region:account:task:your-task-arn" \
     --cdc-start-position "Oracle SCN:6945122"

4. Start Task with New Time Position:

   aws dms start-replication-task \
     --replication-task-arn "arn:aws:dms:region:account:task:your-task-arn" \
     --start-replication-task-type start-replication

5. Verify Recovery Point Success:

   # Monitor initial logs for successful connection to source at recovery point
   aws logs get-log-events \
     --log-group-name /aws/dms/replication-task/<task-id> \
     --log-stream-name <task-log-stream> \
     --limit 100

Step 4: Handling Recovery Failures

1. Check for Oracle Archive Log Availability Issues:

   -- Connect to Oracle and run:
   SELECT
     sequence#,
     first_change#,
     next_change#,
     archived,
     status
   FROM v$archived_log
   WHERE first_change# <= 6916533 AND next_change# > 6916533;

2. Validate SCN Visibility to LogMiner:

   -- Connect to Oracle and run:
   BEGIN
     DBMS_LOGMNR.START_LOGMNR(
       STARTSCN => 6916533,
       OPTIONS => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG +
                 DBMS_LOGMNR.CONTINUOUS_MINE
     );
     
     -- Check if we can see transaction data at this SCN
     FOR rec IN (
       SELECT operation, sql_redo 
       FROM v$logmnr_contents 
       WHERE scn >= 6916533 AND ROWNUM <= 10
     ) LOOP
       DBMS_OUTPUT.PUT_LINE(rec.operation || ': ' || rec.sql_redo);
     END LOOP;
     
     DBMS_LOGMNR.END_LOGMNR;
   EXCEPTION
     WHEN OTHERS THEN
       DBMS_OUTPUT.PUT_LINE('Error: ' || SQLERRM);
       DBMS_LOGMNR.END_LOGMNR;
   END;
   /

3. If Recovery Point Unavailable, Find Nearest Available SCN:

   -- Connect to Oracle and run:
   SELECT
     sequence#,
     first_change# AS start_scn,
     first_time AS start_time,
     next_change# AS end_scn,
     next_time AS end_time
   FROM v$archived_log
   WHERE status = 'A' AND archived = 'YES'
   ORDER BY sequence#;

4. Create a New Recovery Task if Original Can't Be Recovered:

   # If existing task can't be recovered, create a new one
   aws dms create-replication-task \
     --replication-task-identifier "oracle-to-pg-recovery-$(date +%Y%m%d)" \
     --source-endpoint-arn "arn:aws:dms:region:account:endpoint:source-oracle-endpoint" \
     --target-endpoint-arn "arn:aws:dms:region:account:endpoint:target-pg-endpoint" \
     --replication-instance-arn "arn:aws:dms:region:account:rep:instance-name" \
     --migration-type "cdc" \
     --table-mappings file://table-mappings.json \
     --replication-task-settings file://task-settings.json \
     --cdc-start-position "Oracle SCN:7001234"

Part 3: Post-Recovery Validation

1. Verify Data Consistency After Recovery:

   -- Create a PostgreSQL function to check record counts
   CREATE OR REPLACE FUNCTION admin.verify_table_counts(
     start_time TIMESTAMP DEFAULT NULL
   ) RETURNS TABLE (
     table_name TEXT,
     record_count BIGINT,
     updated_since_recovery BIGINT
   ) AS $$
   DECLARE
     tables RECORD;
     query TEXT;
     count_result BIGINT;
     updated_result BIGINT;
   BEGIN
     FOR tables IN 
       SELECT table_schema, table_name 
       FROM information_schema.tables 
       WHERE table_schema = 'public' 
       AND table_type = 'BASE TABLE'
     LOOP
       table_name := tables.table_schema || '.' || tables.table_name;
       
       -- Count all records
       query := 'SELECT COUNT(*) FROM ' || table_name;
       EXECUTE query INTO count_result;
       record_count := count_result;
       
       -- Count recently updated records
       IF start_time IS NOT NULL THEN
         BEGIN
           query := 'SELECT COUNT(*) FROM ' || table_name || 
                    ' WHERE updated_at >= $1';
           EXECUTE query USING start_time INTO updated_result;
           updated_since_recovery := updated_result;
         EXCEPTION WHEN OTHERS THEN
           updated_since_recovery := 0; -- Updated_at column might not exist
         END;
       ELSE
         updated_since_recovery := 0;
       END IF;
       
       RETURN NEXT;
     END LOOP;
     RETURN;
   END;
   $$ LANGUAGE plpgsql;
   
   -- Run the validation
   SELECT * FROM admin.verify_table_counts('2025-02-15 14:30:00');

2. Monitor CDC Performance After Recovery:

   # Create a monitoring script
   cat << 'EOF' > monitor-cdc-latency.sh
   #!/bin/bash
   
   TASK_ARN="arn:aws:dms:region:account:task:your-task-arn"
   
   while true; do
     NOW=$(date +"%Y-%m-%d %H:%M:%S")
     TASK_STATS=$(aws dms describe-replication-tasks \
       --filters Name=replication-task-arn,Values=$TASK_ARN \
       --query 'ReplicationTasks[0].ReplicationTaskStats')
     
     SOURCE_LATENCY=$(echo $TASK_STATS | jq -r '.CdcLatencySource')
     TARGET_LATENCY=$(echo $TASK_STATS | jq -r '.CdcLatencyTarget')
     
     echo "$NOW - Source Latency: $SOURCE_LATENCY seconds, Target Latency: $TARGET_LATENCY seconds"
     
     sleep 60
   done
   EOF
   
   chmod +x monitor-cdc-latency.sh
   ./monitor-cdc-latency.sh

3. Create Regular Data Validation Checks:

   -- Example check to compare row counts in critical tables
   -- Must be adapted to your specific validation needs
   CREATE OR REPLACE PROCEDURE admin.validate_critical_tables()
   LANGUAGE plpgsql
   AS $$
   DECLARE
     validation_passed BOOLEAN := TRUE;
     total_customers INTEGER;
     total_orders INTEGER;
     total_order_items INTEGER;
     validation_msg TEXT := '';
   BEGIN
     -- Get counts from critical tables
     SELECT COUNT(*) INTO total_customers FROM public.customers;
     SELECT COUNT(*) INTO total_orders FROM public.orders;
     SELECT COUNT(*) INTO total_order_items FROM public.order_items;
     
     -- Basic validation rules
     IF total_orders > 0 AND total_order_items = 0 THEN
       validation_passed := FALSE;
       validation_msg := validation_msg || 'Orders exist but no order items. ';
     END IF;
     
     IF total_customers = 0 THEN
       validation_passed := FALSE;
       validation_msg := validation_msg || 'No customers found. ';
     END IF;
     
     -- Log validation results
     INSERT INTO admin.data_validations (
       validation_time, passed, details
     ) VALUES (
       CURRENT_TIMESTAMP, validation_passed, validation_msg
     );
     
     -- Raise notice with results
     RAISE NOTICE 'Validation completed. Passed: %, Message: %', 
       validation_passed, COALESCE(validation_msg, 'No issues found.');
   END;
   $$;
   
   -- Create table for validation results
   CREATE TABLE IF NOT EXISTS admin.data_validations (
     validation_id SERIAL PRIMARY KEY,
     validation_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     passed BOOLEAN,
     details TEXT
   );
   
   -- Run validation
   CALL admin.validate_critical_tables();

Part 4: Automating Recovery for High Availability

1. Create a CDCMonitor Function in AWS Lambda:

   import boto3
   import os
   import json
   import psycopg2
   from datetime import datetime, timedelta
   
   def lambda_handler(event, context):
       # Configuration
       task_arn = os.environ['DMS_TASK_ARN']
       pg_host = os.environ['PG_HOST']
       pg_port = os.environ['PG_PORT']
       pg_db = os.environ['PG_DB']
       pg_user = os.environ['PG_USER']
       pg_password = os.environ['PG_PASSWORD']
       
       # Initialize AWS clients
       dms = boto3.client('dms')
       
       # Get task status
       try:
           response = dms.describe_replication_tasks(
               Filters=[
                   {
                       'Name': 'replication-task-arn',
                       'Values': [task_arn]
                   }
               ]
           )
           
           task = response['ReplicationTasks'][0]
           status = task['Status']
           task_stats = task.get('ReplicationTaskStats', {})
           
           # Check for CDC issues
           source_latency = task_stats.get('CdcLatencySource', 0)
           target_latency = task_stats.get('CdcLatencyTarget', 0)
           
           # Connect to PostgreSQL to log status
           conn = psycopg2.connect(
               host=pg_host,
               port=pg_port,
               dbname=pg_db,
               user=pg_user,
               password=pg_password
           )
           
           cur = conn.cursor()
           
           # Log status to monitoring table
           cur.execute(
               """
               INSERT INTO admin.dms_monitoring
               (task_arn, status, source_latency, target_latency)
               VALUES (%s, %s, %s, %s)
               """,
               (task_arn, status, source_latency, target_latency)
           )
           
           # Check for recovery needs
           if status == 'failed' or source_latency > 1800:  # 30 minutes latency threshold
               # Get last good checkpoint
               cur.execute(
                   """
                   SELECT checkpoint_name, oracle_scn
                   FROM admin.dms_cdc_checkpoints
                   WHERE task_arn = %s AND oracle_scn IS NOT NULL
                   ORDER BY checkpoint_time DESC
                   LIMIT 1
                   """,
                   (task_arn,)
               )
               
               result = cur.fetchone()
               
               if result:
                   checkpoint_name, oracle_scn = result
                   
                   # Log recovery attempt
                   cur.execute(
                       """
                       INSERT INTO admin.dms_recovery_events
                       (task_arn, event_type, checkpoint_name, oracle_scn, notes)
                       VALUES (%s, %s, %s, %s, %s)
                       """,
                       (task_arn, 'AUTO_RECOVERY', checkpoint_name, oracle_scn, 
                        f"Automatic recovery triggered. Status: {status}, Source latency: {source_latency}")
                   )
                   
                   # Stop the task if it's not already stopped
                   if status != 'stopped':
                       dms.stop_replication_task(
                           ReplicationTaskArn=task_arn
                       )
                   
                   # Modify the task to restart from last good SCN
                   dms.modify_replication_task(
                       ReplicationTaskArn=task_arn,
                       CdcStartPosition=f"Oracle SCN:{oracle_scn}"
                   )
                   
                   # Start the task again
                   dms.start_replication_task(
                       ReplicationTaskArn=task_arn,
                       StartReplicationTaskType='start-replication'
                   )
                   
                   message = f"Automatic recovery initiated from SCN: {oracle_scn}"
               else:
                   message = "No valid checkpoint found for recovery"
           else:
               message = "CDC functioning normally"
           
           conn.commit()
           cur.close()
           conn.close()
           
           return {
               'statusCode': 200,
               'body': json.dumps({
                   'task_arn': task_arn,
                   'status': status,
                   'source_latency': source_latency,
                   'target_latency': target_latency,
                   'message': message
               })
           }
       
       except Exception as e:
           return {
               'statusCode': 500,
               'body': json.dumps({
                   'error': str(e)
               })
           }

2. Create Required PostgreSQL Tables for Lambda Function:

   -- Monitoring table
   CREATE TABLE IF NOT EXISTS admin.dms_monitoring (
     monitoring_id SERIAL PRIMARY KEY,
     monitoring_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     task_arn TEXT,
     status TEXT,
     source_latency NUMERIC,
     target_latency NUMERIC
   );
   
   -- Recovery events table
   CREATE TABLE IF NOT EXISTS admin.dms_recovery_events (
     event_id SERIAL PRIMARY KEY,
     event_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     task_arn TEXT,
     event_type TEXT,
     checkpoint_name TEXT,
     oracle_scn NUMERIC,
     notes TEXT
   );

3. Set up CloudWatch Events to Trigger Lambda Function:

   # Create CloudWatch Event Rule
   aws events put-rule \
     --name "DMSCDCMonitoringRule" \
     --schedule-expression "rate(5 minutes)" \
     --state ENABLED
   
   # Configure Lambda as target
   aws events put-targets \
     --rule "DMSCDCMonitoringRule" \
     --targets "Id"="1","Arn"="arn:aws:lambda:region:account:function:CDCMonitor"
   
   # Give CloudWatch Events permission to invoke Lambda
   aws lambda add-permission \
     --function-name CDCMonitor \
     --statement-id CloudWatchEvents \
     --action 'lambda:InvokeFunction' \
     --principal events.amazonaws.com \
     --source-arn arn:aws:events:region:account:rule/DMSCDCMonitoringRule

This comprehensive guide covers the setup, maintenance, and recovery of CDC during an Oracle to PostgreSQL migration using AWS DMS. The procedures are specifically designed for Oracle on-premises to PostgreSQL on AWS migrations, with an emphasis on establishing reliable recovery points and automating the recovery process.

Part 5: Long-Term CDC Maintenance

Step 1: Database-Side Maintenance

1. Oracle Archive Log Management:

   -- Create an Oracle procedure to monitor and alert on archive log space
   CREATE OR REPLACE PROCEDURE admin.monitor_archive_logs
   AS
     v_total_space NUMBER;
     v_used_space NUMBER;
     v_percent_used NUMBER;
     v_oldest_log_date DATE;
     v_alert_threshold NUMBER := 80; -- Alert at 80% usage
   BEGIN
     -- Get recovery file destination space information
     SELECT
       a.space_limit / 1024 / 1024 / 1024,
       a.space_used / 1024 / 1024 / 1024,
       ROUND(a.space_used / a.space_limit * 100, 2)
     INTO
       v_total_space,
       v_used_space,
       v_percent_used
     FROM
       v$recovery_file_dest a;
       
     -- Get oldest archive log date
     SELECT
       MIN(first_time)
     INTO
       v_oldest_log_date
     FROM
       v$archived_log
     WHERE
       archived = 'YES'
       AND deleted = 'NO';
       
     -- Log the information
     INSERT INTO admin.archive_log_monitor
     (total_space_gb, used_space_gb, percent_used, oldest_log_date)
     VALUES
     (v_total_space, v_used_space, v_percent_used, v_oldest_log_date);
     
     -- Alert if approaching capacity
     IF v_percent_used > v_alert_threshold THEN
       -- Send alert (implement your preferred alerting method)
       dbms_output.put_line('WARNING: Archive log space at ' || v_percent_used || '% capacity!');
       
       -- Log the alert
       INSERT INTO admin.dms_alerts
       (alert_type, severity, message)
       VALUES
       ('ARCHIVE_LOG_SPACE', 'WARNING', 
        'Archive log space at ' || v_percent_used || '% capacity. Oldest log: ' || v_oldest_log_date);
     END IF;
     
     COMMIT;
   EXCEPTION
     WHEN OTHERS THEN
       -- Log the error
       INSERT INTO admin.dms_alerts
       (alert_type, severity, message)
       VALUES
       ('ARCHIVE_LOG_SPACE', 'ERROR', 'Error monitoring archive logs: ' || SQLERRM);
       COMMIT;
   END;
   /
   
   -- Create supporting tables
   CREATE TABLE admin.archive_log_monitor (
     monitor_id NUMBER GENERATED ALWAYS AS IDENTITY,
     monitor_time TIMESTAMP DEFAULT SYSTIMESTAMP,
     total_space_gb NUMBER,
     used_space_gb NUMBER,
     percent_used NUMBER,
     oldest_log_date DATE,
     PRIMARY KEY (monitor_id)
   );
   
   CREATE TABLE admin.dms_alerts (
     alert_id NUMBER GENERATED ALWAYS AS IDENTITY,
     alert_time TIMESTAMP DEFAULT SYSTIMESTAMP,
     alert_type VARCHAR2(50),
     severity VARCHAR2(20),
     message VARCHAR2(4000),
     acknowledged CHAR(1) DEFAULT 'N',
     PRIMARY KEY (alert_id)
   );
   
   -- Schedule to run hourly
   BEGIN
     DBMS_SCHEDULER.CREATE_JOB (
       job_name        => 'ARCHIVE_LOG_MONITOR_JOB',
       job_type        => 'STORED_PROCEDURE',
       job_action      => 'admin.monitor_archive_logs',
       start_date      => SYSTIMESTAMP,
       repeat_interval => 'FREQ=HOURLY;INTERVAL=1',
       enabled         => TRUE,
       comments        => 'Job to monitor archive log space hourly',
       auto_drop       => FALSE
     );
   END;
   /

2. PostgreSQL Maintenance Procedures:

   -- Create a maintenance procedure for the target PostgreSQL
   CREATE OR REPLACE PROCEDURE admin.perform_dms_maintenance()
   LANGUAGE plpgsql
   AS $
   DECLARE
     v_current_month TEXT;
     v_old_month TEXT;
     v_table_exists BOOLEAN;
   BEGIN
     -- Create current month partition for monitoring data if needed
     v_current_month := to_char(current_date, 'YYYY_MM');
     
     -- Check if partition exists
     SELECT EXISTS (
       SELECT FROM information_schema.tables 
       WHERE table_schema = 'admin' 
       AND table_name = 'dms_monitoring_' || v_current_month
     ) INTO v_table_exists;
     
     -- Create new partition if needed
     IF NOT v_table_exists THEN
       EXECUTE 'CREATE TABLE admin.dms_monitoring_' || v_current_month || ' (
         LIKE admin.dms_monitoring INCLUDING ALL
       )';
       
       EXECUTE 'ALTER TABLE admin.dms_monitoring_' || v_current_month || 
               ' ADD CONSTRAINT dms_monitoring_' || v_current_month || '_month_check
                CHECK (monitoring_time >= DATE ''' || 
                to_char(date_trunc('month', current_date), 'YYYY-MM-DD') || 
                ''' AND monitoring_time < DATE ''' || 
                to_char(date_trunc('month', current_date + interval '1 month'), 'YYYY-MM-DD') || ''')';
                
       EXECUTE 'CREATE INDEX idx_dms_monitoring_' || v_current_month || '_time
                ON admin.dms_monitoring_' || v_current_month || ' (monitoring_time)';
     END IF;
     
     -- Purge old data (keeping last 90 days)
     DELETE FROM admin.dms_cdc_checkpoints
     WHERE checkpoint_time < current_date - interval '90 days';
     
     -- Move data from main table to partitions
     INSERT INTO admin.dms_monitoring_2025_03
     SELECT * FROM admin.dms_monitoring
     WHERE date_trunc('month', monitoring_time) = date_trunc('month', DATE '2025-03-01')
     ON CONFLICT DO NOTHING;
     
     -- Delete moved data from main table
     DELETE FROM admin.dms_monitoring
     WHERE date_trunc('month', monitoring_time) = date_trunc('month', DATE '2025-03-01');
     
     -- Vacuum tables to reclaim space
     VACUUM ANALYZE admin.dms_cdc_checkpoints;
     VACUUM ANALYZE admin.dms_monitoring;
     VACUUM ANALYZE admin.dms_recovery_events;
     
     -- Log maintenance completion
     INSERT INTO admin.maintenance_log
     (maintenance_type, details)
     VALUES
     ('DMS_TABLES', 'Routine maintenance completed: partition creation and data cleanup');
     
   EXCEPTION WHEN OTHERS THEN
     -- Log error
     INSERT INTO admin.maintenance_log
     (maintenance_type, details, error)
     VALUES
     ('DMS_TABLES', 'Maintenance failed', SQLERRM);
     RAISE;
   END;
   $;
   
   -- Create logging table
   CREATE TABLE IF NOT EXISTS admin.maintenance_log (
     log_id SERIAL PRIMARY KEY,
     log_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
     maintenance_type TEXT,
     details TEXT,
     error TEXT
   );

Step 2: Regular Health Checks

1. Oracle-side Health Check Script:

   #!/bin/bash
   # oracle_cdc_health_check.sh
   
   ORACLE_SID="your_oracle_sid"
   ORACLE_HOME="/path/to/oracle/home"
   ORACLE_USER="system"
   ORACLE_PASSWORD="your_password"
   
   export ORACLE_SID ORACLE_HOME
   
   echo "======== Oracle CDC Health Check: $(date) ========"
   
   # Check archive log mode
   echo "Checking archive log mode:"
   $ORACLE_HOME/bin/sqlplus -s $ORACLE_USER/$ORACLE_PASSWORD << EOF
   SET HEADING OFF FEEDBACK OFF VERIFY OFF LINESIZE 1000
   SELECT 'Archive Log Mode: ' || log_mode FROM v\$database;
   EOF
   
   # Check supplemental logging
   echo -e "\nChecking supplemental logging:"
   $ORACLE_HOME/bin/sqlplus -s $ORACLE_USER/$ORACLE_PASSWORD << EOF
   SET HEADING OFF FEEDBACK OFF VERIFY OFF LINESIZE 1000
   SELECT 'Supplemental Logging: ' || supplemental_log_data_min FROM v\$database;
   SELECT 'PK Supplemental Logging: ' || supplemental_log_data_pk FROM v\$database;
   EOF
   
   # Check archive log destination usage
   echo -e "\nChecking archive log destination usage:"
   $ORACLE_HOME/bin/sqlplus -s $ORACLE_USER/$ORACLE_PASSWORD << EOF
   SET LINESIZE 200 PAGESIZE 100 FEEDBACK OFF VERIFY OFF
   COLUMN name FORMAT A30
   COLUMN space_limit FORMAT A15
   COLUMN space_used FORMAT A15
   COLUMN space_reclaimable FORMAT A15
   COLUMN percent_used FORMAT A15
   
   SELECT name, 
          ROUND(space_limit / 1024 / 1024 / 1024, 2) || ' GB' AS space_limit,
          ROUND(space_used / 1024 / 1024 / 1024, 2) || ' GB' AS space_used,
          ROUND(space_reclaimable / 1024 / 1024 / 1024, 2) || ' GB' AS space_reclaimable,
          ROUND(space_used / space_limit * 100, 2) || '%' AS percent_used
   FROM v\$recovery_file_dest;
   EOF
   
   # Check archive log generation in last 24 hours
   echo -e "\nArchive log generation in last 24 hours:"
   $ORACLE_HOME/bin/sqlplus -s $ORACLE_USER/$ORACLE_PASSWORD << EOF
   SET LINESIZE 200 PAGESIZE 100 FEEDBACK OFF VERIFY OFF
   
   SELECT TRUNC(completion_time, 'HH') AS hour,
          COUNT(*) AS logs_generated,
          ROUND(SUM(blocks * block_size) / 1024 / 1024, 2) || ' MB' AS size_generated
   FROM v\$archived_log
   WHERE completion_time > SYSDATE - 1
   GROUP BY TRUNC(completion_time, 'HH')
   ORDER BY hour;
   EOF
   
   # Check SCN advancement
   echo -e "\nChecking SCN advancement:"
   $ORACLE_HOME/bin/sqlplus -s $ORACLE_USER/$ORACLE_PASSWORD << EOF
   SET LINESIZE 200 PAGESIZE 100 FEEDBACK OFF VERIFY OFF
   
   SELECT checkpoint_name, 
          TO_CHAR(checkpoint_time, 'YYYY-MM-DD HH24:MI:SS') AS time, 
          scn
   FROM admin.dms_scn_checkpoints
   WHERE checkpoint_time > SYSDATE - 1
   ORDER BY checkpoint_time;
   EOF
   
   echo -e "\n======== Health Check Complete ========"

2. PostgreSQL-side Health Check Script:

   #!/bin/bash
   # postgres_cdc_health_check.sh
   
   PG_HOST="your-postgres-host"
   PG_PORT="5432"
   PG_DB="your-postgres-db"
   PG_USER="your-postgres-user"
   PG_PASSWORD="your-postgres-password"
   
   export PGPASSWORD=$PG_PASSWORD
   
   echo "======== PostgreSQL CDC Health Check: $(date) ========"
   
   # Check DMS task status
   echo "DMS Task Status:"
   aws dms describe-replication-tasks \
     --query 'ReplicationTasks[*].{TaskId:ReplicationTaskIdentifier,Status:Status,Migration:MigrationType,Stats:ReplicationTaskStats.{CDCSourceLatency:CdcLatencySource,CDCTargetLatency:CdcLatencyTarget}}' \
     --output table
   
   # Check recent CDC checkpoints
   echo -e "\nRecent CDC Checkpoints:"
   psql -h $PG_HOST -p $PG_PORT -U $PG_USER -d $PG_DB << EOF
   SELECT 
     checkpoint_id,
     checkpoint_time,
     checkpoint_name,
     task_arn,
     oracle_scn,
     notes
   FROM admin.dms_cdc_checkpoints
   ORDER BY checkpoint_time DESC
   LIMIT 5;
   EOF
   
   # Check data consistency across key tables
   echo -e "\nTable Record Counts:"
   psql -h $PG_HOST -p $PG_PORT -U $PG_USER -d $PG_DB << EOF
   WITH prod_counts AS (
     SELECT 'customers' AS table_name, COUNT(*) AS count FROM public.customers
     UNION ALL
     SELECT 'orders' AS table_name, COUNT(*) AS count FROM public.orders
     UNION ALL
     SELECT 'products' AS table_name, COUNT(*) AS count FROM public.products
   )
   SELECT * FROM prod_counts;
   EOF
   
   # Check recent CDC recovery events
   echo -e "\nRecent Recovery Events:"
   psql -h $PG_HOST -p $PG_PORT -U $PG_USER -d $PG_DB << EOF
   SELECT 
     event_id,
     event_time,
     event_type,
     checkpoint_name,
     oracle_scn,
     notes
   FROM admin.dms_recovery_events
   ORDER BY event_time DESC
   LIMIT 5;
   EOF
   
   # Check CDC latency trend
   echo -e "\nCDC Latency Trend (last 24 hours):"
   psql -h $PG_HOST -p $PG_PORT -U $PG_USER -d $PG_DB << EOF
   SELECT 
     date_trunc('hour', monitoring_time) AS hour,
     AVG(source_latency) AS avg_source_latency,
     MAX(source_latency) AS max_source_latency,
     AVG(target_latency) AS avg_target_latency,
     MAX(target_latency) AS max_target_latency
   FROM admin.dms_monitoring
   WHERE monitoring_time > CURRENT_TIMESTAMP - INTERVAL '24 hours'
   GROUP BY date_trunc('hour', monitoring_time)
   ORDER BY hour;
   EOF
   
   echo -e "\n======== Health Check Complete ========"

Step 3: Backup and Disaster Recovery for CDC

1. Oracle Archive Log Backup Strategy:

   #!/bin/bash
   # oracle_archivelog_backup.sh
   
   ORACLE_SID="your_oracle_sid"
   ORACLE_HOME="/path/to/oracle/home"
   BACKUP_DIR="/backup/archivelogs"
   DAYS_TO_KEEP=7
   
   export ORACLE_SID ORACLE_HOME
   
   # Get current date for backup folder
   BACKUP_DATE=$(date +%Y%m%d)
   BACKUP_FOLDER="$BACKUP_DIR/$BACKUP_DATE"
   
   # Create backup directory
   mkdir -p $BACKUP_FOLDER
   
   # Back up archive logs
   $ORACLE_HOME/bin/rman << EOF
   CONNECT TARGET /
   
   # Back up all archive logs not yet backed up
   BACKUP ARCHIVELOG ALL NOT BACKED UP FORMAT '$BACKUP_FOLDER/%d_%T_%s_%p.arc';
   
   # Delete archive logs older than specified days that have been backed up twice
   DELETE ARCHIVELOG ALL COMPLETED BEFORE 'SYSDATE-$DAYS_TO_KEEP' BACKED UP 2 TIMES TO DEVICE TYPE DISK;
   
   # List archive logs
   LIST ARCHIVELOG ALL;
   
   EXIT;
   EOF
   
   # Clean up old backup folders
   find $BACKUP_DIR -type d -mtime +$DAYS_TO_KEEP -exec rm -rf {} \; 2>/dev/null
   
   echo "Archive log backup completed: $(date)"

2. PostgreSQL CDC Configuration Backup:

   #!/bin/bash
   # backup_dms_config.sh
   
   BACKUP_DIR="/backup/dms_config"
   AWS_REGION="your-region"
   PROFILE="your-aws-profile"
   
   # Create timestamp for this backup
   BACKUP_TIME=$(date +%Y%m%d_%H%M%S)
   BACKUP_FOLDER="$BACKUP_DIR/$BACKUP_TIME"
   
   mkdir -p $BACKUP_FOLDER
   
   # Export DMS endpoints
   aws dms describe-endpoints --region $AWS_REGION --profile $PROFILE > "$BACKUP_FOLDER/dms_endpoints.json"
   
   # Export DMS replication instances
   aws dms describe-replication-instances --region $AWS_REGION --profile $PROFILE > "$BACKUP_FOLDER/dms_replication_instances.json"
   
   # Export DMS tasks
   aws dms describe-replication-tasks --region $AWS_REGION --profile $PROFILE > "$BACKUP_FOLDER/dms_replication_tasks.json"
   
   # Export detailed task settings for each task
   for TASK_ARN in $(aws dms describe-replication-tasks --region $AWS_REGION --profile $PROFILE --query 'ReplicationTasks[*].ReplicationTaskArn' --output text); do
     TASK_ID=$(echo $TASK_ARN | awk -F':' '{print $NF}')
     aws dms describe-replication-tasks --region $AWS_REGION --profile $PROFILE --filters Name=replication-task-arn,Values=$TASK_ARN > "$BACKUP_FOLDER/task_${TASK_ID}_config.json"
   done
   
   # Backup SCN checkpoints from PostgreSQL
   PG_HOST="your-postgres-host"
   PG_PORT="5432"
   PG_DB="your-postgres-db"
   PG_USER="your-postgres-user"
   PG_PASSWORD="your-postgres-password"
   
   PGPASSWORD=$PG_PASSWORD psql -h $PG_HOST -p $PG_PORT -U $PG_USER -d $PG_DB -c "\COPY (SELECT * FROM admin.dms_cdc_checkpoints ORDER BY checkpoint_time) TO '$BACKUP_FOLDER/dms_cdc_checkpoints.csv' WITH CSV HEADER"
   
   # Create README with timestamp
   echo "DMS Configuration Backup" > "$BACKUP_FOLDER/README.txt"
   echo "Created: $(date)" >> "$BACKUP_FOLDER/README.txt"
   echo "AWS Region: $AWS_REGION" >> "$BACKUP_FOLDER/README.txt"
   echo "Contents:" >> "$BACKUP_FOLDER/README.txt"
   ls -la $BACKUP_FOLDER | grep -v README >> "$BACKUP_FOLDER/README.txt"
   
   # Compress the backup
   tar -czf "$BACKUP_DIR/dms_backup_$BACKUP_TIME.tar.gz" -C "$BACKUP_DIR" "$BACKUP_TIME"
   
   # Remove the uncompressed directory
   rm -rf "$BACKUP_FOLDER"
   
   echo "DMS configuration backup completed: $(date)"
   echo "Backup file: $BACKUP_DIR/dms_backup_$BACKUP_TIME.tar.gz"
   
   # Clean up old backups (keep last 30 days)
   find $BACKUP_DIR -name "dms_backup_*.tar.gz" -type f -mtime +30 -delete

Part 6: Troubleshooting CDC Issues

Common CDC Issues and Solutions

1. Issue: CDC Task Fails with "ORA-01291: missing logfile"

   Solution:

   -- Check for missing archive logs
   SELECT 
     sequence#, 
     name, 
     status 
   FROM v$archived_log 
   WHERE sequence# = <sequence_number_from_error>;
   
   -- If logs are missing, you may need to restore them from backup
   -- Then restart DMS task with appropriate SCN position

2. Issue: CDC Latency Increasing

   Solution:

   # Check if Oracle is generating logs faster than DMS can process
   
   # First, check log generation rate
   sqlplus -s system/password << EOF
   SELECT 
     TRUNC(completion_time, 'HH') AS hour,
     COUNT(*) AS logs_generated,
     ROUND(SUM(blocks * block_size) / 1024 / 1024, 2) AS size_mb
   FROM v\$archived_log
   WHERE completion_time > SYSDATE - 1
   GROUP BY TRUNC(completion_time, 'HH')
   ORDER BY hour;
   EOF
   
   # Check DMS processing capacity in CloudWatch metrics
   aws cloudwatch get-metric-statistics \
     --namespace AWS/DMS \
     --metric-name CDCLatencySource \
     --dimensions Name=ReplicationInstanceIdentifier,Value=your-instance-id Name=ReplicationTaskIdentifier,Value=your-task-id \
     --start-time $(date -u -d '24 hours ago' +%Y-%m-%dT%H:%M:%SZ) \
     --end-time $(date -u +%Y-%m-%dT%H:%M:%SZ) \
     --period 3600 \
     --statistics Average \
     --region your-region
   
   # If DMS can't keep up, consider scaling up replication instance
   aws dms modify-replication-instance \
     --replication-instance-arn your-instance-arn \
     --replication-instance-class dms.c5.xlarge

3. Issue: Oracle SCN Jumps Unexpectedly

   Solution:

   -- Check for SCN jumps in checkpoint history
   SELECT 
     a.checkpoint_id,
     a.checkpoint_time,
     a.scn,
     a.scn - LAG(a.scn) OVER (ORDER BY a.checkpoint_time) AS scn_jump,
     (a.checkpoint_time - LAG(a.checkpoint_time) OVER (ORDER BY a.checkpoint_time)) DAY TO SECOND AS time_diff
   FROM admin.dms_scn_checkpoints a
   WHERE a.checkpoint_time > SYSDATE - 7
   ORDER BY a.checkpoint_time;
   
   -- If large jumps detected, check for database activities during that time
   SELECT 
     operation, 
     start_time, 
     end_time 
   FROM dba_scheduler_job_run_details
   WHERE start_time > TO_DATE('YYYY-MM-DD HH24:MI:SS', 'time_of_jump');

4. Issue: Inconsistent Data After CDC Recovery

   Solution:

   -- In PostgreSQL, create a validation query to find inconsistencies
   -- Example for order totals validation:
   
   WITH order_totals AS (
     SELECT 
       o.order_id,
       o.total_amount,
       SUM(oi.price * oi.quantity) AS calculated_total
     FROM 
       orders o
     JOIN 
       order_items oi ON o.order_id = oi.order_id
     GROUP BY 
       o.order_id, o.total_amount
   )
   SELECT 
     order_id,
     total_amount,
     calculated_total,
     total_amount - calculated_total AS difference
   FROM 
     order_totals
   WHERE 
     ABS(total_amount - calculated_total) > 0.01
     AND order_id IN (
       SELECT 
         o.order_id 
       FROM 
         orders o
       WHERE 
         updated_at > 'recovery_timestamp'
     );

This comprehensive guide covers the setup, maintenance, and recovery of CDC during an Oracle to PostgreSQL migration using AWS DMS. The procedures are specifically designed for Oracle on-premises to PostgreSQL on AWS migrations, with an emphasis on establishing reliable recovery points and automating the recovery process.