Blocking reads when writes are happening on two flows

Question

I am trying to implement lock by which I want to avoid reads from happening whenever I am doing a write.

My requirements are:

• Reads block until all three maps have been set for the first time.
• Now second time, If I am updating the maps, I can still return all the three old maps value(before the updates are done on all three maps) or it should block and return me all the new three maps value whenever the updates are done on all the three maps.

As I have three Maps - primaryMapping , secondaryMapping and tertiaryMapping so it should return either all the new values of three updated maps or it should return all the old values of the map. Basically, while updating I don't want to return primaryMapping having old values, secondaryMapping having having new values, and tertiaryMapping with new values. It should be consistent, either it should return old values or it should return new values after updating the maps. In my case, updating of maps will happen once in 7 or 8 months (very rarely). I am using Countdown Latch for this and it is working fine so far without any issues.

Now I have two flows as shown below: For each flow, I have a URL from where we get the data for above three maps. In general, I will have three maps for both the flow so we will have different values for those three maps for PROCESS flow as compared to DEVICE flow.

public enum FlowType {
    PROCESS, DEVICE;
}

I have a background thread running every 5 minutes which gets the data from each FLOW url and populate those three maps whenever there is an update. When the application is started for the first time, then it will update the mapping and after that, it will update the mapping 7-8 months afterwards. And it doesn't mean that both the flow mapping will change at the same time. It might be possible PROCESS mapping has change but not DEVICE mapping.

public class DataScheduler {

    private RestTemplate restTemplate = new RestTemplate();
    private static final String PROCESS_URL = "http://process_flow_url/";
    private static final String DEVICE_URL = "http://device_flow_url/";
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    public void startScheduleTask() {
        scheduler.scheduleAtFixedRate(new Runnable() {
            public void run() {
                try {
                    callService();
                } catch (Exception ex) {
                    // logging exception here using logger
                }
            }
        }, 0, 5, TimeUnit.MINUTES);
    }

    public void callService() throws Exception {
        String url = null;
        Map<FlowType, String> holder = new HashMap<FlowType, String>();
        for (FlowType flow : FlowType.values()) {
            try {
                url = getURL(flow);
                String response = restTemplate.getForObject(url, String.class);
                holder.put(flow, response);
            } catch (RestClientException ex) {
                // logging exception here using logger      
            }
        }
        parseResponse(holder);
    }

    private void parseResponse(Map<FlowType, String> responses) throws Exception {
        Map<FlowType, Mapping> partitionMapper = new HashMap<FlowType, Mapping>();
        boolean update = false;

        for (Map.Entry<FlowType, String> responseEntry : responses.entrySet()) {
            FlowType flow = responseEntry.getKey();
            String response = responseEntry.getValue();

            Map<String, Map<Integer, Integer>> primaryMapping = new HashMap<>();
            Map<String, Map<Integer, Integer>> secondaryMapping = new HashMap<>();
            Map<String, Map<Integer, String>> tertiaryMapping = new HashMap<>();

            if (!DataUtils.isEmpty(response)) {
                try (Scanner scanner = new Scanner(response)) {
                    boolean hasProcess = Boolean.parseBoolean(scanner.nextLine().trim().substring(HAS_PROCESS_LEN));
                    if (hasProcess) {
                        update = true;

                        // some code    
                        partitionMapper.put(flow, PartitionHolder.createMapping(
                                            primaryMapping, secondaryMapping, tertiaryMapping, version));
                    }
                }
            }
        }
        // if there is any update, then only update the mappings, otherwise not.
        if (update) {
            PartitionHolder.setMappings(partitionMapper);
        }
    }
}

As you can see above, it updates the mapping by calling setMappings method if any of the flow mapping has been updated.

And below is my PartitionHolder class:

public class PartitionHolder {

    public static class Mapping {
        public final Map<String, Map<Integer, Integer>> primaryMapping;
        public final Map<String, Map<Integer, Integer>> secondaryMapping;
        public final Map<String, Map<Integer, String>> tertiaryMapping;
        public final long version;

        public Mapping(Map<String, Map<Integer, Integer>> primaryMapping,
                Map<String, Map<Integer, Integer>> secondaryMapping,
                Map<String, Map<Integer, String>> tertiaryMapping, long version) {
            this.primaryMapping = primaryMapping;
            this.secondaryMapping = secondaryMapping;
            this.tertiaryMapping = tertiaryMapping;
            this.version = version;
        }

