@Test public void testPut() throws IOException { tearDown(); this.store.put(this.jobName, this.datasetUrn, this.sequence); Assert.assertTrue(this.store.exists(this.jobName, this.datasetUrn)); try { this.store.put(this.jobName, this.datasetUrn, this.sequence); Assert.fail(); } catch (IOException e) { } }
@Override public void put(String jobName, String datasetUrn, CommitSequence commitSequence) throws IOException { datasetUrn = sanitizeDatasetUrn(datasetUrn); if (exists(jobName, datasetUrn)) { throw new IOException(String.format("CommitSequence already exists for job %s, dataset %s", jobName, datasetUrn)); } Path jobPath = new Path(this.rootPath, jobName); this.fs.mkdirs(jobPath); Path datasetPath = new Path(jobPath, datasetUrn); try (DataOutputStream dos = this.fs.create(datasetPath)) { dos.writeBytes(GSON.toJson(commitSequence)); } }
FsCommitSequenceStore implements CommitSequenceStore { @Override public void put(String jobName, String datasetUrn, CommitSequence commitSequence) throws IOException { datasetUrn = sanitizeDatasetUrn(datasetUrn); if (exists(jobName, datasetUrn)) { throw new IOException(String.format("CommitSequence already exists for job %s, dataset %s", jobName, datasetUrn)); } Path jobPath = new Path(this.rootPath, jobName); this.fs.mkdirs(jobPath); Path datasetPath = new Path(jobPath, datasetUrn); try (DataOutputStream dos = this.fs.create(datasetPath)) { dos.writeBytes(GSON.toJson(commitSequence)); } } }
FsCommitSequenceStore implements CommitSequenceStore { @Override public void put(String jobName, String datasetUrn, CommitSequence commitSequence) throws IOException { datasetUrn = sanitizeDatasetUrn(datasetUrn); if (exists(jobName, datasetUrn)) { throw new IOException(String.format("CommitSequence already exists for job %s, dataset %s", jobName, datasetUrn)); } Path jobPath = new Path(this.rootPath, jobName); this.fs.mkdirs(jobPath); Path datasetPath = new Path(jobPath, datasetUrn); try (DataOutputStream dos = this.fs.create(datasetPath)) { dos.writeBytes(GSON.toJson(commitSequence)); } } FsCommitSequenceStore(FileSystem fs, Path rootPath); }
FsCommitSequenceStore implements CommitSequenceStore { @Override public void put(String jobName, String datasetUrn, CommitSequence commitSequence) throws IOException { datasetUrn = sanitizeDatasetUrn(datasetUrn); if (exists(jobName, datasetUrn)) { throw new IOException(String.format("CommitSequence already exists for job %s, dataset %s", jobName, datasetUrn)); } Path jobPath = new Path(this.rootPath, jobName); this.fs.mkdirs(jobPath); Path datasetPath = new Path(jobPath, datasetUrn); try (DataOutputStream dos = this.fs.create(datasetPath)) { dos.writeBytes(GSON.toJson(commitSequence)); } } FsCommitSequenceStore(FileSystem fs, Path rootPath); @Override boolean exists(String jobName); @Override boolean exists(String jobName, String datasetUrn); @Override void delete(String jobName); @Override void delete(String jobName, String datasetUrn); @Override void put(String jobName, String datasetUrn, CommitSequence commitSequence); @Override Collection<String> get(String jobName); @Override Optional<CommitSequence> get(String jobName, String datasetUrn); }
FsCommitSequenceStore implements CommitSequenceStore { @Override public void put(String jobName, String datasetUrn, CommitSequence commitSequence) throws IOException { datasetUrn = sanitizeDatasetUrn(datasetUrn); if (exists(jobName, datasetUrn)) { throw new IOException(String.format("CommitSequence already exists for job %s, dataset %s", jobName, datasetUrn)); } Path jobPath = new Path(this.rootPath, jobName); this.fs.mkdirs(jobPath); Path datasetPath = new Path(jobPath, datasetUrn); try (DataOutputStream dos = this.fs.create(datasetPath)) { dos.writeBytes(GSON.toJson(commitSequence)); } } FsCommitSequenceStore(FileSystem fs, Path rootPath); @Override boolean exists(String jobName); @Override boolean exists(String jobName, String datasetUrn); @Override void delete(String jobName); @Override void delete(String jobName, String datasetUrn); @Override void put(String jobName, String datasetUrn, CommitSequence commitSequence); @Override Collection<String> get(String jobName); @Override Optional<CommitSequence> get(String jobName, String datasetUrn); static final String GOBBLIN_RUNTIME_COMMIT_SEQUENCE_STORE_FS_URI; static final String GOBBLIN_RUNTIME_COMMIT_SEQUENCE_STORE_DIR; }