Add forward_to support to opensearch sink by kkondaka · Pull Request #6349 · opensearch-project/data-prepper

kkondaka · 2025-12-11T05:44:14Z

Description

forward_to option to all sinks is added via SinkContext but the sink level implementation for opensearch sink is not done. This PR adds the support for forwarding events after successful delivery to opensearch sink.

Also added check to not allow more than one forwarding pipeline because that would require acknowledgement support. Forwarding to multiple pipelines an be done after initially forwarding to one pipeline, if it is really needed.

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

[X ] New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
[ X] Commits are signed with a real name per the DCO

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.

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dlvenable · 2025-12-15T17:43:26Z

        @JsonProperty("with_data") final Map<String, Object> withData,
        @JsonProperty("with_metadata") final Map<String, Object> withMetadata) {
+        if (pipelineNames.size() != 1) {
+            throw new RuntimeException("Supports only one forwarding pipeline");


This should throw InvalidPipelineConfigurationException.

@dlvenable I do not think it can throw InvalidPipelineConfigurationException because it is in a different package that depends on data-prepper-api package. It will introduce circular dependency between the packages.

Changed it to IllegalArgumentException

@kkondaka ,

We can use InvalidPluginConfigurationException instead.

dlvenable · 2025-12-15T17:45:52Z

    }
+
+    @Test
+    void testInvalidValues() {


I'm unclear on this. Is this testing all invalid values? When testing invalid values, you should use all valid values except one specific invalid value. Otherwise, you don't know what you tested against or if it works.

Additionally, you should test invalid values on both ends: empty list and list with two pipelines.

Also, make the names clearer.

e.g.

empty_pipelines_list_throws_exception() pipelines_list_with_two_or_more_pipelines_throws_exception()

Fair point. Added two negative tests to test 0 and two pipelines. Added one positive test to test with one pipeline.

dlvenable · 2025-12-15T17:46:33Z

        assertThat(sinkContext.forwardRecords(sinkForwardRecordsContext, Map.of("datakey1", "datavalue1"), Map.of("metadataKey1", "metadataValue1")), equalTo(true));
-        verify(forwardPipeline1, times(1)).sendEvents(eq(records));
-        verify(forwardPipeline2, times(1)).sendEvents(eq(records));
+        verify(forwardPipeline1, times(1)).sendEvents(any());


Why are we loosening these verifies? This doesn't seem right.

No. We are not loosening the verifies. Since the forwardRecords() clears the records It can't be verified like that anymore. The original records would now be empty and the verification would fail

dlvenable · 2025-12-15T17:47:18Z

        sinkForwardRecordsContext.addRecords(List.of(record2, record3));
        List<Record<Event>> records = sinkForwardRecordsContext.getRecords();
        assertThat(records.size(), equalTo(3));
+        sinkForwardRecordsContext.clearRecords();


This should be it's own test.

sinkForwardRecordsContext_clearRecords_removes_all_records()

dlvenable · 2025-12-15T17:49:03Z

    }

+    @Test
+    public void testForwardingRecords() throws IOException, InterruptedException {


This this a better name and scope down what it is doing.

output_forwards_created_documents_to...()

Also, as this is an IT, I think it would be better to actually forward these to another OpenSearch index.

@dlvenable Yes, it is IT but to test forwarding to another Opensearch sink requires complete pipeline functionality. We are not forwarding to another sink but forwarding to a pipeline. Goal is to verify that it is forwarding to pipelines. I do not think we need to mock entire pipeline. We could probably send to a sink directly from
sinkContext.forwardRecords() But sending another opensearch sink really doesn't test any more functionality than what is being tested already.

Yea, I think you are right here. This is an integration test of OpenSearch functionality, not core Data Prepper.

dlvenable · 2025-12-15T17:49:43Z

            final AccumulatingBulkRequest requestToReissue = bulkRequestSupplier.get();
            final ImmutableList.Builder<FailedBulkOperation> nonRetryableFailures = ImmutableList.builder();
            int index = 0;
+            List<BulkOperationWrapper> successfulOperations = new ArrayList<>();


Initialize this array with the total number of operations, assuming the best case.

@dlvenable You mean this?

successfulOperations = new ArrayList<>(request.getOperations.size());

dlvenable · 2025-12-15T17:51:05Z

    });
  }

