[BUG] Lambda processing failure stored as document

Describe the bug

I have a pipeline that reads data from S3, processes it via CSV and then uses the Lambda processor to create the documents to ingest into OpenSearch

In one case, the Lambda invocation failed, the failure message got ingested into OS for some reason:

{
  "took": 0,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 1,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "test-index",
        "_id": "_id",
        "_score": 1,
        "_source": {
          "errorMessage": "'table'",
          "errorType": "KeyError",
          "requestId": "d303b986-8f1c-4be1-9723-3b0184f60ffd",
          "stackTrace": [
            "  File \"/var/task/lambda_function.py\", line 33, in handler\n    tracks_ids, playlists_ids, artists_ids = parse_records(records, datadog_env)\n",
            "  File \"/var/task/code/records_parser.py\", line 15, in parse_records\n    table_name = record['table']\n"
          ]
        }
      }
    ]
  }
}

To Reproduce

This is my pipeline (I am evaluating Data Prepper so I just have a basic test pipeline)

simple-sample-pipeline:
  workers: 2
  source:
    s3:
      # Prevent data loss by only considering logs to be processed successfully after they are received by the opensearch sink
      acknowledgments: true
      notification_type: "sqs"
      # Provide compression property, can be "none", "gzip", or "automatic"
      compression: "none"
      codec:
        newline:
      sqs:
        # Provide a SQS Queue URL to read from
        queue_url: ""
        # Lower maximum_messages depending on the size of your S3 objects
        maximum_messages: 10
        # Modify the visibility_timeout of the sqs messages depending on the size of your access log S3 objects.
        # Objects that are small (< 0.5 GB) and evenly distributed in size will result in the best performance
        # It is recommended to allocate a minimum of 30 seconds, and to add 30 seconds for every 0.25 GB of data in each S3 Object
        visibility_timeout: "30s"
        # Enable this flag to allow the visibility timeout to be extended if an object has not yet finished processing.
        # This helps prevent duplicate processing of SQS messages when the visibility timeout is lower than the amount of time required to process an object.
        visibility_duplication_protection: true
      aws:
        # Provide the region to use for aws credentials
        region: "ap-southeast-2"
        # Provide the role to assume for requests to SQS and S3
        sts_role_arn: ""
  processor:
    - csv:
        column_names: ["op", "table", "schema"]
    - aws_lambda:
        function_name: "function_name"
        invocation_type: "request-response"
        aws:
          region: "ap-southeast-2"
          sts_role_arn: ""
        max_retries: 3
        batch:
          key_name: "records"
          threshold:
            event_count: 50
            maximum_size: "5mb"
            event_collect_timeout: PT10S
  sink:
    - opensearch:
        hosts: ["http://127.0.0.1:9200"]
        username: empty
        password: empty
        insecure: true
        index_type: management_disabled
        index: test-index
        document_id: _id
        max_retries: 20
        bulk_size: 4

Expected behavior

The failed batch of records is ignored/sent to DLQ

Environment (please complete the following information):

• OS: [e.g. Ubuntu 20.04 LTS] Ubuntu 22.04.3 LTS
• Version [e.g. 22] Latest version (opensearchproject/data-prepper:latest)