private ConfigData get(String path, Predicate<Path> fileFilter) {
        if (allowedPaths == null) {
            throw new IllegalStateException("The provider has not been configured yet.");
        }

        Map<String, String> map = emptyMap();

        if (path != null && !path.isEmpty()) {
            Path dir = allowedPaths.parseUntrustedPath(path);
            if (dir == null) {
                log.warn("The path {} is not allowed to be accessed", path);
                return new ConfigData(map);
            }

            if (!Files.isDirectory(dir)) {
                log.warn("The path {} is not a directory", path);
            } else {
                try (Stream<Path> stream = Files.list(dir)) {
                    map = stream
                        .filter(fileFilter)
                        .collect(Collectors.toMap(
                            p -> p.getFileName().toString(),
                            p -> read(p)));
                } catch (IOException e) {
                    log.error("Could not list directory {}", dir, e);
                    throw new ConfigException("Could not list directory " + dir);
                }
            }
        }
        return new ConfigData(map);
    }
Retrieves the data contained in the regular files named by {@code keys} in the directory given by {@code path}.
Non-regular files (such as directories) in the given directory are silently ignored.
@param path the directory where data files reside.
@param keys the keys whose values will be retrieved.
@return the configuration data.
clients/src/main/java/org/apache/kafka/common/config/provider/DirectoryConfigProvider.java