Collect and combine results of async calls returning completable futures in Java in a cleaner way?

Question

I need an abstract way to implement this below functionality because in my code I have to use it in multiple places with different REST endpoints. If I ignore it will cause code duplication.
Please suggest a good solution.

This below code is working fine. I need a clean and simple way to write such calls every time I need to call an REST API call in batch and then collect the result of the batch into a list. A default way in Java or Spring or an own abstractions to handle such calls and retrievals.

I want to a call a REST API then collect the result into a list and I want to make async calls to the API endpoint that will return list of objects in JSON.

I need to call this REST endpoint more many times with particular batch size.

      /**
         * Retrieve basic Student Details(Student) details from Platform API service
         * asynchronously
         * API can process only 50 students so passing 50 studentIds at a time
         * to fetch and collect into a completable list and then process
         */
        private List<Student> getstudentRecords(List<Integer> studentIds)
                throws InterruptedException, ExecutionException {
            List<List<Integer>> studentIdsPartitions = Lists.partition(studentIds, FETCH_STUDENT_BATCH_SIZE);

            List<Student> studentRecords = new ArrayList<>();

            List<CompletableFuture<List<Student>>> studentRecordsOfBatchedstudentIdsFutureList = new ArrayList<>();

            // get studentRecord details from pst REST API asynchronously and collect them
            // in completable future list
            for (List<Integer> batchedstudentIds : studentIdsPartitions) { 
                CompletableFuture<List<Student>> studentRecordsOfBatchedstudentIdsFuture = 
                        studentFetchService.retrieveStudents(batchedstudentIds);
                studentRecordsOfBatchedstudentIdsFutureList.add(studentRecordsOfBatchedstudentIdsFuture);
            }

            // collect studentRecord details synchronously(by blocking thread) to collect
            // from collected completable future list
            for (CompletableFuture<List<Student>> studentRecordsFuture : studentRecordsOfBatchedstudentIdsFutureList) {
                List<Student> StudentList = studentRecordsFuture.get();
                studentRecords.addAll(StudentList);
            }
            return studentRecords;
          }

@Component
 public class StudentFetchService{
 @Async
    public CompletableFuture<List<Student>> retrieveStudents(List<Integer> studentIds) {

        return CompletableFuture.completedFuture(
            getStudentsFromPlatform(studentIds);
        );
    }
 }

private ResponseEntity<List<Student>> getStudentsFromPlatform(List<Integer> stdIds) {
        ResponseEntity<List<Student>> response;
        String stdIdsQueryParams = Optional.ofNullable(stdIds).orElseGet(Collections::emptyList)
                .stream().map(x->x.toString()).collect(Collectors.joining(","));
        URI uri = UriComponentsBuilder.fromUriString("http://www.tomsheldondev.com/testapi/student")
                .queryParam("stdId", stdIdsQueryParams).build().encode().toUri();
        response = restTemplate.exchange(uri, HttpMethod.GET, null,
                new ParameterizedTypeReference<List<Student>>() {
                });
        return response;
    }

Answer 1

Consider making the method generic and creating an interface for doing the REST API calls:

    static interface RequestService<T,S> {
        CompletableFuture<List<T>> request(List<S> inputs);
    }
    
    private <T,S> List<T> batchedFetch(RequestService<T,S> service, List<S> inputs)
            throws InterruptedException, ExecutionException {
        // maybe make BATCH_SIZE an input to this method 
        List<List<S>> partitionedInput = Lists.partition(inputs, BATCH_SIZE);


        List<CompletableFuture<List<T>>> futureList = new ArrayList<>();

        // get details from REST API asynchronously and collect them
        // in completable future list
        for (List<S> batchedInputs : partitionedInput) {
            CompletableFuture<List<T>> batchedFutures = service.request(batchedInputs);
            futureList.add(batchedFutures);
        }

        // collect responses synchronously(by blocking thread) to collect
        // from collected completable future list
        List<T> results = new ArrayList<>(futureList.size());
        for (CompletableFuture<List<T>> future : futureList) {
            List<T> batchResults = future.get();
            results.addAll(batchResults);
        }
        return results;
    }

Now you can make calls that take different inputs and get different outputs by providing a different implementation for the RequestService:

@Component
 public class StudentFetchService implements RequestService<Student,Integer> {
    @Override
    @Async
    public CompletableFuture<List<Student>> request(List<Integer> studentIds) {

        return CompletableFuture.completedFuture(
            getStudentsFromPlatform(studentIds)
        );
    }
 }

    static class DoSomethingElseService implements RequestService<Boolean,String> {
       @Override
       public CompletableFuture<List<Boolean>> request(List<String> inputs) {

           return CompletableFuture.completedFuture(
               getResultsFromRestCall(inputs)
           );
       }
       private List<Boolean> getResultsFromRestCall(List<String> inputs) {
           return Collections.emptyList(); // get real results here
       }
    }

Try it out:

    void testIt() throws InterruptedException, ExecutionException {
        List<Student> students = batchedFetch(new StudentFetchService(), List.of(1,2,3));
        List<Boolean> results = batchedFetch(new DoSomethingElseService(), List.of("one", "two", "three"));
    }