Synchronous / asynchronous REST client

Question

I have working code with original design, and now I had a slight design change so trying to code review that as well. I already had code review on my original design here.

Original Design:

I am using RestTemplate as my HttpClient to execute a URL and then my server will return a JSON string as the response. The customer will call this library by passing DataKey object which has userId in it.

• Using the given userId, I will find out what are the machines that I can hit to get the data and then store those machines in a ArrayList, so that I can execute them sequentially.
• After that I will check whether the first hostname is in block list or not. If it is not there in the block list, then I will make a URL with the first hostname in the list and execute it and if the response is successful then return the response. But let's say if that first hostname is in the block list, then I will try to get the second hostname in the list and make the url and execute it, so basically, first find the hostname which is not in block list before making the URL.
• Now, let's say if we selected first hostname which was not in the block list and executed the URL and somehow server was down or not responding, then I will execute the second hostname in the list and keep doing this until you get a successful response. But make sure they were not in the block list as well so we need to follow above point.
• If all servers are down or in block list, then I can simply log and return the error that service is unavailable.

I am making a library in which I will have synchronous (getSyncData) and asynchronous (getAsyncData) methods in it.

• getSyncData() - waits until I have a result, returns the result.
• getAsyncData() - returns a Future immediately which can be processed after other things are done, if needed.

Below is my DataClient class which will be called by customer and they will pass DataKey object to either getSyncData or getAsyncData method depending on what they want to call. In general some customer will call getSyncData method and some customer might call getAsyncData method.

public class DataClient implements Client {

    private RestTemplate restTemplate = new RestTemplate();
    private ExecutorService service = Executors.newFixedThreadPool(15);

    @Override
    public DataResponse getSyncData(DataKey key) {
        DataResponse response = null;
        Future<DataResponse> responseFuture = null;

        try {
            responseFuture = getAsyncData(key);
            response = responseFuture.get(key.getTimeout(), key.getTimeoutUnit());
        } catch (TimeoutException ex) {
            response = new DataResponse(DataErrorEnum.CLIENT_TIMEOUT, DataStatusEnum.ERROR);
            responseFuture.cancel(true); // terminating the tasks that have got timed out
            // logging exception here               
        } catch (Exception ex) {
            response = new DataResponse(DataErrorEnum.ERROR_CLIENT, DataStatusEnum.ERROR);
            // logging exception here
        }

        return response;
    }   

    @Override
    public Future<DataResponse> getAsyncData(DataKey key) {
        DataFetcherTask task = new DataFetcherTask(key, restTemplate);
        Future<DataResponse> future = service.submit(task);

        return future;
    }
}

DataFetcherTask class:

public class DataFetcherTask implements Callable<DataResponse> {

    private DataKey key;
    private RestTemplate restTemplate;

    public DataFetcherTask(DataKey key, RestTemplate restTemplate) {
        this.key = key;
        this.restTemplate = restTemplate;
    }

    @Override
    public DataResponse call() throws Exception {
        ResponseEntity<String> response = null;

        Mappings mappings = ShardMapping.getMappings(key.getFlowType());
        List<String> hostnames = mappings.getAllHostnames(key);

        for (String hostname : hostnames) {
            if (DataUtils.isEmpty(hostname) || ShardMapping.isBlockHost(hostname)) {
                continue;
            }
            try {
                String url = generateUrl(hostname);
                URI uri = URI.create(url);
                response = restTemplate.exchange(uri, HttpMethod.GET, key.getEntity(), String.class);

                ShardMapping.unblockHost(hostname);
                if (response.getStatusCode() == HttpStatus.NO_CONTENT) {
                    return new DataResponse(response.getBody(), DataErrorEnum.NO_CONTENT,
                            DataStatusEnum.SUCCESS);
                } else {
                    return new DataResponse(response.getBody(), DataErrorEnum.OK, DataStatusEnum.SUCCESS);
                }
            } catch (HttpClientErrorException ex) {
                HttpStatusCodeException httpException = ex;
                DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
                String errorMessage = httpException.getResponseBodyAsString();
                return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
                // logging exception here                   
            } catch (HttpServerErrorException ex) {
                HttpStatusCodeException httpException = ex;
                DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
                String errorMessage = httpException.getResponseBodyAsString();
                return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
                // logging exception here                                       
            } catch (RestClientException ex) {
                ShardMapping.blockHost(hostname);
                // logging exception here                                       
            }
        }

        return new DataResponse(DataErrorEnum.SERVICE_UNAVAILABLE, DataStatusEnum.ERROR);
    }
}

