[email protected]
Let's continue with the previous article AclEnforcement class .

The code analysis

verifyParentWithTopoChildrenDeleteDead() Method

First, the parameters passed in by this method are ：
CuratorFramework zk ：zookeeper Client framework for
ACL superUserAcl： User permissions
String path： route
Map<String, Id> topoToZkCreds：id
boolean fixUp: If you want to repair, the incoming fixUp by true, If no, yes false

private static void verifyParentWithTopoChildrenDeleteDead(CuratorFramework zk, ACL superUserAcl, String path,
Map<String, Id> topoToZkCreds, boolean fixUp,int perms) throws Exception {

if (zk.checkExists().forPath(path) != null) {

verifyAclStrict(zk, Arrays.asList(superUserAcl), path, fixUp);
// It may be a bad node id Set 
Set<String> possiblyBadIds = new HashSet<>();
// Through one for Loop to replace the bad node id Join the group 
for (String topoId : zk.getChildren().forPath(path)) {

String childPath = path + ClusterUtils.ZK_SEPERATOR + topoId;
if (!topoToZkCreds.containsKey(topoId)) {

//topoToZkCreds Does not exist in the topoId Key value 
possiblyBadIds.add(topoId);
} else {

List<ACL> rwAcl = getTopoAcl(path, topoId, topoToZkCreds, superUserAcl, fixUp, perms);
verifyAclStrictRecursive(zk, rwAcl, childPath, fixUp);
}
}

1、 First zk.checkExists().forPath(path) Determine whether the node exists .
If the node exists , Call verifyAclStrict() Method , Verify the of this node acl jurisdiction .
2、 Through one for The loop will the... Of the bad node id Add to a collection ,topoToZkCreds.containsKey(topoId) If in topoToZkCreds Does not exist in the topoId Key value , Then put his id Join the collection . If it exists, get acl jurisdiction , And then through verifyAclStrictRecursive() Method validation acl Strictly recursive .

if (!possiblyBadIds.isEmpty()) {

//Lets reread the children in STORMS as the source of truth and see if a new one was created in the background
possiblyBadIds.removeAll(zk.getChildren().forPath(ClusterUtils.STORMS_SUBTREE));
for (String topoId : possiblyBadIds) {

//Now we know for sure that this is a bad id
String childPath = path + ClusterUtils.ZK_SEPERATOR + topoId;
zk.delete().deletingChildrenIfNeeded().forPath(childPath);
}
}
}
}

3、 If possiblyBadIds Collection is empty , Just look at one side of the child node again , See if a new... Is created in the background .
4、 Remove compliance zk.getChildren().forPath(ClusterUtils.STORMS_SUBTREE) Of the child nodes of id. adopt for Loop through the rest of the collection id, Then delete childPath The node of the path .

Various verification methods

private static void verifyParentWithTopoChildren(CuratorFramework zk, ACL superUserAcl, String path,
Map<String, Id> topoToZkCreds, boolean fixUp, int perms) throws Exception {

if (zk.checkExists().forPath(path) != null) {

verifyAclStrict(zk, Arrays.asList(superUserAcl), path, fixUp);
for (String topoId : zk.getChildren().forPath(path)) {

String childPath = path + ClusterUtils.ZK_SEPERATOR + topoId;
List<ACL> rwAcl = getTopoAcl(path, topoId, topoToZkCreds, superUserAcl, fixUp, perms);
verifyAclStrictRecursive(zk, rwAcl, childPath, fixUp);
}
}
}

Verify the parent node with child nodes .

private static void verifyParentWithReadOnlyTopoChildren(CuratorFramework zk, ACL superUserAcl, String path,
Map<String, Id> topoToZkCreds, boolean fixUp) throws Exception {

verifyParentWithTopoChildren(zk, superUserAcl, path, topoToZkCreds, fixUp, ZooDefs.Perms.READ);
}
private static void verifyParentWithReadWriteTopoChildren(CuratorFramework zk, ACL superUserAcl, String path,
Map<String, Id> topoToZkCreds, boolean fixUp) throws Exception {

verifyParentWithTopoChildren(zk, superUserAcl, path, topoToZkCreds, fixUp, ZooDefs.Perms.ALL);
}

Verify the parent-child points of read-only child nodes .
Verify the parent node of the child node with read and write .

private static void verifyAclStrictRecursive(CuratorFramework zk, List<ACL> strictAcl, String path, boolean fixUp) throws Exception {

verifyAclStrict(zk, strictAcl, path, fixUp);
for (String child : zk.getChildren().forPath(path)) {

String newPath = path + ClusterUtils.ZK_SEPERATOR + child;
verifyAclStrictRecursive(zk, strictAcl, newPath, fixUp);
}
}

Recursive check acl.

private static void verifyAclStrict(CuratorFramework zk, List<ACL> strictAcl, String path, boolean fixUp) throws Exception {

try {

List<ACL> foundAcl = zk.getACL().forPath(path);
if (!equivalent(foundAcl, strictAcl)) {

if (fixUp) {

LOG.warn("{} expected to have ACL {}, but has {}. Fixing...", path, strictAcl, foundAcl);
zk.setACL().withACL(strictAcl).forPath(path);
} else {

throw new IllegalStateException(path + " did not have the correct ACL found " + foundAcl + " expected " + strictAcl);
}
}
} catch (KeeperException.NoNodeException ne) {

LOG.debug("{} removed in the middle of checking it", ne);
}
}

check acl.

private static boolean equivalent(List<ACL> a, List<ACL> b) {

if (a.size() == b.size()) {

for (ACL acl : a) {

if (!b.contains(acl)) {

return false;
}
}
return true;
}
return false;
}

Whether two acl Whether the permission list is equal .

Come here AclEnforcement Class is over .