@Test public void testCreateImportDir() throws Exception { Master master = EasyMock.createMock(Master.class); ServerContext context = EasyMock.createMock(ServerContext.class); VolumeManager volumeManager = EasyMock.createMock(VolumeManager.class); UniqueNameAllocator uniqueNameAllocator = EasyMock.createMock(UniqueNameAllocator.class); String[] expDirs = {"hdfs: "hdfs: String[] tableDirs = {"hdfs: "hdfs: Set<String> tableDirSet = Set.of(tableDirs); String dirName = "abcd"; EasyMock.expect(master.getContext()).andReturn(context); EasyMock.expect(master.getVolumeManager()).andReturn(volumeManager).times(3); EasyMock.expect(context.getUniqueNameAllocator()).andReturn(uniqueNameAllocator); EasyMock.expect(volumeManager.matchingFileSystem(EasyMock.eq(new Path(expDirs[0])), EasyMock.eq(tableDirSet))).andReturn(new Path(tableDirs[0])); EasyMock.expect(volumeManager.matchingFileSystem(EasyMock.eq(new Path(expDirs[1])), EasyMock.eq(tableDirSet))).andReturn(new Path(tableDirs[1])); EasyMock.expect(volumeManager.matchingFileSystem(EasyMock.eq(new Path(expDirs[2])), EasyMock.eq(tableDirSet))).andReturn(new Path(tableDirs[2])); EasyMock.expect(uniqueNameAllocator.getNextName()).andReturn(dirName).times(3); ImportedTableInfo ti = new ImportedTableInfo(); ti.tableId = TableId.of("5b"); ti.directories = ImportTable.parseExportDir(Set.of(expDirs)); assertEquals(3, ti.directories.size()); EasyMock.replay(master, context, volumeManager, uniqueNameAllocator); CreateImportDir ci = new CreateImportDir(ti); ci.create(tableDirSet, master); assertEquals(3, ti.directories.size()); for (ImportedTableInfo.DirectoryMapping dm : ti.directories) { assertNotNull(dm.exportDir); assertNotNull(dm.importDir); assertTrue(dm.importDir.contains(Constants.HDFS_TABLES_DIR)); assertMatchingFilesystem(dm.exportDir, dm.importDir); assertTrue( dm.importDir.contains(ti.tableId.canonical() + "/" + Constants.BULK_PREFIX + dirName)); } EasyMock.verify(master, context, volumeManager, uniqueNameAllocator); }
static List<ImportedTableInfo.DirectoryMapping> parseExportDir(Set<String> exportDirs) { if (exportDirs == null || exportDirs.isEmpty()) { return Collections.emptyList(); } return exportDirs.stream().filter(not(String::isEmpty)) .map(ImportedTableInfo.DirectoryMapping::new).collect(Collectors.toList()); }
ImportTable extends MasterRepo { static List<ImportedTableInfo.DirectoryMapping> parseExportDir(Set<String> exportDirs) { if (exportDirs == null || exportDirs.isEmpty()) { return Collections.emptyList(); } return exportDirs.stream().filter(not(String::isEmpty)) .map(ImportedTableInfo.DirectoryMapping::new).collect(Collectors.toList()); } }
ImportTable extends MasterRepo { static List<ImportedTableInfo.DirectoryMapping> parseExportDir(Set<String> exportDirs) { if (exportDirs == null || exportDirs.isEmpty()) { return Collections.emptyList(); } return exportDirs.stream().filter(not(String::isEmpty)) .map(ImportedTableInfo.DirectoryMapping::new).collect(Collectors.toList()); } ImportTable(String user, String tableName, Set<String> exportDirs,
      NamespaceId namespaceId); }
ImportTable extends MasterRepo { static List<ImportedTableInfo.DirectoryMapping> parseExportDir(Set<String> exportDirs) { if (exportDirs == null || exportDirs.isEmpty()) { return Collections.emptyList(); } return exportDirs.stream().filter(not(String::isEmpty)) .map(ImportedTableInfo.DirectoryMapping::new).collect(Collectors.toList()); } ImportTable(String user, String tableName, Set<String> exportDirs,
      NamespaceId namespaceId); @Override long isReady(long tid, Master environment); @Override Repo<Master> call(long tid, Master env); @SuppressFBWarnings(value = "OS_OPEN_STREAM", justification = "closing intermediate readers would close the ZipInputStream") void checkVersions(Master env); @Override void undo(long tid, Master env); }
ImportTable extends MasterRepo { static List<ImportedTableInfo.DirectoryMapping> parseExportDir(Set<String> exportDirs) { if (exportDirs == null || exportDirs.isEmpty()) { return Collections.emptyList(); } return exportDirs.stream().filter(not(String::isEmpty)) .map(ImportedTableInfo.DirectoryMapping::new).collect(Collectors.toList()); } ImportTable(String user, String tableName, Set<String> exportDirs,
      NamespaceId namespaceId); @Override long isReady(long tid, Master environment); @Override Repo<Master> call(long tid, Master env); @SuppressFBWarnings(value = "OS_OPEN_STREAM", justification = "closing intermediate readers would close the ZipInputStream") void checkVersions(Master env); @Override void undo(long tid, Master env);  }