@Test public void testGetSpecRobustness() throws Exception { File specDir = Files.createTempDir(); Properties properties = new Properties(); properties.setProperty(FSSpecStore.SPECSTORE_FS_DIR_KEY, specDir.getAbsolutePath()); SpecSerDe serde = Mockito.mock(SpecSerDe.class); TestFsSpecStore fsSpecStore = new TestFsSpecStore(ConfigUtils.propertiesToConfig(properties), serde); File specFileFail = new File(specDir, "spec_fail"); Assert.assertTrue(specFileFail.createNewFile()); File specFile1 = new File(specDir, "spec0"); Assert.assertTrue(specFile1.createNewFile()); File specFile2 = new File(specDir, "spec1"); Assert.assertTrue(specFile2.createNewFile()); File specFile3 = new File(specDir, "serDeFail"); Assert.assertTrue(specFile3.createNewFile()); FileSystem fs = FileSystem.getLocal(new Configuration()); Assert.assertEquals(fs.getFileStatus(new Path(specFile3.getAbsolutePath())).getLen(), 0); Collection<Spec> specList = fsSpecStore.getSpecs(); Assert.assertEquals(specList.size(), 2); for (Spec spec: specList) { Assert.assertTrue(!spec.getDescription().contains("spec_fail")); Assert.assertTrue(!spec.getDescription().contains("serDeFail")); } }
private void getSpecs(Path directory, Collection<Spec> specs) throws Exception { FileStatus[] fileStatuses = fs.listStatus(directory); for (FileStatus fileStatus : fileStatuses) { if (fileStatus.isDirectory()) { getSpecs(fileStatus.getPath(), specs); } else { try { specs.add(readSpecFromFile(fileStatus.getPath())); } catch (Exception e) { log.warn(String.format("Path[%s] cannot be correctly deserialized as Spec", fileStatus.getPath()), e); } } } }
FSSpecStore extends InstrumentedSpecStore { private void getSpecs(Path directory, Collection<Spec> specs) throws Exception { FileStatus[] fileStatuses = fs.listStatus(directory); for (FileStatus fileStatus : fileStatuses) { if (fileStatus.isDirectory()) { getSpecs(fileStatus.getPath(), specs); } else { try { specs.add(readSpecFromFile(fileStatus.getPath())); } catch (Exception e) { log.warn(String.format("Path[%s] cannot be correctly deserialized as Spec", fileStatus.getPath()), e); } } } } }
FSSpecStore extends InstrumentedSpecStore { private void getSpecs(Path directory, Collection<Spec> specs) throws Exception { FileStatus[] fileStatuses = fs.listStatus(directory); for (FileStatus fileStatus : fileStatuses) { if (fileStatus.isDirectory()) { getSpecs(fileStatus.getPath(), specs); } else { try { specs.add(readSpecFromFile(fileStatus.getPath())); } catch (Exception e) { log.warn(String.format("Path[%s] cannot be correctly deserialized as Spec", fileStatus.getPath()), e); } } } } FSSpecStore(GobblinInstanceEnvironment env, SpecSerDe specSerDe);  FSSpecStore(Config sysConfig, SpecSerDe specSerDe);  FSSpecStore(GobblinInstanceEnvironment env, SpecSerDe specSerDe, Optional<Logger> log);  FSSpecStore(Config sysConfig, SpecSerDe specSerDe, Optional<Logger> log); }
FSSpecStore extends InstrumentedSpecStore { private void getSpecs(Path directory, Collection<Spec> specs) throws Exception { FileStatus[] fileStatuses = fs.listStatus(directory); for (FileStatus fileStatus : fileStatuses) { if (fileStatus.isDirectory()) { getSpecs(fileStatus.getPath(), specs); } else { try { specs.add(readSpecFromFile(fileStatus.getPath())); } catch (Exception e) { log.warn(String.format("Path[%s] cannot be correctly deserialized as Spec", fileStatus.getPath()), e); } } } } FSSpecStore(GobblinInstanceEnvironment env, SpecSerDe specSerDe);  FSSpecStore(Config sysConfig, SpecSerDe specSerDe);  FSSpecStore(GobblinInstanceEnvironment env, SpecSerDe specSerDe, Optional<Logger> log);  FSSpecStore(Config sysConfig, SpecSerDe specSerDe, Optional<Logger> log); static String getSpecGroup(Path specUri); static String getSpecName(Path specUri); @Override Collection<Spec> getAllVersionsOfSpec(URI specUri); @Override boolean existsImpl(URI specUri); @Override void addSpecImpl(Spec spec); @Override boolean deleteSpec(Spec spec); @Override boolean deleteSpecImpl(URI specUri); @Override boolean deleteSpec(URI specUri, String version); @Override Spec updateSpecImpl(Spec spec); @Override Spec getSpecImpl(URI specUri); @Override Spec getSpec(URI specUri, String version); @Override Collection<Spec> getSpecsImpl(); @Override Iterator<URI> getSpecURIsImpl(); @Override Iterator<URI> getSpecURIsWithTag(String tag); @Override Optional<URI> getSpecStoreURI(); }
FSSpecStore extends InstrumentedSpecStore { private void getSpecs(Path directory, Collection<Spec> specs) throws Exception { FileStatus[] fileStatuses = fs.listStatus(directory); for (FileStatus fileStatus : fileStatuses) { if (fileStatus.isDirectory()) { getSpecs(fileStatus.getPath(), specs); } else { try { specs.add(readSpecFromFile(fileStatus.getPath())); } catch (Exception e) { log.warn(String.format("Path[%s] cannot be correctly deserialized as Spec", fileStatus.getPath()), e); } } } } FSSpecStore(GobblinInstanceEnvironment env, SpecSerDe specSerDe);  FSSpecStore(Config sysConfig, SpecSerDe specSerDe);  FSSpecStore(GobblinInstanceEnvironment env, SpecSerDe specSerDe, Optional<Logger> log);  FSSpecStore(Config sysConfig, SpecSerDe specSerDe, Optional<Logger> log); static String getSpecGroup(Path specUri); static String getSpecName(Path specUri); @Override Collection<Spec> getAllVersionsOfSpec(URI specUri); @Override boolean existsImpl(URI specUri); @Override void addSpecImpl(Spec spec); @Override boolean deleteSpec(Spec spec); @Override boolean deleteSpecImpl(URI specUri); @Override boolean deleteSpec(URI specUri, String version); @Override Spec updateSpecImpl(Spec spec); @Override Spec getSpecImpl(URI specUri); @Override Spec getSpec(URI specUri, String version); @Override Collection<Spec> getSpecsImpl(); @Override Iterator<URI> getSpecURIsImpl(); @Override Iterator<URI> getSpecURIsWithTag(String tag); @Override Optional<URI> getSpecStoreURI(); static final String SPECSTORE_FS_DIR_KEY; }