Kinesis Firehose -> Transform Lambda -> OpenSearch Service: Kinesis Firehose does not support ElasticSearch Bulk API commands inside the Firehose record

• We are running an opentelemetry collector that exports the OTEL traces to Kinesis
• A Kinesis Firehose Stream transforms the records with a lambda and saves them to the OpenSearch Service destination
• The lambda is transforming a single Kinesis record into many OpenSearch documents by setting the index and document ID in the returned JSON body for the record data.

This kind of worked at first. The destination error logs were flooded with {"type":"mapper_parsing_exception","reason":"failed to parse, document is empty"} 400 responses from OpenSearch, but the documents got saved into OpenSearch. Now we've started to see a new error, which increased in frequency to the point we see this error every time in our production pipeline:

Kinesis Firehose does not support ElasticSearch Bulk API commands inside the Firehose record (code: ES.MalformedData)

I've had no luck googling this, but if I understand it correctly, this error means that Kinesis is complaining about the { "index": { "_index": "foo", "_id": "123" } } lines we've added above each document in the record data returned by the lambda transformation. I've noticed the index rotation is not working due to our hard coded index, and Kinesis seems confused about the records delivered to OpenSearch (it says none have been delivered but the index has 250 million documents so far). Buffer settings seem unreliable, as we turning what Kinesis thinks is a single record into potentially thousands of records.

The odd things is data still seems to be making its way into OpenSearch most of the time, but seeing these errors and no way that I know of to resolve them has me hesitant (to say the least) about relying on this solution. From what I can tell in answers such as this one or this one, the core of the issue is we are doing something Kinesis does not support: transforming a single record into multiple records.

It should also be noted, we need to set the document ID manually because we need to rely on upsert logic in some cases, such as the collector sending us the same span for the same trace more than once.

When searching for answers on this, I came across Data Prepper which appears to be an OpenSearch/AWS native solution for our exact use case. At this point I think the only reliable/feasible option is to change to Data Prepper, but I want to make sure I'm not missing something simple first.

Has anyone encountered this error before, or can someone just tell me what it means? Is there a viable way to make Kinesis do what we want, or can we even rely on Kinesis to keep working with how we are transforming the records?