Send acknowledgements to source when events are forwarded to remote peer #4305
Conversation
kkondaka
requested review from
chenqi0805,
engechas,
graytaylor0,
dinujoh,
asifsmohammed,
dlvenable and
oeyh
as code owners
| @@ -59,6 +59,7 @@ public class PipelineTransformer { | |||
| private final EventFactory eventFactory; | |||
| private final AcknowledgementSetManager acknowledgementSetManager; | |||
| private final SourceCoordinatorFactory sourceCoordinatorFactory; | |||
| private boolean acknowledgementsEnabled; | |||
There was a problem hiding this comment.
would be good to have some basic unit test coverage
| @@ -117,6 +119,8 @@ private void buildPipelineFromConfiguration( | |||
| LOG.info("Building buffer for the pipeline [{}]", pipelineName); | |||
| final Buffer pipelineDefinedBuffer = pluginFactory.loadPlugin(Buffer.class, pipelineConfiguration.getBufferPluginSetting(), source.getDecoder()); | |||
|
|
|||
| if (pipelineDefinedBuffer.isByteBuffer()) | |||
| acknowledgementsEnabled = true; | |||
There was a problem hiding this comment.
Should this be set to true for all sub-pipelines just because one sub-pipeline uses persistent buffer?
There was a problem hiding this comment.
Yes. I have added comment and the test case also validates this.
| @@ -50,7 +50,7 @@ public ProcessWorker( | |||
| this.pipeline = pipeline; | |||
| this.pluginMetrics = PluginMetrics.fromNames("ProcessWorker", pipeline.getName()); | |||
| this.invalidEventHandlesCounter = pluginMetrics.counter(INVALID_EVENT_HANDLES); | |||
| this.acknowledgementsEnabled = pipeline.getSource().areAcknowledgementsEnabled(); | |||
| this.acknowledgementsEnabled = pipeline.getSource().areAcknowledgementsEnabled() || readBuffer.isByteBuffer(); | |||
There was a problem hiding this comment.
I'm not sure this logic is right. A byte buffer in itself does not require acknowledgements. Can we add a new default method on Buffer to express the actual need?
boolean isPersistentBuffer();
Or some other boolean?
Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>
Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>
Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>
Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>
| @@ -78,6 +79,7 @@ public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel, | |||
| this.eventFactory = eventFactory; | |||
| this.acknowledgementSetManager = acknowledgementSetManager; | |||
| this.sourceCoordinatorFactory = sourceCoordinatorFactory; | |||
| this.acknowledgementsEnabled = false; | |||
There was a problem hiding this comment.
This is related to @graytaylor0 's comment below. This should probably not be a field variable and should be set for each pipeline independently.
There was a problem hiding this comment.
No. It should be for all pipelines.
There was a problem hiding this comment.
@kkondaka , If I understand correctly, then a downstream pipeline needs to have acknowledgements enabled. This makes sense.
But, the current code ends up changing all pipelines. Some pipelines can run fully independently of each other without any direct relationship. Those would have acknowledgements enabled incorrectly.
| @@ -78,6 +79,7 @@ public PipelineTransformer(final PipelinesDataFlowModel pipelinesDataFlowModel, | |||
| this.eventFactory = eventFactory; | |||
| this.acknowledgementSetManager = acknowledgementSetManager; | |||
| this.sourceCoordinatorFactory = sourceCoordinatorFactory; | |||
| this.acknowledgementsEnabled = false; | |||
There was a problem hiding this comment.
@kkondaka , If I understand correctly, then a downstream pipeline needs to have acknowledgements enabled. This makes sense.
But, the current code ends up changing all pipelines. Some pipelines can run fully independently of each other without any direct relationship. Those would have acknowledgements enabled incorrectly.
Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>
Description
Send acknowledgements to source when events are forwarded to remote peer
Also, propagate the acknowledgements enabled flag in pipeline correctly when ByteBuffer is used.
Issues Resolved
Resolves #[Issue number to be closed when this PR is merged]
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.