+  private void successfulOperationsHandler(final List<BulkOperationWrapper> successfulOperations) {


This needs unit tests. There are a lot of conditions to cover as well.

dlvenable · 2025-12-15T17:54:08Z

        return new BulkRetryStrategy(
                requestFunction,
                logFailure,
+                (operations) -> {


This moves some unnecessary logic into the tests. I think you can just make a mock(Consumer.class) and then call verify(successOperationsHandler).accept(successOperations).

We probably need tests to verify that failed operations are not included.

@dlvenable Unfortunately, that wont work. The tests count on eventhandle getting released. So, it is important to release the event handle in this Consumer.

graytaylor0 · 2025-12-15T21:25:14Z

    });
  }

+  private void successfulOperationsHandler(final List<BulkOperationWrapper> successfulOperations) {


Did you consider making this handle List and moving the logic to a higher level to make it re-usable between the sinks? Use bulk operation wrapper makes this very coupled to opensearch sink

@graytaylor0 by "handle list", you mean "event handle list"? We can't forward event handles. It needs to be events. I don't think this part of the code would be that much re-usable in other Sinks. We are moving towards using only pipeline DLQ, which means all sinks need to keep the events, and just event handles. Due to legacy code we have multiple conditions in opensearch.

I meant to say List<Event> sorry

graytaylor0 · 2025-12-15T21:28:30Z

The OpenSearchSinkIT are all failing with this error

 OpenSearchSinkIT > testOpenSearchIndexWithDate(String) > [2] yyyy-MM-dd FAILED
    java.lang.NullPointerException at OpenSearchSinkIT.java:159

OpenSearchSinkIT > testOpenSearchIndexWithDate(String) > [3] dd-MM-yyyy FAILED
    java.lang.NullPointerException at OpenSearchSinkIT.java:159

OpenSearchSinkIT > testOpenSearchRouting(String) > [1]  FAILED
    java.lang.NullPointerException at OpenSearchSinkIT.java:159

OpenSearchSinkIT > testOpenSearchRouting(String) > [2] info/ids/rid FAILED
    java.lang.NullPointerException at OpenSearchSinkIT.java:159

OpenSearchSinkIT > testOpenSearchRouting(String) > [3] rid FAILED
    java.lang.NullPointerException at OpenSearchSinkIT.java:159

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dlvenable

Thanks!

* Add forward_to support to opensearch sink Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Added integration test Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> --------- Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> Signed-off-by: Nathan Wand <wandna@amazon.com>

* Add forward_to support to opensearch sink Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Added integration test Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> --------- Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> Signed-off-by: Simon ELBAZ <elbazsimon9@gmail.com>

* Add forward_to support to opensearch sink Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Added integration test Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> --------- Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

Add forward_to support to opensearch sink

269fd9b

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

kkondaka requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, oeyh, san81, sb2k16 and srikanthjg as code owners December 11, 2025 05:44

Added integration test

0c958f4

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dlvenable requested changes Dec 15, 2025

View reviewed changes

graytaylor0 reviewed Dec 15, 2025

View reviewed changes

kkondaka added 2 commits December 16, 2025 06:55

Addressed review comments

7768d87

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

Addressed review comments

f2bee24

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

graytaylor0 approved these changes Dec 16, 2025

View reviewed changes

dlvenable approved these changes Dec 16, 2025

View reviewed changes

kkondaka merged commit 37d65f8 into opensearch-project:main Dec 16, 2025
69 of 70 checks passed

tillwulfram mentioned this pull request Mar 16, 2026

[BUG] The Default "dlq_pipeline" no longer receives failed events from OpenSearch sink #6643

Closed

Uh oh!

Conversation

kkondaka commented Dec 11, 2025

Description

Issues Resolved

Check List

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kkondaka Dec 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

graytaylor0 commented Dec 15, 2025

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

kkondaka Dec 16, 2025 •

edited

Loading