fix: Change the return type of Heartbeat::getEstimatedLowWatermark to long (#1631)

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://togithub.com/googleapis/java-bigtable/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the [samples format](
https://togithub.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).

Author: tengzhonger

google-cloud-bigtable/clirr-ignored-differences.xml

@@ -86,4 +86,11 @@
         <className>com/google/cloud/bigtable/gaxx/reframing/ReframingResponseObserver</className>
         <to>com/google/api/gax/rpc/StateCheckingResponseObserver</to>
     </difference>
+    <!-- change method return type is ok because Heartbeat is InternalApi -->
+    <difference>
+        <differenceType>7006</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/models/Heartbeat</className>
+        <method>*getEstimatedLowWatermark*</method>
+        <to>long</to>
+    </difference>
 </differences>

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java

@@ -18,7 +18,7 @@
 import com.google.api.core.InternalApi;
 import com.google.auto.value.AutoValue;
 import com.google.bigtable.v2.ReadChangeStreamResponse;
-import com.google.protobuf.Timestamp;
+import com.google.protobuf.util.Timestamps;
 import java.io.Serializable;
 import javax.annotation.Nonnull;
 
@@ -29,21 +29,20 @@ public abstract class Heartbeat implements ChangeStreamRecord, Serializable {
   private static final long serialVersionUID = 7316215828353608504L;
 
   private static Heartbeat create(
-      ChangeStreamContinuationToken changeStreamContinuationToken,
-      Timestamp estimatedLowWatermark) {
+      ChangeStreamContinuationToken changeStreamContinuationToken, long estimatedLowWatermark) {
     return new AutoValue_Heartbeat(changeStreamContinuationToken, estimatedLowWatermark);
   }
 
   /** Wraps the protobuf {@link ReadChangeStreamResponse.Heartbeat}. */
   static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) {
     return create(
         ChangeStreamContinuationToken.fromProto(heartbeat.getContinuationToken()),
-        heartbeat.getEstimatedLowWatermark());
+        Timestamps.toNanos(heartbeat.getEstimatedLowWatermark()));
   }
 
   @InternalApi("Intended for use by the BigtableIO in apache/beam only.")
   public abstract ChangeStreamContinuationToken getChangeStreamContinuationToken();
 
   @InternalApi("Intended for use by the BigtableIO in apache/beam only.")
-  public abstract Timestamp getEstimatedLowWatermark();
+  public abstract long getEstimatedLowWatermark();
 }

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java

@@ -24,6 +24,7 @@
 import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Timestamp;
+import com.google.protobuf.util.Timestamps;
 import com.google.rpc.Status;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -119,7 +120,8 @@ public void heartbeatTest() {
             .build();
     Heartbeat actualHeartbeat = Heartbeat.fromProto(heartbeatProto);
 
-    assertThat(actualHeartbeat.getEstimatedLowWatermark()).isEqualTo(lowWatermark);
+    assertThat(actualHeartbeat.getEstimatedLowWatermark())
+        .isEqualTo(Timestamps.toNanos(lowWatermark));
     assertThat(actualHeartbeat.getChangeStreamContinuationToken().getPartition())
         .isEqualTo(ByteStringRange.create(rowRange.getStartKeyClosed(), rowRange.getEndKeyOpen()));
     assertThat(actualHeartbeat.getChangeStreamContinuationToken().getToken()).isEqualTo(token);

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/changestream/ChangeStreamRecordMergingCallableTest.java

@@ -32,6 +32,7 @@
 import com.google.cloud.bigtable.gaxx.testing.FakeStreamingApi.ServerStreamingStashCallable;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Timestamp;
+import com.google.protobuf.util.Timestamps;
 import com.google.rpc.Status;
 import java.util.Collections;
 import java.util.List;
@@ -80,7 +81,7 @@ public void heartbeatTest() {
     assertThat(heartbeat.getChangeStreamContinuationToken().getToken())
         .isEqualTo(heartbeatProto.getContinuationToken().getToken());
     assertThat(heartbeat.getEstimatedLowWatermark())
-        .isEqualTo(heartbeatProto.getEstimatedLowWatermark());
+        .isEqualTo(Timestamps.toNanos(heartbeatProto.getEstimatedLowWatermark()));
   }
 
   @Test

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/changestream/ReadChangeStreamMergingAcceptanceTest.java

@@ -141,7 +141,8 @@ public void test() throws Exception {
                                   .build())
                           .setToken(heartbeat.getChangeStreamContinuationToken().getToken())
                           .build())
-                  .setEstimatedLowWatermark(heartbeat.getEstimatedLowWatermark())
+                  .setEstimatedLowWatermark(
+                      Timestamps.fromNanos(heartbeat.getEstimatedLowWatermark()))
                   .build();
           actualResults.add(
               ReadChangeStreamTest.Result.newBuilder()