DMS CDC Replication Task from RDS MariaDb is not failing but records are not appearing in Kinesis

Question

I'm trying to CDC data from RDS MariaDb to a Kinesis Stream.

The DMS Replication Task itself is successful. I also see no errors in the logs. However, no records actually arrive at the Kinesis Stream.

I followed the docs on https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html.

What am I doing wrong?

Replication task config:

{
    "Logging": {
        "EnableLogging": true,
        "EnableLogContext": true,
        "LogComponents": [
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "TRANSFORMATION"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "SOURCE_UNLOAD"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "IO"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "TARGET_LOAD"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "PERFORMANCE"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "SOURCE_CAPTURE"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "SORTER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "REST_SERVER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "VALIDATOR_EXT"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "TARGET_APPLY"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "TASK_MANAGER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "TABLES_MANAGER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "METADATA_MANAGER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "FILE_FACTORY"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "COMMON"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "ADDONS"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "DATA_STRUCTURE"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "COMMUNICATION"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "FILE_TRANSFER"
            }
        ],
        "CloudWatchLogGroup": "dms-tasks-data-exporter-dev",
        "CloudWatchLogStream": "dms-task-3ODT6664IUAQLWF5J7SV3YZHK7TWAAGBJTVB3SY"
    },
    "StreamBufferSettings": {
        "StreamBufferCount": 3,
        "CtrlStreamBufferSizeInMB": 5,
        "StreamBufferSizeInMB": 8
    },
    "ErrorBehavior": {
        "FailOnNoTablesCaptured": true,
        "ApplyErrorUpdatePolicy": "LOG_ERROR",
        "FailOnTransactionConsistencyBreached": false,
        "RecoverableErrorThrottlingMax": 1800,
        "DataErrorEscalationPolicy": "SUSPEND_TABLE",
        "ApplyErrorEscalationCount": 0,
        "RecoverableErrorStopRetryAfterThrottlingMax": true,
        "RecoverableErrorThrottling": true,
        "ApplyErrorFailOnTruncationDdl": false,
        "DataTruncationErrorPolicy": "LOG_ERROR",
        "ApplyErrorInsertPolicy": "LOG_ERROR",
        "EventErrorPolicy": "IGNORE",
        "ApplyErrorEscalationPolicy": "LOG_ERROR",
        "RecoverableErrorCount": 0,
        "DataErrorEscalationCount": 50,
        "TableErrorEscalationPolicy": "STOP_TASK",
        "RecoverableErrorInterval": 5,
        "ApplyErrorDeletePolicy": "IGNORE_RECORD",
        "TableErrorEscalationCount": 50,
        "FullLoadIgnoreConflicts": true,
        "DataErrorPolicy": "LOG_ERROR",
        "TableErrorPolicy": "SUSPEND_TABLE"
    },
    "TTSettings": {
        "TTS3Settings": null,
        "TTRecordSettings": null,
        "EnableTT": false
    },
    "FullLoadSettings": {
        "CommitRate": 10000,
        "StopTaskCachedChangesApplied": false,
        "StopTaskCachedChangesNotApplied": false,
        "MaxFullLoadSubTasks": 8,
        "TransactionConsistencyTimeout": 600,
        "CreatePkAfterFullLoad": false,
        "TargetTablePrepMode": "DO_NOTHING"
    },
    "TargetMetadata": {
        "ParallelApplyBufferSize": 100,
        "ParallelApplyQueuesPerThread": 1,
        "ParallelApplyThreads": 0,
        "TargetSchema": "",
        "InlineLobMaxSize": 0,
        "ParallelLoadQueuesPerThread": 1,
        "SupportLobs": true,
        "LobChunkSize": 64,
        "TaskRecoveryTableEnabled": false,
        "ParallelLoadThreads": 0,
        "LobMaxSize": 32,
        "BatchApplyEnabled": false,
        "FullLobMode": false,
        "LimitedSizeLobMode": true,
        "LoadMaxFileSize": 0,
        "ParallelLoadBufferSize": 0
    },
    "BeforeImageSettings": {
        "EnableBeforeImage": false,
        "ColumnFilter": "pk-only",
        "FieldName": ""
    },
    "ControlTablesSettings": {
        "historyTimeslotInMinutes": 5,
        "HistoryTimeslotInMinutes": 5,
        "StatusTableEnabled": false,
        "SuspendedTablesTableEnabled": false,
        "HistoryTableEnabled": false,
        "ControlSchema": "",
        "FullLoadExceptionTableEnabled": false
    },
    "LoopbackPreventionSettings": null,
    "CharacterSetSettings": null,
    "FailTaskWhenCleanTaskResourceFailed": false,
    "ChangeProcessingTuning": {
        "StatementCacheSize": 50,
        "CommitTimeout": 1,
        "BatchApplyPreserveTransaction": true,
        "BatchApplyTimeoutMin": 1,
        "BatchSplitSize": 0,
        "BatchApplyTimeoutMax": 30,
        "MinTransactionSize": 1000,
        "MemoryKeepTime": 60,
        "BatchApplyMemoryLimit": 500,
        "MemoryLimitTotal": 1024
    },
    "ChangeProcessingDdlHandlingPolicy": {
        "HandleSourceTableDropped": false,
        "HandleSourceTableTruncated": false,
        "HandleSourceTableAltered": false
    },
    "PostProcessingRules": null
}

Table mappings:

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "fosanis_cdc_test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "fosanis_cdc_test",
                "table-name": "customers"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "3",
            "rule-name": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "fosanis_cdc_test",
                "table-name": "contacts"
            }
        }
    ]
}

Kinesis endpoint:

{
    "StreamArn": "arn:aws:kinesis:eu-central-1:614797193252:stream/data_exporter_dev",
    "MessageFormat": "json-unformatted",
    "ServiceAccessRoleArn": "arn:aws:iam::614797193252:role/data_exporter_kinesis_dev",
    "IncludeTransactionDetails": true,
    "IncludePartitionValue": true,
    "PartitionIncludeSchemaTable": true,
    "IncludeTableAlterOperations": true,
    "IncludeControlDetails": true,
    "IncludeNullAndEmpty": false
}

---

EDIT

This replication tasks works when writing to an S3 bucket with the following configs:

{
    "ServiceAccessRoleArn": "arn:aws:iam::614797193252:role/S3AdminForDms",
    "CsvRowDelimiter": "\\n",
    "CsvDelimiter": ",",
    "BucketName": "mika-dms-test",
    "CompressionType": "NONE",
    "EnableStatistics": true,
    "DatePartitionEnabled": false
}

And the extra connection attributes are bucketName=mika-dms-test;compressionType=NONE;csvDelimiter=,;csvRowDelimiter=\n;datePartitionEnabled=false;.

Answer 1

To see if info is sent, you can try these:

• Make sure your database version is okay. Look here: AWS Database Versions.

• Check settings for MariaDB. Here's an example: AWS MariaDB Settings (use binlog_checksum: NONE and binlog_format: ROW).

• Test if your replication instance can talk to your RDS. I often forget to give Secrets Manager permissions.

• Look at replication logs like binlog. Go here: Enable Binary Log or run SHOW MASTER STATUS.