in elastic-db-tools/src/main/java/com/microsoft/azure/elasticdb/shard/recovery/RecoveryManager.java [538:761]
public List<RecoveryToken> detectMappingDifferences(ShardLocation location,
String shardMapName) {
ExceptionUtils.disallowNullArgument(location, "location");
List<RecoveryToken> listOfTokens = new ArrayList<>();
StoreResults getShardsLocalResult;
try (IStoreOperationLocal op = this.getShardMapManager().getStoreOperationFactory().createGetShardsLocalOperation(this.getShardMapManager(),
location, "DetectMappingDifferences")) {
getShardsLocalResult = op.doLocal();
}
catch (IOException e) {
e.printStackTrace();
throw (ShardManagementException) e.getCause();
}
assert getShardsLocalResult.getResult() == StoreResult.Success;
List<StoreShardMap> shardMaps = shardMapName == null ? getShardsLocalResult.getStoreShardMaps()
: getShardsLocalResult.getStoreShardMaps().stream().filter(s -> shardMapName.equals(s.getName())).collect(Collectors.toList());
List<Pair<StoreShardMap, StoreShard>> shardInfos = shardMaps.stream()
.map(sm -> new ImmutablePair<>(sm,
getShardsLocalResult.getStoreShards().stream().filter(s -> s.getShardMapId().equals(sm.getId())).findFirst().orElse(null)))
.collect(Collectors.toList());
for (Pair<StoreShardMap, StoreShard> shardInfo : shardInfos) {
RecoveryToken token = new RecoveryToken();
listOfTokens.add(token);
this.getStoreShardMaps().put(token, shardInfo);
this.getLocations().put(token, location);
this.getInconsistencies().put(token, new HashMap<>());
StoreShard ssLocal = shardInfo.getRight();
StoreShardMap ssmLocal = shardInfo.getLeft();
StoreShard dss = new StoreShard(ssLocal.getId(), ssLocal.getVersion(), ssLocal.getShardMapId(), ssLocal.getLocation(),
ssLocal.getStatus());
// First get all local mappings.
StoreResults lsmMappings;
try (IStoreOperationLocal op = this.getShardMapManager().getStoreOperationFactory().createGetMappingsByRangeLocalOperation(
this.getShardMapManager(), location, "DetectMappingDifferences", ssmLocal, dss, null, true)) {
lsmMappings = op.doLocal();
if (lsmMappings.getResult() == StoreResult.ShardMapDoesNotExist) {
// The shard needs to be re-attached. We are ignoring these errors in
// DetectMappingDifferences, since corruption is more profound than
// just inconsistent mappings.
// Alternatively, this shard belongs to a different shard map manager.
// Either way, we can't do anything about it here.
continue;
}
}
catch (IOException e) {
e.printStackTrace();
throw (ShardManagementException) e.getCause();
}
// Next build up a set of relevant global mappings.
// This is the union of those mappings that are associated with this local shard
// and those mappings which intersect with mappings found in the local shard.
// We will partition these mappings based on ranges.
Map<ShardRange, StoreMapping> relevantGsmMappings = new HashMap<>();
StoreResults gsmMappingsByMap;
try (IStoreOperationGlobal op = this.getShardMapManager().getStoreOperationFactory().createGetMappingsByRangeGlobalOperation(
this.getShardMapManager(), "DetectMappingDifferences", ssmLocal, dss, null, ShardManagementErrorCategory.Recovery, false, true)) {
gsmMappingsByMap = op.doGlobal();
}
catch (Exception e) {
e.printStackTrace();
throw (ShardManagementException) e.getCause();
}
if (gsmMappingsByMap.getResult() == StoreResult.ShardMapDoesNotExist) {
// The shard map is not properly attached to this GSM.
// This is beyond what we can handle resolving mappings.
continue;
}
for (StoreMapping gsmMapping : gsmMappingsByMap.getStoreMappings()) {
ShardKey min = ShardKey.fromRawValue(ssmLocal.getKeyType(), gsmMapping.getMinValue());
ShardKey max;
switch (ssmLocal.getMapType()) {
case Range:
max = ShardKey.fromRawValue(ssmLocal.getKeyType(), gsmMapping.getMaxValue());
break;
default:
assert ssmLocal.getMapType() == ShardMapType.List;
max = ShardKey.fromRawValue(ssmLocal.getKeyType(), gsmMapping.getMinValue()).getNextKey();
break;
}
ShardRange range = new ShardRange(min, max);
relevantGsmMappings.put(range, gsmMapping);
}
// Next, for each of the mappings in lsmMappings, we need to augment
// the gsmMappingsByMap by intersecting ranges.
for (StoreMapping lsmMapping : lsmMappings.getStoreMappings()) {
ShardKey min = ShardKey.fromRawValue(ssmLocal.getKeyType(), lsmMapping.getMinValue());
StoreResults gsmMappingsByRange;
if (ssmLocal.getMapType() == ShardMapType.Range) {
ShardKey max = ShardKey.fromRawValue(ssmLocal.getKeyType(), lsmMapping.getMaxValue());
ShardRange range = new ShardRange(min, max);
try (IStoreOperationGlobal op = this.getShardMapManager().getStoreOperationFactory().createGetMappingsByRangeGlobalOperation(
this.getShardMapManager(), "DetectMappingDifferences", ssmLocal, null, range, ShardManagementErrorCategory.Recovery,
false, true)) {
gsmMappingsByRange = op.doGlobal();
}
catch (Exception e) {
e.printStackTrace();
throw (ShardManagementException) e.getCause();
}
if (gsmMappingsByRange.getResult() == StoreResult.ShardMapDoesNotExist) {
// The shard was not properly attached.
// This is more than we can deal with in mapping resolution.
continue;
}
}
else {
assert ssmLocal.getMapType() == ShardMapType.List;
try (IStoreOperationGlobal op = this.getShardMapManager().getStoreOperationFactory().createFindMappingByKeyGlobalOperation(
this.getShardMapManager(), "DetectMappingDifferences", ssmLocal, min, CacheStoreMappingUpdatePolicy.OverwriteExisting,
ShardManagementErrorCategory.Recovery, false, true)) {
gsmMappingsByRange = op.doGlobal();
if (gsmMappingsByRange.getResult() == StoreResult.MappingNotFoundForKey
|| gsmMappingsByRange.getResult() == StoreResult.ShardMapDoesNotExist) {
// * No intersections being found is fine. Skip to the next mapping.
// * The shard was not properly attached.
// This is more than we can deal with in mapping resolution.
continue;
}
}
catch (Exception e) {
e.printStackTrace();
throw (ShardManagementException) e.getCause();
}
}
for (StoreMapping gsmMapping : gsmMappingsByRange.getStoreMappings()) {
ShardKey retrievedMin = ShardKey.fromRawValue(ssmLocal.getKeyType(), gsmMapping.getMinValue());
ShardRange retrievedRange;
switch (ssmLocal.getMapType()) {
case Range:
ShardKey retrievedMax = ShardKey.fromRawValue(ssmLocal.getKeyType(), gsmMapping.getMaxValue());
retrievedRange = new ShardRange(retrievedMin, retrievedMax);
break;
default:
assert ssmLocal.getMapType() == ShardMapType.List;
retrievedMax = ShardKey.fromRawValue(ssmLocal.getKeyType(), gsmMapping.getMinValue()).getNextKey();
retrievedRange = new ShardRange(retrievedMin, retrievedMax);
break;
}
relevantGsmMappings.put(retrievedRange, gsmMapping);
}
}
List<MappingComparisonResult> comparisonResults;
Map<ShardRange, MappingDifference> innerMap = new HashMap<>();
switch (ssmLocal.getMapType()) {
case Range:
comparisonResults = MappingComparisonUtils.compareRangeMappings(ssmLocal, new ArrayList<>(relevantGsmMappings.values()),
lsmMappings.getStoreMappings());
break;
default:
assert ssmLocal.getMapType() == ShardMapType.List;
comparisonResults = MappingComparisonUtils.comparePointMappings(ssmLocal, new ArrayList<>(relevantGsmMappings.values()),
lsmMappings.getStoreMappings());
break;
}
// Now we have 2 sets of mappings. Each sub mapping generated from this function is
// 1.) in the GSM only: report.
// 2.) in the LSM only: report.
// 3.) in both but with different version number: report.
// 4.) in both with the same version number: skip.
for (MappingComparisonResult r : comparisonResults) {
switch (r.getMappingLocation()) {
case MappingInShardMapOnly:
case MappingInShardOnly:
break;
default:
assert r.getMappingLocation() == MappingLocation.MappingInShardMapAndShard;
if (r.getShardMapManagerMapping().getId().equals(r.getShardMapping().getId())) {
// No conflict found, skip to the next range.
continue;
}
break;
}
// Store the inconsistency for later reporting.
innerMap.put(r.getRange(), new MappingDifference(MappingDifferenceType.Range, r.getMappingLocation(), r.getShardMap(),
r.getShardMapManagerMapping(), r.getShardMapping()));
}
this.getInconsistencies().get(token).putAll(innerMap);
}
return listOfTokens;
}