opensearch-project / opensearch-project/data-prepper
[BUG] Lambda list output is treated as a single document
Nobody has claimed this yet.
- 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:
- Create a lambda function with the code I wrote above
- Create a pipeline using my template, replace the vars with yours
- Start it
- 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
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- 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