ZihanLi58 commented on a change in pull request #3071:
URL: https://github.com/apache/incubator-gobblin/pull/3071#discussion_r462690694
##########
File path:
gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/InputRecordCountHelper.java
##########
@@ -123,7 +125,8 @@ public void saveState (Path dir, State state) throws
IOException {
saveState(this.fs, dir, state);
}
- private static void saveState (FileSystem fs, Path dir, State state) throws
IOException {
+ @VisibleForTesting
Review comment:
It's because test class and InputRecordCountHelper are not in the same
package
##########
File path:
gobblin-compaction/src/main/java/org/apache/gobblin/compaction/verify/CompactionTimeRangeVerifier.java
##########
@@ -73,10 +75,28 @@ public Result verify (FileSystemDataset dataset) {
Period minTimeAgo = formatter.parsePeriod(minTimeAgoStr);
latest = compactionStartTime.minus(minTimeAgo);
+ // get latest last run start time, we want to limit the duration between
two compaction for the same dataset
+ if
(state.contains(TimeBasedSubDirDatasetsFinder.MIN_RECOMPACTION_DURATION)) {
+ String minDurationStrList =
this.state.getProp(TimeBasedSubDirDatasetsFinder.MIN_RECOMPACTION_DURATION);
+ String minDurationStr = getMachedLookbackTime(datasetName,
minDurationStrList,
TimeBasedSubDirDatasetsFinder.DEFAULT_MIN_RECOMPACTION_DURATION);
+ Period minDurationTime = formatter.parsePeriod(minDurationStr);
+ DateTime latestLastRunTime =
compactionStartTime.minus(minDurationTime);
+ InputRecordCountHelper helper = new InputRecordCountHelper(state);
+ State compactState = helper.loadState(new
Path(result.getDstAbsoluteDir()));
+ if (compactState.contains(CompactionSlaEventHelper.LAST_RUN_START_TIME)
+ &&
compactState.getPropAsLong(CompactionSlaEventHelper.LAST_RUN_START_TIME) >
latestLastRunTime.getMillis()) {
+ log.warn("Last compaction for {} is {}, not before {}",
dataset.datasetRoot(), new
DateTime(compactState.getPropAsLong(CompactionSlaEventHelper.LAST_RUN_START_TIME),
timeZone), latestLastRunTime);
+ return new Result(false, "Last compaction for " +
dataset.datasetRoot() + " is not before" + latestLastRunTime);
+ }
+
+ }
+
if (earliest.isBefore(folderTime) && latest.isAfter(folderTime)) {
log.debug("{} falls in the user defined time range",
dataset.datasetRoot());
return new Result(true, "");
}
+ } catch (RuntimeException e) {
Review comment:
It's due to the findBugsMain, seems we catch exception and not throw it
intentionally, I will change the rule to pass it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
