Inject PeerForwarderServer in to DataPrepper class

[asifsmohammed]

Signed-off-by: Asif Sohail Mohammed nsifmoh@amazon.com

Description

Start and stop peer forwarder server in DataPrepper

Issues Resolved

resolves #1608

Check List

• New functionality includes testing.
• New functionality has been documented.
  • 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.

[dlvenable]

We abstract that Data Prepper starts this server, but then expose the shutdown.

It appears that the current split between shutdown and shutdownDataPrepperServer is to change the status code returned. Basically, if we shutdown the pipeline, we return 200, even if there is a problem shutting down the Data Prepper Core HTTP Server. What do we want to do when Core Peer Forwarder fails to shutdown? There might be some implications with the buffers here.

[asifsmohammed]

I don't think there's a different status code for shutdown and shutdownDataPrepperServer. If shotdwon is successful we return 200 but if it fails we return 500. In both the cases we call shutdownDataPrepperServer in finally block

[dlvenable]

Yes, you are correct.

So my question is: What should we use for the behavior of the peer forwarder server? If it fails to shutdown should we return 500 or 200?

[asifsmohammed]

Status code is dependent on whether we are able to shutdown the pipeline successfully or not.

My thinking was if shutdown call to Data Prepper server doesn't affect the status code, why should shutdown call to Peer Forwarder server affect the return status code?

[asifsmohammed]

If we need to abstract the shutdown call to Peer Forwarder Server, we can include that in the shutdownDataPrepperServer() method instead of having a separate method.

public void shutdownDataPrepperServer() {
    dataPrepperServer.stop();
    peerForwarderServer.stop();
}

[dlvenable]

My thinking was if shutdown call to Data Prepper server doesn't affect the status code, why should shutdown call to Peer Forwarder server affect the return status code?

The only reason I would think would be if it affects the buffers.

If we need to abstract the shutdown call to Peer Forwarder Server, we can include that in the shutdownDataPrepperServer() method instead of having a separate method.

I think I agree with this. Maybe we should rename the method as well since it is now shutting down multiple other components.

[asifsmohammed]

One way to make sure that all the data in buffers are empty is add it as part of shouldStop() method in ProcessWorker. We have to iterate all the peer forwarder buffers to make sure they are empty.

data-prepper/data-prepper-core/src/main/java/org/opensearch/dataprepper/pipeline/ProcessWorker.java

Lines 80 to 87 in b02a15e

	private boolean shouldStop() {
	return pipeline.isStopRequested() && areComponentsReadyForShutdown();
	}
	private boolean areComponentsReadyForShutdown() {
	return readBuffer.isEmpty() && processors.stream()
	.map(Processor::isReadyForShutdown)
	.allMatch(result -> result == true);

[codecov-commenter]

Codecov Report

Merging #1764 (b716a62) into main (b02a15e) will increase coverage by 0.00%.
The diff coverage is 100.00%.

@@            Coverage Diff            @@
##               main    #1764   +/-   ##
=========================================
  Coverage     93.99%   93.99%           
  Complexity     1414     1414           
=========================================
  Files           187      187           
  Lines          4162     4165    +3     
  Branches        330      330           
=========================================
+ Hits           3912     3915    +3     
  Misses          170      170           
  Partials         80       80           

Impacted Files	Coverage Δ
...n/java/org/opensearch/dataprepper/DataPrepper.java	96.55% <100.00%> (+0.39%)	⬆️
...h/dataprepper/pipeline/server/ShutdownHandler.java	100.00% <100.00%> (ø)

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

[dlvenable]

Thanks for combining these shutdown calls!