My block list keeps getting updated from another background thread every 1 minute. If any server is down and not responding, then I need to block that server by using this -

ShardMapping.blockHost(hostname);

And to check whether any server is in block list or not, I use this -

ShardMapping.isBlockHost(hostname);

I am returning SERVICE_UNAVAILABLE if all the servers are down or in block list at the end. Now the above code is working fine in production from a long time.

Design Change:

Change is - People will pass DataKey (for example keyA) object to my library and then I will make a http call to another service (which I am not doing in my current design) by using user id present in DataKey (keyA) object which will give me back list of user id's so I will use those user id's and make another DataKey (keyB, keyC, keyD) object one for each user id returned in the response. And then I will have List<DataKey> object which will have keyB, keyC and keyD DataKey object. Max element in the List<DataKey> will be three, that's all.

Now for each of those DataKey object in List<DataKey> I want to execute above DataFetcherTask.call method in parallel and then make List<DataResponse> by adding each DataResponse. So now my DataFetcherTask class will return back List<DataResponse> object instead of DataResponse and then signature of getSyncData and getAsyncData method will change as well. Idea behind this parallel call is to get the data for all those max three keys in the same global timeout value.

As you can see I have a global timeout set in getSyncData method and I want to keep that way because I want to throttle the request instead of asking people to bombard our servers. So I want to apply the same global timeout on this new http call as well which will get list of user id's. Meaning let's say if this http call is taking some time, then I want to timeout the way I was doing earlier bcoz it's happening in the client that's why.

Below is the changes I have made in the code:

public class DataFetcherTask implements Callable<List<DataResponse>> {

    private DataKey key;
    private RestTemplate restTemplate;
    private ExecutorService executorService = Executors.newFixedThreadPool(10);

    public DataFetcherTask(DataKey key, RestTemplate restTemplate) {
        this.key = key;
        this.restTemplate = restTemplate;
    }

    @Override
    public List<DataResponse> call() throws Exception {
        List<DataKey> keys = performKeyRequest();
        List<Future<DataResponse>> responseFutureList = new ArrayList<Future<DataResponse>>();

        for (final DataKey key : keys) {
            responseFutureList.add(executorService.submit(new Callable<DataResponse>() {
                @Override
                public DataResponse call() throws Exception {
                    return performDataRequest(key);
                }
            }));
        }

        List<DataResponse> responseList = new ArrayList<DataResponse>();
        for (Future<DataResponse> future : responseFutureList) {
            responseList.add(future.get());
        }

        return responseList;
    }

    // In this method I am making an HTTP call to another service
    // and then I will make List<DataKey> accordingly.
    private List<DataKey> performKeyRequest() {
        List<DataKey> keys = new ArrayList<>();
        // use key object which is passed in contructor to make HTTP call to another service
        // and then make List of DataKey object and return keys.
        return keys;
    }       

    private DataResponse performDataRequest(DataKey key) {
        Mappings mappings = ShardMapping.getMappings(key.getFlowType());
        List<String> hostnames = mappings.getAllHostnames(key);

        for (String hostname : hostnames) {
            if (DataUtils.isEmpty(hostname) || ShardMapping.isBlockHost(hostname)) {
                continue;
            }
            try {
                String url = generateUrl(hostname);
                URI uri = URI.create(url);
                ResponseEntity<String> response = restTemplate.exchange(uri, HttpMethod.GET, key.getEntity(), String.class);

                ShardMapping.unblockHost(hostname);
                if (response.getStatusCode() == HttpStatus.NO_CONTENT) {
                    return new DataResponse(response.getBody(), DataErrorEnum.NO_CONTENT,
                            DataStatusEnum.SUCCESS);
                } else {
                    return new DataResponse(response.getBody(), DataErrorEnum.OK, DataStatusEnum.SUCCESS);
                }
            } catch (HttpClientErrorException ex) {
                HttpStatusCodeException httpException = ex;
                DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
                String errorMessage = httpException.getResponseBodyAsString();
                return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
                // logging exception here                   
            } catch (HttpServerErrorException ex) {
                HttpStatusCodeException httpException = ex;
                DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
                String errorMessage = httpException.getResponseBodyAsString();
                return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
                // logging exception here                                       
            } catch (RestClientException ex) {
                ShardMapping.blockHost(hostname);
                // logging exception here                                       
            }
        }

        return new DataResponse(DataErrorEnum.SERVICE_UNAVAILABLE, DataStatusEnum.ERROR);       
    }
}

