@Test public void testGetLowWatermarkOnAppendExtract() { SourceState sourceState = new SourceState(); String startValue = "20140101000000"; sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE, startValue); TestPartitioner partitioner = new TestPartitioner(sourceState); ExtractType extractType = ExtractType.APPEND_DAILY; int delta = 1; Assert.assertEquals( partitioner.getLowWatermark(extractType, null, ConfigurationKeys.DEFAULT_WATERMARK_VALUE, delta), Long.parseLong(startValue), "Low watermark should be " + startValue); long previousWatermark = 20140101000050L; long expected = previousWatermark + delta; Assert.assertEquals( partitioner.getLowWatermark(extractType, WatermarkType.SIMPLE, previousWatermark, delta), expected, "Low watermark should be " + expected); Assert.assertEquals( partitioner.getLowWatermark(extractType, WatermarkType.TIMESTAMP, previousWatermark, delta), expected, "Low watermark should be " + expected); int backupSecs = 10; sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_LOW_WATERMARK_BACKUP_SECS, backupSecs); Assert.assertEquals( partitioner.getLowWatermark(extractType, WatermarkType.SIMPLE, previousWatermark, delta), expected, "Low watermark should be " + expected); Assert.assertEquals( partitioner.getLowWatermark(extractType, WatermarkType.TIMESTAMP, previousWatermark, delta), expected, "Low watermark should be " + expected); }
@VisibleForTesting protected long getLowWatermark(ExtractType extractType, WatermarkType watermarkType, long previousWatermark, int deltaForNextWatermark) { long lowWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isFullDump() || this.isWatermarkOverride()) { String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE, ConfigurationKeys.DEFAULT_SOURCE_TIMEZONE); lowWatermark = Utils.getLongWithCurrentDate(this.state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE), timeZone); LOG.info("Overriding low water mark with the given start value: " + lowWatermark); } else { if (isSnapshot(extractType)) { lowWatermark = this.getSnapshotLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } else { lowWatermark = this.getAppendLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } } return (lowWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : lowWatermark); }
Partitioner { @VisibleForTesting protected long getLowWatermark(ExtractType extractType, WatermarkType watermarkType, long previousWatermark, int deltaForNextWatermark) { long lowWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isFullDump() || this.isWatermarkOverride()) { String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE, ConfigurationKeys.DEFAULT_SOURCE_TIMEZONE); lowWatermark = Utils.getLongWithCurrentDate(this.state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE), timeZone); LOG.info("Overriding low water mark with the given start value: " + lowWatermark); } else { if (isSnapshot(extractType)) { lowWatermark = this.getSnapshotLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } else { lowWatermark = this.getAppendLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } } return (lowWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : lowWatermark); } }
Partitioner { @VisibleForTesting protected long getLowWatermark(ExtractType extractType, WatermarkType watermarkType, long previousWatermark, int deltaForNextWatermark) { long lowWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isFullDump() || this.isWatermarkOverride()) { String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE, ConfigurationKeys.DEFAULT_SOURCE_TIMEZONE); lowWatermark = Utils.getLongWithCurrentDate(this.state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE), timeZone); LOG.info("Overriding low water mark with the given start value: " + lowWatermark); } else { if (isSnapshot(extractType)) { lowWatermark = this.getSnapshotLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } else { lowWatermark = this.getAppendLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } } return (lowWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : lowWatermark); } Partitioner(SourceState state); }
Partitioner { @VisibleForTesting protected long getLowWatermark(ExtractType extractType, WatermarkType watermarkType, long previousWatermark, int deltaForNextWatermark) { long lowWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isFullDump() || this.isWatermarkOverride()) { String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE, ConfigurationKeys.DEFAULT_SOURCE_TIMEZONE); lowWatermark = Utils.getLongWithCurrentDate(this.state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE), timeZone); LOG.info("Overriding low water mark with the given start value: " + lowWatermark); } else { if (isSnapshot(extractType)) { lowWatermark = this.getSnapshotLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } else { lowWatermark = this.getAppendLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } } return (lowWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : lowWatermark); } Partitioner(SourceState state); Partition getGlobalPartition(long previousWatermark); @Deprecated HashMap<Long, Long> getPartitions(long previousWatermark); List<Partition> getPartitionList(long previousWatermark); boolean isFullDump(); boolean isWatermarkOverride(); @VisibleForTesting DateTime getCurrentTime(String timeZone); }
Partitioner { @VisibleForTesting protected long getLowWatermark(ExtractType extractType, WatermarkType watermarkType, long previousWatermark, int deltaForNextWatermark) { long lowWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isFullDump() || this.isWatermarkOverride()) { String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE, ConfigurationKeys.DEFAULT_SOURCE_TIMEZONE); lowWatermark = Utils.getLongWithCurrentDate(this.state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE), timeZone); LOG.info("Overriding low water mark with the given start value: " + lowWatermark); } else { if (isSnapshot(extractType)) { lowWatermark = this.getSnapshotLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } else { lowWatermark = this.getAppendLowWatermark(watermarkType, previousWatermark, deltaForNextWatermark); } } return (lowWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : lowWatermark); } Partitioner(SourceState state); Partition getGlobalPartition(long previousWatermark); @Deprecated HashMap<Long, Long> getPartitions(long previousWatermark); List<Partition> getPartitionList(long previousWatermark); boolean isFullDump(); boolean isWatermarkOverride(); @VisibleForTesting DateTime getCurrentTime(String timeZone); static final String WATERMARKTIMEFORMAT; static final String HAS_USER_SPECIFIED_PARTITIONS; static final String USER_SPECIFIED_PARTITIONS; static final String IS_EARLY_STOPPED; static final String ALLOW_EQUAL_WATERMARK_BOUNDARY; static final Comparator<Partition> ascendingComparator; }