        public String getLocalPrimaryAddress(final String localDataCenterPath, final int partitionId) {
            final int localPrimaryHostId = primaryMapping.get(localDataCenterPath).get(partitionId);
            final String localPrimaryHostname = getHostname(localDataCenterPath, localPrimaryHostId);
            return localPrimaryHostname;
        }


        public String getRemotePrimaryAddress(final String remoteDataCenterPath, final int partitionId) {
            final int remotePrimaryHostId = primaryMapping.get(remoteDataCenterPath).get(partitionId);
            final String remotePrimaryHostname = getHostname(remoteDataCenterPath, remotePrimaryHostId);
            return remotePrimaryHostname;
        }

        public String getLocalSecondaryHostIPAddress(final String localDataCenterPath, final int partitionId) {
            final int localSecondaryHostId = secondaryMapping.get(localDataCenterPath).get(partitionId);
            final String localSecondaryHostname = getHostname(localDataCenterPath, localSecondaryHostId);
            return localSecondaryHostname;
        }

        public String getRemoteSecondaryHostIPAddress(final String remoteDataCenterPath, final int partitionId) {
            final int remoteSecondaryHostId = secondaryMapping.get(remoteDataCenterPath).get(
                    partitionId);
            final String remoteSecondaryHostname = getHostname(remoteDataCenterPath, remoteSecondaryHostId);
            return remoteSecondaryHostname;
        }

        private String getHostname(final String dataCenterPath, final int hostId) {
            final String hostname = tertiaryMapping.get(dataCenterPath).get(hostId);
            return hostname;
        }

        public List<String> getServerNames(final DataKey key) {
            LinkedList<String> allPaths =   getFullPath();
            String localPath = allPaths.removeFirst();

            int partitionId = key.getPartition();

            String localPrimaryHostIPAdress = getLocalPrimaryAddress(localPath, partitionId);
            String localSecondaryHostIPAdress = getLocalSecondaryHostIPAddress(localPath, partitionId);

            String remotePath = getPath();
            String remotePrimaryHostIPAdress = getRemotePrimaryAddress(remoteZookPath, partitionId);
            String remoteSecondaryHostIPAdress = getRemoteSecondaryHostIPAddress(remoteZookPath, partitionId);

            List<String> hostnames = new LinkedList<String>();

            // some code
            return hostnames;
        }
    }

    private static final AtomicReference<Map<FlowType, Mapping>> mappings = new AtomicReference<Map<FlowType, Mapping>>();
    private static final CountDownLatch hasInitialized = new CountDownLatch(1);

    public static Mapping getFlowMapping(FlowType flowType) {
        try {
            hasInitialized.await();
            return mappings.get().get(flowType);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }

    public static void setMappings(Map<FlowType, Mapping> newMapData) {
        mappings.set(newMapData);
        hasInitialized.countDown();
    }

    public static Mapping createMapping(
            Map<String, Map<Integer, Integer>> primaryMapping,
            Map<String, Map<Integer, Integer>> secondaryMapping,
            Map<String, Map<Integer, String>> tertiaryMapping, long version) {
        return new Mapping(primaryMapping, secondaryMapping,
                tertiaryMapping, version);
    }
}

Now this is the way I am using PartitionHolder class from the main thread. In a single call, we will get the mapping for one flow only. In below code dataKey.getFlowType() can be PROCESS OR DEVICE.

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

    Mapping mappings = PartitionHolder.getFlowMapping(dataKey.getFlowType());
    List<String> hostnames = mappings.getServerNames(dataKey);

    // use hostnnames here      
}

Opting for a code review here. I am specifically interested in the design I have for my background thread running every 5 minutes and populating those three maps for each flow whenever there is an update. And then from main thread, I am extracting mapping object out for each flow type, thinking it will be thread safe since I am using same mapping object. Is there any better way of doing as the way I am doing currently?

Should I use abstract factory pattern here or anything else for those three maps for each flow? Also as of now, I am updating both the flow mapping if any one of the flow mapping is change so that might not be a good design.

Answer 1

DataScheduler

Since FlowType is an enum, I will suggest using an EnumMap inside callService(), plus inlining your variables:

