Make Watermark checking configurable in distcpNG-replication (#1997)

Author: autumnust

gobblin-data-management/src/main/java/gobblin/data/management/copy/replication/ConfigBasedDataset.java

@@ -70,11 +70,14 @@ public class ConfigBasedDataset implements CopyableDataset {
   private final CopyRoute copyRoute;
   private final ReplicationConfiguration rc;
   private String datasetURN;
+  private boolean watermarkEnabled;
 
   public ConfigBasedDataset(ReplicationConfiguration rc, Properties props, CopyRoute copyRoute) {
     this.props = props;
     this.copyRoute = copyRoute;
     this.rc = rc;
+    this.watermarkEnabled = Boolean.parseBoolean
+        (this.props.getProperty(ConfigBasedDatasetsFinder.WATERMARK_ENABLE, "true"));
     calculateDatasetURN();
   }
 
@@ -110,14 +113,16 @@ public Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs, Co
       return copyableFiles;
     }
 
-    if ((!copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent())
-        || (copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent()
-            && copyFromRaw.getWatermark().get().compareTo(copyToRaw.getWatermark().get()) <= 0)) {
-      log.info(
-          "No need to copy as destination watermark >= source watermark with source watermark {}, for dataset with metadata {}",
-          copyFromRaw.getWatermark().isPresent() ? copyFromRaw.getWatermark().get().toJson() : "N/A",
-          this.rc.getMetaData());
-      return copyableFiles;
+    if (this.watermarkEnabled) {
+      if ((!copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent()) || (
+          copyFromRaw.getWatermark().isPresent() && copyToRaw.getWatermark().isPresent()
+              && copyFromRaw.getWatermark().get().compareTo(copyToRaw.getWatermark().get()) <= 0)) {
+        log.info(
+            "No need to copy as destination watermark >= source watermark with source watermark {}, for dataset with metadata {}",
+            copyFromRaw.getWatermark().isPresent() ? copyFromRaw.getWatermark().get().toJson() : "N/A",
+            this.rc.getMetaData());
+        return copyableFiles;
+      }
     }
 
     HadoopFsEndPoint copyFrom = (HadoopFsEndPoint) copyFromRaw;
@@ -194,7 +199,7 @@ public boolean apply(FileStatus input) {
           deleteCommitStep, 0));
     }
 
-    // generate the watermark file
+    // generate the watermark file even if watermark checking is disabled. Make sure it can come into functional once disired.
     if ((!watermarkMetadataCopied) && copyFrom.getWatermark().isPresent()) {
       copyableFiles.add(new PostPublishStep(copyTo.getDatasetPath().toString(), Maps.<String, String> newHashMap(),
           new WatermarkMetadataGenerationCommitStep(copyTo.getFsURI().toString(), copyTo.getDatasetPath(),

gobblin-data-management/src/main/java/gobblin/data/management/copy/replication/ConfigBasedDatasetsFinder.java

@@ -95,6 +95,11 @@ public abstract class ConfigBasedDatasetsFinder implements DatasetsFinder {
   public static final String JOB_LEVEL_BLACKLIST = CopyConfiguration.COPY_PREFIX + ".configBased.blacklist" ;
   public static final String JOB_LEVEL_WHITELIST = CopyConfiguration.COPY_PREFIX + ".configBased.whitelist" ;
 
+  // There are some cases that WATERMARK checking is desired, like
+  // Unexpected data loss on target while not changing watermark accordingly.
+  // This configuration make WATERMARK checking configurable for operation convenience, default true
+  public static final String WATERMARK_ENABLE = CopyConfiguration.COPY_PREFIX + ".configBased.watermark.enabled" ;
+
 
   protected final String storeRoot;
   protected final Path commonRoot;