Repository navigation
Add ByteDecoder for opensearch_api source to support Kafka buffer - #6879
divakarsingh wants to merge 1 commit into
Conversation
6fbd388 to
220f714
Compare
✅ License Header Check PassedAll newly added files have proper license headers. Great work! 🎉 |
220f714 to
2a982d8
Compare
dlvenable
left a comment
dlvenable
left a comment
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) { |
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.
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>
2a982d8 to
504f703
Compare
|
@dlvenable can you pls check and advise next steps? |
dlvenable
left a comment
dlvenable
left a comment
There was a problem hiding this comment.
@divakarsingh , Thank you for taking the refactor. This is an improvement. I do want to be sure that we continue to return 400 for invalid data.
| @ValueSource(booleans = {false, true}) | ||
| public void testBulkRequestJsonResponse400WithInvalidPayload(boolean includeIndexInPath) throws Exception { | ||
| // Prepare | ||
| public void testBulkRequestWithInvalidPayloadReturns200(boolean includeIndexInPath) throws Exception { |
There was a problem hiding this comment.
The source should continue to respond with a 400 for invalid payload. The change has it dropping the events.
It may be a good idea to add a BulkErrorHandler interface. You can have the one given to Kafka to skip data. And the one given to the API service to throw.
| event.getMetadata().setAttribute( | ||
| BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION, action); | ||
|
|
||
| // Set index: action line takes priority, then URL-level fallback |
There was a problem hiding this comment.
This is a breaking change, but I think it is good. It is the way OpenSearch works. I'm going to add the breaking change label to this PR because of this. No need to make any changes. I'm just pointing it out.
srikanthpadakanti
added
the
breaking change
| } | ||
|
|
||
| public void parse(InputStream inputStream, Instant timeReceived, BulkAPIRequestParams requestParams, Consumer<Record<Event>> eventConsumer) throws IOException { | ||
| final BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); |
There was a problem hiding this comment.
Wrap the reader in try-with-resources
3 participants
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