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.