### Question:
FileSystemStorage extends Configured implements Storage { @Override public void validateACL(AccessControlList acl) throws FalconException { try { for (Location location : getLocations()) { String pathString = getRelativePath(location); Path path = new Path(pathString); FileSystem fileSystem = HadoopClientFactory.get().createProxiedFileSystem(path.toUri(), getConf()); if (fileSystem.exists(path)) { FileStatus fileStatus = fileSystem.getFileStatus(path); Set<String> groups = CurrentUser.getGroupNames(); if (fileStatus.getOwner().equals(acl.getOwner()) || groups.contains(acl.getGroup())) { return; } LOG.error("Permission denied: Either Feed ACL owner {} or group {} doesn't " + "match the actual file owner {} or group {} for file {}", acl, acl.getGroup(), fileStatus.getOwner(), fileStatus.getGroup(), path); throw new FalconException("Permission denied: Either Feed ACL owner " + acl + " or group " + acl.getGroup() + " doesn't match the actual " + "file owner " + fileStatus.getOwner() + " or group " + fileStatus.getGroup() + " for file " + path); } } } catch (IOException e) { LOG.error("Can't validate ACL on storage {}", getStorageUrl(), e); throw new RuntimeException("Can't validate storage ACL (URI " + getStorageUrl() + ")", e); } } FileSystemStorage(Feed feed); protected  FileSystemStorage(String storageUrl, Locations locations); protected  FileSystemStorage(String storageUrl, List<Location> locations); protected  FileSystemStorage(String uriTemplate); @Override TYPE getType(); String getStorageUrl(); List<Location> getLocations(); @Override String getUriTemplate(); @Override String getUriTemplate(LocationType locationType); String getUriTemplate(LocationType locationType, List<Location> locationList); Path getWorkingDir(); @Override boolean isIdentical(Storage toCompareAgainst); static Location getLocation(List<Location> locations, LocationType type); @Override void validateACL(AccessControlList acl); @Override StringBuilder evict(String retentionLimit, String timeZone, Path logFilePath); @Override @SuppressWarnings("MagicConstant") List<FeedInstanceStatus> getListing(Feed feed, String clusterName, LocationType locationType,
                                               Date start, Date end); @Override FeedInstanceStatus.AvailabilityStatus getInstanceAvailabilityStatus(Feed feed, String clusterName,
                                                                   LocationType locationType,
                                                                   Date instanceTime); FileStatus getFileStatus(FileSystem fileSystem, Path feedInstancePath); Configuration getConf(); @Override String toString(); static final String FEED_PATH_SEP; static final String LOCATION_TYPE_SEP; static final String FILE_SYSTEM_URL; }### Answer:
@Test public void testValidateACL() throws Exception { final Location location = new Location(); Path path = new Path("/foo/bar"); location.setPath(path.toString()); location.setType(LocationType.DATA); List<Location> locations = new ArrayList<Location>(); locations.add(location); String user = System.getProperty("user.name"); EmbeddedCluster cluster = EmbeddedCluster.newCluster(user); FileSystem fs = cluster.getFileSystem(); fs.mkdirs(path); FileSystemStorage storage = new FileSystemStorage( cluster.getConf().get(HadoopClientFactory.FS_DEFAULT_NAME_KEY), locations); storage.validateACL(new TestACL(user, user, "0x755")); try { storage.validateACL(new TestACL("random", user, "0x755")); Assert.fail("Validation should have failed"); } catch(FalconException e) { } location.setPath("/foo/bar/${YEAR}/${MONTH}/${DAY}"); storage.validateACL(new TestACL(user, user, "rrr")); try { storage.validateACL(new TestACL("random", user, "0x755")); Assert.fail("Validation should have failed"); } catch(FalconException e) { } }