Add common sink framework for DataPrepper sinks by kkondaka · Pull Request #6183 · opensearch-project/data-prepper

kkondaka · 2025-10-16T22:06:40Z

Description

Add common sink framework for DataPrepper sinks. Include main execution loop, flexible flushing, sink level metrics.

Issues Resolved

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

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
New functionality has javadoc added
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: Kondaka <krishkdk@amazon.com>

KarstenSchnitter

Thanks for this PR. Looks good. I have some questions on error handling, nothing to major.

KarstenSchnitter · 2025-10-17T11:54:54Z

+public interface SinkFlushContext {
+}


What is the generic idea behind this empty interface?

Yes, shouldn't there be a flush() method?

flush() is in the SinkFlushableBuffer class. This is to carry sink specific information like s3Client or sqsClient or httpSender or opensearchClient. The empty class just gives a name. I felt this is better than using Object. Also, it is possible that it will have some methods in future.

So, for example, in case of sqs sink, it will have something like

public class SqsSinkFlushContext implements SinkFlushContext { final SqsClient sqsClient; public SqsSinkFlushContext(final SqsClient sqsClient) { this.sqsClient = sqsClient; } public SqsClient getSqsClient() { return sqsClient; } }

The flush() in SqsSinkFlushableBuffer class will get sqsClient from SqsSinkFlushContext to actually push the messages to SQS.

The idea is that SinkFlushableBuffer can be flush()ed, not SinkFlushContext. SinkFlushContext just gives some context to do the flush

I assumed the context to be a placeholder for more specialiced interfaces. I was just wondering, why this abstraction is introduced, when it is currently not used. I am fine with having it available to migrate the existing sinks to this pattern.

KarstenSchnitter · 2025-10-17T12:55:02Z

+            if (sinkBuffer.exceedsFlushTimeInterval()) {
+                flushBuffer();
+            }


You want to flush even when there are no records? Is this reasonable?

Good catch. We should check that there is something present.

There may not be records in the current execution but the buffer may have records. Ofcourse. If the buffer doesn't have anything, this does nothing.

This is very important because imagine what happens if sink output() is called with 1 record every time and it never hits maxEvents or maxRequestSize limits (or it takes an hour to hit these limits). We have the third "time based" limit to flush. So, whatever accumulated in the sink buffer will be flushed if the time limit is reached.

I wanted to make sure, this is intended behavior. It could have been an unintended side-effect. Thanks for clarifying this.

KarstenSchnitter · 2025-10-17T12:56:55Z

+                    }
+
+                    if (!sinkBuffer.addToBuffer(bufferEntry)) {
+                        throw new RuntimeException("Failed to add event to sink buffer");


Same question on dropped events here.

Same answer as above.

dlvenable · 2025-10-17T16:32:50Z

+import java.time.Instant;
+
+public class DefaultSinkBuffer implements SinkBuffer {
+    final SinkBufferWriter sinkBufferWriter;


These should all be private.

dlvenable · 2025-10-17T16:34:17Z

+    private final DistributionSummary sinkEventSize;
+
+    public DefaultSinkMetrics(final PluginMetrics pluginMetrics, final String sinkPrefix, final String eventName) {
+        this.sinkRequestsSucceeded = pluginMetrics.counter(sinkPrefix + SINK_REQUESTS_SUCCEEDED);


Why do we need a sinkPrefix? The plugin name should be part of the name already.

Ok, will delete the prefix.

dlvenable · 2025-10-17T16:35:04Z

+public interface SinkFlushContext {
+}


Yes, shouldn't there be a flush() method?

dlvenable · 2025-10-17T16:37:42Z

+    public abstract void flushDLQList();
+    public abstract void addFailedEventsToDLQ(final List<Event> events, final Throwable ex);
+    public abstract SinkBufferEntry getSinkBufferEntry(final Event event) throws Exception;
+    public abstract boolean exceedsMaxEventSizeThreshold(final long estimatedSize);


This would be better as long getMaxEventSizeThreshold(). Then the comparison happens in the common code.

flush() is in SinkFlushableBuffer class

dlvenable · 2025-10-17T16:40:41Z

+        }                   
+    }
+
+    public abstract void flushDLQList();


I still think these should be handled as injected interfaces. Think about writing a sink. You don't need to mix these into the same class because they are not strongly coupled together.

For example, getting a SinkBufferEntry is about serialization. This is a very different concern from writing to the DLQ.

e.g.

public interface SinkBufferEntryProvider { SinkBufferEntry getSinkBufferEntry(final Event event) throws Exception; }

@dlvenable OK, I will modify to add interfaces

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

dlvenable · 2025-10-23T20:27:31Z

+
+public interface SinkBufferEntryProvider {
+    SinkBufferEntry getSinkBufferEntry(final Event event) throws Exception;
+    boolean exceedsMaxEventSizeThreshold(final long estimatedSize);


I'd recommend that this take in the SinkBufferEntry. That way, if max size is somehow dynamic you'd have more context.

dlvenable · 2025-10-23T20:28:18Z

+
+import java.util.List;
+
+public interface SinkDLQHandler {


Please rename to use Dlq to match existing naming conventions.

SinkDlqHandler

dlvenable · 2025-10-23T20:28:27Z

+import java.util.List;
+
+public interface SinkDLQHandler {
+    void flushDLQList();


Similar as above: use Dlq

dlvenable · 2025-10-23T20:28:30Z

+
+public interface SinkDLQHandler {
+    void flushDLQList();
+    void addFailedEventsToDLQ(final List<Event> events, final Throwable ex);


Similar as above: use Dlq

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

dlvenable

Thanks @kkondaka for this improvement to help with consistency!

graytaylor0 · 2025-10-28T22:33:28Z

+
+import java.time.Instant;
+
+public class DefaultSinkBuffer implements SinkBuffer {


Curious how this will be applied to all sinks? Or is it only meant to be applied for some?

@graytaylor0 DefaultSinkBuffer is expected to extended by each sink.

Add common sink framework for DataPrepper sinks

28a85fb

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 October 16, 2025 22:06

kkondaka mentioned this pull request Oct 16, 2025

Add Common Sink Output Strategies #6100

Closed

3 tasks

KarstenSchnitter reviewed Oct 17, 2025

View reviewed changes

dlvenable reviewed Oct 17, 2025

View reviewed changes

kkondaka added 2 commits October 17, 2025 12:34

Addressed review comments

57cd9f9

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

Removed unnecessary file

2043667

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

dlvenable reviewed Oct 23, 2025

View reviewed changes

Addressed review comments

f39f647

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

dlvenable approved these changes Oct 28, 2025

View reviewed changes

graytaylor0 approved these changes Oct 28, 2025

View reviewed changes

kkondaka merged commit 5dd9eb1 into opensearch-project:main Oct 28, 2025
45 of 47 checks passed


		import java.time.Instant;

		public class DefaultSinkBuffer implements SinkBuffer {

Conversation

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

Uh oh!

Description

Issues Resolved

Check List

Uh oh!

KarstenSchnitter left a comment

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 Oct 17, 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!

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!

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 Oct 17, 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!

kkondaka Oct 17, 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!

dlvenable left a comment

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!

Uh oh!

Reviewers

Assignees

Labels

Projects

kkondaka commented Oct 16, 2025 •

edited by KarstenSchnitter

Loading

kkondaka Oct 17, 2025 •

edited

Loading

kkondaka Oct 17, 2025 •

edited

Loading

kkondaka Oct 17, 2025 •

edited

Loading