Add stateful buffer for lambda sink #5354
Conversation
srikanthjg
requested review from
sb2k16,
chenqi0805,
engechas,
san81,
graytaylor0,
dinujoh,
kkondaka,
KarstenSchnitter,
dlvenable and
oeyh
as code owners
|
|
||
| // The partial buffer that may not yet have reached threshold. | ||
| // Access must be synchronized | ||
| private Buffer statefulBuffer; |
There was a problem hiding this comment.
This should handle concurrent actions?
There was a problem hiding this comment.
yes, all i have used "synchronized" on the methods where this is used. In s3 sink, they use reentrant lock to handle something similar.
| for (Buffer buf : buffersToFlush) { | ||
| combinedRecords.addAll(buf.getRecords()); | ||
| } |
There was a problem hiding this comment.
I may be not clear on the overall approach here but Why are we combining the buffers here? also, may be not a big deal but we are creating additional copy of the payload here.
There was a problem hiding this comment.
i have addressed the comment, this might not be needed for flushing to lambda but needed for handlingFailures. Lets say the "full buffer" records have some failures, only those have to be tagged and forwarded(failure behaviour), the remaning records should persist for the next doOutput should not go through the failure handling mechanism. I have refactored the lambdaCommonHandler to take in Buffer and moved this logic within the catch.
|
|
||
| verify(numberOfRecordsFailedCounter, times(1)).increment(1); | ||
| // Utility to set private fields | ||
| private static void setPrivateField(Object target, String fieldName, Object value) { |
There was a problem hiding this comment.
There is a data-prepper-test-common module that has these kind of common reusable methods used for test cases. We can reuse these utilities from there and avoid adding these methods in every plugin.
Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>
Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>
Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>
Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>
Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>
srikanthjg
force-pushed
the
lambda-sink-stateful
branch
from
ced7a2d to
3d9fb37
Compare
Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>
srikanthjg
force-pushed
the
lambda-sink-stateful
branch
from
0795cdc to
debad91
Compare
Description
Add Stateful Buffering to LambdaSink
Issues Resolved
Resolves #5353
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.