@Test public void testGetHighWatermarkOnAppendExtract() { String endValue = "20140101000000"; SourceState sourceState = new SourceState(); sourceState.setProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY, true); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE, endValue); ExtractType extractType = ExtractType.APPEND_DAILY; TestPartitioner partitioner = new TestPartitioner(sourceState); Assert.assertEquals( partitioner.getHighWatermark(extractType, null), Long.parseLong(endValue), "High watermark should be " + endValue); Assert.assertEquals( partitioner.getUserSpecifiedHighWatermark(), true, "Should mark as user specified high watermark"); partitioner.reset(); sourceState.removeProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY); Assert.assertEquals( partitioner.getHighWatermark(ExtractType.APPEND_BATCH, null), ConfigurationKeys.DEFAULT_WATERMARK_VALUE, "High watermark should be " + ConfigurationKeys.DEFAULT_WATERMARK_VALUE); Assert.assertEquals( partitioner.getUserSpecifiedHighWatermark(), false, "Should not mark as user specified high watermark"); long expected = Long.parseLong(TestPartitioner.currentTimeString); Assert.assertEquals( partitioner.getHighWatermark(extractType, null), expected, "High watermark should be " + expected); Assert.assertEquals( partitioner.getUserSpecifiedHighWatermark(), false, "Should not mark as user specified high watermark"); String maxLimit = "CURRENTDATE-1"; sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_APPEND_MAX_WATERMARK_LIMIT, maxLimit); Assert.assertEquals( partitioner.getHighWatermark(extractType, null), 20161231235959L, "High watermark should be 20161231235959"); Assert.assertEquals( partitioner.getUserSpecifiedHighWatermark(), true, "Should not mark as user specified high watermark"); partitioner.reset(); maxLimit = "CURRENTHOUR-1"; sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_APPEND_MAX_WATERMARK_LIMIT, maxLimit); Assert.assertEquals( partitioner.getHighWatermark(extractType, null), 20161231235959L, "High watermark should be 20161231235959"); Assert.assertEquals( partitioner.getUserSpecifiedHighWatermark(), true, "Should not mark as user specified high watermark"); partitioner.reset(); maxLimit = "CURRENTMINUTE-1"; sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_APPEND_MAX_WATERMARK_LIMIT, maxLimit); Assert.assertEquals( partitioner.getHighWatermark(extractType, null), 20161231235959L, "High watermark should be 20161231235959"); Assert.assertEquals( partitioner.getUserSpecifiedHighWatermark(), true, "Should not mark as user specified high watermark"); partitioner.reset(); maxLimit = "CURRENTSECOND-1"; sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_APPEND_MAX_WATERMARK_LIMIT, maxLimit); Assert.assertEquals( partitioner.getHighWatermark(extractType, null), 20161231235959L, "High watermark should be 20161231235959"); Assert.assertEquals( partitioner.getUserSpecifiedHighWatermark(), true, "Should not mark as user specified high watermark"); }
@VisibleForTesting protected long getHighWatermark(ExtractType extractType, WatermarkType watermarkType) { LOG.debug("Getting high watermark"); String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE); long highWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isWatermarkOverride()) { highWatermark = this.state.getPropAsLong(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE, 0); if (highWatermark == 0) { highWatermark = Long.parseLong(Utils.dateTimeToString(getCurrentTime(timeZone), WATERMARKTIMEFORMAT, timeZone)); } else { hasUserSpecifiedHighWatermark = true; } LOG.info("Overriding high water mark with the given end value:" + highWatermark); } else { if (isSnapshot(extractType)) { highWatermark = this.getSnapshotHighWatermark(watermarkType); } else { highWatermark = this.getAppendHighWatermark(extractType); } } return (highWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : highWatermark); }
Partitioner { @VisibleForTesting protected long getHighWatermark(ExtractType extractType, WatermarkType watermarkType) { LOG.debug("Getting high watermark"); String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE); long highWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isWatermarkOverride()) { highWatermark = this.state.getPropAsLong(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE, 0); if (highWatermark == 0) { highWatermark = Long.parseLong(Utils.dateTimeToString(getCurrentTime(timeZone), WATERMARKTIMEFORMAT, timeZone)); } else { hasUserSpecifiedHighWatermark = true; } LOG.info("Overriding high water mark with the given end value:" + highWatermark); } else { if (isSnapshot(extractType)) { highWatermark = this.getSnapshotHighWatermark(watermarkType); } else { highWatermark = this.getAppendHighWatermark(extractType); } } return (highWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : highWatermark); } }
Partitioner { @VisibleForTesting protected long getHighWatermark(ExtractType extractType, WatermarkType watermarkType) { LOG.debug("Getting high watermark"); String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE); long highWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isWatermarkOverride()) { highWatermark = this.state.getPropAsLong(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE, 0); if (highWatermark == 0) { highWatermark = Long.parseLong(Utils.dateTimeToString(getCurrentTime(timeZone), WATERMARKTIMEFORMAT, timeZone)); } else { hasUserSpecifiedHighWatermark = true; } LOG.info("Overriding high water mark with the given end value:" + highWatermark); } else { if (isSnapshot(extractType)) { highWatermark = this.getSnapshotHighWatermark(watermarkType); } else { highWatermark = this.getAppendHighWatermark(extractType); } } return (highWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : highWatermark); } Partitioner(SourceState state); }
Partitioner { @VisibleForTesting protected long getHighWatermark(ExtractType extractType, WatermarkType watermarkType) { LOG.debug("Getting high watermark"); String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE); long highWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isWatermarkOverride()) { highWatermark = this.state.getPropAsLong(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE, 0); if (highWatermark == 0) { highWatermark = Long.parseLong(Utils.dateTimeToString(getCurrentTime(timeZone), WATERMARKTIMEFORMAT, timeZone)); } else { hasUserSpecifiedHighWatermark = true; } LOG.info("Overriding high water mark with the given end value:" + highWatermark); } else { if (isSnapshot(extractType)) { highWatermark = this.getSnapshotHighWatermark(watermarkType); } else { highWatermark = this.getAppendHighWatermark(extractType); } } return (highWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : highWatermark); } 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 getHighWatermark(ExtractType extractType, WatermarkType watermarkType) { LOG.debug("Getting high watermark"); String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE); long highWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; if (this.isWatermarkOverride()) { highWatermark = this.state.getPropAsLong(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE, 0); if (highWatermark == 0) { highWatermark = Long.parseLong(Utils.dateTimeToString(getCurrentTime(timeZone), WATERMARKTIMEFORMAT, timeZone)); } else { hasUserSpecifiedHighWatermark = true; } LOG.info("Overriding high water mark with the given end value:" + highWatermark); } else { if (isSnapshot(extractType)) { highWatermark = this.getSnapshotHighWatermark(watermarkType); } else { highWatermark = this.getAppendHighWatermark(extractType); } } return (highWatermark == 0 ? ConfigurationKeys.DEFAULT_WATERMARK_VALUE : highWatermark); } 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; }