So now as you can see I have two executors, one outside the DataFetcherTask class and other inside DataFetcherTask class. Outside one is to control the amount of threads that can do the job and have global level timeout. Inside one is to parallelize the call to performDataRequest method so that in the same global timeout value, I can query in parallel for each DataKey object and get the response.

Do we need to have two executors like I have in my code? Is there any better way to solve this problem or any kind of simplification/design change we can do? I feel little weird having call method inside another call method not sure whether it's a good design.

Please review my code. If you have any questions, please ask. If you have any comments, please give them. I am still on Java 7.

Answer 1

In call(), you have duplicate code:

        } catch (HttpClientErrorException ex) {
            HttpStatusCodeException httpException = ex;
            DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
            String errorMessage = httpException.getResponseBodyAsString();
            return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
            // logging exception here                   
        } catch (HttpServerErrorException ex) {
            HttpStatusCodeException httpException = ex;
            DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
            String errorMessage = httpException.getResponseBodyAsString();
            return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
            // logging exception here                                       
        }

Since Java 7, you can remove the duplicate code with a multiple-catch:

        } catch (HttpClientErrorException | HttpServerErrorException ex) {
            HttpStatusCodeException httpException = ex;
            DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
            String errorMessage = httpException.getResponseBodyAsString();
            return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
            // logging exception here                                       
        }

In the design change, it is in performDataRequest().

---

In the call() method of the design change, you have this line:

    List<Future<DataResponse>> responseFutureList = new ArrayList<Future<DataResponse>>();

If the code that is calling the method performs minimal gets to the list, it is preferred to use LinkedList instead, as LinkedList is faster at adding but slower at getting.

Same thing here:

    List<DataKey> keys = new ArrayList<>();

I don't know much about this topic; I'll leave that to the more experienced programmers.

Answer 2

On every getAsyncData call you create new DataFetcherTask with

private ExecutorService executorService = Executors.newFixedThreadPool(10);

(why 10?) then create 3 threads and then abandon them all. (Note that it's better to shutdown threadpool manually than rely on ThreadPoolExecutor.finalize() method.)
Creating threads on each getAsyncData call is expensive. Make this threads reusable, for example by creating another thread pool in DataClient and passing it into DataFetcherTask constructor.

---

My block list keeps getting updated from another background thread every 1 minute. If any server is down and not responding, then I need to block that server by using this...

But you didn't give more details about background thread so I don't know how exactly it works.
Inside performDataRequest function you call ShardMapping.unblockHost(hostname); it's safer to rely on health checks in your background thread instead.
I'll briefly explain what can happen: after receiving response from the host in 1st thread, this host goes down, the 2nd thread detects it and adds host to blockList, and only after this (thread scheduling is quite complex), the 1st thread invokes ShardMapping.unblockHost(hostname); and removes host (which is currently down) from blockList. So one more request will fail to reach this host.

---

It would be convinient to create a shutdown function in DataClient with service .shutdown() or service.shutdownNow() inside it.

I hope you wouldn't use hardcoded constants Executors.newFixedThreadPool(10); in your final version and give users ability to tune your library.

Answer 3

CompletionService

Your call() method can be simplified by using a CompletionService. This allows the code to complete in a first-man-home scenario, if you want too.

@Override
public List<DataResponse> call() throws Exception {
    List<DataKey> keys = performKeyRequest();
    CompletionService<DataResponse> comp = new ExecutorCompletionService<>(executorService);


    int count = 0;
    for (final DataKey key : keys) {
        comp.submit(new Callable<DataResponse>() {
            @Override
            public DataResponse call() throws Exception {
                return performDataRequest(key);
            }
        });
    }


    List<DataResponse> responseList = new ArrayList<DataResponse>();
    while (count-- > 0) {
        Future<DataResponse> future = comp.take();
        responseList.add(future.get());
    }
    return responseList;
}

Note how the comp.take() will return the (completed) Futures in completion order.

Catching Exception

This "just" makes the performDataRequest method a problem still. It has too much error handling ;-) That handing makes it hard to read. You have already been pointed to the multi-catch system, so I won't repeat that. But, what people have missed, is that you auto-assign the Client or Server exceptions to be an HttpStatusCodeException .... so, I looked, and this is the documentation:

