Introduce support for Opensearch data streams#6249
Merged
dlvenable merged 9 commits intoNov 18, 2025
Conversation
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
JonahCalvo
requested review from
KarstenSchnitter,
chenqi0805,
dinujoh,
dlvenable,
engechas,
graytaylor0,
kkondaka,
oeyh,
san81,
sb2k16 and
srikanthjg
as code owners
dlvenable
requested changes
Nov 6, 2025
dlvenable
left a comment
dlvenable
left a comment
Member
There was a problem hiding this comment.
Thanks @JonahCalvo for this change!
Do we also need to verify that the documentId is not set? What about routing? Is this allowed with data streams?
graytaylor0
reviewed
Nov 7, 2025
| BulkOperation bulkOperation; | ||
|
|
||
| try { | ||
| if (dataStreamDetector.isDataStream(indexName) && !event.containsKey("@timestamp")) { |
Member
There was a problem hiding this comment.
Will this always be @timestamp for data streams? If not should probably make this configurable right?
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
dlvenable
requested changes
Nov 10, 2025
dlvenable
left a comment
dlvenable
left a comment
Member
There was a problem hiding this comment.
Thanks @JonahCalvo for making the improvements. I have only a few comments left.
| @@ -0,0 +1,66 @@ | |||
| /* | |||
Member
There was a problem hiding this comment.
We have a new license header for new files.
https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md#license-headers
Please use this for all new files.
| } | ||
| } | ||
|
|
||
| public void validateDataStreamCompatibility(final String indexName, final String documentId, final String routing) { |
Member
There was a problem hiding this comment.
I don't think you are using this.
Also, it would be better to pass an instance of IndexConfiguration. This keeps to caller from having to know which configurations the data streams care about.
| import org.opensearch.dataprepper.plugins.sink.opensearch.dlq.FailedBulkOperationConverter; | ||
| import org.opensearch.dataprepper.plugins.sink.opensearch.dlq.FailedDlqData; | ||
| import org.opensearch.dataprepper.plugins.sink.opensearch.index.ClusterSettingsParser; | ||
| import org.opensearch.dataprepper.plugins.sink.opensearch.index.DataStreamDetector; |
Member
There was a problem hiding this comment.
We should probably have an integration test in OpenSearchSinkIT.
Collaborator
There was a problem hiding this comment.
It would also be beneficial to document the required permissions for the Data Prepper user in OpenSearch.
JonahCalvo
force-pushed
the
data-stream-support
branch
12 times, most recently
from
58e013e to
71573c4
Compare
JonahCalvo
force-pushed
the
data-stream-support
branch
6 times, most recently
from
4a45fb1 to
61998e2
Compare
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
JonahCalvo
force-pushed
the
data-stream-support
branch
from
61998e2 to
81f2386
Compare
added 3 commits
November 17, 2025 11:57
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
JonahCalvo
force-pushed
the
data-stream-support
branch
from
0d53a7e to
7b5b46b
Compare
Signed-off-by: Jonah Calvo <caljonah@amazon.com>
JonahCalvo
force-pushed
the
data-stream-support
branch
from
7b5b46b to
b392748
Compare
graytaylor0
approved these changes
Nov 17, 2025
| if (configuredAction != null && | ||
| !configuredAction.equals(OpenSearchBulkActions.CREATE.toString()) && | ||
| !configuredAction.equals(OpenSearchBulkActions.INDEX.toString())) { | ||
| LOG.warn("Data Stream '{}' requires 'create' action, but '{}' was configured. Using 'create' action.", |
Member
There was a problem hiding this comment.
This should probably be marked as NOISY. Can be a fast-follow
3 tasks
eatulban
pushed a commit
to eatulban/data-prepper
that referenced
this pull request
Dec 11, 2025
Add OpenSearch Data Stream support with automatic action selection Signed-off-by: Jonah Calvo <caljonah@amazon.com>
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
4 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
This PR adds native support for OpenSearch Data Streams to the OpenSearch sink, enabling automatic detection and proper handling of data stream indices with zero configuration changes required.
Upon initialization, uses GetDataStreamRequest API to identify when an index name refers to a data stream, caching the result with index name
Overrides configured actions to use
createfor data streams (required by OpenSearch)Automatically adds @timestamp field for data streams when missing
Issues Resolved
Resolves #2037 #2038
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.