Add support for forwarding successful records from sinks using SinkContext by kkondaka · Pull Request #5994 · opensearch-project/data-prepper

kkondaka · 2025-08-14T16:07:51Z

Description

Add support for forwarding successful records from sinks using SinkContext.
A new config option is added to all sinks

forward_to: 
  pipelines: [ <list of sub-pipepine names> ]
  with_data:
       key: value
  with_metadata:
       key: value

Issues Resolved

Resolves #5985

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.

graytaylor0 · 2025-08-19T21:27:18Z

+        this.forwardToPipelines = new HashMap<>();
+        if (forwardPipelineNames != null) {
+            for (final String forwardPipelineName: forwardPipelineNames) {
+                this.forwardToPipelines.put(forwardPipelineName, null);


Why is value null here?

Initially, there are no "pipeline"s attached to to a given pipeline name. It's possible that the pipeline name is incorrect. The code below will match the names with the pipelines.

graytaylor0 · 2025-08-19T21:28:06Z

+            if (pipeline != null) {
+                forwardToPipelines.put(key, pipeline);
+            } else {
+                throw new RuntimeException(String.format("forwarding pipeline {} doesn't exist", key));


Can we only catch this at runtime? We could validate on startup if a pipeline configured in forward_to doesn't exist right?

Sure. We can add additional validation. But the order of events may make it difficult. The sink gets instantiated first and then the pipelines are instantiated so at the time of sink construction we do not know if a pipeline is present or not.

graytaylor0 · 2025-08-22T16:18:11Z

+    @JsonProperty("with_metadata")
+    Map<String, Object> withMetadata;
+
+    @JsonProperty("with_data")
+    Map<String, Object> withData;


Trying to understand the use case for this?

It's just like an add_entries?

Not really. These will be specific to sink and can only be determined by the sink - for example, documentId after a document is successfully stored in opensearch. The values of the map may need to support reserved words like documentId which needs to be implemented

dlvenable · 2025-08-27T23:55:52Z

                }
            }

+            if (doForward) {


I think it would be nice to call the SinkContext to get a "forwardTo" batch. Then you populate it and forward it. This can move these conditions in there so we don't repeat a lot of possibly fragile code.

ForwardBatch forwardBatch = sinkContext.getForwardBatch(); ... checkTypeAndWriteObject(record.getData(), writer); forwardBatch(record); ... forwardBatch.forward();

dlvenable · 2025-08-27T23:56:48Z

+
+        final List<Record<Event>> sinkRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER_FORWARD);
+
+        for (int i = 0; i < sinkRecords.size(); i++) {


Assert the size of sinkRecords first.

dlvenable · 2025-08-27T23:57:00Z

+
+        final List<Record<Event>> sinkRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER_DLQ);
+
+        for (int i = 0; i < sinkRecords.size(); i++) {


Assert the size of sinkRecords first.

dlvenable · 2025-08-27T23:57:52Z

+    }
+
+    @Test
+    void pipeline_forward_test() {


While both sink forwarding and DLQ use similar code underneath, they are different end functions. I think we should have different IT test suites for each to reflect that.

I have one above for DLQ and this one is for forwarding. Is that what you meant?

A test suite is a collection of tests. So I mean to have different files. PipelineDlqIT and ForwardPipelinesIT.

Oh. I can split them into two

dlvenable · 2025-08-28T19:28:32Z

+package org.opensearch.dataprepper.model.sink;
+
+public class SinkForwardRecordsContext {
+    public SinkForwardRecordsContext() {}


Perhaps adding in here would make sense. What if a sink only writes some records to the destination?

This context would collect all of the successful ones.

dlvenable · 2025-08-29T00:04:44Z

                return;

            for (final Record<Object> record : records) {
+                if (record.getData() instanceof Event) {


This logic can go into SinkForwardRecordsContext.

public void addRecord(Record<?> record) { if(! (record.getData() instanceof Event)) { return; } ...

That means we can't forward Record<Object> records. I was testing with File sink forwarding to another file sink to verify the forwarding logic.

dlvenable · 2025-08-29T00:04:52Z

            for (final Record<Object> record : records) {
+                if (record.getData() instanceof Event) {
+                    Event event = (Event)record.getData();
+                    sinkForwardRecordsContext.addRecord(new Record<>(event));


Why do you create a new Record?

@dlvenable because FileSink uses Record<Object> and not Record<Event>. It looks like I can't simply cast it. I don't think we should change forwardRecords to accept Record<Object>

kkondaka · 2025-08-29T16:10:28Z

@dlvenable I think this can be merged as is and if you want the last comment to be addressed, I can do in a different PR

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

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

kkondaka requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, oeyh, san81, sb2k16 and srikanthjg as code owners August 14, 2025 16:07

graytaylor0 previously approved these changes Aug 19, 2025

View reviewed changes

kkondaka dismissed graytaylor0’s stale review via 2ed73b9 August 22, 2025 15:49

graytaylor0 reviewed Aug 22, 2025

View reviewed changes

dlvenable requested changes Aug 27, 2025

View reviewed changes

dlvenable reviewed Aug 28, 2025

View reviewed changes

dlvenable reviewed Aug 29, 2025

View reviewed changes

graytaylor0 previously approved these changes Oct 1, 2025

View reviewed changes

graytaylor0 self-requested a review October 1, 2025 22:37

dlvenable previously approved these changes Oct 1, 2025

View reviewed changes

kkondaka dismissed stale reviews from dlvenable and graytaylor0 via 70d543c October 2, 2025 00:05

kkondaka force-pushed the forward-pipelines branch from 8f194b7 to 70d543c Compare October 2, 2025 00:05

kkondaka added 7 commits October 2, 2025 09:56

Add support for forwarding successful records from sinks using SinkCo…

811a132

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

Modified to addressed modified forward_to config

ff8a07f

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

Fixed build failures

fb60535

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

Fixed build failures

e50a5d6

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

Fixing code to address failing tests

2cbd1af

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

Addressed review comments and fixed acknowledgements issue

9466560

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

Addressed review comments

020f6fd

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

Addressed review comments

ef7aaa3

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

kkondaka force-pushed the forward-pipelines branch from 70d543c to ef7aaa3 Compare October 2, 2025 17:41

dlvenable approved these changes Oct 2, 2025

View reviewed changes

graytaylor0 approved these changes Oct 2, 2025

View reviewed changes

kkondaka merged commit 3e71902 into opensearch-project:main Oct 3, 2025
44 of 47 checks passed


		final List<Record<Event>> sinkRecords = inMemorySinkAccessor.get(IN_MEMORY_IDENTIFIER_FORWARD);

		for (int i = 0; i < sinkRecords.size(); i++) {

Conversation

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

Uh oh!

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!

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 commented Aug 29, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

kkondaka commented Aug 14, 2025 •

edited

Loading