Having trouble processing dynamodb stream with AWS Opensearch Ingestion Pipeline

[martinpsz]

I am trying to index a dynamodb stream (old and new image) into opensearch serverless with an opensearch ingestion pipeline. Unfortunately, I am running into errors like this:

"org.opensearch.dataprepper.plugins.sink.opensearch.BulkRetryStrategy - Bulk Operation Failed"

To diagnose the underlying issue, I took a look at the dlq files generated for these error events. What I am seeing is one or a few records not a large batch and the records look as I would expect them to. So my hypothesis is that the volume of dynamodb stream events is overwhelming the ingestion process. If I am right, and seeing as how persistent buffering and bulk sizing are not supported by AWS' OpenSearch Serverless service, what can I do to avoid these bulk operation errors?

For reference, here is my OS ingestion pipeline configuration:

     {
   "version": "2",
   "dynamodb-pipeline": {
       "source": {
           "dynamodb": {
               "acknowledgments": true,
               "tables": [
                   {
                       "table_arn": "arn:aws:dynamodb:us-east-1:table/member-search-api-person-nexus",
                       "stream": {
                           "start_position": "LATEST"
                       },
                       "export": {
                           "s3_bucket": "member-search-serverless",
                           "s3_region": "us-east-1",
                           "s3_prefix": "dynamo-to-aoss-export/"
                       }
                   }
               ],
               "aws": {
                   "sts_role_arn": "arn:aws:iam:role/member-search-api-pipeline-role-df625a3",
                   "region": "us-east-1"
               }
           }
       },
       "processor": [
           {
               "delete_entries": {
                   "with_keys": [
                       "original",
                       "affiliate_ids",
                       "office",
                       "ts"
                   ]
               }
           }
       ],
       "sink": [
           {
               "opensearch": {
                   "hosts": [
                       "<poensearch endpoint>"
                   ],
                   "index": "member-search",
                   "index_type": "custom",
                   "document_id": "${getMetadata(\"primary_key\")}",
                   "action": "${getMetadata(\"opensearch_action\")}",
                   "document_version": "${getMetadata(\"document_version\")}",
                   "document_version_type": "external",
                   "max_retries": 16,
                   "aws": {
                       "sts_role_arn": "arn:aws:iam:role/member-search-api-pipeline-role-df625a3",
                       "region": "us-east-1",
                       "serverless": true,
                       "serverless_options": {
                           "network_policy_name": "serverless-network"
                       }
                   },
                   "dlq": {
                       "s3": {
                           "bucket": "member-search-serverless",
                           "key_path_prefix": "dynamodb-pipeline/dlq",
                           "region": "us-east-1",
                           "sts_role_arn": "arn:aws:iam:role/member-search-api-pipeline-role-df625a3"
                       }
                   }
               }
           }
       ]
   }
}