@Test public void testGetLowWatermarkOnUserOverride() { String startValue = "20140101000000"; SourceState sourceState = new SourceState(); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_IS_WATERMARK_OVERRIDE, true); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE, startValue); TestPartitioner partitioner = new TestPartitioner(sourceState); Assert.assertEquals( partitioner.getLowWatermark(null, null, -1, 0), Long.parseLong(startValue), "Low watermark should be " + startValue); sourceState.removeProp(ConfigurationKeys.SOURCE_QUERYBASED_IS_WATERMARK_OVERRIDE); sourceState.setProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, true); Assert.assertEquals( partitioner.getLowWatermark(null, null, -1, 0), Long.parseLong(startValue), "Low watermark should be " + startValue); sourceState.removeProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE); Assert.assertEquals( partitioner.getLowWatermark(null, null, -1, 0), ConfigurationKeys.DEFAULT_WATERMARK_VALUE, "Low watermark should be " + ConfigurationKeys.DEFAULT_WATERMARK_VALUE); }
@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; }