opensearch-project / opensearch-project/data-prepper
[BUG] opensearch_api source does not work with Kafka buffer - no ByteDecoder registered
@divakarsingh is already working on this.
Since May 26, 2026.
- Dominant language
- Java
- Stars
- 374
- Forks
- 355
- Avg merge
- 3d 18h
- Merged PRs (30d)
- 8
Description
Describe the bug
The opensearch_api source does not work when configured with a kafka buffer. Traffic stops flowing — events are written to Kafka but never consumed/reconstructed on the other side. It works correctly with bounded_blocking buffer.
To Reproduce
- Configure Data Prepper with
opensearch_apisource andkafkabuffer - Send a
_bulkrequest to Data Prepper - Observe: events are written to Kafka topic but never reach the sink
- Switch buffer to
bounded_blocking— traffic flows correctly
Expected behavior
The opensearch_api source should register a ByteDecoder that can parse NDJSON bulk format back into Data Prepper events with correct metadata attributes (opensearch_action, opensearch_index, opensearch_id, etc.) when using Kafka buffer.
Screenshots
N/A
Environment (please complete the following information):
- OS: Linux
- Version: Data Prepper 2.15.1 / main branch
Additional context
Forum report: https://forum.opensearch.org/t/opensearch-api-source-doesnt-work-with-kafka-buffer/28059
Root cause: When buffer.isByteBuffer() == true (Kafka buffer), the source writes raw bytes via buffer.writeBytes(content.array(), ...). On the consumer side, the buffer needs a ByteDecoder to reconstruct Record<Event> objects from those bytes. OpenSearchAPISource does not override getDecoder(), so no decoder is registered. The Kafka buffer consumer cannot parse the NDJSON bulk format back into events.
The bounded_blocking buffer works because it stores Record<Event> objects directly in a BlockingQueue — no serialization/deserialization needed.
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.
Assessment
This issue has not been assessed yet.