Add ByteDecoder for opensearch_api source to support Kafka buffer#6879
Open
divakarsingh wants to merge 1 commit into
Open
Add ByteDecoder for opensearch_api source to support Kafka buffer#6879divakarsingh wants to merge 1 commit into
divakarsingh wants to merge 1 commit into
Conversation
divakarsingh
requested review from
KarstenSchnitter,
Zhangxunmt,
dinujoh,
divbok,
dlvenable,
graytaylor0,
kkondaka,
oeyh,
san81,
sb2k16,
srikanthjg and
srikanthpadakanti
as code owners
divakarsingh
force-pushed
the
fix/opensearch-api-kafka-buffer
branch
from
6fbd388 to
220f714
Compare
✅ License Header Check PassedAll newly added files have proper license headers. Great work! 🎉 |
divakarsingh
force-pushed
the
fix/opensearch-api-kafka-buffer
branch
from
220f714 to
2a982d8
Compare
dlvenable
requested changes
Jun 5, 2026
dlvenable
left a comment
dlvenable
left a comment
Member
There was a problem hiding this comment.
Thank you @divakarsingh for this contribution! I have a few comments on the code and the approach.
| public void parse(InputStream inputStream, Instant timeReceived, Consumer<Record<Event>> eventConsumer) throws IOException { | ||
| final BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); | ||
| String line; | ||
| while ((line = reader.readLine()) != null) { |
Member
There was a problem hiding this comment.
This code should be the same as the existing code in OpenSearchAPIService for handling the response for non-byte buffers. The current structure of this class looks reusable. I think we might be able to refactor OpenSearchAPIService to use this in processBulkRequest.
Author
There was a problem hiding this comment.
@dlvenable Refactored OpenSearchAPIService.processBulkRequest to delegate to OpenSearchBulkByteDecoder for the non-byte-buffer path, so there's a single copy of the parsing logic. Let me know if you had a different structure in mind.
The opensearch_api source did not work with Kafka buffer because no ByteDecoder was registered. When buffer.isByteBuffer()=true, raw bytes are written but the consumer side had no way to reconstruct events from the NDJSON bulk format. This adds OpenSearchBulkByteDecoder which parses NDJSON bulk format back into Data Prepper events with correct metadata attributes. OpenSearchAPIService now delegates to the decoder for the non-byte-buffer path as well, eliminating duplicate parsing logic. Resolves opensearch-project#6876 Signed-off-by: Divakar Pratap Singh <divakar.p.singh@gmail.com>
divakarsingh
force-pushed
the
fix/opensearch-api-kafka-buffer
branch
from
2a982d8 to
504f703
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
The
opensearch_apisource did not work with Kafka buffer because noByteDecoderwas registered. Whenbuffer.isByteBuffer()=true, raw bytes are written but the consumer side had no way to reconstruct events from the NDJSON bulkformat.
This adds
OpenSearchBulkByteDecoderwhich parses NDJSON bulk format back into Data Prepper events with correct metadataattributes.
Issues Resolved
Resolves #6876
Check List