Only update watermark if the new value is larger than the existing value

sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/KafkaSourceConsumerFnTest.java

@@ -159,7 +159,7 @@ public void testKafkaOffsetHolderEquality() {
             null));
     tester.testEquals();
   }
-};
+}
 
 class CounterSourceConnector extends SourceConnector {
   public static class CounterSourceConnectorConfig extends AbstractConfig {

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

@@ -528,14 +528,22 @@ public Void updateToFinished(String partitionToken) {
     }
 
     /**
-     * Update the partition watermark to the given timestamp.
+     * Update the partition watermark to the given timestamp if the given timestamp is larger than
+     * the existing value.
      *
      * @param partitionToken the partition unique identifier
      * @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));
+      Timestamp currentWatermark = getWatermarkForPartition(partitionToken);
+      if (currentWatermark == null) {
+        LOG.debug("Partiton {} cannot be found.", partitionToken);
+        return null;
+      }
+      if (watermark.compareTo(currentWatermark) > 0) {
+        transaction.buffer(createUpdateMetadataWatermarkMutationFrom(partitionToken, watermark));
+      }
       return null;
     }
 
@@ -582,6 +590,52 @@ public Void updateWatermark(String partitionToken, Timestamp watermark) {
       }
     }
 
+    /**
+     * Fetches the watermark for a given partition token.
+     *
+     * @param partitionToken the partition unique identifier
+     * @return the watermark for the given token if it exists. Otherwise, it returns null.
+     */
+    public @Nullable Timestamp getWatermarkForPartition(String partitionToken) {
+      Struct row =
+   *             transaction.readRow(metadataTableName, Key.of(singerId), Collections.singleton(column));
+      final Statement statement;
+      if (this.dialect == Dialect.POSTGRESQL) {
+        statement =
+            Statement.newBuilder(
+                    "SELECT \""
+                        + COLUMN_WATERMARK
+                        + "\" FROM \""
+                        + metadataTableName
+                        + "\" WHERE \""
+                        + COLUMN_PARTITION_TOKEN
+                        + "\" = $1")
+                .bind("p1")
+                .to(partitionToken)
+                .build();
+      } else {
+        statement =
+            Statement.newBuilder(
+                    "SELECT "
+                        + COLUMN_WATERMARK
+                        + " FROM "
+                        + metadataTableName
+                        + " WHERE "
+                        + COLUMN_PARTITION_TOKEN
+                        + " = @partition")
+                .bind("partition")
+                .to(partitionToken)
+                .build();
+      }
+      try (ResultSet resultSet =
+          transaction.executeQuery(statement, Options.tag("query=getWatermarkForPartition"))) {
+        if (resultSet.next()) {
+          return resultSet.getTimestamp(COLUMN_WATERMARK);
+        }
+        return null;
+      }
+    }
+
     private Mutation createInsertMetadataMutationFrom(PartitionMetadata partitionMetadata) {
       return Mutation.newInsertBuilder(metadataTableName)
           .set(COLUMN_PARTITION_TOKEN)

test.cc

@@ -0,0 +1,2 @@
+#include "usr/local/google/home/xuechen/Documents/beam-master/test.h"
+Test