Adding hostname to block list after x consecutive failures in multithreading application

Question

I am using Callable in my code which will be called by multiple threads as shown below. Given a user id in DataKey object, I will find out what are the hostnames I can use to get the data and then I will iterate that hostnames list one by one to get the data.

As of now, whenever any RestClientException is thrown on a particular hostname then I am adding that hostname to blockList so that no other thread can make a call to that hostname.

public class Task implements Callable<DataResponse> {
 private final DataKey key;
 private final RestTemplate restTemplate;
 private final ConcurrentHashMap<String, AtomicInteger> failedCallCount = new ConcurrentHashMap<>();
 public Task(DataKey key, RestTemplate restTemplate) {
 this.key = key;
 this.restTemplate = restTemplate;
 }
 @Override
 public DataResponse call() {
 ResponseEntity<String> response = null;
 // construct what are the hostnames I can call basis on user id
 List<String> hostnames = some_code_here;
 for (String hostname : hostnames) {
 // If host name is null or host name is in block list, skip sending request to this host
 if (DataUtils.isEmpty(hostname) || DataMapping.isBlocked(hostname)) {
 continue;
 }
 try {
 String url = createURL(hostname);
 response = restTemplate.exchange(url, HttpMethod.GET, key.getEntity(), String.class);
 resetFailedCallCount(hostname);
 // some code here to return the response if successful
 } catch (HttpClientErrorException ex) {
 // log exception
 return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
 } catch (HttpServerErrorException ex) {
 // log exception
 return new DataResponse(errorMessage, error, DataStatusEnum.ERROR);
 } catch (RestClientException ex) {
 registerFailedCall(hostname);
 if (shouldBeBlocked(hostname)) {
 // adding to block list if condition met
 DataMapping.blockHost(hostname);
 }
 }
 }
 return new DataResponse(DataErrorEnum.SERVER_UNAVAILABLE, DataStatusEnum.ERROR); 
 }
 // can we abstract all these logics from Task class to DataMapping since that's where we are handling 
 // all the hostname whether it is blocked or not.
 private boolean shouldBeBlocked(String hostname) {
 AtomicInteger count = failedCallCount.getOrDefault(hostname, new AtomicInteger());
 return count.get() >= 5;
 }
 private void registerFailedCall(String hostname) {
 AtomicInteger newValue = new AtomicInteger();
 AtomicInteger val = failedCallCount.putIfAbsent(hostname, newValue);
 if (val == null) {
 val = newValue;
 }
 if (val.get() < 5) {
 val.incrementAndGet();
 }
 }
 private void resetFailedCallCount(String hostname) {
 AtomicInteger count = failedCallCount.get(hostname);
 if (count != null) {
 count.set(0);
 }
 } 
}

Below is what I have in DataMapping class:

private static final AtomicReference<ConcurrentHashMap<String, String>> blockedHosts = 
 new AtomicReference<ConcurrentHashMap<String, String>>(new ConcurrentHashMap<String, String>());
public static boolean isBlocked(String hostName) {
 return blockedHosts.get().containsKey(hostName);
}
public static void blockHost(String hostName) {
 blockedHosts.get().put(hostName, hostName);
}
// this is being updated from my background thread which runs every 2 minutes
public static void replaceBlockedHosts(List<String> hostNames) {
 ConcurrentHashMap<String, String> newBlockedHosts = new ConcurrentHashMap<>();
 for (String hostName : hostNames) {
 newBlockedHosts.put(hostName, hostName);
 }
 blockedHosts.set(newBlockedHosts);
}

In the call method as shown above, I will block the hostname if it has thrown RestClientException five times consecutively by calling this line DataMapping.blockHost(hostname); otherwise I will not add it to the blockList.

Also I have a background thread in my library which runs every 2 minutes and that replaces the entire set of blockedHosts and that background thread gets the data from another service whether any hostname is really blocked or not. And I guess because of that reason I am using Atomic Reference on blockedHosts.

Now as my background thread is running every 2 minutes so I might get late to get the information about which machine is down or not so because of that reason, I am implementing local blocking as well. Meaning if any machine throws RestClientException 5 times consecutively, then only add it to block list otherwise don't add it. And if let's say machineA throws RestClientException 4 times but next time it was ok and started serving response, then the count should be reset back to zero for that machine.

In this case, my call method will be called from multiple threads so I need to make sure I am keeping the count properly for each hostname in case they throw RestClientException.

Can we simplify this code better since this library will be used under very heavy load so it has to be fast? Maximum, I will have 70-100 unique machines in total.

I am doing all the failed hostname logic in Task class itself which I don't think I should be doing there. Can we simplify that part and may be do everything in DataMapping class?

Answer 1

Can we simplify that part and may be do everything in DataMapping class?

Sure, but what Data and what Mapping? There's neither, it's a BlockingManager or whatever.

As it stands, you could simply move the code there and be done. However, static is mostly bad and you should create an instance and it do the work (you could use the singleton (anti-)pattern or something better).

---

You're using the map in blockedHosts as a Set. This is fine, but why to put there some dummy data when you can put there the counters?

So I'd combine and reduce it to something like

private final AtomicReference<ConcurrentHashMap<String, AtomicInteger>> blockedHosts =
 new AtomicReference<>(new ConcurrentHashMap<>());
boolean isBlocked(String hostName) {
 AtomicInteger count = blockedHosts.get();
 return count != null && count.get() >= BLOCKING_THRESHOLD;
}
void onFailure(String hostname) {
 AtomicInteger newValue = new AtomicInteger();
 AtomicInteger val = blockedHosts.get().putIfAbsent(hostname, newValue);
 // no need to care about over-reaching 5 here
 (val == null ? newValue : val).incrementAndGet();
}
void onSuccess(String hostname) {
 blockedHosts.get().remove(hostname);
}
// this is being updated from my background thread which runs every 2 minutes
public static void replaceBlockedHosts(List<String> hostNames) {
 ConcurrentHashMap<String, AtomicInteger> newBlockedHosts =
 new ConcurrentHashMap<>();
 for (String hostName : hostNames) {
 newBlockedHosts.put(hostName, new AtomicInteger(BLOCKING_THRESHOLD));
 }
 blockedHosts.set(newBlockedHosts);
}

I'm pretty sure, that you need no AtomicReference as volatile would suffice. You could also use Guava's AtomicLongMap<K> which is just like ConcurrentHashMap<K, AtomicLong> but simplifies the operations you need.

• arsenal: Thanks for suggestion. When I try to use generics as the way you suggested to initialize blockedHosts - I see this error : Type mismatch: cannot convert from AtomicReference<ConcurrentHashMap<Object,Object>> to AtomicReference<ConcurrentHashMap<String,AtomicInteger>> Any idea why? I am using Java 7 btw.
• maaartinus: @lining The type inference seems to work for assignments only, not for arguments. So you need the type parameters for the ConcurrentHashMap, but (I hope) not for the AtomicReference.
• arsenal: Yes this works fine private final AtomicReference<ConcurrentHashMap<String, AtomicInteger>> blockedHosts = new AtomicReference<>( new ConcurrentHashMap<String, AtomicInteger>()); I figured it out after some time.
• arsenal: I do see compilation error in this method isBlocked method. I guess it should be like this right? AtomicInteger count = blockedHosts.get().get(hostname);
• maaartinus: @lining Sure! I didn't try to compile it. so take it lightly. The idea should be clear, the details are up to you. ;)