HdfsFileSystemAdapter implements FileSystemAdapter { public List<FileMetadata> getAllFiles(String streamName) { List<FileMetadata> ret = new ArrayList<>(); try { Path streamPath = new Path(streamName); FileSystem fileSystem = streamPath.getFileSystem(new Configuration()); FileStatus[] fileStatuses = fileSystem.listStatus(streamPath); for (FileStatus fileStatus : fileStatuses) { if (!fileStatus.isDirectory()) { ret.add(new FileMetadata(fileStatus.getPath().toString(), fileStatus.getLen())); } else { ret.addAll(getAllFiles(fileStatus.getPath().toString())); } } } catch (IOException e) { LOG.error("Failed to get the list of files for " + streamName, e); throw new SamzaException(e); } return ret; }  List<FileMetadata> getAllFiles(String streamName);  }
@Test public void testGetAllFiles() throws Exception { URL url = this.getClass().getResource("/partitioner"); FileSystemAdapter adapter = new HdfsFileSystemAdapter(); List<FileSystemAdapter.FileMetadata> result = adapter.getAllFiles(url.getPath()); Assert.assertEquals(3, result.size()); }