Fallback Implementation

[parveensania]

Please add a meaningful description for your change here

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[m-trieu]

some minor updates needed

[m-trieu]

just the test then LGTM

[scwhittle]

Can you merge in the removal of the throttling tracker and other changes?

[scwhittle]

ditto, should be

(version < pending) || (version==pending && type=activeType)

[parveensania]

We can't add the check version==pending here. As in the caller function pending version gets set to new endpoints version. So in this function for the most latest endpoints, pending version will always be equal to new endpoints version(irrespective of the enpoint type)

[scwhittle]

I don't really like the duplication here between this an SingleSourceWorkerHarness in SE mode.
Could this just create a SingleSourceWorkerHarness and start it?

It seems like it is just missing setBudget support, but perhaps that could be added to SingleSourceWorkerHarness? Or alternatively could we have some static method to construct this receiver that we share between the two classes?

[parveensania]

Refactored the class. PTAL

One thing I want clarification on, when close() is called, SSWH shuts down the executor service and workCommitter but does not call shutdown of getWorkStream explicitly. Should we be calling getWorkStream.shutdown() for graceful termination or it is irrelevant?

[scwhittle]

@arunpandianp If you have cycles to help review that would be great

[scwhittle]

I don't see where this is used, I think you could just move the definition into the different branches.

[scwhittle]

all of these deps that are just for the fallback case I'm wondering if we can create lazily so that we don't startup threads etc unless we need to.

One maybe easy way to do that is replace all the ones that are only used in the fallback path with a Supplier and Suppliers.memoize

example
Supplier ... = Suppliers.memoize(...);
Supplier ... = Supppliers.memoize(()->new StreamPoolGetDataClient(getDataMetricTracker, getDataStreamPool.get());

[scwhittle]

can this be removed and just fixed?

[scwhittle]

this seems racy, it's not synchronized and we're passing this method to some other class to call which could be simultaneous with it's usage in setBudget

[arunpandianp]

@arunpandianp If you have cycles to help review that would be great

on it.

[arunpandianp]

To fallback, will it be simpler to shutdown the current streamingWorkerHarness, reinitialize it with SingleSourceWorkerHarness and call start?

void switchStreamingWorkerHarness(boolean directPath) {
if (directPath) {
  if(!(streamingWorkerHarness instanceof FanOutStreamingEngineWorkerHarness)) {
     streamingWorkerHarness.shutdown();
     streamingWorkerHarness = createFanOutStreamingEngineWorkerHarness();
     streamingWorkerHarness.start();
     return;
  } 
} else {
  if (!(streamingWorkerHarness instanceof SingleSourceWorkerHarness)) {
     streamingWorkerHarness.shutdown();
     streamingWorkerHarness = createSingleSourceWorkerHarness();
     streamingWorkerHarness.start();
     return;
  } 
}
}

The directpath boolean can be read from StreamingGlobalConfigHandle::getConfig().