public void callService() throws Exception {
    // note: no need for early, redundant declarations when you can inline
    // String url = null;
    Map<FlowType, String> holder = new EnumMap<>(FlowType.class);
    for (FlowType flow : FlowType.values()) {
        try {
            // inline-d the following
            // url = getURL(flow);
            // String response = restTemplate.getForObject(url, String.class);
            holder.put(flow, restTemplate.getForObject(getURL(flow), String.class));
        } catch (RestClientException ex) {
            // logging exception here using logger      
        }
    }
    parseResponse(holder);
}

Is it possible to use more specific Exceptions to throw from parseResponse()? They may influence how callers of this method can recover from such errors, e.g. just log, do a timeout-and-retry, or to hit the big-red "Panic!!!" button.

You can also consider re-working the if-statements a little bit to reduce the nesting, and also to cut back on the variable declarations again...

for (Map.Entry<FlowType, String> responseEntry : responses.entrySet()) {
    FlowType flow = responseEntry.getKey();
    String response = responseEntry.getValue();
    if (DataUtils.isEmpty(response)) {
        continue;
    }
    try (Scanner scanner = new Scanner(response)) {
        if (Boolean.parseBoolean(scanner.nextLine().trim().substring(HAS_PROCESS_LEN))) {
            update = true;
            // some code
            // BTW is 'version' some kind of constant?
            // if so, then name it VERSION perhaps?
            partitionMapper.put(flow, PartitionHolder.createMapping(
                    new HashMap<>(), new HashMap<>(), new HashMap<>(), version));
        }
    }
}

PartitionHolder

The implementation for local/remote primary/secondary addresses all look extremely familiar... to the refactoring machine!

public String getLocalPrimaryAddress(final String localDataCenterPath,
                    final int partitionId) {
    return lookup(primaryMapping, localDataCenterPath, partitionId);
}

public String getRemotePrimaryAddress(final String remoteDataCenterPath,           
                    final int partitionId) {
    return lookup(primaryMapping, remoteDataCenterPath, partitionId);
}

public String getLocalSecondaryHostIPAddress(final String localDataCenterPath,
                    final int partitionId) {
    return lookup(secondaryMapping, localDataCenterPath, partitionId);
}

public String getRemoteSecondaryHostIPAddress(final String remoteDataCenterPath,
                    final int partitionId) {
    return lookup(secondaryMapping, remoteDataCenterPath, partitionId);
}

private String lookup(final Map<String, Map<Integer, Integer>> mapping,
                    final String path, final int partitionId) {
    return getHostname(path, mapping.get(path).get(partitionId));
}

private String getHostname(final String dataCenterPath, final int hostId) {
    return tertiaryMapping.get(dataCenterPath).get(hostId);
}

Instead of duplicating the code, you can introduce a private method lookup(Map, String, int) that accepts the relevant Map, the path and the partitionId to standardize the lookup code. Again, instead of using a temporary variable final String hostname = ... in getHostname(String, int), it will be shorter to simply return the result of performing the get() operation.

Also, as you should know, Java 7 supports type inference for generic instance creation aka diamond operator, so you should use <> consistently, where applicable.

Answer 2

1. Try to have methods doing only one thing. Here I think parseResponse should only take one URL (instead of Map<FlowType, String>) and return a Mapping. You can deal with the various flow types within callServices by calling parseResponse many times.

2. EnumMap is meant for cases like Map<FlowType, X>.

3. Declare variables only when needed. In parseResponse the three mappings are declares before if (!DataUtils.isEmpty(response)), but they are only used within that if-statement. (Maybe it is not the case in the real code and the true usage of those mappings was lost in editing.)

Concurrency

The code seems thread-safe. I would probably add in the doc that Mapping is an immutable class. I might also cast all inner maps to immutable maps when they are set in the constructor, just to be safe. The maps cannot be changed with the current API of Mapping, but someone might change it at some point and allow inner map modifications by mistake.

I'm not sure I like your use of the CountDownLatch. I think the API would be better if getFlowMapping would just return null if the values are not yet initialized. I think it's more polite to let the client decide what to do (very likely try again in a while), rather than jam the client thread.

You could also consider returning a Future.

I don't understand your whole architecture, but maybe you could use a subscriber/publisher model to dispatch updates. See Reactive Java.