Direct Known Subclasses:

HttpClientErrorException, HttpServerErrorException

So, since both of your "caught" exceptions are children of the base class, you may as well just catch be base exception:

} catch (HttpStatusCodeException ex) {

Now you don't need the assignment either... get rid of:

HttpStatusCodeException httpException = ex;

Gratuitous variables

You don't have to make everything a variable - one-time use variables sometimes make things harder to read - and imply data that "sticks around". These lines here:

Mappings mappings = ShardMapping.getMappings(key.getFlowType());
List<String> hostnames = mappings.getAllHostnames(key);

for (String hostname : hostnames) {

I would restructure as just:

Mappings mappings = ShardMapping.getMappings(key.getFlowType());

for (String hostname : mappings.getAllHostnames(key)) {

The reason I "excised" the hostnames variable is because the List<String> generic value makes it relatively complicated. Now there's just mappings, and a String loop. No apparent generics.

Similarly, I would excise the uri variable. It is not useful as a standalone value.

if-return

The inner most if/else block should be reverted to just an if-block. Before:

if (response.getStatusCode() == HttpStatus.NO_CONTENT) {
    return new DataResponse(response.getBody(), DataErrorEnum.NO_CONTENT,
            DataStatusEnum.SUCCESS);
} else {
    return new DataResponse(response.getBody(), DataErrorEnum.OK, DataStatusEnum.SUCCESS);
}

after:

if (response.getStatusCode() == HttpStatus.NO_CONTENT) {
    return new DataResponse(response.getBody(), DataErrorEnum.NO_CONTENT,
            DataStatusEnum.SUCCESS);
}
return new DataResponse(response.getBody(), DataErrorEnum.OK, DataStatusEnum.SUCCESS);

Unblocked

There is no need to unset the "blocked" status. You will never run the inner loop if the host is blocked, so you will never be able to unblock it. This code is a never-run:

ShardMapping.unblockHost(hostname);

Remove it.

Summary

This boils the method down to something like:

private DataResponse performDataRequest(DataKey key) {
    Mappings mappings = ShardMapping.getMappings(key.getFlowType());

    for (String hostname : mappings.getAllHostnames(key)) {
        if (DataUtils.isEmpty(hostname) || ShardMapping.isBlockHost(hostname)) {
            continue;
        }
        try {
            String url = generateUrl(hostname);
            ResponseEntity<String> response = restTemplate.exchange(
                    URI.create(url), HttpMethod.GET, key.getEntity(), String.class);

            if (response.getStatusCode() == HttpStatus.NO_CONTENT) {
                return new DataResponse(response.getBody(), DataErrorEnum.NO_CONTENT, DataStatusEnum.SUCCESS);
            }
            return new DataResponse(response.getBody(), DataErrorEnum.OK, DataStatusEnum.SUCCESS);

        } catch (HttpStatusCodeException httpException) {
            DataErrorEnum error = DataErrorEnum.getErrorEnumByException(httpException);
            String errorMessage = httpException.getResponseBodyAsString();
            return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
            // logging exception here
        } catch (RestClientException ex) {
            ShardMapping.blockHost(hostname);
            // logging exception here
        }
    }

    return new DataResponse(DataErrorEnum.SERVICE_UNAVAILABLE, DataStatusEnum.ERROR);
}