[9.4](backport #52077) Fix silent event drop when a parser clears message content

[mergify]

Proposed commit message

Fix silent event drop when a parser clears message content

A parser can move the decoded data into the event fields and clear the
content, for example an ndjson parser configured without a message_key. When
such an object was read through the readFile path (any content-type other than
application/json or application/x-ndjson), the input dropped every event with
no error logged, because it only published when the message content was
non-empty. Events are now published when either the content or the fields are
populated.

The bug was first seen in as SDH affecting a customer.

The aws-s3 input picks a read path based on each object's Content-Type. Objects served as application/json and application/x-ndjson go through a JSON reader that ignores the configured parsers, so the missing message_key never mattered. Any other content-type routes the object through the generic file-reader path, which does run the ndjson parser.

Stage 1: the parser doesn't drop anything. With no message_key, the parser returns empty content plus the parsed JSON, and writes those parsed keys into the event's fields:

decode() returns empty content when no message_key
the parsed JSON is written into the event fields
So at this point the data is fully intact. It's in message.Fields, only message.Content has been cleared.

Stage 2: the event is dropped here. In the S3 file-reader loop, there is a guard that only checks whether message.Content is non-empty. Since the parser emptied Content, the whole block is skipped, eventCallback is never called, and the fields that hold the data are discarded with no log line:

readFile only emits when len(message.Content) > 0

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works. Where relevant, I have used the stresstest.sh script to run them under stress conditions and race detector to verify their stability.
• I have added an entry in ./changelog/fragments using the changelog tool.

Disruptive User Impact

No. They actually fix a bug!

How to test this PR locally

This reproduces the customer's case: an integration policy with an ndjson
parser and no message_key, reading S3 objects whose Content-Type is not
application/json / application/x-ndjson (e.g. text/plain).
Before the fix no events are ingested and nothing is logged while after the fix the events are ingested with the JSON
decoded into their fields.

Related issues

Use cases

Screenshots

Logs

---

This is an automatic backport of pull request #52077 done by [Mergify](https://mergify.com).

[github-actions]

🤖 GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)
• /test : Run the Buildkite pipeline.

[infra-vault-gh-plugin-prod]

Pinging @elastic/obs-ds-hosted-services (Team:obs-ds-hosted-services)