@Test public void testVerify() { ZonedDateTime time = ZonedDateTime.of(2019, 12, 1, 18, 16, 0, 0, zone); State state = new State(); state.setProp(CompactionSource.COMPACTION_INIT_TIME, time.toInstant().toEpochMilli()); state.setProp(CompactionAuditCountVerifier.COMPACTION_COMMPLETENESS_GRANULARITY, "DAY"); state.setProp(CompactionWatermarkChecker.TIME_FORMAT, "yyyy/MM/dd"); FileSystemDataset dataset1201 = new SimpleFileSystemDataset(new Path("/dataset/2019/12/01")); FileSystemDataset dataset1130 = new SimpleFileSystemDataset(new Path("/dataset/2019/11/30")); FileSystemDataset datasetDash = new SimpleFileSystemDataset(new Path("/dataset/datepartition=2019-11-30")); state.setProp(CompactionAuditCountVerifier.COMPACTION_COMMPLETENESS_ENABLED, false); doVerifyDataset(new State(state), dataset1201, null, null); doVerifyDataset(new State(state), dataset1130, "1575187199999", null); doVerifyDataset(new State(state), datasetDash, null, null); state.setProp(CompactionAuditCountVerifier.COMPACTION_COMMPLETENESS_ENABLED, true); doVerifyDataset(new State(state), dataset1201, null, null); doVerifyDataset(new State(state), dataset1130, "1575187199999", "1575187199999"); doVerifyDataset(new State(state), datasetDash, null, null); }
@Override public Result verify(FileSystemDataset dataset) { Result res = super.verify(dataset); if (!dataset.datasetRoot().toString().contains(precedingTimeDatasetPartitionName)) { return res; } this.state.setProp(COMPACTION_WATERMARK, watermarkTime); if (enabled && res.isSuccessful()) { log.info("Set dataset {} complete and compaction watermark {}", dataset.datasetRoot(), watermarkTime); this.state.setProp(COMPLETION_COMPACTION_WATERMARK, watermarkTime); } else { log.info("Set dataset {} compaction watermark {}", dataset.datasetRoot(), watermarkTime); } return res; }
CompactionWatermarkChecker extends CompactionAuditCountVerifier { @Override public Result verify(FileSystemDataset dataset) { Result res = super.verify(dataset); if (!dataset.datasetRoot().toString().contains(precedingTimeDatasetPartitionName)) { return res; } this.state.setProp(COMPACTION_WATERMARK, watermarkTime); if (enabled && res.isSuccessful()) { log.info("Set dataset {} complete and compaction watermark {}", dataset.datasetRoot(), watermarkTime); this.state.setProp(COMPLETION_COMPACTION_WATERMARK, watermarkTime); } else { log.info("Set dataset {} compaction watermark {}", dataset.datasetRoot(), watermarkTime); } return res; } }
CompactionWatermarkChecker extends CompactionAuditCountVerifier { @Override public Result verify(FileSystemDataset dataset) { Result res = super.verify(dataset); if (!dataset.datasetRoot().toString().contains(precedingTimeDatasetPartitionName)) { return res; } this.state.setProp(COMPACTION_WATERMARK, watermarkTime); if (enabled && res.isSuccessful()) { log.info("Set dataset {} complete and compaction watermark {}", dataset.datasetRoot(), watermarkTime); this.state.setProp(COMPLETION_COMPACTION_WATERMARK, watermarkTime); } else { log.info("Set dataset {} compaction watermark {}", dataset.datasetRoot(), watermarkTime); } return res; } CompactionWatermarkChecker(State state); }
CompactionWatermarkChecker extends CompactionAuditCountVerifier { @Override public Result verify(FileSystemDataset dataset) { Result res = super.verify(dataset); if (!dataset.datasetRoot().toString().contains(precedingTimeDatasetPartitionName)) { return res; } this.state.setProp(COMPACTION_WATERMARK, watermarkTime); if (enabled && res.isSuccessful()) { log.info("Set dataset {} complete and compaction watermark {}", dataset.datasetRoot(), watermarkTime); this.state.setProp(COMPLETION_COMPACTION_WATERMARK, watermarkTime); } else { log.info("Set dataset {} compaction watermark {}", dataset.datasetRoot(), watermarkTime); } return res; } CompactionWatermarkChecker(State state); @Override Result verify(FileSystemDataset dataset); }
CompactionWatermarkChecker extends CompactionAuditCountVerifier { @Override public Result verify(FileSystemDataset dataset) { Result res = super.verify(dataset); if (!dataset.datasetRoot().toString().contains(precedingTimeDatasetPartitionName)) { return res; } this.state.setProp(COMPACTION_WATERMARK, watermarkTime); if (enabled && res.isSuccessful()) { log.info("Set dataset {} complete and compaction watermark {}", dataset.datasetRoot(), watermarkTime); this.state.setProp(COMPLETION_COMPACTION_WATERMARK, watermarkTime); } else { log.info("Set dataset {} compaction watermark {}", dataset.datasetRoot(), watermarkTime); } return res; } CompactionWatermarkChecker(State state); @Override Result verify(FileSystemDataset dataset); static final String TIME_FORMAT; static final String COMPACTION_WATERMARK; static final String COMPLETION_COMPACTION_WATERMARK; }