Add change stream new record models and tests.

Include PartitionStartRecord, PartitionEndRecord, PartitionEventRecord,
MoveInEvent, MoveOutEvent. These will be used in the following PRs - the
returned change stream record will be parsed into these models, and
corresponding actions will be added and called later.

Author: changliiu

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

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.model;
+
+import java.io.Serializable;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
+
+/** Describes move-in of the key ranges into the change stream partition. */
+@SuppressWarnings("initialization.fields.uninitialized") // Avro requires the default constructor
+@DefaultCoder(AvroCoder.class)
+public class MoveInEvent implements Serializable {
+
+  private static final long serialVersionUID = 7954905180615424094L;
+
+  private String sourcePartitionToken;
+
+  /** Default constructor for serialization only. */
+  private MoveInEvent() {}
+
+  /**
+   * Constructs a MoveInEvent from the source partition token.
+   *
+   * @param sourcePartitionToken An unique partition identifier describing the source change stream
+   *     partition that recorded changes for the key range that is moving into this partition.
+   */
+  public MoveInEvent(String sourcePartitionToken) {
+    this.sourcePartitionToken = sourcePartitionToken;
+  }
+
+  /**
+   * The source change stream partition token that recorded changes for the key range that is moving
+   * into this partition.
+   *
+   * @return source partition token string.
+   */
+  public String getSourcePartitionToken() {
+    return sourcePartitionToken;
+  }
+
+  @Override
+  public boolean equals(@Nullable Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof MoveInEvent)) {
+      return false;
+    }
+    MoveInEvent moveInEvent = (MoveInEvent) o;
+    return Objects.equals(sourcePartitionToken, moveInEvent.sourcePartitionToken);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(sourcePartitionToken);
+  }
+
+  @Override
+  public String toString() {
+    return "MoveInEvent{" + "sourcePartitionToken=" + sourcePartitionToken + '}';
+  }
+}

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

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.model;
+
+import java.io.Serializable;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
+
+/** Describes move-out of the key ranges into the change stream partition. */
+@SuppressWarnings("initialization.fields.uninitialized") // Avro requires the default constructor
+@DefaultCoder(AvroCoder.class)
+public class MoveOutEvent implements Serializable {
+
+  private static final long serialVersionUID = 1961571231618408326L;
+
+  private String destinationPartitionToken;
+
+  /** Default constructor for serialization only. */
+  private MoveOutEvent() {}
+
+  /**
+   * Constructs a MoveOutEvent from the destination partition token.
+   *
+   * @param destinationPartitionToken An unique partition identifier describing the destination
+   *     change stream partition that recorded changes for the key range that is moving into this
+   *     partition.
+   */
+  public MoveOutEvent(String destinationPartitionToken) {
+    this.destinationPartitionToken = destinationPartitionToken;
+  }
+
+  /**
+   * The destination change stream partition token that recorded changes for the key range that is
+   * moving out this partition.
+   *
+   * @return destination partition token string.
+   */
+  public String getDestinationPartitionToken() {
+    return destinationPartitionToken;
+  }
+
+  @Override
+  public boolean equals(@Nullable Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof MoveOutEvent)) {
+      return false;
+    }
+    MoveOutEvent moveOutEvent = (MoveOutEvent) o;
+    return Objects.equals(destinationPartitionToken, moveOutEvent.destinationPartitionToken);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(destinationPartitionToken);
+  }
+
+  @Override
+  public String toString() {
+    return "MoveOutEvent{" + "destinationPartitionToken=" + destinationPartitionToken + '}';
+  }
+}

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

@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.model;
+
+import com.google.cloud.Timestamp;
+import java.util.Objects;
+import org.apache.avro.reflect.AvroEncode;
+import org.apache.avro.reflect.Nullable;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.encoder.TimestampEncoding;
+
+/**
+ * A partition end record serves as a notification that the client should stop reading the
+ * partition. No further records are expected to be retrieved on it.
+ */
+@SuppressWarnings("initialization.fields.uninitialized") // Avro requires the default constructor
+@DefaultCoder(AvroCoder.class)
+public class PartitionEndRecord implements ChangeStreamRecord {
+
+  private static final long serialVersionUID = 5406538761724655621L;
+
+  @AvroEncode(using = TimestampEncoding.class)
+  private Timestamp endTimestamp;
+
+  private String recordSequence;
+
+  @Nullable private ChangeStreamRecordMetadata metadata;
+
+  /** Default constructor for serialization only. */
+  private PartitionEndRecord() {}
+
+  /**
+   * Constructs the partition end record with the given timestamp, record sequence and metadata.
+   *
+   * @param endTimestamp end timestamp at which the change stream partition is terminated
+   * @param recordSequence the order within a partition and a transaction in which the record was
+   *     put to the stream
+   * @param metadata connector execution metadata for the given record
+   */
+  public PartitionEndRecord(
+      Timestamp endTimestamp, String recordSequence, ChangeStreamRecordMetadata metadata) {
+    this.endTimestamp = endTimestamp;
+    this.recordSequence = recordSequence;
+    this.metadata = metadata;
+  }
+
+  /**
+   * Indicates the timestamp for which the change stream partition is terminated.
+   *
+   * @return the timestamp for which the change stream partition is terminated
+   */
+  @Override
+  public Timestamp getRecordTimestamp() {
+    return getEndTimestamp();
+  }
+
+  /**
+   * The end timestamp at which the change stream partition is terminated.
+   *
+   * @return the timestamp for which the change stream partition is terminated
+   */
+  public Timestamp getEndTimestamp() {
+    return endTimestamp;
+  }
+
+  /**
+   * Indicates the order in which a record was put to the stream. Is unique and increasing within a
+   * partition. It is relative to the scope of partition, commit timestamp, and
+   * server_transaction_id. It is useful for readers downstream to dedup any duplicate records that
+   * were read/recorded.
+   *
+   * @return record sequence of the record
+   */
+  public String getRecordSequence() {
+    return recordSequence;
+  }
+
+  @Override
+  public boolean equals(@javax.annotation.Nullable Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof PartitionEndRecord)) {
+      return false;
+    }
+    PartitionEndRecord that = (PartitionEndRecord) o;
+    return Objects.equals(endTimestamp, that.endTimestamp)
+        && Objects.equals(recordSequence, that.recordSequence);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(endTimestamp, recordSequence);
+  }
+
+  @Override
+  public String toString() {
+    return "PartitionEndRecord{"
+        + "endTimestamp="
+        + endTimestamp
+        + ", recordSequence='"
+        + recordSequence
+        + '\''
+        + ", metadata="
+        + metadata
+        + '}';
+  }
+}