Zero Buffer Implementation and Tests #5416
Conversation
Signed-off-by: Mohammed Aghil Puthiyottil <[email protected]>
MohammedAghil
requested review from
sb2k16,
chenqi0805,
engechas,
san81,
srikanthjg,
graytaylor0,
dinujoh,
kkondaka,
KarstenSchnitter,
dlvenable and
oeyh
as code owners
There was a problem hiding this comment.
Thank you @MohammedAghil . This is an exciting new feature!
| Metrics.addRegistry(new SimpleMeterRegistry()); | ||
| } | ||
|
|
||
| /*-------------------------Tests for Writing to ZeroBuffer---------------------------*/ |
There was a problem hiding this comment.
I don't think these comments will remain valid as we modify this code.
One approach you can use is to create JUnit @Nested tests. These will even report tests in groups. And you may even find opportunities to create common setup.
Here is an example:
|
|
||
| dependencies { | ||
| implementation project(':data-prepper-api') | ||
| implementation project(':data-prepper-core') |
There was a problem hiding this comment.
I don't think we should depend on data-prepper-core here.
I think it may make the most sense to move this project (zero-buffer) into data-prepper-core.
|
|
||
| @DataPrepperPluginConstructor | ||
| public ZeroBuffer(PipelineDescription pipelineDescription) { | ||
| this.pluginMetrics = PluginMetrics.fromNames(PLUGIN_COMPONENT_ID, pipelineDescription.getPipelineName()); |
There was a problem hiding this comment.
I think we can now get this from dependency injection. @san81 , is that correct?
| readRecordsCounter.increment(storedRecords.size() * 1.0); | ||
| } | ||
|
|
||
| return new AbstractMap.SimpleEntry<>(storedRecords, checkpointState); |
There was a problem hiding this comment.
Please use Map.entry(storedRecords, checkpointState) instead.
| if (storedRecords!= null && !storedRecords.isEmpty()) { | ||
| checkpointState = new CheckpointState(storedRecords.size()); | ||
| threadLocalStore.remove(); | ||
| readRecordsCounter.increment(storedRecords.size() * 1.0); |
There was a problem hiding this comment.
Why * 1.0? If it is make a double, may just cast.
| * Represents the base class for zero buffer implementation and implements {@link Buffer} interface. | ||
| * It provides a common functionality to run all processors and sinks within the same thread context. | ||
| */ | ||
| public abstract class AbstractZeroBuffer <T extends Record<?>> implements Buffer<T> { |
There was a problem hiding this comment.
Rather than an abstract class, make this an interface. Maybe something like SupportsPipelineRunner.
interface SupportsPipelineRunner {
PipelineRunner getPipelineRunner();
void setPipelineRunner(PipelineRunner pipelineRunner);
}
Then move the implementation into ZeroBuffer.
| import java.util.Map; | ||
| import java.util.concurrent.TimeoutException; | ||
|
|
||
| @DataPrepperPlugin(name = "zero_buffer", pluginType = Buffer.class) |
There was a problem hiding this comment.
Should this just be named zero? We have moved away from including the type suffix. For example, we went from otel_trace_source to otel_trace.
| } | ||
|
|
||
| if (threadLocalStore.get() == null) { | ||
| threadLocalStore.set(records); |
There was a problem hiding this comment.
As you are possibly reusing the thread local store, you actually are modifying this input list. You should make a copy of this. new ArrayList<>(records).
Description
Added Zero Buffer Plugin Implementation and Tests.
Issues Resolved
Resolves [#5415]
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.