From b562176cfa10d6163ea6d77a36ba5dc8a6311149 Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Tue, 26 Jul 2022 16:35:15 -0400 Subject: [PATCH 1/7] Add ChangeStreamMutation which is a ChangeStreamRecord A ChangeStreamMutation holds a list of mods, represented by List, where an Entry is one of DeleteFamily/DeleteCells/SetCell. --- .../data/v2/models/ChangeStreamMutation.java | 275 +++++++++++ .../bigtable/data/v2/models/DeleteCells.java | 81 ++++ .../bigtable/data/v2/models/DeleteFamily.java | 58 +++ .../cloud/bigtable/data/v2/models/Entry.java | 26 ++ .../bigtable/data/v2/models/SetCell.java | 92 ++++ .../v2/models/ChangeStreamMutationTest.java | 426 ++++++++++++++++++ .../v2/models/ChangeStreamRecordTest.java | 3 + .../bigtable/data/v2/models/EntryTest.java | 101 +++++ 8 files changed, 1062 insertions(+) create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Entry.java create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java create mode 100644 google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java create mode 100644 google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/EntryTest.java diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java new file mode 100644 index 00000000000..1e23a5b7f2e --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java @@ -0,0 +1,275 @@ +/* + * Copyright 2022 Google LLC + * + * Licensed 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 + * + * https://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 com.google.cloud.bigtable.data.v2.models; + +import com.google.api.core.InternalApi; +import com.google.bigtable.v2.ReadChangeStreamResponse.DataChange.Type; +import com.google.cloud.bigtable.data.v2.models.Range.TimestampRange; +import com.google.common.base.MoreObjects; +import com.google.common.base.Objects; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.ByteString; +import com.google.protobuf.Timestamp; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; +import javax.annotation.Nonnull; + +/** + * A ChangeStreamMutation represents a list of mods(represented by List<{@link Entry}>) targeted at + * a single row, which is concatenated by (TODO:ChangeStreamRecordMerger). + * + *

It's meant to be used in {@link com.google.cloud.bigtable.data.v2.models.ChangeStreamRecord}. + */ +public final class ChangeStreamMutation implements ChangeStreamRecord, Serializable { + private static final long serialVersionUID = 8419520253162024218L; + + private final ByteString rowKey; + + /** Possible values: USER/GARBAGE_COLLECTION. */ + private final Type type; + + /** This should only be set when type==USER. */ + private final String sourceClusterId; + + private final Timestamp commitTimestamp; + + private final int tieBreaker; + + private transient ImmutableList.Builder entries = ImmutableList.builder(); + + /** Token and lowWatermark are set when we finish building the logical mutation. */ + private String token; + + private Timestamp lowWatermark; + + private ChangeStreamMutation( + ByteString rowKey, + Type type, + String sourceClusterId, + Timestamp commitTimestamp, + int tieBreaker) { + this.rowKey = rowKey; + this.type = type; + this.sourceClusterId = sourceClusterId; + this.commitTimestamp = commitTimestamp; + this.tieBreaker = tieBreaker; + } + + /** Creates a new instance of a user initiated mutation. */ + static ChangeStreamMutation create( + @Nonnull ByteString rowKey, + @Nonnull Type type, + @Nonnull String sourceClusterId, + @Nonnull Timestamp commitTimestamp, + int tieBreaker) { + Preconditions.checkArgument( + type == Type.USER, + "ChangeStreamMutation with a specified source cluster id must be a user initiated mutation."); + return new ChangeStreamMutation(rowKey, type, sourceClusterId, commitTimestamp, tieBreaker); + } + + /** Creates a new instance of a GC mutation. */ + static ChangeStreamMutation create( + @Nonnull ByteString rowKey, + @Nonnull Type type, + @Nonnull Timestamp commitTimestamp, + int tieBreaker) { + Preconditions.checkArgument( + type == Type.GARBAGE_COLLECTION, + "ChangeStreamMutation without source cluster id must be a garbage collection mutation."); + return new ChangeStreamMutation(rowKey, type, null, commitTimestamp, tieBreaker); + } + + private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException { + input.defaultReadObject(); + + @SuppressWarnings("unchecked") + ImmutableList deserialized = (ImmutableList) input.readObject(); + this.entries = ImmutableList.builder().addAll(deserialized); + } + + private void writeObject(ObjectOutputStream output) throws IOException { + output.defaultWriteObject(); + output.writeObject(entries.build()); + } + + @Nonnull + public ByteString getRowKey() { + return this.rowKey; + } + + @Nonnull + public Type getType() { + return this.type; + } + + public String getSourceClusterId() { + return this.sourceClusterId; + } + + @Nonnull + public Timestamp getCommitTimestamp() { + return this.commitTimestamp; + } + + public int getTieBreaker() { + return this.tieBreaker; + } + + public ChangeStreamMutation setToken(@Nonnull String token) { + this.token = token; + return this; + } + + public String getToken() { + return this.token; + } + + public ChangeStreamMutation setLowWatermark(@Nonnull Timestamp lowWatermark) { + this.lowWatermark = lowWatermark; + return this; + } + + public Timestamp getLowWatermark() { + return this.lowWatermark; + } + + @Nonnull + public List getEntries() { + return this.entries.build(); + } + + ChangeStreamMutation setCell( + @Nonnull String familyName, + @Nonnull ByteString qualifier, + long timestamp, + @Nonnull ByteString value) { + this.entries.add(new SetCell(familyName, qualifier, timestamp, value)); + return this; + } + + ChangeStreamMutation deleteCells( + @Nonnull String familyName, + @Nonnull ByteString qualifier, + @Nonnull TimestampRange timestampRange) { + this.entries.add(new DeleteCells(familyName, qualifier, timestampRange)); + return this; + } + + ChangeStreamMutation deleteFamily(@Nonnull String familyName) { + this.entries.add(new DeleteFamily(familyName)); + return this; + } + + @InternalApi("Used in Changestream beam pipeline.") + public RowMutation toRowMutation(@Nonnull String tableId) { + Preconditions.checkArgument( + token != null && lowWatermark != null, + "ChangeStreamMutation must have a continuation token and low watermark."); + RowMutation rowMutation = RowMutation.create(tableId, rowKey); + for (Entry entry : this.entries.build()) { + if (entry instanceof DeleteFamily) { + rowMutation.deleteFamily(((DeleteFamily) entry).getFamilyName()); + } else if (entry instanceof DeleteCells) { + DeleteCells deleteCells = (DeleteCells) entry; + rowMutation.deleteCells( + deleteCells.getFamilyName(), + deleteCells.getQualifier(), + deleteCells.getTimestampRange()); + } else if (entry instanceof SetCell) { + SetCell setCell = (SetCell) entry; + rowMutation.setCell( + setCell.getFamilyName(), + setCell.getQualifier(), + setCell.getTimestamp(), + setCell.getValue()); + } else { + throw new IllegalArgumentException("Unexpected Entry type."); + } + } + return rowMutation; + } + + @InternalApi("Used in Changestream beam pipeline.") + public RowMutationEntry toRowMutationEntry() { + Preconditions.checkArgument( + token != null && lowWatermark != null, + "ChangeStreamMutation must have a continuation token and low watermark."); + RowMutationEntry rowMutationEntry = RowMutationEntry.create(rowKey); + for (Entry entry : this.entries.build()) { + if (entry instanceof DeleteFamily) { + rowMutationEntry.deleteFamily(((DeleteFamily) entry).getFamilyName()); + } else if (entry instanceof DeleteCells) { + DeleteCells deleteCells = (DeleteCells) entry; + rowMutationEntry.deleteCells( + deleteCells.getFamilyName(), + deleteCells.getQualifier(), + deleteCells.getTimestampRange()); + } else if (entry instanceof SetCell) { + SetCell setCell = (SetCell) entry; + rowMutationEntry.setCell( + setCell.getFamilyName(), + setCell.getQualifier(), + setCell.getTimestamp(), + setCell.getValue()); + } else { + throw new IllegalArgumentException("Unexpected Entry type."); + } + } + return rowMutationEntry; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ChangeStreamMutation otherChangeStreamMutation = (ChangeStreamMutation) o; + return Objects.equal(this.toString(), otherChangeStreamMutation.toString()); + } + + @Override + public int hashCode() { + return Objects.hashCode( + rowKey, type, sourceClusterId, commitTimestamp, tieBreaker, token, lowWatermark, entries); + } + + @Override + public String toString() { + List entriesAsStrings = new ArrayList<>(); + for (Entry entry : this.entries.build()) { + entriesAsStrings.add(entry.toString()); + } + String entryString = "[" + String.join(";\t", entriesAsStrings) + "]"; + return MoreObjects.toStringHelper(this) + .add("rowKey", this.rowKey.toStringUtf8()) + .add("type", this.type) + .add("sourceClusterId", this.sourceClusterId) + .add("commitTimestamp", this.commitTimestamp.toString()) + .add("token", this.token) + .add("lowWatermark", this.lowWatermark) + .add("entries", entryString) + .toString(); + } +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java new file mode 100644 index 00000000000..9928e684b5b --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java @@ -0,0 +1,81 @@ +/* + * Copyright 2022 Google LLC + * + * Licensed 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 + * + * https://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 com.google.cloud.bigtable.data.v2.models; + +import com.google.cloud.bigtable.data.v2.models.Range.TimestampRange; +import com.google.common.base.MoreObjects; +import com.google.common.base.Objects; +import com.google.protobuf.ByteString; +import java.io.Serializable; +import javax.annotation.Nonnull; + +/** Representation of a DeleteCells mod in a data change. */ +public final class DeleteCells implements Entry, Serializable { + private static final long serialVersionUID = 851772158721462017L; + + private final String familyName; + private final ByteString qualifier; + private final TimestampRange timestampRange; + + DeleteCells( + @Nonnull String familyName, + @Nonnull ByteString qualifier, + @Nonnull TimestampRange timestampRange) { + this.familyName = familyName; + this.qualifier = qualifier; + this.timestampRange = timestampRange; + } + + public String getFamilyName() { + return this.familyName; + } + + public ByteString getQualifier() { + return this.qualifier; + } + + public TimestampRange getTimestampRange() { + return this.timestampRange; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + DeleteCells otherDeleteCell = (DeleteCells) o; + return Objects.equal(familyName, otherDeleteCell.getFamilyName()) + && Objects.equal(qualifier, otherDeleteCell.getQualifier()) + && Objects.equal(timestampRange, otherDeleteCell.getTimestampRange()); + } + + @Override + public int hashCode() { + return Objects.hashCode(familyName); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("familyName", familyName) + .add("qualifier", qualifier.toStringUtf8()) + .add("timestampRange", timestampRange) + .toString(); + } +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java new file mode 100644 index 00000000000..1d22321055e --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java @@ -0,0 +1,58 @@ +/* + * Copyright 2022 Google LLC + * + * Licensed 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 + * + * https://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 com.google.cloud.bigtable.data.v2.models; + +import com.google.common.base.MoreObjects; +import com.google.common.base.Objects; +import java.io.Serializable; +import javax.annotation.Nonnull; + +/** Representation of a DeleteFamily mod in a data change. */ +public final class DeleteFamily implements Entry, Serializable { + private static final long serialVersionUID = 81806775917145615L; + + private final String familyName; + + DeleteFamily(@Nonnull String familyName) { + this.familyName = familyName; + } + + public String getFamilyName() { + return this.familyName; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + DeleteFamily otherDeleteFamily = (DeleteFamily) o; + return Objects.equal(familyName, otherDeleteFamily.getFamilyName()); + } + + @Override + public int hashCode() { + return Objects.hashCode(familyName); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this).add("familyName", familyName).toString(); + } +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Entry.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Entry.java new file mode 100644 index 00000000000..c5c30016f40 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Entry.java @@ -0,0 +1,26 @@ +/* + * Copyright 2022 Google LLC + * + * Licensed 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 + * + * https://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 com.google.cloud.bigtable.data.v2.models; + +import com.google.api.core.InternalExtensionOnly; + +/** + * Default representation of a mod in a data change, which can be a {@link DeleteFamily}, a {@link + * DeleteCells}, or a {@link SetCell} This class will be used by {@link ChangeStreamMutation} to + * represent a list of mods in a logical change stream mutation. + */ +@InternalExtensionOnly +public interface Entry {} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java new file mode 100644 index 00000000000..29312ea6841 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java @@ -0,0 +1,92 @@ +/* + * Copyright 2022 Google LLC + * + * Licensed 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 + * + * https://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 com.google.cloud.bigtable.data.v2.models; + +import com.google.common.base.MoreObjects; +import com.google.common.base.Objects; +import com.google.protobuf.ByteString; +import java.io.Serializable; +import javax.annotation.Nonnull; + +/** + * Representation of a SetCell mod in a data change, whose value is concatenated by + * (TODO:ChangeStreamRecordMerger) in case of SetCell value chunking. + */ +public final class SetCell implements Entry, Serializable { + private static final long serialVersionUID = 77123872266724154L; + + private final String familyName; + private final ByteString qualifier; + private final long timestamp; + private final ByteString value; + + SetCell( + @Nonnull String familyName, + @Nonnull ByteString qualifier, + long timestamp, + @Nonnull ByteString value) { + this.familyName = familyName; + this.qualifier = qualifier; + this.timestamp = timestamp; + this.value = value; + } + + public String getFamilyName() { + return this.familyName; + } + + public ByteString getQualifier() { + return this.qualifier; + } + + public long getTimestamp() { + return this.timestamp; + } + + public ByteString getValue() { + return this.value; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + SetCell otherSetCell = (SetCell) o; + return Objects.equal(familyName, otherSetCell.getFamilyName()) + && Objects.equal(qualifier, otherSetCell.getQualifier()) + && Objects.equal(timestamp, otherSetCell.getTimestamp()) + && Objects.equal(value, otherSetCell.getValue()); + } + + @Override + public int hashCode() { + return Objects.hashCode(familyName); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("familyName", familyName) + .add("qualifier", qualifier.toStringUtf8()) + .add("timestamp", timestamp) + .add("value", value.toStringUtf8()) + .toString(); + } +} diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java new file mode 100644 index 00000000000..c4b582d1594 --- /dev/null +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java @@ -0,0 +1,426 @@ +/* + * Copyright 2022 Google LLC + * + * Licensed 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 + * + * https://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 com.google.cloud.bigtable.data.v2.models; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.bigtable.v2.MutateRowRequest; +import com.google.bigtable.v2.MutateRowsRequest; +import com.google.bigtable.v2.ReadChangeStreamResponse; +import com.google.cloud.bigtable.data.v2.internal.NameUtil; +import com.google.cloud.bigtable.data.v2.internal.RequestContext; +import com.google.common.primitives.Longs; +import com.google.protobuf.ByteString; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class ChangeStreamMutationTest { + private static final String PROJECT_ID = "fake-project"; + private static final String INSTANCE_ID = "fake-instance"; + private static final String TABLE_ID = "fake-table"; + private static final String APP_PROFILE_ID = "fake-profile"; + private static final RequestContext REQUEST_CONTEXT = + RequestContext.create(PROJECT_ID, INSTANCE_ID, APP_PROFILE_ID); + + @Rule public ExpectedException expect = ExpectedException.none(); + + @Test + public void userInitiatedMutationTest() throws IOException, ClassNotFoundException { + // Create a user initiated logical mutation. + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")) + .deleteFamily("fake-family") + .deleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark); + + // Test the getters. + Assert.assertEquals(changeStreamMutation.getRowKey(), ByteString.copyFromUtf8("key")); + Assert.assertEquals( + changeStreamMutation.getType(), ReadChangeStreamResponse.DataChange.Type.USER); + Assert.assertEquals(changeStreamMutation.getSourceClusterId(), "fake-source-cluster-id"); + Assert.assertEquals(changeStreamMutation.getCommitTimestamp(), fakeCommitTimestamp); + Assert.assertEquals(changeStreamMutation.getTieBreaker(), 0); + Assert.assertEquals(changeStreamMutation.getToken(), "fake-token"); + Assert.assertEquals(changeStreamMutation.getLowWatermark(), fakeLowWatermark); + + // Test serialization. + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(bos); + oos.writeObject(changeStreamMutation); + oos.close(); + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray())); + ChangeStreamMutation actual = (ChangeStreamMutation) ois.readObject(); + assertThat(actual.toString()).isEqualTo(changeStreamMutation.toString()); + } + + @Test + public void gcMutationTest() throws IOException, ClassNotFoundException { + // Create a GC mutation. + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.GARBAGE_COLLECTION, + fakeCommitTimestamp, + 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")) + .deleteFamily("fake-family") + .deleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark); + + // Test the getters. + Assert.assertEquals(changeStreamMutation.getRowKey(), ByteString.copyFromUtf8("key")); + Assert.assertEquals( + changeStreamMutation.getType(), + ReadChangeStreamResponse.DataChange.Type.GARBAGE_COLLECTION); + Assert.assertNull(changeStreamMutation.getSourceClusterId()); + Assert.assertEquals(changeStreamMutation.getCommitTimestamp(), fakeCommitTimestamp); + Assert.assertEquals(changeStreamMutation.getTieBreaker(), 0); + Assert.assertEquals(changeStreamMutation.getToken(), "fake-token"); + Assert.assertEquals(changeStreamMutation.getLowWatermark(), fakeLowWatermark); + + // Test serialization. + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(bos); + oos.writeObject(changeStreamMutation); + oos.close(); + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray())); + ChangeStreamMutation actual = (ChangeStreamMutation) ois.readObject(); + assertThat(actual.toString()).isEqualTo(changeStreamMutation.toString()); + } + + @Test(expected = IllegalArgumentException.class) + public void userInitiatedMutationHasSourceClusterIdTest() { + // Create a user initiated logical mutation. + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + fakeCommitTimestamp, + 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")) + .deleteFamily("fake-family") + .deleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark); + expect.expect(IllegalArgumentException.class); + } + + @Test(expected = IllegalArgumentException.class) + public void gcMutationHasNoSourceClusterIdTest() { + // Create a GC mutation. + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.GARBAGE_COLLECTION, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")) + .deleteFamily("fake-family") + .deleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark); + expect.expect(IllegalArgumentException.class); + } + + @Test(expected = IllegalArgumentException.class) + public void invalidTypeTest() { + // Create a ChangeStreamMutation with CONTINUATION type. + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.CONTINUATION, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")) + .deleteFamily("fake-family") + .deleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark); + expect.expect(IllegalArgumentException.class); + } + + @Test + public void toRowMutationTest() { + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")) + .deleteFamily("fake-family") + .deleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark); + + // Convert it to a rowMutation and construct a MutateRowRequest. + RowMutation rowMutation = changeStreamMutation.toRowMutation(TABLE_ID); + MutateRowRequest mutateRowRequest = rowMutation.toProto(REQUEST_CONTEXT); + String tableName = + NameUtil.formatTableName( + REQUEST_CONTEXT.getProjectId(), REQUEST_CONTEXT.getInstanceId(), TABLE_ID); + assertThat(mutateRowRequest.getTableName()).isEqualTo(tableName); + assertThat(mutateRowRequest.getMutationsList()).hasSize(3); + assertThat(mutateRowRequest.getMutations(0).getSetCell().getValue()) + .isEqualTo(ByteString.copyFromUtf8("fake-value")); + assertThat(mutateRowRequest.getMutations(1).getDeleteFromFamily().getFamilyName()) + .isEqualTo("fake-family"); + assertThat(mutateRowRequest.getMutations(2).getDeleteFromColumn().getFamilyName()) + .isEqualTo("fake-family"); + assertThat(mutateRowRequest.getMutations(2).getDeleteFromColumn().getColumnQualifier()) + .isEqualTo(ByteString.copyFromUtf8("fake-qualifier")); + } + + @Test(expected = IllegalArgumentException.class) + public void toRowMutationWithoutTokenShouldFailTest() { + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .deleteFamily("fake-family") + .setLowWatermark(fakeLowWatermark); + + // Convert it to a rowMutation. + RowMutation rowMutation = changeStreamMutation.toRowMutation(TABLE_ID); + expect.expect(IllegalArgumentException.class); + } + + @Test(expected = IllegalArgumentException.class) + public void toRowMutationWithoutLowWatermarkShouldFailTest() { + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .deleteFamily("fake-family") + .setToken("fake-token"); + + // Convert it to a rowMutation. + RowMutation rowMutation = changeStreamMutation.toRowMutation(TABLE_ID); + expect.expect(IllegalArgumentException.class); + } + + @Test + public void toRowMutationEntryTest() { + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")) + .deleteFamily("fake-family") + .deleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark); + + // Convert it to a rowMutationEntry and construct a MutateRowRequest. + RowMutationEntry rowMutationEntry = changeStreamMutation.toRowMutationEntry(); + MutateRowsRequest.Entry mutateRowsRequestEntry = rowMutationEntry.toProto(); + assertThat(mutateRowsRequestEntry.getRowKey()).isEqualTo(ByteString.copyFromUtf8("key")); + assertThat(mutateRowsRequestEntry.getMutationsList()).hasSize(3); + assertThat(mutateRowsRequestEntry.getMutations(0).getSetCell().getValue()) + .isEqualTo(ByteString.copyFromUtf8("fake-value")); + assertThat(mutateRowsRequestEntry.getMutations(1).getDeleteFromFamily().getFamilyName()) + .isEqualTo("fake-family"); + assertThat(mutateRowsRequestEntry.getMutations(2).getDeleteFromColumn().getFamilyName()) + .isEqualTo("fake-family"); + assertThat(mutateRowsRequestEntry.getMutations(2).getDeleteFromColumn().getColumnQualifier()) + .isEqualTo(ByteString.copyFromUtf8("fake-qualifier")); + } + + @Test(expected = IllegalArgumentException.class) + public void toRowMutationEntryWithoutTokenShouldFailTest() { + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .deleteFamily("fake-family") + .setLowWatermark(fakeLowWatermark); + + // Convert it to a rowMutationEntry and construct a MutateRowRequest. + RowMutationEntry rowMutationEntry = changeStreamMutation.toRowMutationEntry(); + expect.expect(IllegalArgumentException.class); + } + + @Test(expected = IllegalArgumentException.class) + public void toRowMutationEntryWithoutLowWatermarkShouldFailTest() { + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .deleteFamily("fake-family") + .setToken("fake-token"); + + // Convert it to a rowMutationEntry and construct a MutateRowRequest. + RowMutationEntry rowMutationEntry = changeStreamMutation.toRowMutationEntry(); + expect.expect(IllegalArgumentException.class); + } + + @Test + public void testWithLongValue() { + com.google.protobuf.Timestamp fakeCommitTimestamp = + com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + com.google.protobuf.Timestamp fakeLowWatermark = + com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.create( + ByteString.copyFromUtf8("key"), + ReadChangeStreamResponse.DataChange.Type.USER, + "fake-source-cluster-id", + fakeCommitTimestamp, + 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000L, + ByteString.copyFrom(Longs.toByteArray(1L))) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark); + + RowMutation rowMutation = changeStreamMutation.toRowMutation(TABLE_ID); + MutateRowRequest mutateRowRequest = rowMutation.toProto(REQUEST_CONTEXT); + String tableName = + NameUtil.formatTableName( + REQUEST_CONTEXT.getProjectId(), REQUEST_CONTEXT.getInstanceId(), TABLE_ID); + assertThat(mutateRowRequest.getTableName()).isEqualTo(tableName); + assertThat(mutateRowRequest.getMutationsList()).hasSize(1); + assertThat(mutateRowRequest.getMutations(0).getSetCell().getValue()) + .isEqualTo(ByteString.copyFromUtf8("\000\000\000\000\000\000\000\001")); + } +} diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java index c82aae73308..8c00d7ae726 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java @@ -97,6 +97,9 @@ public void closeStreamSerializationTest() throws IOException, ClassNotFoundExce assertThat(actual.getStatus()).isEqualTo(closeStream.getStatus()); } + @Test + public void changeStreamMutationSerializationTest() {} + @Test public void heartbeatTest() { Timestamp lowWatermark = Timestamp.newBuilder().setSeconds(1000).build(); diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/EntryTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/EntryTest.java new file mode 100644 index 00000000000..1e581ce1c5f --- /dev/null +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/EntryTest.java @@ -0,0 +1,101 @@ +/* + * Copyright 2022 Google LLC + * + * Licensed 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 + * + * https://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 com.google.cloud.bigtable.data.v2.models; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.protobuf.ByteString; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class EntryTest { + private void validateSerializationRoundTrip(Object obj) + throws IOException, ClassNotFoundException { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(bos); + oos.writeObject(obj); + oos.close(); + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray())); + assertThat(ois.readObject()).isEqualTo(obj); + } + + @Test + public void serializationTest() throws IOException, ClassNotFoundException { + // DeleteFamily + Entry deleteFamilyEntry = new DeleteFamily("fake-family"); + validateSerializationRoundTrip(deleteFamilyEntry); + + // DeleteCell + Entry deleteCellsEntry = + new DeleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)); + validateSerializationRoundTrip(deleteCellsEntry); + + // SetCell + Entry setCellEntry = + new SetCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")); + validateSerializationRoundTrip(setCellEntry); + } + + @Test + public void deleteFamilyTest() { + Entry deleteFamilyEntry = new DeleteFamily("fake-family"); + DeleteFamily deleteFamily = (DeleteFamily) deleteFamilyEntry; + Assert.assertEquals("fake-family", deleteFamily.getFamilyName()); + } + + @Test + public void deleteCellsTest() { + Entry deleteCellEntry = + new DeleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)); + DeleteCells deleteCells = (DeleteCells) deleteCellEntry; + Assert.assertEquals("fake-family", deleteCells.getFamilyName()); + Assert.assertEquals(ByteString.copyFromUtf8("fake-qualifier"), deleteCells.getQualifier()); + Assert.assertEquals(Range.TimestampRange.create(1000L, 2000L), deleteCells.getTimestampRange()); + } + + @Test + public void setSellTest() { + Entry setCellEntry = + new SetCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")); + SetCell setCell = (SetCell) setCellEntry; + Assert.assertEquals("fake-family", setCell.getFamilyName()); + Assert.assertEquals(ByteString.copyFromUtf8("fake-qualifier"), setCell.getQualifier()); + Assert.assertEquals(1000, setCell.getTimestamp()); + Assert.assertEquals(ByteString.copyFromUtf8("fake-value"), setCell.getValue()); + } +} From f5444f10d328608f17373304d18ec7ab114b57c0 Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Tue, 26 Jul 2022 22:53:01 -0400 Subject: [PATCH 2/7] fix: Fix styles --- .../data/v2/models/ChangeStreamMutation.java | 3 - .../bigtable/data/v2/models/DeleteCells.java | 3 + .../bigtable/data/v2/models/DeleteFamily.java | 1 + .../bigtable/data/v2/models/SetCell.java | 3 + .../v2/models/ChangeStreamMutationTest.java | 71 +++++++------------ .../v2/models/ChangeStreamRecordTest.java | 8 +-- 6 files changed, 35 insertions(+), 54 deletions(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java index 1e23a5b7f2e..059f46d8b40 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java @@ -15,7 +15,6 @@ */ package com.google.cloud.bigtable.data.v2.models; -import com.google.api.core.InternalApi; import com.google.bigtable.v2.ReadChangeStreamResponse.DataChange.Type; import com.google.cloud.bigtable.data.v2.models.Range.TimestampRange; import com.google.common.base.MoreObjects; @@ -179,7 +178,6 @@ ChangeStreamMutation deleteFamily(@Nonnull String familyName) { return this; } - @InternalApi("Used in Changestream beam pipeline.") public RowMutation toRowMutation(@Nonnull String tableId) { Preconditions.checkArgument( token != null && lowWatermark != null, @@ -208,7 +206,6 @@ public RowMutation toRowMutation(@Nonnull String tableId) { return rowMutation; } - @InternalApi("Used in Changestream beam pipeline.") public RowMutationEntry toRowMutationEntry() { Preconditions.checkArgument( token != null && lowWatermark != null, diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java index 9928e684b5b..0e909f238ba 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java @@ -39,14 +39,17 @@ public final class DeleteCells implements Entry, Serializable { this.timestampRange = timestampRange; } + @Nonnull public String getFamilyName() { return this.familyName; } + @Nonnull public ByteString getQualifier() { return this.qualifier; } + @Nonnull public TimestampRange getTimestampRange() { return this.timestampRange; } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java index 1d22321055e..a95316473e3 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java @@ -30,6 +30,7 @@ public final class DeleteFamily implements Entry, Serializable { this.familyName = familyName; } + @Nonnull public String getFamilyName() { return this.familyName; } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java index 29312ea6841..1683595c2ab 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java @@ -44,10 +44,12 @@ public final class SetCell implements Entry, Serializable { this.value = value; } + @Nonnull public String getFamilyName() { return this.familyName; } + @Nonnull public ByteString getQualifier() { return this.qualifier; } @@ -56,6 +58,7 @@ public long getTimestamp() { return this.timestamp; } + @Nonnull public ByteString getValue() { return this.value; } diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java index c4b582d1594..d5f77352fef 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java @@ -24,6 +24,7 @@ import com.google.cloud.bigtable.data.v2.internal.RequestContext; import com.google.common.primitives.Longs; import com.google.protobuf.ByteString; +import com.google.protobuf.Timestamp; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -50,10 +51,8 @@ public class ChangeStreamMutationTest { @Test public void userInitiatedMutationTest() throws IOException, ClassNotFoundException { // Create a user initiated logical mutation. - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -97,10 +96,8 @@ public void userInitiatedMutationTest() throws IOException, ClassNotFoundExcepti @Test public void gcMutationTest() throws IOException, ClassNotFoundException { // Create a GC mutation. - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -144,10 +141,8 @@ public void gcMutationTest() throws IOException, ClassNotFoundException { @Test(expected = IllegalArgumentException.class) public void userInitiatedMutationHasSourceClusterIdTest() { // Create a user initiated logical mutation. - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -172,10 +167,8 @@ public void userInitiatedMutationHasSourceClusterIdTest() { @Test(expected = IllegalArgumentException.class) public void gcMutationHasNoSourceClusterIdTest() { // Create a GC mutation. - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -201,10 +194,8 @@ public void gcMutationHasNoSourceClusterIdTest() { @Test(expected = IllegalArgumentException.class) public void invalidTypeTest() { // Create a ChangeStreamMutation with CONTINUATION type. - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -229,10 +220,8 @@ public void invalidTypeTest() { @Test public void toRowMutationTest() { - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -273,10 +262,8 @@ public void toRowMutationTest() { @Test(expected = IllegalArgumentException.class) public void toRowMutationWithoutTokenShouldFailTest() { - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -294,8 +281,7 @@ public void toRowMutationWithoutTokenShouldFailTest() { @Test(expected = IllegalArgumentException.class) public void toRowMutationWithoutLowWatermarkShouldFailTest() { - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -313,10 +299,8 @@ public void toRowMutationWithoutLowWatermarkShouldFailTest() { @Test public void toRowMutationEntryTest() { - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -354,10 +338,8 @@ public void toRowMutationEntryTest() { @Test(expected = IllegalArgumentException.class) public void toRowMutationEntryWithoutTokenShouldFailTest() { - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -368,15 +350,14 @@ public void toRowMutationEntryWithoutTokenShouldFailTest() { .deleteFamily("fake-family") .setLowWatermark(fakeLowWatermark); - // Convert it to a rowMutationEntry and construct a MutateRowRequest. + // Convert it to a rowMutationEntry. RowMutationEntry rowMutationEntry = changeStreamMutation.toRowMutationEntry(); expect.expect(IllegalArgumentException.class); } @Test(expected = IllegalArgumentException.class) public void toRowMutationEntryWithoutLowWatermarkShouldFailTest() { - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), @@ -387,17 +368,15 @@ public void toRowMutationEntryWithoutLowWatermarkShouldFailTest() { .deleteFamily("fake-family") .setToken("fake-token"); - // Convert it to a rowMutationEntry and construct a MutateRowRequest. + // Convert it to a rowMutationEntry. RowMutationEntry rowMutationEntry = changeStreamMutation.toRowMutationEntry(); expect.expect(IllegalArgumentException.class); } @Test public void testWithLongValue() { - com.google.protobuf.Timestamp fakeCommitTimestamp = - com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build(); - com.google.protobuf.Timestamp fakeLowWatermark = - com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build(); + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = ChangeStreamMutation.create( ByteString.copyFromUtf8("key"), diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java index 8c00d7ae726..05df6039598 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java @@ -23,6 +23,7 @@ import com.google.bigtable.v2.StreamPartition; import com.google.protobuf.ByteString; import com.google.protobuf.Timestamp; +import com.google.rpc.Status; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -57,7 +58,7 @@ public void heartbeatSerializationTest() throws IOException, ClassNotFoundExcept @Test public void closeStreamSerializationTest() throws IOException, ClassNotFoundException { - com.google.rpc.Status status = com.google.rpc.Status.newBuilder().setCode(0).build(); + Status status = Status.newBuilder().setCode(0).build(); RowRange rowRange1 = RowRange.newBuilder() .setStartKeyClosed(ByteString.copyFromUtf8("")) @@ -97,9 +98,6 @@ public void closeStreamSerializationTest() throws IOException, ClassNotFoundExce assertThat(actual.getStatus()).isEqualTo(closeStream.getStatus()); } - @Test - public void changeStreamMutationSerializationTest() {} - @Test public void heartbeatTest() { Timestamp lowWatermark = Timestamp.newBuilder().setSeconds(1000).build(); @@ -127,7 +125,7 @@ public void heartbeatTest() { @Test public void closeStreamTest() { - com.google.rpc.Status status = com.google.rpc.Status.newBuilder().setCode(0).build(); + Status status = Status.newBuilder().setCode(0).build(); RowRange rowRange1 = RowRange.newBuilder() .setStartKeyClosed(ByteString.copyFromUtf8("")) From decb0ad2d4a9c9ffa960067386652f173d4445b6 Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Thu, 28 Jul 2022 16:15:13 -0400 Subject: [PATCH 3/7] fix: Address comments --- .../data/v2/models/ChangeStreamMutation.java | 146 ++++++++------ .../bigtable/data/v2/models/DeleteCells.java | 52 +---- .../bigtable/data/v2/models/DeleteFamily.java | 34 +--- .../bigtable/data/v2/models/SetCell.java | 61 ++---- .../v2/models/ChangeStreamMutationTest.java | 190 ++++-------------- .../bigtable/data/v2/models/EntryTest.java | 12 +- 6 files changed, 162 insertions(+), 333 deletions(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java index 059f46d8b40..d1faf048411 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java @@ -59,42 +59,32 @@ public final class ChangeStreamMutation implements ChangeStreamRecord, Serializa private Timestamp lowWatermark; - private ChangeStreamMutation( - ByteString rowKey, - Type type, - String sourceClusterId, - Timestamp commitTimestamp, - int tieBreaker) { - this.rowKey = rowKey; - this.type = type; - this.sourceClusterId = sourceClusterId; - this.commitTimestamp = commitTimestamp; - this.tieBreaker = tieBreaker; + private ChangeStreamMutation(Builder builder) { + this.rowKey = builder.rowKey; + this.type = builder.type; + this.sourceClusterId = builder.sourceClusterId; + this.commitTimestamp = builder.commitTimestamp; + this.tieBreaker = builder.tieBreaker; + this.token = builder.token;; + this.lowWatermark = builder.lowWatermark; + this.entries = builder.entries; } /** Creates a new instance of a user initiated mutation. */ - static ChangeStreamMutation create( + static Builder createUserMutation( @Nonnull ByteString rowKey, - @Nonnull Type type, @Nonnull String sourceClusterId, @Nonnull Timestamp commitTimestamp, int tieBreaker) { - Preconditions.checkArgument( - type == Type.USER, - "ChangeStreamMutation with a specified source cluster id must be a user initiated mutation."); - return new ChangeStreamMutation(rowKey, type, sourceClusterId, commitTimestamp, tieBreaker); + return new Builder(rowKey, Type.USER, sourceClusterId, commitTimestamp, tieBreaker); } /** Creates a new instance of a GC mutation. */ - static ChangeStreamMutation create( + static Builder createGcMutation( @Nonnull ByteString rowKey, - @Nonnull Type type, @Nonnull Timestamp commitTimestamp, int tieBreaker) { - Preconditions.checkArgument( - type == Type.GARBAGE_COLLECTION, - "ChangeStreamMutation without source cluster id must be a garbage collection mutation."); - return new ChangeStreamMutation(rowKey, type, null, commitTimestamp, tieBreaker); + return new Builder(rowKey, Type.GARBAGE_COLLECTION, null, commitTimestamp, tieBreaker); } private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException { @@ -110,78 +100,123 @@ private void writeObject(ObjectOutputStream output) throws IOException { output.writeObject(entries.build()); } + /** Get the row key of the current mutation. */ @Nonnull public ByteString getRowKey() { return this.rowKey; } + /** Get the type of the current mutation. */ @Nonnull public Type getType() { return this.type; } + /** Get the source cluster id of the current mutation. Null for Garbage collection mutation. */ public String getSourceClusterId() { return this.sourceClusterId; } + /** Get the commit timestamp of the current mutation. */ @Nonnull public Timestamp getCommitTimestamp() { return this.commitTimestamp; } + /** Get the tie breaker of the current mutation. This is used to resolve conflicts when multiple mutations + * are applied to different clusters at the same time. + * */ public int getTieBreaker() { return this.tieBreaker; } - public ChangeStreamMutation setToken(@Nonnull String token) { - this.token = token; - return this; - } - + /** Get the token of the current mutation, which can be used to resume the changestream. */ public String getToken() { return this.token; } - public ChangeStreamMutation setLowWatermark(@Nonnull Timestamp lowWatermark) { - this.lowWatermark = lowWatermark; - return this; - } - + /** Get the low watermark of the current mutation. */ public Timestamp getLowWatermark() { return this.lowWatermark; } + /** Get the list of mods of the current mutation. */ @Nonnull public List getEntries() { return this.entries.build(); } - ChangeStreamMutation setCell( - @Nonnull String familyName, - @Nonnull ByteString qualifier, - long timestamp, - @Nonnull ByteString value) { - this.entries.add(new SetCell(familyName, qualifier, timestamp, value)); - return this; - } + /** Helper class to create a ChangeStreamMutation. */ + public static class Builder { + private final ByteString rowKey; - ChangeStreamMutation deleteCells( - @Nonnull String familyName, - @Nonnull ByteString qualifier, - @Nonnull TimestampRange timestampRange) { - this.entries.add(new DeleteCells(familyName, qualifier, timestampRange)); - return this; - } + private final Type type; + + private final String sourceClusterId; + + private final Timestamp commitTimestamp; + + private final int tieBreaker; + + private transient ImmutableList.Builder entries = ImmutableList.builder(); + + private String token; + + private Timestamp lowWatermark; - ChangeStreamMutation deleteFamily(@Nonnull String familyName) { - this.entries.add(new DeleteFamily(familyName)); - return this; + private Builder(ByteString rowKey, + Type type, + String sourceClusterId, + Timestamp commitTimestamp, + int tieBreaker) { + this.rowKey = rowKey; + this.type = type; + this.sourceClusterId = sourceClusterId; + this.commitTimestamp = commitTimestamp; + this.tieBreaker = tieBreaker; + } + + Builder setCell( + @Nonnull String familyName, + @Nonnull ByteString qualifier, + long timestamp, + @Nonnull ByteString value) { + this.entries.add(SetCell.create(familyName, qualifier, timestamp, value)); + return this; + } + + Builder deleteCells( + @Nonnull String familyName, + @Nonnull ByteString qualifier, + @Nonnull TimestampRange timestampRange) { + this.entries.add(DeleteCells.create(familyName, qualifier, timestampRange)); + return this; + } + + Builder deleteFamily(@Nonnull String familyName) { + this.entries.add(DeleteFamily.create(familyName)); + return this; + } + + public Builder setToken(@Nonnull String token) { + this.token = token; + return this; + } + + public Builder setLowWatermark(@Nonnull Timestamp lowWatermark) { + this.lowWatermark = lowWatermark; + return this; + } + + public ChangeStreamMutation build() { + Preconditions.checkArgument( + token != null && lowWatermark != null, + "ChangeStreamMutation must have a continuation token and low watermark."); + return new ChangeStreamMutation(this); + } } public RowMutation toRowMutation(@Nonnull String tableId) { - Preconditions.checkArgument( - token != null && lowWatermark != null, - "ChangeStreamMutation must have a continuation token and low watermark."); RowMutation rowMutation = RowMutation.create(tableId, rowKey); for (Entry entry : this.entries.build()) { if (entry instanceof DeleteFamily) { @@ -207,9 +242,6 @@ public RowMutation toRowMutation(@Nonnull String tableId) { } public RowMutationEntry toRowMutationEntry() { - Preconditions.checkArgument( - token != null && lowWatermark != null, - "ChangeStreamMutation must have a continuation token and low watermark."); RowMutationEntry rowMutationEntry = RowMutationEntry.create(rowKey); for (Entry entry : this.entries.build()) { if (entry instanceof DeleteFamily) { diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java index 0e909f238ba..68f3fa1e19d 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java @@ -15,70 +15,40 @@ */ package com.google.cloud.bigtable.data.v2.models; +import com.google.auto.value.AutoValue; import com.google.cloud.bigtable.data.v2.models.Range.TimestampRange; import com.google.common.base.MoreObjects; -import com.google.common.base.Objects; import com.google.protobuf.ByteString; import java.io.Serializable; import javax.annotation.Nonnull; /** Representation of a DeleteCells mod in a data change. */ -public final class DeleteCells implements Entry, Serializable { +@AutoValue +public abstract class DeleteCells implements Entry, Serializable { private static final long serialVersionUID = 851772158721462017L; - private final String familyName; - private final ByteString qualifier; - private final TimestampRange timestampRange; - - DeleteCells( + public static DeleteCells create( @Nonnull String familyName, @Nonnull ByteString qualifier, @Nonnull TimestampRange timestampRange) { - this.familyName = familyName; - this.qualifier = qualifier; - this.timestampRange = timestampRange; + return new AutoValue_DeleteCells(familyName, qualifier, timestampRange); } @Nonnull - public String getFamilyName() { - return this.familyName; - } + public abstract String getFamilyName(); @Nonnull - public ByteString getQualifier() { - return this.qualifier; - } + public abstract ByteString getQualifier(); @Nonnull - public TimestampRange getTimestampRange() { - return this.timestampRange; - } - - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o == null || getClass() != o.getClass()) { - return false; - } - DeleteCells otherDeleteCell = (DeleteCells) o; - return Objects.equal(familyName, otherDeleteCell.getFamilyName()) - && Objects.equal(qualifier, otherDeleteCell.getQualifier()) - && Objects.equal(timestampRange, otherDeleteCell.getTimestampRange()); - } - - @Override - public int hashCode() { - return Objects.hashCode(familyName); - } + public abstract TimestampRange getTimestampRange(); @Override public String toString() { return MoreObjects.toStringHelper(this) - .add("familyName", familyName) - .add("qualifier", qualifier.toStringUtf8()) - .add("timestampRange", timestampRange) + .add("familyName", getFamilyName()) + .add("qualifier", getQualifier().toStringUtf8()) + .add("timestampRange", getTimestampRange()) .toString(); } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java index a95316473e3..32735844b91 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java @@ -15,45 +15,25 @@ */ package com.google.cloud.bigtable.data.v2.models; +import com.google.auto.value.AutoValue; import com.google.common.base.MoreObjects; -import com.google.common.base.Objects; import java.io.Serializable; import javax.annotation.Nonnull; /** Representation of a DeleteFamily mod in a data change. */ -public final class DeleteFamily implements Entry, Serializable { +@AutoValue +public abstract class DeleteFamily implements Entry, Serializable { private static final long serialVersionUID = 81806775917145615L; - private final String familyName; - - DeleteFamily(@Nonnull String familyName) { - this.familyName = familyName; + public static DeleteFamily create(@Nonnull String familyName) { + return new AutoValue_DeleteFamily(familyName); } @Nonnull - public String getFamilyName() { - return this.familyName; - } - - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o == null || getClass() != o.getClass()) { - return false; - } - DeleteFamily otherDeleteFamily = (DeleteFamily) o; - return Objects.equal(familyName, otherDeleteFamily.getFamilyName()); - } - - @Override - public int hashCode() { - return Objects.hashCode(familyName); - } + public abstract String getFamilyName(); @Override public String toString() { - return MoreObjects.toStringHelper(this).add("familyName", familyName).toString(); + return MoreObjects.toStringHelper(this).add("familyName", getFamilyName()).toString(); } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java index 1683595c2ab..a406f6af674 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java @@ -15,8 +15,8 @@ */ package com.google.cloud.bigtable.data.v2.models; +import com.google.auto.value.AutoValue; import com.google.common.base.MoreObjects; -import com.google.common.base.Objects; import com.google.protobuf.ByteString; import java.io.Serializable; import javax.annotation.Nonnull; @@ -25,71 +25,36 @@ * Representation of a SetCell mod in a data change, whose value is concatenated by * (TODO:ChangeStreamRecordMerger) in case of SetCell value chunking. */ -public final class SetCell implements Entry, Serializable { +@AutoValue +public abstract class SetCell implements Entry, Serializable { private static final long serialVersionUID = 77123872266724154L; - private final String familyName; - private final ByteString qualifier; - private final long timestamp; - private final ByteString value; - - SetCell( + public static SetCell create( @Nonnull String familyName, @Nonnull ByteString qualifier, long timestamp, @Nonnull ByteString value) { - this.familyName = familyName; - this.qualifier = qualifier; - this.timestamp = timestamp; - this.value = value; + return new AutoValue_SetCell(familyName, qualifier, timestamp, value); } @Nonnull - public String getFamilyName() { - return this.familyName; - } + public abstract String getFamilyName(); @Nonnull - public ByteString getQualifier() { - return this.qualifier; - } + public abstract ByteString getQualifier(); - public long getTimestamp() { - return this.timestamp; - } + public abstract long getTimestamp(); @Nonnull - public ByteString getValue() { - return this.value; - } - - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o == null || getClass() != o.getClass()) { - return false; - } - SetCell otherSetCell = (SetCell) o; - return Objects.equal(familyName, otherSetCell.getFamilyName()) - && Objects.equal(qualifier, otherSetCell.getQualifier()) - && Objects.equal(timestamp, otherSetCell.getTimestamp()) - && Objects.equal(value, otherSetCell.getValue()); - } - - @Override - public int hashCode() { - return Objects.hashCode(familyName); - } + public abstract ByteString getValue(); @Override public String toString() { return MoreObjects.toStringHelper(this) - .add("familyName", familyName) - .add("qualifier", qualifier.toStringUtf8()) - .add("timestamp", timestamp) - .add("value", value.toStringUtf8()) + .add("familyName", getFamilyName()) + .add("qualifier", getQualifier().toStringUtf8()) + .add("timestamp", getTimestamp()) + .add("value", getValue().toStringUtf8()) .toString(); } } diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java index d5f77352fef..39e9982b43a 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java @@ -54,12 +54,8 @@ public void userInitiatedMutationTest() throws IOException, ClassNotFoundExcepti Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) .setCell( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), @@ -71,7 +67,8 @@ public void userInitiatedMutationTest() throws IOException, ClassNotFoundExcepti ByteString.copyFromUtf8("fake-qualifier"), Range.TimestampRange.create(1000L, 2000L)) .setToken("fake-token") - .setLowWatermark(fakeLowWatermark); + .setLowWatermark(fakeLowWatermark) + .build(); // Test the getters. Assert.assertEquals(changeStreamMutation.getRowKey(), ByteString.copyFromUtf8("key")); @@ -99,11 +96,8 @@ public void gcMutationTest() throws IOException, ClassNotFoundException { Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.GARBAGE_COLLECTION, - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createGcMutation( + ByteString.copyFromUtf8("key"), fakeCommitTimestamp, 0) .setCell( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), @@ -115,7 +109,8 @@ public void gcMutationTest() throws IOException, ClassNotFoundException { ByteString.copyFromUtf8("fake-qualifier"), Range.TimestampRange.create(1000L, 2000L)) .setToken("fake-token") - .setLowWatermark(fakeLowWatermark); + .setLowWatermark(fakeLowWatermark) + .build(); // Test the getters. Assert.assertEquals(changeStreamMutation.getRowKey(), ByteString.copyFromUtf8("key")); @@ -138,97 +133,13 @@ public void gcMutationTest() throws IOException, ClassNotFoundException { assertThat(actual.toString()).isEqualTo(changeStreamMutation.toString()); } - @Test(expected = IllegalArgumentException.class) - public void userInitiatedMutationHasSourceClusterIdTest() { - // Create a user initiated logical mutation. - Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); - Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); - ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - fakeCommitTimestamp, - 0) - .setCell( - "fake-family", - ByteString.copyFromUtf8("fake-qualifier"), - 1000, - ByteString.copyFromUtf8("fake-value")) - .deleteFamily("fake-family") - .deleteCells( - "fake-family", - ByteString.copyFromUtf8("fake-qualifier"), - Range.TimestampRange.create(1000L, 2000L)) - .setToken("fake-token") - .setLowWatermark(fakeLowWatermark); - expect.expect(IllegalArgumentException.class); - } - - @Test(expected = IllegalArgumentException.class) - public void gcMutationHasNoSourceClusterIdTest() { - // Create a GC mutation. - Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); - Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); - ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.GARBAGE_COLLECTION, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) - .setCell( - "fake-family", - ByteString.copyFromUtf8("fake-qualifier"), - 1000, - ByteString.copyFromUtf8("fake-value")) - .deleteFamily("fake-family") - .deleteCells( - "fake-family", - ByteString.copyFromUtf8("fake-qualifier"), - Range.TimestampRange.create(1000L, 2000L)) - .setToken("fake-token") - .setLowWatermark(fakeLowWatermark); - expect.expect(IllegalArgumentException.class); - } - - @Test(expected = IllegalArgumentException.class) - public void invalidTypeTest() { - // Create a ChangeStreamMutation with CONTINUATION type. - Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); - Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); - ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.CONTINUATION, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) - .setCell( - "fake-family", - ByteString.copyFromUtf8("fake-qualifier"), - 1000, - ByteString.copyFromUtf8("fake-value")) - .deleteFamily("fake-family") - .deleteCells( - "fake-family", - ByteString.copyFromUtf8("fake-qualifier"), - Range.TimestampRange.create(1000L, 2000L)) - .setToken("fake-token") - .setLowWatermark(fakeLowWatermark); - expect.expect(IllegalArgumentException.class); - } - @Test public void toRowMutationTest() { Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) .setCell( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), @@ -240,7 +151,8 @@ public void toRowMutationTest() { ByteString.copyFromUtf8("fake-qualifier"), Range.TimestampRange.create(1000L, 2000L)) .setToken("fake-token") - .setLowWatermark(fakeLowWatermark); + .setLowWatermark(fakeLowWatermark) + .build(); // Convert it to a rowMutation and construct a MutateRowRequest. RowMutation rowMutation = changeStreamMutation.toRowMutation(TABLE_ID); @@ -265,17 +177,11 @@ public void toRowMutationWithoutTokenShouldFailTest() { Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) .deleteFamily("fake-family") - .setLowWatermark(fakeLowWatermark); - - // Convert it to a rowMutation. - RowMutation rowMutation = changeStreamMutation.toRowMutation(TABLE_ID); + .setLowWatermark(fakeLowWatermark) + .build(); expect.expect(IllegalArgumentException.class); } @@ -283,17 +189,11 @@ public void toRowMutationWithoutTokenShouldFailTest() { public void toRowMutationWithoutLowWatermarkShouldFailTest() { Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) .deleteFamily("fake-family") - .setToken("fake-token"); - - // Convert it to a rowMutation. - RowMutation rowMutation = changeStreamMutation.toRowMutation(TABLE_ID); + .setToken("fake-token") + .build(); expect.expect(IllegalArgumentException.class); } @@ -302,12 +202,8 @@ public void toRowMutationEntryTest() { Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) .setCell( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), @@ -319,7 +215,8 @@ public void toRowMutationEntryTest() { ByteString.copyFromUtf8("fake-qualifier"), Range.TimestampRange.create(1000L, 2000L)) .setToken("fake-token") - .setLowWatermark(fakeLowWatermark); + .setLowWatermark(fakeLowWatermark) + .build(); // Convert it to a rowMutationEntry and construct a MutateRowRequest. RowMutationEntry rowMutationEntry = changeStreamMutation.toRowMutationEntry(); @@ -341,17 +238,11 @@ public void toRowMutationEntryWithoutTokenShouldFailTest() { Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) .deleteFamily("fake-family") - .setLowWatermark(fakeLowWatermark); - - // Convert it to a rowMutationEntry. - RowMutationEntry rowMutationEntry = changeStreamMutation.toRowMutationEntry(); + .setLowWatermark(fakeLowWatermark) + .build(); expect.expect(IllegalArgumentException.class); } @@ -359,17 +250,11 @@ public void toRowMutationEntryWithoutTokenShouldFailTest() { public void toRowMutationEntryWithoutLowWatermarkShouldFailTest() { Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) .deleteFamily("fake-family") - .setToken("fake-token"); - - // Convert it to a rowMutationEntry. - RowMutationEntry rowMutationEntry = changeStreamMutation.toRowMutationEntry(); + .setToken("fake-token") + .build(); expect.expect(IllegalArgumentException.class); } @@ -378,19 +263,16 @@ public void testWithLongValue() { Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); ChangeStreamMutation changeStreamMutation = - ChangeStreamMutation.create( - ByteString.copyFromUtf8("key"), - ReadChangeStreamResponse.DataChange.Type.USER, - "fake-source-cluster-id", - fakeCommitTimestamp, - 0) + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) .setCell( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), 1000L, ByteString.copyFrom(Longs.toByteArray(1L))) .setToken("fake-token") - .setLowWatermark(fakeLowWatermark); + .setLowWatermark(fakeLowWatermark) + .build(); RowMutation rowMutation = changeStreamMutation.toRowMutation(TABLE_ID); MutateRowRequest mutateRowRequest = rowMutation.toProto(REQUEST_CONTEXT); diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/EntryTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/EntryTest.java index 1e581ce1c5f..11ff0a9f028 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/EntryTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/EntryTest.java @@ -43,12 +43,12 @@ private void validateSerializationRoundTrip(Object obj) @Test public void serializationTest() throws IOException, ClassNotFoundException { // DeleteFamily - Entry deleteFamilyEntry = new DeleteFamily("fake-family"); + Entry deleteFamilyEntry = DeleteFamily.create("fake-family"); validateSerializationRoundTrip(deleteFamilyEntry); // DeleteCell Entry deleteCellsEntry = - new DeleteCells( + DeleteCells.create( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), Range.TimestampRange.create(1000L, 2000L)); @@ -56,7 +56,7 @@ public void serializationTest() throws IOException, ClassNotFoundException { // SetCell Entry setCellEntry = - new SetCell( + SetCell.create( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), 1000, @@ -66,7 +66,7 @@ public void serializationTest() throws IOException, ClassNotFoundException { @Test public void deleteFamilyTest() { - Entry deleteFamilyEntry = new DeleteFamily("fake-family"); + Entry deleteFamilyEntry = DeleteFamily.create("fake-family"); DeleteFamily deleteFamily = (DeleteFamily) deleteFamilyEntry; Assert.assertEquals("fake-family", deleteFamily.getFamilyName()); } @@ -74,7 +74,7 @@ public void deleteFamilyTest() { @Test public void deleteCellsTest() { Entry deleteCellEntry = - new DeleteCells( + DeleteCells.create( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), Range.TimestampRange.create(1000L, 2000L)); @@ -87,7 +87,7 @@ public void deleteCellsTest() { @Test public void setSellTest() { Entry setCellEntry = - new SetCell( + SetCell.create( "fake-family", ByteString.copyFromUtf8("fake-qualifier"), 1000, From ac05fef0aacac3c613f8bc3f7814620039be53e6 Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Thu, 28 Jul 2022 16:35:00 -0400 Subject: [PATCH 4/7] fix: Update Heartbeat to use AutoValue --- .../data/v2/models/ChangeStreamMutation.java | 42 +++++++------- .../bigtable/data/v2/models/Heartbeat.java | 55 +++++-------------- 2 files changed, 35 insertions(+), 62 deletions(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java index d1faf048411..6e803c32ea4 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java @@ -65,7 +65,7 @@ private ChangeStreamMutation(Builder builder) { this.sourceClusterId = builder.sourceClusterId; this.commitTimestamp = builder.commitTimestamp; this.tieBreaker = builder.tieBreaker; - this.token = builder.token;; + this.token = builder.token; this.lowWatermark = builder.lowWatermark; this.entries = builder.entries; } @@ -81,9 +81,7 @@ static Builder createUserMutation( /** Creates a new instance of a GC mutation. */ static Builder createGcMutation( - @Nonnull ByteString rowKey, - @Nonnull Timestamp commitTimestamp, - int tieBreaker) { + @Nonnull ByteString rowKey, @Nonnull Timestamp commitTimestamp, int tieBreaker) { return new Builder(rowKey, Type.GARBAGE_COLLECTION, null, commitTimestamp, tieBreaker); } @@ -123,9 +121,10 @@ public Timestamp getCommitTimestamp() { return this.commitTimestamp; } - /** Get the tie breaker of the current mutation. This is used to resolve conflicts when multiple mutations - * are applied to different clusters at the same time. - * */ + /** + * Get the tie breaker of the current mutation. This is used to resolve conflicts when multiple + * mutations are applied to different clusters at the same time. + */ public int getTieBreaker() { return this.tieBreaker; } @@ -164,11 +163,12 @@ public static class Builder { private Timestamp lowWatermark; - private Builder(ByteString rowKey, - Type type, - String sourceClusterId, - Timestamp commitTimestamp, - int tieBreaker) { + private Builder( + ByteString rowKey, + Type type, + String sourceClusterId, + Timestamp commitTimestamp, + int tieBreaker) { this.rowKey = rowKey; this.type = type; this.sourceClusterId = sourceClusterId; @@ -177,18 +177,18 @@ private Builder(ByteString rowKey, } Builder setCell( - @Nonnull String familyName, - @Nonnull ByteString qualifier, - long timestamp, - @Nonnull ByteString value) { + @Nonnull String familyName, + @Nonnull ByteString qualifier, + long timestamp, + @Nonnull ByteString value) { this.entries.add(SetCell.create(familyName, qualifier, timestamp, value)); return this; } Builder deleteCells( - @Nonnull String familyName, - @Nonnull ByteString qualifier, - @Nonnull TimestampRange timestampRange) { + @Nonnull String familyName, + @Nonnull ByteString qualifier, + @Nonnull TimestampRange timestampRange) { this.entries.add(DeleteCells.create(familyName, qualifier, timestampRange)); return this; } @@ -210,8 +210,8 @@ public Builder setLowWatermark(@Nonnull Timestamp lowWatermark) { public ChangeStreamMutation build() { Preconditions.checkArgument( - token != null && lowWatermark != null, - "ChangeStreamMutation must have a continuation token and low watermark."); + token != null && lowWatermark != null, + "ChangeStreamMutation must have a continuation token and low watermark."); return new ChangeStreamMutation(this); } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java index 73876f887b7..9498369501c 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java @@ -15,65 +15,38 @@ */ package com.google.cloud.bigtable.data.v2.models; -import com.google.api.core.InternalApi; +import com.google.auto.value.AutoValue; import com.google.bigtable.v2.ReadChangeStreamResponse; import com.google.common.base.MoreObjects; -import com.google.common.base.Objects; import com.google.protobuf.Timestamp; import java.io.Serializable; import javax.annotation.Nonnull; -public final class Heartbeat implements ChangeStreamRecord, Serializable { +@AutoValue +public abstract class Heartbeat implements ChangeStreamRecord, Serializable { private static final long serialVersionUID = 7316215828353608504L; - private final Timestamp lowWatermark; - private final ChangeStreamContinuationToken changeStreamContinuationToken; - private Heartbeat( - Timestamp lowWatermark, ChangeStreamContinuationToken changeStreamContinuationToken) { - this.lowWatermark = lowWatermark; - this.changeStreamContinuationToken = changeStreamContinuationToken; - } - - @InternalApi("Used in Changestream beam pipeline.") - public ChangeStreamContinuationToken getChangeStreamContinuationToken() { - return changeStreamContinuationToken; - } - - @InternalApi("Used in Changestream beam pipeline.") - public Timestamp getLowWatermark() { - return lowWatermark; + public static Heartbeat create( + ChangeStreamContinuationToken changeStreamContinuationToken, Timestamp lowWatermark) { + return new AutoValue_Heartbeat(changeStreamContinuationToken, lowWatermark); } /** Wraps the protobuf {@link ReadChangeStreamResponse.Heartbeat}. */ - static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) { - return new Heartbeat( - heartbeat.getLowWatermark(), - ChangeStreamContinuationToken.fromProto(heartbeat.getContinuationToken())); + public static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) { + return create( + ChangeStreamContinuationToken.fromProto(heartbeat.getContinuationToken()), + heartbeat.getLowWatermark()); } - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o == null || getClass() != o.getClass()) { - return false; - } - Heartbeat record = (Heartbeat) o; - return Objects.equal(lowWatermark, record.getLowWatermark()) - && Objects.equal(changeStreamContinuationToken, record.getChangeStreamContinuationToken()); - } + public abstract ChangeStreamContinuationToken getChangeStreamContinuationToken(); - @Override - public int hashCode() { - return Objects.hashCode(lowWatermark, changeStreamContinuationToken); - } + public abstract Timestamp getLowWatermark(); @Override public String toString() { return MoreObjects.toStringHelper(this) - .add("lowWatermark", lowWatermark) - .add("changeStreamContinuationToken", changeStreamContinuationToken) + .add("lowWatermark", getLowWatermark()) + .add("changeStreamContinuationToken", getChangeStreamContinuationToken()) .toString(); } } From 354d043959e859281f070dd93db37f31e97453ca Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Thu, 28 Jul 2022 16:42:48 -0400 Subject: [PATCH 5/7] fix: Add more comments --- .../google/cloud/bigtable/data/v2/models/DeleteCells.java | 3 +++ .../google/cloud/bigtable/data/v2/models/DeleteFamily.java | 1 + .../com/google/cloud/bigtable/data/v2/models/Heartbeat.java | 5 ++++- .../com/google/cloud/bigtable/data/v2/models/SetCell.java | 4 ++++ 4 files changed, 12 insertions(+), 1 deletion(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java index 68f3fa1e19d..c76a32c1ef2 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java @@ -34,12 +34,15 @@ public static DeleteCells create( return new AutoValue_DeleteCells(familyName, qualifier, timestampRange); } + /** Get the column family of the current DeleteCells. */ @Nonnull public abstract String getFamilyName(); + /** Get the column qualifier of the current DeleteCells. */ @Nonnull public abstract ByteString getQualifier(); + /** Get the timestamp range of the current DeleteCells. */ @Nonnull public abstract TimestampRange getTimestampRange(); diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java index 32735844b91..5022a648e0f 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java @@ -29,6 +29,7 @@ public static DeleteFamily create(@Nonnull String familyName) { return new AutoValue_DeleteFamily(familyName); } + /** Get the column family of the current DeleteFamily. */ @Nonnull public abstract String getFamilyName(); diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java index 9498369501c..255c6669787 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java @@ -15,6 +15,7 @@ */ package com.google.cloud.bigtable.data.v2.models; +import com.google.api.core.InternalApi; import com.google.auto.value.AutoValue; import com.google.bigtable.v2.ReadChangeStreamResponse; import com.google.common.base.MoreObjects; @@ -32,14 +33,16 @@ public static Heartbeat create( } /** Wraps the protobuf {@link ReadChangeStreamResponse.Heartbeat}. */ - public static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) { + static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) { return create( ChangeStreamContinuationToken.fromProto(heartbeat.getContinuationToken()), heartbeat.getLowWatermark()); } + @InternalApi("Used in Changestream beam pipeline.") public abstract ChangeStreamContinuationToken getChangeStreamContinuationToken(); + @InternalApi("Used in Changestream beam pipeline.") public abstract Timestamp getLowWatermark(); @Override diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java index a406f6af674..3b61d76fdf9 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java @@ -37,14 +37,18 @@ public static SetCell create( return new AutoValue_SetCell(familyName, qualifier, timestamp, value); } + /** Get the column family of the current SetCell. */ @Nonnull public abstract String getFamilyName(); + /** Get the column qualifier of the current SetCell. */ @Nonnull public abstract ByteString getQualifier(); + /** Get the timestamp of the current SetCell. */ public abstract long getTimestamp(); + /** Get the value of the current SetCell. */ @Nonnull public abstract ByteString getValue(); From 301f1ec5566b27730d01346c6e748b5e724434fe Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Mon, 1 Aug 2022 10:30:34 -0400 Subject: [PATCH 6/7] fix: Address comments --- .../data/v2/models/ChangeStreamMutation.java | 57 +++++++++++++++++-- .../bigtable/data/v2/models/DeleteCells.java | 10 ---- .../bigtable/data/v2/models/DeleteFamily.java | 6 -- .../bigtable/data/v2/models/Heartbeat.java | 9 --- .../bigtable/data/v2/models/SetCell.java | 11 ---- .../v2/models/ChangeStreamMutationTest.java | 27 +++++++++ 6 files changed, 79 insertions(+), 41 deletions(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java index 6e803c32ea4..d62cf62784c 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java @@ -33,9 +33,33 @@ /** * A ChangeStreamMutation represents a list of mods(represented by List<{@link Entry}>) targeted at - * a single row, which is concatenated by (TODO:ChangeStreamRecordMerger). + * a single row, which is concatenated by (TODO:ChangeStreamRecordMerger). It represents a logical + * row mutation and can be converted to the original write request(i.e. {@link RowMutation} or + * {@link RowMutationEntry}. * - *

It's meant to be used in {@link com.google.cloud.bigtable.data.v2.models.ChangeStreamRecord}. + *

A ChangeStreamMutation can be constructed in two ways, depending on whether it's a user + * initiated mutation or a Garbage Collection mutation. Either way, the caller should explicitly set + * `token` and `lowWatermark` before build(), otherwise it'll raise an error. + * + *

Case 1) User initiated mutation. + * + * 

{@code
+ * ChangeStreamMutation.Builder builder = ChangeStreamMutation.createUserMutation(...);
+ * builder.setCell(...);
+ * builder.deleteFamily(...);
+ * builder.deleteCells(...);
+ * ChangeStreamMutation changeStreamMutation = builder.setToken(...).setLowWatermark().build();
+ * }
+ * + * Case 2) Garbage Collection mutation. + * + * 
{@code
+ * ChangeStreamMutation.Builder builder = ChangeStreamMutation.createGcMutation(...);
+ * builder.setCell(...);
+ * builder.deleteFamily(...);
+ * builder.deleteCells(...);
+ * ChangeStreamMutation changeStreamMutation = builder.setToken(...).setLowWatermark().build();
+ * }
*/ public final class ChangeStreamMutation implements ChangeStreamRecord, Serializable { private static final long serialVersionUID = 8419520253162024218L; @@ -54,7 +78,6 @@ public final class ChangeStreamMutation implements ChangeStreamRecord, Serializa private transient ImmutableList.Builder entries = ImmutableList.builder(); - /** Token and lowWatermark are set when we finish building the logical mutation. */ private String token; private Timestamp lowWatermark; @@ -70,7 +93,11 @@ private ChangeStreamMutation(Builder builder) { this.entries = builder.entries; } - /** Creates a new instance of a user initiated mutation. */ + /** + * Creates a new instance of a user initiated mutation. It returns a builder instead of a + * ChangeStreamMutation because `token` and `loWatermark` must be set later when we finish + * building the logical mutation. + */ static Builder createUserMutation( @Nonnull ByteString rowKey, @Nonnull String sourceClusterId, @@ -79,7 +106,11 @@ static Builder createUserMutation( return new Builder(rowKey, Type.USER, sourceClusterId, commitTimestamp, tieBreaker); } - /** Creates a new instance of a GC mutation. */ + /** + * Creates a new instance of a GC mutation. It returns a builder instead of a ChangeStreamMutation + * because `token` and `loWatermark` must be set later when we finish building the logical + * mutation. + */ static Builder createGcMutation( @Nonnull ByteString rowKey, @Nonnull Timestamp commitTimestamp, int tieBreaker) { return new Builder(rowKey, Type.GARBAGE_COLLECTION, null, commitTimestamp, tieBreaker); @@ -145,6 +176,11 @@ public List getEntries() { return this.entries.build(); } + /** Returns a builder containing all the values of this ChangeStreamMutation class. */ + Builder toBuilder() { + return new Builder(this); + } + /** Helper class to create a ChangeStreamMutation. */ public static class Builder { private final ByteString rowKey; @@ -176,6 +212,17 @@ private Builder( this.tieBreaker = tieBreaker; } + private Builder(ChangeStreamMutation changeStreamMutation) { + this.rowKey = changeStreamMutation.rowKey; + this.type = changeStreamMutation.type; + this.sourceClusterId = changeStreamMutation.sourceClusterId; + this.commitTimestamp = changeStreamMutation.commitTimestamp; + this.tieBreaker = changeStreamMutation.tieBreaker; + this.entries = changeStreamMutation.entries; + this.token = changeStreamMutation.token; + this.lowWatermark = changeStreamMutation.lowWatermark; + } + Builder setCell( @Nonnull String familyName, @Nonnull ByteString qualifier, diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java index c76a32c1ef2..238ddb1638e 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteCells.java @@ -17,7 +17,6 @@ import com.google.auto.value.AutoValue; import com.google.cloud.bigtable.data.v2.models.Range.TimestampRange; -import com.google.common.base.MoreObjects; import com.google.protobuf.ByteString; import java.io.Serializable; import javax.annotation.Nonnull; @@ -45,13 +44,4 @@ public static DeleteCells create( /** Get the timestamp range of the current DeleteCells. */ @Nonnull public abstract TimestampRange getTimestampRange(); - - @Override - public String toString() { - return MoreObjects.toStringHelper(this) - .add("familyName", getFamilyName()) - .add("qualifier", getQualifier().toStringUtf8()) - .add("timestampRange", getTimestampRange()) - .toString(); - } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java index 5022a648e0f..171ecccb416 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/DeleteFamily.java @@ -16,7 +16,6 @@ package com.google.cloud.bigtable.data.v2.models; import com.google.auto.value.AutoValue; -import com.google.common.base.MoreObjects; import java.io.Serializable; import javax.annotation.Nonnull; @@ -32,9 +31,4 @@ public static DeleteFamily create(@Nonnull String familyName) { /** Get the column family of the current DeleteFamily. */ @Nonnull public abstract String getFamilyName(); - - @Override - public String toString() { - return MoreObjects.toStringHelper(this).add("familyName", getFamilyName()).toString(); - } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java index 255c6669787..db82657e492 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java @@ -18,7 +18,6 @@ import com.google.api.core.InternalApi; import com.google.auto.value.AutoValue; import com.google.bigtable.v2.ReadChangeStreamResponse; -import com.google.common.base.MoreObjects; import com.google.protobuf.Timestamp; import java.io.Serializable; import javax.annotation.Nonnull; @@ -44,12 +43,4 @@ static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat @InternalApi("Used in Changestream beam pipeline.") public abstract Timestamp getLowWatermark(); - - @Override - public String toString() { - return MoreObjects.toStringHelper(this) - .add("lowWatermark", getLowWatermark()) - .add("changeStreamContinuationToken", getChangeStreamContinuationToken()) - .toString(); - } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java index 3b61d76fdf9..a157b5cd73a 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/SetCell.java @@ -16,7 +16,6 @@ package com.google.cloud.bigtable.data.v2.models; import com.google.auto.value.AutoValue; -import com.google.common.base.MoreObjects; import com.google.protobuf.ByteString; import java.io.Serializable; import javax.annotation.Nonnull; @@ -51,14 +50,4 @@ public static SetCell create( /** Get the value of the current SetCell. */ @Nonnull public abstract ByteString getValue(); - - @Override - public String toString() { - return MoreObjects.toStringHelper(this) - .add("familyName", getFamilyName()) - .add("qualifier", getQualifier().toStringUtf8()) - .add("timestamp", getTimestamp()) - .add("value", getValue().toStringUtf8()) - .toString(); - } } diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java index 39e9982b43a..8ead371bd5e 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java @@ -284,4 +284,31 @@ public void testWithLongValue() { assertThat(mutateRowRequest.getMutations(0).getSetCell().getValue()) .isEqualTo(ByteString.copyFromUtf8("\000\000\000\000\000\000\000\001")); } + + @Test + public void toBuilderTest() { + // Create a user initiated logical mutation. + Timestamp fakeCommitTimestamp = Timestamp.newBuilder().setSeconds(1000).build(); + Timestamp fakeLowWatermark = Timestamp.newBuilder().setSeconds(2000).build(); + ChangeStreamMutation changeStreamMutation = + ChangeStreamMutation.createUserMutation( + ByteString.copyFromUtf8("key"), "fake-source-cluster-id", fakeCommitTimestamp, 0) + .setCell( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + 1000, + ByteString.copyFromUtf8("fake-value")) + .deleteFamily("fake-family") + .deleteCells( + "fake-family", + ByteString.copyFromUtf8("fake-qualifier"), + Range.TimestampRange.create(1000L, 2000L)) + .setToken("fake-token") + .setLowWatermark(fakeLowWatermark) + .build(); + + // Test round-trip of a ChangeStreamMutation through `toBuilder().build()`. + ChangeStreamMutation otherMutation = changeStreamMutation.toBuilder().build(); + assertThat(changeStreamMutation).isEqualTo(otherMutation); + } } From 0c189568c0f81c7904d7b90739bd8e589451a8a5 Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Mon, 1 Aug 2022 11:00:47 -0400 Subject: [PATCH 7/7] fix: Fix unit test due to toString(). Can't compare ByteString.toString() directly even though the contents are the same. So we compare their fields and toRowMutation. --- .../data/v2/models/ChangeStreamMutation.java | 2 +- .../v2/models/ChangeStreamMutationTest.java | 20 +++++++++++++++++-- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java index d62cf62784c..b79b184e7a4 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutation.java @@ -322,7 +322,7 @@ public boolean equals(Object o) { return false; } ChangeStreamMutation otherChangeStreamMutation = (ChangeStreamMutation) o; - return Objects.equal(this.toString(), otherChangeStreamMutation.toString()); + return Objects.equal(this.hashCode(), otherChangeStreamMutation.hashCode()); } @Override diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java index 8ead371bd5e..938213fb365 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamMutationTest.java @@ -87,7 +87,15 @@ public void userInitiatedMutationTest() throws IOException, ClassNotFoundExcepti oos.close(); ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray())); ChangeStreamMutation actual = (ChangeStreamMutation) ois.readObject(); - assertThat(actual.toString()).isEqualTo(changeStreamMutation.toString()); + Assert.assertEquals(actual.getRowKey(), changeStreamMutation.getRowKey()); + Assert.assertEquals(actual.getType(), changeStreamMutation.getType()); + Assert.assertEquals(actual.getSourceClusterId(), changeStreamMutation.getSourceClusterId()); + Assert.assertEquals(actual.getCommitTimestamp(), changeStreamMutation.getCommitTimestamp()); + Assert.assertEquals(actual.getTieBreaker(), changeStreamMutation.getTieBreaker()); + Assert.assertEquals(actual.getToken(), changeStreamMutation.getToken()); + Assert.assertEquals(actual.getLowWatermark(), changeStreamMutation.getLowWatermark()); + assertThat(actual.toRowMutation(TABLE_ID).toProto(REQUEST_CONTEXT)) + .isEqualTo(changeStreamMutation.toRowMutation(TABLE_ID).toProto(REQUEST_CONTEXT)); } @Test @@ -130,7 +138,15 @@ public void gcMutationTest() throws IOException, ClassNotFoundException { oos.close(); ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray())); ChangeStreamMutation actual = (ChangeStreamMutation) ois.readObject(); - assertThat(actual.toString()).isEqualTo(changeStreamMutation.toString()); + Assert.assertEquals(actual.getRowKey(), changeStreamMutation.getRowKey()); + Assert.assertEquals(actual.getType(), changeStreamMutation.getType()); + Assert.assertEquals(actual.getSourceClusterId(), changeStreamMutation.getSourceClusterId()); + Assert.assertEquals(actual.getCommitTimestamp(), changeStreamMutation.getCommitTimestamp()); + Assert.assertEquals(actual.getTieBreaker(), changeStreamMutation.getTieBreaker()); + Assert.assertEquals(actual.getToken(), changeStreamMutation.getToken()); + Assert.assertEquals(actual.getLowWatermark(), changeStreamMutation.getLowWatermark()); + assertThat(actual.toRowMutation(TABLE_ID).toProto(REQUEST_CONTEXT)) + .isEqualTo(changeStreamMutation.toRowMutation(TABLE_ID).toProto(REQUEST_CONTEXT)); } @Test