Why main thread is not waiting for other async process(threads) to complete. allOff not working properly

user_8182851 0
Viewed 529

I am calling one async method inside for loop and adding future of it into a list. I am not sure why allOff is not waiting to complete all futures and returning partial result. Have a look of my code.

I have one overridden method 

@Overide
@Async
CompletableFuture<someType> fetchData()
{
returns new CompletableFuture<someType>();
}

I am calling above method in a for loop with different instances.

get all beans which implments one interface which has mehod fetchData.

Map<String, SomeClass> allBeans =context.getBeansOfType(SomeClass.class);

List<SomeClass> list=
    allBeans.values().stream().collect(SomeClass.toList());

for (SomeClass classInstance: list) {
  classInstance.fetchData().thenApply(x->{
    //Some DB persistence Call
    futureList.add(x);
  });
}

after that I am applying allOff so that all future can be completed but it is not waiting for all and main thread excuting rest of flow.

CompletableFuture<Void> combinedFutures = CompletableFuture.allOf(
      futureList.toArray(new CompletableFuture[futureList.size()]));

  CompletableFuture<List<futureResponse>> finalList=
      combinedFutures.thenApply(v -> {
        return futureList.stream().map(m -> 
        m.join()).collect(Collectors.toList());
       });

finalList- in this List I want all the completed futures returned by fetch invocation.

In finalList I am always getting 2 objects but fetchData is getting run 5 times( based on number of instances), I saw the log after all of remaining async call are getting completed. Could someone help here.

Observation:- After putting main thread on sleep for 30 sec, I could see I have all 5 objects in the list. Could some one please tell why main thread is not waiting at allOff for all futures to complete.

completable-future asynchronous multithreading java
2 Answers

You have a race condition : 

classInstance.fetchData().thenApply(x->{
    futureList.add(x);
});

This code means that only when the future x is completed, then x will be added to futureList. This might be in 10 milliseconds, or 2 hours, who knows ? (It might be never if the future fails exceptionnaly).

So, when the code reaches

CompletableFuture.allOf(futureList....

There is no "guarantee" that the thenApply have been called. futureList could even be empty.

One way you could correct this code is like so : 

Map<String, SomeClass> allBeans =context.getBeansOfType(SomeClass.class);


List<SomeClass> list=
    allBeans.values().stream().collect(SomeClass.toList());

 for (SomeClass classInstance: list) {
    futureList.add(classInstance.fetchData());
 }

Or if you actually need to do something in a thenApply: 

 for (SomeClass classInstance: list) {
    futureList.add(
        classInstance.fetchData().thenApply(x -> whatever(x))
    );
 }

This way, your futureList is populated not when an async result returns (which is unknown, and might even fail with an exception an never occur), but as soon as the async call is created, which is what you actually want.

user_21310740

IIUC, what you want to do can be done simpler by doing

CompletableFuture<List<FutureResponse>> = CompletableFuture.supplyAsync(() -> {
    // start fetches and collect the individual futures
    List<CompletableFuture> fetches = 
       allBeans.values().stream()
               .map(SomeClass::fetchData)
               .collect(toList());
    // join all futures and return the list of the results
    return fetches.stream()
               .map(CompletableFuture::join)
               .collect(toList());
}

I think you can't do it in a single stream (ie map to fetch, then immediately to join) because that might wait for the join before the next future is created.

user_28377410
Related