Only update the partition watermark iff the partition watermark in metadata table is smaller than the given watermark.

Author: jiangzzhu

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dao/PartitionMetadataDao.java

@@ -528,14 +528,24 @@ public Void updateToFinished(String partitionToken) {
     }
 
     /**
-     * Update the partition watermark to the given timestamp.
+     * Update the partition watermark to the given timestamp iff the partition
+     * watermark in metadata table is smaller than the given watermark.
      *
      * @param partitionToken the partition unique identifier
-     * @param watermark the new partition watermark
+     * @param watermark      the new partition watermark
      * @return the commit timestamp of the read / write transaction
      */
     public Void updateWatermark(String partitionToken, Timestamp watermark) {
-      transaction.buffer(createUpdateMetadataWatermarkMutationFrom(partitionToken, watermark));
+      Struct row = transaction.readRow(metadataTableName, KeySet.singleKey(partitionToken),
+          Collections.singleton(COLUMN_WATERMARK));
+      if (row == null) {
+        LOG.error("[{}] Failed to read Watermark column", partitionToken);
+        return null;
+      }
+      Timestamp partitionWatermark = row.getTimestamp(COLUMN_WATERMARK);
+      if (partitionWatermark.compareTo(watermark) < 0) {
+        transaction.buffer(createUpdateMetadataWatermarkMutationFrom(partitionToken, watermark));
+      }
       return null;
     }