CosmosDb: Remove blocking calls from CFP

[milismsft]

Remove blocking calls from ChangeFeedProcessor implementation and other various fixes.

[christopheranderson]

/azp run java-cosmos-tests

[azure-pipelines]

No pipelines are associated with this pull request.

[christopheranderson]

/azp run java - cosmos - tests

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[moderakh]

1. there are still some blocking call in the code (please check my comments).
2. please add unit tests. There are a lot of code here which is not tested.

[kushagraThapar]

Few questions and minor comments on coding standards.
One thing I noticed throughout the code is, one liner if conditions statements are not surrounded by braces. I believe that reduces the code readability by a big factor. One liner if condition statements should be surrounded by braces. :)

[moderakh]

in response to your question:

milismsft: We need to loop/wait before making another check to retrieve the state. Can you please recommend another way to do this in a non-blocking fashion?

Take a look at Flux.repeat() and Flux.takeUntil() you can build a non-blocking while loop using them.

[moderakh]

two general comments:

1. in some classes we have class fields which are not atomic/synchronized/volatile, those class field may get updated on one thread and accessed on another thread, as the access is not atomic/synchronized/volatile I am not sure if the recent most changes will be visible by the second thread. Please note even for a single flux when you do multiple transformations based on your operators the actions will take place on different thread so we should consider thread safety.

2. we should use reactor scheduler to have the same async pattern in our sdk. I see in some places you decided to use ExecutorService instead of Reactor pattern. If we decide to go with ExecutorService at least we should let ExecutorService handle threads and not create any threads manually.

[milismsft]

1. The oder of the operations ensures that the non-atomic class members are not accessed from threads that are executing in parallel.
2. The ExecutorService will handle the threads.

[moderakh]

as discussed offline,

there is a potential bug:
I see this pattern that you have two threads

1. thread 1 is keep looping till either cancellation is requested or a failure is observed by a PartitionProcessor instance
2. thread 2 is setting the failure

as exception setting/getting are not done synchronized/atomically the most recent value set by thread 2 will not be visible by thread 1. and thread 1. may keep looping anware of the failure.

this applies to all implementors of the interface PartitionProcessor.
we should fix this bug.

for long term we should consider merging the exception returned by getResultException to the Mono returned by run(.) if that makes sense. Millis please create work items for this.

[moderakh]

Please create work items for any work/comment you plan to address outside of this PR; so we can track them.

I see three work items:

1. the remaining blocking calls (Thread.sleep) need to change to non-blocking pattern
2. more mocking unit tests (see bellow) this is very important IMO
3. please consider merging exception to Mono terminal result as discussed.

We need more unit test and integration tests. Ideally every class should be unit tested, or at least the main classes. You can mock the input to the class and test the behaviour given the mock behaviour.
doing just end to end testing is hard to simulate different scenarios.

We should consider mock unit tests for:

• Normal Scenario
• Partition Split (I assume we have some special logic? we should test that)
• network failure, etc

Please see DocumentProducerTest which does just unit test using mocking against different scenarios (partition split, etc)
https://github.com/Azure/azure-cosmosdb-java/blob/cc5d007aa7e5984ceb4919b910fe2e691bd91647/sdk/src/test/java/com/microsoft/azure/cosmosdb/rx/internal/query/DocumentProducerTest.java

[milismsft]

Work items created for two remaining issues:
#5015
#5013

There are two E2E unit tests in practice will ensure coverage for most of the functionality in CFP. There's one work item (in progress) which adds the coverage for extensive usage and handles potential splits. Additionally I'm looking at adding tests similar to what we have in .NET for CFP on opportunistic bases.