opensearch-project / opensearch-project/data-prepper

[BUG] Lambda list output is treated as a single document

Open
#5,510 5 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

bug
Dominant language
Java
Stars
374
Forks
354
Avg merge
3d 18h
Merged PRs (30d)
8

Description

Describe the bug

I have a Lambda processor that receives the records (N num of records) to process and outputs (M num of records) to ingest in OS

The Lambda returns the list of documents, this is the code to replicate the issue:

import json

def lambda_handler(event, context):
    return [
        {'a': 1},
        {'a': 2},
        {'a': 3},
    ]

My expectation is that 3 events will flow to my pipeline after the Lambda for further processing/to be ingested. The reality I see right now is that Data Prepper receives the array and treats it as a single event.

This is the output from Data Prepper showing this:

2025-03-08T12:14:12,590 [s3-source-sqs-1] INFO  org.opensearch.dataprepper.plugins.source.s3.S3ObjectWorker - Read S3 object: [bucketName=7sybr6i6-search-indexer-replication, key=cdc/public/content_artist/20250308-121411717.csv]
2025-03-08T12:14:16,803 [simple-sample-pipeline-processor-worker-1-thread-1] INFO  org.opensearch.dataprepper.plugins.lambda.processor.LambdaProcessor - Flush to Lambda check: currentBuffer.size=176, currentBuffer.events=1, currentBuffer.duration=PT58.743S
[{"a":1},{"a":2},{"a":3}]
2025-03-08T12:14:18,008 [acknowledgement-callback-4] INFO  org.opensearch.dataprepper.plugins.source.s3.SqsWorker - Deleted 1 messages from SQS. [e42e8b50-bea9-4753-9832-cf6092173e88]
2025-03-08T12:14:29,118 [simple-sample-pipeline-sink-worker-2-thread-1] WARN  org.opensearch.dataprepper.plugins.sink.opensearch.BulkRetryStrategy - Bulk Operation Failed.
org.opensearch.client.opensearch._types.OpenSearchException: Request failed: [x_content_parse_exception] [1:8] [UpdateRequest] doc doesn't support values of type: START_ARRAY

You can see in the output [{"a":1},{"a":2},{"a":3}] being seen as one item (this is the stdout output), and then you can see OS failing to import the object because it sees an array

My pipeline:

simple-sample-pipeline:
  workers: 1
  source:
    s3:
      acknowledgments: true
      notification_type: "sqs"
      compression: "none"
      codec:
        csv:
      sqs:
        queue_url: "queue_url"
        maximum_messages: 10
        visibility_timeout: "30s"
        visibility_duplication_protection: true
      aws:
        region: "region"
        sts_role_arn: "role"
  processor:
    - aws_lambda:
        function_name: "function_name"
        invocation_type: "request-response"
        aws:
          region: "region"
          sts_role_arn: "role"
        max_retries: 3
        batch:
          key_name: "records"
          threshold:
            event_count: 5
            maximum_size: "5mb"
            event_collect_timeout: PT10S
  sink:
      - stdout:
      - opensearch:
          hosts: ["http://127.0.0.1:9200"]
          username: username
          password: password
          # aws_sigv4: true
          insecure: true
          index_type: management_disabled
          index: test-index
          document_id: "${/id}"
          max_retries: 20
          bulk_size: 4
          action: upsert

To Reproduce
Steps to reproduce the behavior:

  1. Create a lambda function with the code I wrote above
  2. Create a pipeline using my template, replace the vars with yours
  3. Start it
  4. See error

Expected behavior

Data Prepper should see 3 events

{"a": 1}
{"a": 2}
{"a": 3}

Environment (please complete the following information):

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

Additional context

In my lambda I will need the ability to return a dynamic number of documents (that will likely be less than the number of records I receive in input)
I will also need it to return "delete actions" and not just upsert ones, but shouldn't matter for this bug

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with the aws_lambda processor and trace how the Lambda response is handled after invocation. Reproduce the issue with the provided lambda_handler and pipeline configuration, then verify that a list response is represented as three events rather than one array-valued event, including the documented ingestion behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
aws, java
Domain
backend
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.