From 8b95d37dc104dc22ae6761e9cf0658cb6c19f9c7 Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Thu, 21 Jul 2022 14:03:20 -0400 Subject: [PATCH 1/3] feat: Add ReadChangeStreamQuery and ChangeStreamRecord::Heartbeat/CloseStream 1. ReadChangeStreamQuery will be used by readChangeStream(TODO) 2. ChangeStreamRecord is one of: Heartbeat, CloseStream, or a ChangeStreamMutation(TODO) --- .../models/ChangeStreamContinuationToken.java | 101 +++++ .../data/v2/models/ChangeStreamRecord.java | 25 ++ .../bigtable/data/v2/models/CloseStream.java | 103 +++++ .../bigtable/data/v2/models/Heartbeat.java | 80 ++++ .../data/v2/models/ReadChangeStreamQuery.java | 267 +++++++++++++ .../v2/models/ChangeStreamRecordTest.java | 165 ++++++++ .../v2/models/ReadChangeStreamQueryTest.java | 368 ++++++++++++++++++ 7 files changed, 1109 insertions(+) create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQuery.java create mode 100644 google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java create mode 100644 google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java new file mode 100644 index 000000000000..80fe627cc71d --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java @@ -0,0 +1,101 @@ +/* + * Copyright 2018 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.RowRange; +import com.google.bigtable.v2.StreamContinuationToken; +import com.google.common.base.MoreObjects; +import com.google.common.base.Objects; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import javax.annotation.Nonnull; + +/** A simple wrapper for {@link StreamContinuationToken}. */ +public final class ChangeStreamContinuationToken implements Serializable { + private static final long serialVersionUID = 524679926247095L; + + private transient StreamContinuationToken.Builder builder; + + private ChangeStreamContinuationToken(@Nonnull StreamContinuationToken.Builder builder) { + this.builder = builder; + } + + private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException { + input.defaultReadObject(); + builder = StreamContinuationToken.newBuilder().mergeFrom(input); + } + + private void writeObject(ObjectOutputStream output) throws IOException { + output.defaultWriteObject(); + builder.build().writeTo(output); + } + + public RowRange getRowRange() { + return this.builder.getPartition().getRowRange(); + } + + public String getToken() { + return this.builder.getToken(); + } + + /** + * Creates the protobuf. This method is considered an internal implementation detail and not meant + * to be used by applications. + */ + @InternalApi("Used in Changestream veneer client.") + public StreamContinuationToken toProto() { + return builder.build(); + } + + /** Wraps the protobuf {@link StreamContinuationToken}. */ + public static ChangeStreamContinuationToken fromProto( + @Nonnull StreamContinuationToken streamContinuationToken) { + return new ChangeStreamContinuationToken(streamContinuationToken.toBuilder()); + } + + public ChangeStreamContinuationToken clone() { + return new ChangeStreamContinuationToken(this.builder.clone()); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ChangeStreamContinuationToken otherToken = (ChangeStreamContinuationToken) o; + return Objects.equal(getRowRange(), otherToken.getRowRange()) + && Objects.equal(getToken(), otherToken.getToken()); + } + + @Override + public int hashCode() { + return Objects.hashCode(getRowRange(), getToken()); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("rowRange", getRowRange()) + .add("token", getToken()) + .toString(); + } +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java new file mode 100644 index 000000000000..0811fda08d71 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java @@ -0,0 +1,25 @@ +/* + * Copyright 2018 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 change stream record, which can be a Heartbeat, a CloseStream, or a + * logical mutation. + */ +@InternalExtensionOnly +public interface ChangeStreamRecord {} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java new file mode 100644 index 000000000000..38797e44a79e --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java @@ -0,0 +1,103 @@ +/* + * Copyright 2018 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; +import com.google.bigtable.v2.StreamContinuationToken; +import com.google.common.base.MoreObjects; +import com.google.common.base.Objects; +import com.google.common.collect.ImmutableList; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.List; +import javax.annotation.Nonnull; + +public final class CloseStream implements ChangeStreamRecord, Serializable { + private static final long serialVersionUID = 7316215828353608505L; + private final com.google.rpc.Status status; + private transient ImmutableList.Builder + changeStreamContinuationTokens = ImmutableList.builder(); + + private CloseStream( + com.google.rpc.Status status, List continuationTokens) { + this.status = status; + for (StreamContinuationToken streamContinuationToken : continuationTokens) { + changeStreamContinuationTokens.add( + ChangeStreamContinuationToken.fromProto(streamContinuationToken)); + } + } + + @InternalApi("Used in Changestream beam pipeline.") + public com.google.rpc.Status getStatus() { + return this.status; + } + + @InternalApi("Used in Changestream beam pipeline.") + public List getChangeStreamContinuationTokens() { + return changeStreamContinuationTokens.build(); + } + + private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException { + input.defaultReadObject(); + + @SuppressWarnings("unchecked") + ImmutableList deserialized = + (ImmutableList) input.readObject(); + this.changeStreamContinuationTokens = + ImmutableList.builder().addAll(deserialized); + } + + private void writeObject(ObjectOutputStream output) throws IOException { + output.defaultWriteObject(); + output.writeObject(changeStreamContinuationTokens.build()); + } + + /** Wraps the protobuf {@link ReadChangeStreamResponse.CloseStream}. */ + @InternalApi("Used in Changestream veneer client.") + public static CloseStream fromProto(@Nonnull ReadChangeStreamResponse.CloseStream closeStream) { + return new CloseStream(closeStream.getStatus(), closeStream.getContinuationTokensList()); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + CloseStream record = (CloseStream) o; + return Objects.equal(status, record.getStatus()) + && Objects.equal( + changeStreamContinuationTokens.build(), record.getChangeStreamContinuationTokens()); + } + + @Override + public int hashCode() { + return Objects.hashCode(status, changeStreamContinuationTokens); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("status", status) + .add("changeStreamContinuationTokens", changeStreamContinuationTokens) + .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 new file mode 100644 index 000000000000..5e49e7ee9418 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java @@ -0,0 +1,80 @@ +/* + * Copyright 2018 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; +import com.google.common.base.MoreObjects; +import com.google.common.base.Objects; +import java.io.Serializable; +import javax.annotation.Nonnull; + +public final class Heartbeat implements ChangeStreamRecord, Serializable { + private static final long serialVersionUID = 7316215828353608504L; + private final com.google.protobuf.Timestamp lowWatermark; + private final ChangeStreamContinuationToken changeStreamContinuationToken; + + private Heartbeat( + com.google.protobuf.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 com.google.protobuf.Timestamp getLowWatermark() { + return lowWatermark; + } + + /** Wraps the protobuf {@link ReadChangeStreamResponse.Heartbeat}. */ + @InternalApi("Used in Changestream veneer client.") + public static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) { + return new Heartbeat( + heartbeat.getLowWatermark(), + ChangeStreamContinuationToken.fromProto(heartbeat.getContinuationToken())); + } + + @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()); + } + + @Override + public int hashCode() { + return Objects.hashCode(lowWatermark, changeStreamContinuationToken); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("lowWatermark", lowWatermark) + .add("changeStreamContinuationToken", changeStreamContinuationToken) + .toString(); + } +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQuery.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQuery.java new file mode 100644 index 000000000000..6713f55d2224 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQuery.java @@ -0,0 +1,267 @@ +/* + * Copyright 2018 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.ReadChangeStreamRequest; +import com.google.bigtable.v2.RowRange; +import com.google.bigtable.v2.StreamContinuationTokens; +import com.google.bigtable.v2.StreamPartition; +import com.google.cloud.bigtable.data.v2.internal.NameUtil; +import com.google.cloud.bigtable.data.v2.internal.RequestContext; +import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange; +import com.google.common.base.MoreObjects; +import com.google.common.base.Objects; +import com.google.common.base.Preconditions; +import com.google.protobuf.ByteString; +import com.google.protobuf.Duration; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.List; +import javax.annotation.Nonnull; + +/** A simple wrapper to construct a query for the ReadChangeStream RPC. */ +public final class ReadChangeStreamQuery implements Serializable { + private static final long serialVersionUID = 948588515749969176L; + + private final String tableId; + private transient ReadChangeStreamRequest.Builder builder = ReadChangeStreamRequest.newBuilder(); + + /** + * Constructs a new ReadChangeStreamQuery object for the specified table id. The table id will be + * combined with the instance name specified in the {@link + * com.google.cloud.bigtable.data.v2.BigtableDataSettings}. + */ + public static ReadChangeStreamQuery create(String tableId) { + return new ReadChangeStreamQuery(tableId); + } + + private ReadChangeStreamQuery(String tableId) { + this.tableId = tableId; + } + + private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException { + input.defaultReadObject(); + builder = ReadChangeStreamRequest.newBuilder().mergeFrom(input); + } + + private void writeObject(ObjectOutputStream output) throws IOException { + output.defaultWriteObject(); + builder.build().writeTo(output); + } + + /** + * Adds a partition. + * + * @param rowRange Represents the partition in the form [startKey, endKey). startKey can be null + * to represent negative infinity. endKey can be null to represent positive infinity. + */ + public ReadChangeStreamQuery streamPartition(@Nonnull RowRange rowRange) { + builder.setPartition(StreamPartition.newBuilder().setRowRange(rowRange).build()); + return this; + } + + /** + * Adds a partition. + * + * @param start The beginning of the range (inclusive). Can be null to represent negative + * infinity. + * @param end The end of the range (exclusive). Can be null to represent positive infinity. + */ + public ReadChangeStreamQuery streamPartition(String start, String end) { + return streamPartition(wrapKey(start), wrapKey(end)); + } + + /** + * Adds a partition. + * + * @param start The beginning of the range (inclusive). Can be null to represent negative + * infinity. + * @param end The end of the range (exclusive). Can be null to represent positive infinity. + */ + public ReadChangeStreamQuery streamPartition(ByteString start, ByteString end) { + RowRange.Builder rangeBuilder = RowRange.newBuilder(); + if (start != null) { + rangeBuilder.setStartKeyClosed(start); + } + if (end != null) { + rangeBuilder.setEndKeyOpen(end); + } + return streamPartition(rangeBuilder.build()); + } + + /** Adds a partition. */ + public ReadChangeStreamQuery streamPartition(ByteStringRange range) { + RowRange.Builder rangeBuilder = RowRange.newBuilder(); + + switch (range.getStartBound()) { + case OPEN: + throw new IllegalStateException("Start bound should be closed."); + case CLOSED: + rangeBuilder.setStartKeyClosed(range.getStart()); + break; + case UNBOUNDED: + rangeBuilder.clearStartKey(); + break; + default: + throw new IllegalStateException("Unknown start bound: " + range.getStartBound()); + } + + switch (range.getEndBound()) { + case OPEN: + rangeBuilder.setEndKeyOpen(range.getEnd()); + break; + case CLOSED: + throw new IllegalStateException("End bound should be open."); + case UNBOUNDED: + rangeBuilder.clearEndKey(); + break; + default: + throw new IllegalStateException("Unknown end bound: " + range.getEndBound()); + } + + return streamPartition(rangeBuilder.build()); + } + + /** Sets the startTime to read the change stream. */ + public ReadChangeStreamQuery startTime(com.google.protobuf.Timestamp value) { + Preconditions.checkArgument( + !builder.hasContinuationTokens(), + "startTime and continuationTokens can't be specified together"); + builder.setStartTime(value); + return this; + } + + /** Sets the endTime to read the change stream. */ + public ReadChangeStreamQuery endTime(com.google.protobuf.Timestamp value) { + builder.setEndTime(value); + return this; + } + + /** Sets the stream continuation tokens to read the change stream. */ + public ReadChangeStreamQuery continuationTokens( + List changeStreamContinuationTokens) { + Preconditions.checkArgument( + !builder.hasStartTime(), "startTime and continuationTokens can't be specified together"); + StreamContinuationTokens.Builder streamContinuationTokensBuilder = + StreamContinuationTokens.newBuilder(); + for (ChangeStreamContinuationToken changeStreamContinuationToken : + changeStreamContinuationTokens) { + builder.setContinuationTokens( + streamContinuationTokensBuilder.addTokens(changeStreamContinuationToken.toProto())); + } + builder.setContinuationTokens(streamContinuationTokensBuilder.build()); + return this; + } + + /** Sets the heartbeat duration for the change stream. */ + public ReadChangeStreamQuery heartbeatDuration(long seconds) { + return heartbeatDuration(seconds, 0); + } + + /** Sets the heartbeat duration for the change stream. */ + public ReadChangeStreamQuery heartbeatDuration(long seconds, int nanos) { + builder.setHeartbeatDuration(Duration.newBuilder().setSeconds(seconds).setNanos(nanos).build()); + return this; + } + + /** + * Creates the request protobuf. This method is considered an internal implementation detail and + * not meant to be used by applications. + */ + @InternalApi("Used in Changestream beam pipeline.") + public ReadChangeStreamRequest toProto(RequestContext requestContext) { + String tableName = + NameUtil.formatTableName( + requestContext.getProjectId(), requestContext.getInstanceId(), tableId); + + return builder + .setTableName(tableName) + .setAppProfileId(requestContext.getAppProfileId()) + .build(); + } + + /** + * Wraps the protobuf {@link ReadChangeStreamRequest}. + * + *

WARNING: Please note that the project id & instance id in the table name will be overwritten + * by the configuration in the BigtableDataClient. + */ + public static ReadChangeStreamQuery fromProto(@Nonnull ReadChangeStreamRequest request) { + ReadChangeStreamQuery query = + new ReadChangeStreamQuery(NameUtil.extractTableIdFromTableName(request.getTableName())); + query.builder = request.toBuilder(); + + return query; + } + + public ReadChangeStreamQuery clone() { + ReadChangeStreamQuery query = ReadChangeStreamQuery.create(tableId); + query.builder = this.builder.clone(); + return query; + } + + private static ByteString wrapKey(String key) { + if (key == null) { + return null; + } + return ByteString.copyFromUtf8(key); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ReadChangeStreamQuery query = (ReadChangeStreamQuery) o; + return Objects.equal(tableId, query.tableId) + && Objects.equal(builder.getPartition(), query.builder.getPartition()) + && Objects.equal(builder.getStartTime(), query.builder.getStartTime()) + && Objects.equal(builder.getEndTime(), query.builder.getEndTime()) + && Objects.equal(builder.getContinuationTokens(), query.builder.getContinuationTokens()) + && Objects.equal(builder.getHeartbeatDuration(), query.builder.getHeartbeatDuration()); + } + + @Override + public int hashCode() { + return Objects.hashCode( + tableId, + builder.getPartition(), + builder.getStartTime(), + builder.getEndTime(), + builder.getContinuationTokens(), + builder.getHeartbeatDuration()); + } + + @Override + public String toString() { + ReadChangeStreamRequest request = builder.build(); + + return MoreObjects.toStringHelper(this) + .add("tableId", tableId) + .add("partition", request.getPartition()) + .add("startTime", request.getStartTime()) + .add("endTime", request.getEndTime()) + .add("continuationTokens", request.getContinuationTokens()) + .add("heartbeatDuration", request.getHeartbeatDuration()) + .toString(); + } +} 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 new file mode 100644 index 000000000000..be362dd03f14 --- /dev/null +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java @@ -0,0 +1,165 @@ +/* + * Copyright 2018 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.ReadChangeStreamResponse; +import com.google.bigtable.v2.RowRange; +import com.google.bigtable.v2.StreamContinuationToken; +import com.google.bigtable.v2.StreamPartition; +import com.google.protobuf.ByteString; +import com.google.protobuf.Timestamp; +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 ChangeStreamRecordTest { + + @Test + public void heartbeatSerializationTest() throws IOException, ClassNotFoundException { + ReadChangeStreamResponse.Heartbeat heartbeatProto = + ReadChangeStreamResponse.Heartbeat.newBuilder() + .setLowWatermark(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build()) + .setContinuationToken( + StreamContinuationToken.newBuilder().setToken("random-token").build()) + .build(); + Heartbeat heartbeat = Heartbeat.fromProto(heartbeatProto); + + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(bos); + oos.writeObject(heartbeat); + oos.close(); + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray())); + Heartbeat actual = (Heartbeat) ois.readObject(); + assertThat(actual).isEqualTo(heartbeat); + } + + @Test + public void closeStreamSerializationTest() throws IOException, ClassNotFoundException { + com.google.rpc.Status status = com.google.rpc.Status.newBuilder().setCode(0).build(); + RowRange rowRange1 = + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("")) + .setEndKeyOpen(ByteString.copyFromUtf8("apple")) + .build(); + String token1 = "close-stream-token-1"; + RowRange rowRange2 = + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("apple")) + .setEndKeyOpen(ByteString.copyFromUtf8("")) + .build(); + String token2 = "close-stream-token-2"; + ReadChangeStreamResponse.CloseStream closeStreamProto = + ReadChangeStreamResponse.CloseStream.newBuilder() + .addContinuationTokens( + StreamContinuationToken.newBuilder() + .setPartition(StreamPartition.newBuilder().setRowRange(rowRange1).build()) + .setToken(token1) + .build()) + .addContinuationTokens( + StreamContinuationToken.newBuilder() + .setPartition(StreamPartition.newBuilder().setRowRange(rowRange2).build()) + .setToken(token2) + .build()) + .setStatus(status) + .build(); + CloseStream closeStream = CloseStream.fromProto(closeStreamProto); + + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(bos); + oos.writeObject(closeStream); + oos.close(); + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray())); + CloseStream actual = (CloseStream) ois.readObject(); + assertThat(actual.getChangeStreamContinuationTokens()) + .isEqualTo(closeStream.getChangeStreamContinuationTokens()); + } + + @Test + public void heartbeatTest() { + Timestamp lowWatermark = Timestamp.newBuilder().setSeconds(1000).build(); + RowRange rowRange = + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("apple")) + .setEndKeyOpen(ByteString.copyFromUtf8("banana")) + .build(); + String token = "heartbeat-token"; + ReadChangeStreamResponse.Heartbeat heartbeatProto = + ReadChangeStreamResponse.Heartbeat.newBuilder() + .setLowWatermark(lowWatermark) + .setContinuationToken( + StreamContinuationToken.newBuilder() + .setPartition(StreamPartition.newBuilder().setRowRange(rowRange).build()) + .setToken(token) + .build()) + .build(); + Heartbeat actualHeartbeat = Heartbeat.fromProto(heartbeatProto); + + Assert.assertEquals(actualHeartbeat.getLowWatermark(), lowWatermark); + Assert.assertEquals(actualHeartbeat.getChangeStreamContinuationToken().getRowRange(), rowRange); + Assert.assertEquals(actualHeartbeat.getChangeStreamContinuationToken().getToken(), token); + } + + @Test + public void closeStreamTest() { + com.google.rpc.Status status = com.google.rpc.Status.newBuilder().setCode(0).build(); + RowRange rowRange1 = + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("")) + .setEndKeyOpen(ByteString.copyFromUtf8("apple")) + .build(); + String token1 = "close-stream-token-1"; + RowRange rowRange2 = + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("apple")) + .setEndKeyOpen(ByteString.copyFromUtf8("")) + .build(); + String token2 = "close-stream-token-2"; + ReadChangeStreamResponse.CloseStream closeStreamProto = + ReadChangeStreamResponse.CloseStream.newBuilder() + .addContinuationTokens( + StreamContinuationToken.newBuilder() + .setPartition(StreamPartition.newBuilder().setRowRange(rowRange1).build()) + .setToken(token1) + .build()) + .addContinuationTokens( + StreamContinuationToken.newBuilder() + .setPartition(StreamPartition.newBuilder().setRowRange(rowRange2).build()) + .setToken(token2) + .build()) + .setStatus(status) + .build(); + CloseStream actualCloseStream = CloseStream.fromProto(closeStreamProto); + + Assert.assertEquals(status, actualCloseStream.getStatus()); + Assert.assertEquals( + rowRange1, actualCloseStream.getChangeStreamContinuationTokens().get(0).getRowRange()); + Assert.assertEquals( + token1, actualCloseStream.getChangeStreamContinuationTokens().get(0).getToken()); + Assert.assertEquals( + rowRange2, actualCloseStream.getChangeStreamContinuationTokens().get(1).getRowRange()); + Assert.assertEquals( + token2, actualCloseStream.getChangeStreamContinuationTokens().get(1).getToken()); + } +} diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java new file mode 100644 index 000000000000..33a23c0387da --- /dev/null +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java @@ -0,0 +1,368 @@ +/* + * Copyright 2018 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.ReadChangeStreamRequest; +import com.google.bigtable.v2.ReadChangeStreamRequest.Builder; +import com.google.bigtable.v2.RowRange; +import com.google.bigtable.v2.StreamContinuationToken; +import com.google.bigtable.v2.StreamContinuationTokens; +import com.google.bigtable.v2.StreamPartition; +import com.google.cloud.bigtable.data.v2.internal.NameUtil; +import com.google.cloud.bigtable.data.v2.internal.RequestContext; +import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange; +import com.google.protobuf.ByteString; +import com.google.protobuf.Duration; +import com.google.protobuf.Timestamp; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.Collections; +import org.junit.Before; +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 ReadChangeStreamQueryTest { + 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-id"; + private RequestContext requestContext; + + @Rule public ExpectedException expect = ExpectedException.none(); + + @Before + public void setUp() { + requestContext = RequestContext.create(PROJECT_ID, INSTANCE_ID, APP_PROFILE_ID); + } + + @Test + public void requestContextTest() { + ReadChangeStreamQuery query = ReadChangeStreamQuery.create(TABLE_ID); + + ReadChangeStreamRequest proto = query.toProto(requestContext); + assertThat(proto).isEqualTo(expectedProtoBuilder().build()); + } + + @Test + public void streamPartitionTest() { + // Case 1: String. + ReadChangeStreamQuery query1 = + ReadChangeStreamQuery.create(TABLE_ID).streamPartition("simple-begin", "simple-end"); + ReadChangeStreamRequest actualProto1 = query1.toProto(requestContext); + Builder expectedProto1 = expectedProtoBuilder(); + expectedProto1.setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("simple-begin")) + .setEndKeyOpen(ByteString.copyFromUtf8("simple-end")) + .build()) + .build()); + assertThat(actualProto1).isEqualTo(expectedProto1.build()); + + // Case 2: ByteString. + ReadChangeStreamQuery query2 = + ReadChangeStreamQuery.create(TABLE_ID) + .streamPartition( + ByteString.copyFromUtf8("byte-begin"), ByteString.copyFromUtf8("byte-end")); + ReadChangeStreamRequest actualProto2 = query2.toProto(requestContext); + Builder expectedProto2 = expectedProtoBuilder(); + expectedProto2.setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("byte-begin")) + .setEndKeyOpen(ByteString.copyFromUtf8("byte-end")) + .build()) + .build()); + assertThat(actualProto2).isEqualTo(expectedProto2.build()); + + // Case 3: ByteStringRange. + ReadChangeStreamQuery query3 = + ReadChangeStreamQuery.create(TABLE_ID) + .streamPartition(ByteStringRange.create("range-begin", "range-end")); + ReadChangeStreamRequest actualProto3 = query3.toProto(requestContext); + Builder expectedProto3 = expectedProtoBuilder(); + expectedProto3.setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("range-begin")) + .setEndKeyOpen(ByteString.copyFromUtf8("range-end")) + .build()) + .build()); + assertThat(actualProto3).isEqualTo(expectedProto3.build()); + } + + @Test + public void startTimeTest() { + ReadChangeStreamQuery query = + ReadChangeStreamQuery.create(TABLE_ID) + .startTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build()); + + Builder expectedProto = + expectedProtoBuilder() + .setStartTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build()); + + ReadChangeStreamRequest actualProto = query.toProto(requestContext); + assertThat(actualProto).isEqualTo(expectedProto.build()); + } + + @Test + public void endTimeTest() { + ReadChangeStreamQuery query = + ReadChangeStreamQuery.create(TABLE_ID) + .endTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build()); + + Builder expectedProto = + expectedProtoBuilder() + .setEndTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build()); + + ReadChangeStreamRequest actualProto = query.toProto(requestContext); + assertThat(actualProto).isEqualTo(expectedProto.build()); + } + + @Test + public void heartbeatDurationTest() { + ReadChangeStreamQuery query = ReadChangeStreamQuery.create(TABLE_ID).heartbeatDuration(5); + + Builder expectedProto = + expectedProtoBuilder() + .setHeartbeatDuration(com.google.protobuf.Duration.newBuilder().setSeconds(5).build()); + + ReadChangeStreamRequest actualProto = query.toProto(requestContext); + assertThat(actualProto).isEqualTo(expectedProto.build()); + } + + @Test + public void continuationTokensTest() { + StreamContinuationToken tokenProto = + StreamContinuationToken.newBuilder() + .setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("start")) + .setEndKeyOpen(ByteString.copyFromUtf8("end")) + .build()) + .build()) + .setToken("random-token") + .build(); + ChangeStreamContinuationToken token = ChangeStreamContinuationToken.fromProto(tokenProto); + ReadChangeStreamQuery query = + ReadChangeStreamQuery.create(TABLE_ID).continuationTokens(Collections.singletonList(token)); + + Builder expectedProto = + expectedProtoBuilder() + .setContinuationTokens( + StreamContinuationTokens.newBuilder().addTokens(tokenProto).build()); + + ReadChangeStreamRequest actualProto = query.toProto(requestContext); + assertThat(actualProto).isEqualTo(expectedProto.build()); + } + + @Test(expected = IllegalArgumentException.class) + public void createWithStartTimeAndContinuationTokensTest() { + StreamContinuationToken tokenProto = + StreamContinuationToken.newBuilder() + .setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("start")) + .setEndKeyOpen(ByteString.copyFromUtf8("end")) + .build()) + .build()) + .setToken("random-token") + .build(); + ChangeStreamContinuationToken token = ChangeStreamContinuationToken.fromProto(tokenProto); + ReadChangeStreamQuery query = + ReadChangeStreamQuery.create(TABLE_ID) + .startTime(Timestamp.newBuilder().setSeconds(5).build()) + .continuationTokens(Collections.singletonList(token)); + expect.expect(IllegalArgumentException.class); + expect.expectMessage("startTime and continuationTokens can't be specified together"); + } + + @Test + public void serializationTest() throws IOException, ClassNotFoundException { + StreamContinuationToken tokenProto = + StreamContinuationToken.newBuilder() + .setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("start")) + .setEndKeyOpen(ByteString.copyFromUtf8("end")) + .build()) + .build()) + .setToken("random-token") + .build(); + ChangeStreamContinuationToken token = ChangeStreamContinuationToken.fromProto(tokenProto); + ReadChangeStreamQuery expected = + ReadChangeStreamQuery.create(TABLE_ID) + .streamPartition("simple-begin", "simple-end") + .continuationTokens(Collections.singletonList(token)) + .endTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build()) + .heartbeatDuration(5); + + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(bos); + oos.writeObject(expected); + oos.close(); + + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray())); + + ReadChangeStreamQuery actual = (ReadChangeStreamQuery) ois.readObject(); + assertThat(actual.toProto(requestContext)).isEqualTo(expected.toProto(requestContext)); + } + + private static ReadChangeStreamRequest.Builder expectedProtoBuilder() { + return ReadChangeStreamRequest.newBuilder() + .setTableName(NameUtil.formatTableName(PROJECT_ID, INSTANCE_ID, TABLE_ID)) + .setAppProfileId(APP_PROFILE_ID); + } + + @Test + public void testFromProto() { + StreamContinuationToken token = + StreamContinuationToken.newBuilder() + .setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("start")) + .setEndKeyOpen(ByteString.copyFromUtf8("end")) + .build()) + .build()) + .setToken("random-token") + .build(); + ReadChangeStreamRequest request = + ReadChangeStreamRequest.newBuilder() + .setTableName(NameUtil.formatTableName(PROJECT_ID, INSTANCE_ID, TABLE_ID)) + .setAppProfileId(APP_PROFILE_ID) + .setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("")) + .setEndKeyClosed(ByteString.copyFromUtf8("")) + .build())) + .setContinuationTokens(StreamContinuationTokens.newBuilder().addTokens(token).build()) + .setEndTime(Timestamp.newBuilder().setSeconds(2000).build()) + .setHeartbeatDuration(Duration.newBuilder().setSeconds(5).build()) + .build(); + ReadChangeStreamQuery query = ReadChangeStreamQuery.fromProto(request); + assertThat(query.toProto(requestContext)).isEqualTo(request); + } + + @Test(expected = IllegalArgumentException.class) + public void testFromProtoWithEmptyTableId() { + ReadChangeStreamQuery.fromProto(ReadChangeStreamRequest.getDefaultInstance()); + + expect.expect(IllegalArgumentException.class); + expect.expectMessage("Invalid table name:"); + } + + @Test + public void testEquality() { + ReadChangeStreamQuery request = + ReadChangeStreamQuery.create(TABLE_ID) + .streamPartition("simple-begin", "simple-end") + .startTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build()) + .endTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build()) + .heartbeatDuration(5); + + // ReadChangeStreamQuery#toProto should not change the ReadChangeStreamQuery instance state + request.toProto(requestContext); + assertThat(request) + .isEqualTo( + ReadChangeStreamQuery.create(TABLE_ID) + .streamPartition("simple-begin", "simple-end") + .startTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build()) + .endTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(2000).build()) + .heartbeatDuration(5)); + + assertThat(ReadChangeStreamQuery.create(TABLE_ID).streamPartition("begin-1", "end-1")) + .isNotEqualTo(ReadChangeStreamQuery.create(TABLE_ID).streamPartition("begin-2", "end-1")); + assertThat( + ReadChangeStreamQuery.create(TABLE_ID) + .startTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build())) + .isNotEqualTo( + ReadChangeStreamQuery.create(TABLE_ID) + .startTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1001).build())); + assertThat( + ReadChangeStreamQuery.create(TABLE_ID) + .endTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1000).build())) + .isNotEqualTo( + ReadChangeStreamQuery.create(TABLE_ID) + .endTime(com.google.protobuf.Timestamp.newBuilder().setSeconds(1001).build())); + assertThat(ReadChangeStreamQuery.create(TABLE_ID).heartbeatDuration(5)) + .isNotEqualTo(ReadChangeStreamQuery.create(TABLE_ID).heartbeatDuration(6)); + } + + @Test + public void testClone() { + StreamContinuationToken tokenProto = + StreamContinuationToken.newBuilder() + .setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("start")) + .setEndKeyOpen(ByteString.copyFromUtf8("end")) + .build()) + .build()) + .setToken("random-token") + .build(); + ChangeStreamContinuationToken token = ChangeStreamContinuationToken.fromProto(tokenProto); + ReadChangeStreamQuery query = + ReadChangeStreamQuery.create(TABLE_ID) + .streamPartition("begin", "end") + .continuationTokens(Collections.singletonList(token)) + .endTime(Timestamp.newBuilder().setSeconds(2000).build()) + .heartbeatDuration(5); + ReadChangeStreamRequest request = + ReadChangeStreamRequest.newBuilder() + .setTableName(NameUtil.formatTableName(PROJECT_ID, INSTANCE_ID, TABLE_ID)) + .setAppProfileId(APP_PROFILE_ID) + .setPartition( + StreamPartition.newBuilder() + .setRowRange( + RowRange.newBuilder() + .setStartKeyClosed(ByteString.copyFromUtf8("begin")) + .setEndKeyOpen(ByteString.copyFromUtf8("end")) + .build())) + .setContinuationTokens( + StreamContinuationTokens.newBuilder().addTokens(tokenProto).build()) + .setEndTime(Timestamp.newBuilder().setSeconds(2000).build()) + .setHeartbeatDuration(Duration.newBuilder().setSeconds(5).build()) + .build(); + + ReadChangeStreamQuery clonedReq = query.clone(); + assertThat(clonedReq).isEqualTo(query); + assertThat(clonedReq.toProto(requestContext)).isEqualTo(request); + } +} From e03c06bfc2e1aef00193bb975be00952097f93dc Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Fri, 22 Jul 2022 13:44:05 -0400 Subject: [PATCH 2/3] fix: Address comments about styles --- .../v2/models/ChangeStreamContinuationToken.java | 7 ++++--- .../bigtable/data/v2/models/ChangeStreamRecord.java | 2 +- .../cloud/bigtable/data/v2/models/CloseStream.java | 12 ++++++------ .../cloud/bigtable/data/v2/models/Heartbeat.java | 12 ++++++------ .../data/v2/models/ReadChangeStreamQuery.java | 7 ++++--- .../data/v2/models/ChangeStreamRecordTest.java | 2 +- .../data/v2/models/ReadChangeStreamQueryTest.java | 2 +- 7 files changed, 23 insertions(+), 21 deletions(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java index 80fe627cc71d..5c86662f1dae 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 Google LLC + * 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. @@ -59,12 +59,13 @@ public String getToken() { * to be used by applications. */ @InternalApi("Used in Changestream veneer client.") - public StreamContinuationToken toProto() { + StreamContinuationToken toProto() { return builder.build(); } /** Wraps the protobuf {@link StreamContinuationToken}. */ - public static ChangeStreamContinuationToken fromProto( + @InternalApi("Used in Changestream veneer client.") + static ChangeStreamContinuationToken fromProto( @Nonnull StreamContinuationToken streamContinuationToken) { return new ChangeStreamContinuationToken(streamContinuationToken.toBuilder()); } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java index 0811fda08d71..0bf5e0c31eef 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 Google LLC + * 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. diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java index 38797e44a79e..f49c560c6d15 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 Google LLC + * 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. @@ -21,6 +21,7 @@ import com.google.common.base.MoreObjects; import com.google.common.base.Objects; import com.google.common.collect.ImmutableList; +import com.google.rpc.Status; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; @@ -30,12 +31,11 @@ public final class CloseStream implements ChangeStreamRecord, Serializable { private static final long serialVersionUID = 7316215828353608505L; - private final com.google.rpc.Status status; + private final Status status; private transient ImmutableList.Builder changeStreamContinuationTokens = ImmutableList.builder(); - private CloseStream( - com.google.rpc.Status status, List continuationTokens) { + private CloseStream(Status status, List continuationTokens) { this.status = status; for (StreamContinuationToken streamContinuationToken : continuationTokens) { changeStreamContinuationTokens.add( @@ -44,7 +44,7 @@ private CloseStream( } @InternalApi("Used in Changestream beam pipeline.") - public com.google.rpc.Status getStatus() { + public Status getStatus() { return this.status; } @@ -70,7 +70,7 @@ private void writeObject(ObjectOutputStream output) throws IOException { /** Wraps the protobuf {@link ReadChangeStreamResponse.CloseStream}. */ @InternalApi("Used in Changestream veneer client.") - public static CloseStream fromProto(@Nonnull ReadChangeStreamResponse.CloseStream closeStream) { + static CloseStream fromProto(@Nonnull ReadChangeStreamResponse.CloseStream closeStream) { return new CloseStream(closeStream.getStatus(), closeStream.getContinuationTokensList()); } 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 5e49e7ee9418..8ce08e7a6017 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018 Google LLC + * 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. @@ -19,17 +19,17 @@ 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 { private static final long serialVersionUID = 7316215828353608504L; - private final com.google.protobuf.Timestamp lowWatermark; + private final Timestamp lowWatermark; private final ChangeStreamContinuationToken changeStreamContinuationToken; private Heartbeat( - com.google.protobuf.Timestamp lowWatermark, - ChangeStreamContinuationToken changeStreamContinuationToken) { + Timestamp lowWatermark, ChangeStreamContinuationToken changeStreamContinuationToken) { this.lowWatermark = lowWatermark; this.changeStreamContinuationToken = changeStreamContinuationToken; } @@ -40,13 +40,13 @@ public ChangeStreamContinuationToken getChangeStreamContinuationToken() { } @InternalApi("Used in Changestream beam pipeline.") - public com.google.protobuf.Timestamp getLowWatermark() { + public Timestamp getLowWatermark() { return lowWatermark; } /** Wraps the protobuf {@link ReadChangeStreamResponse.Heartbeat}. */ @InternalApi("Used in Changestream veneer client.") - public static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) { + static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) { return new Heartbeat( heartbeat.getLowWatermark(), ChangeStreamContinuationToken.fromProto(heartbeat.getContinuationToken())); diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQuery.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQuery.java index 6713f55d2224..5ac3a743f6ad 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQuery.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQuery.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 Google LLC + * 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. @@ -28,6 +28,7 @@ import com.google.common.base.Preconditions; import com.google.protobuf.ByteString; import com.google.protobuf.Duration; +import com.google.protobuf.Timestamp; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; @@ -139,7 +140,7 @@ public ReadChangeStreamQuery streamPartition(ByteStringRange range) { } /** Sets the startTime to read the change stream. */ - public ReadChangeStreamQuery startTime(com.google.protobuf.Timestamp value) { + public ReadChangeStreamQuery startTime(Timestamp value) { Preconditions.checkArgument( !builder.hasContinuationTokens(), "startTime and continuationTokens can't be specified together"); @@ -148,7 +149,7 @@ public ReadChangeStreamQuery startTime(com.google.protobuf.Timestamp value) { } /** Sets the endTime to read the change stream. */ - public ReadChangeStreamQuery endTime(com.google.protobuf.Timestamp value) { + public ReadChangeStreamQuery endTime(Timestamp value) { builder.setEndTime(value); return this; } 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 be362dd03f14..cbf2d5600f81 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018 Google LLC + * 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. diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java index 33a23c0387da..5dc0b95df182 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 Google LLC + * 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. From 652d7288851495c7c1e7a80edc83a11bae1628a8 Mon Sep 17 00:00:00 2001 From: Teng Zhong Date: Tue, 26 Jul 2022 13:00:59 -0400 Subject: [PATCH 3/3] fix: Remove `InternalApi` tag for package private methods in veneer client --- .../data/v2/models/ChangeStreamContinuationToken.java | 3 --- .../com/google/cloud/bigtable/data/v2/models/CloseStream.java | 1 - .../com/google/cloud/bigtable/data/v2/models/Heartbeat.java | 1 - .../cloud/bigtable/data/v2/models/ChangeStreamRecordTest.java | 1 + .../bigtable/data/v2/models/ReadChangeStreamQueryTest.java | 4 ++-- 5 files changed, 3 insertions(+), 7 deletions(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java index 5c86662f1dae..f499a94e453f 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java @@ -15,7 +15,6 @@ */ package com.google.cloud.bigtable.data.v2.models; -import com.google.api.core.InternalApi; import com.google.bigtable.v2.RowRange; import com.google.bigtable.v2.StreamContinuationToken; import com.google.common.base.MoreObjects; @@ -58,13 +57,11 @@ public String getToken() { * Creates the protobuf. This method is considered an internal implementation detail and not meant * to be used by applications. */ - @InternalApi("Used in Changestream veneer client.") StreamContinuationToken toProto() { return builder.build(); } /** Wraps the protobuf {@link StreamContinuationToken}. */ - @InternalApi("Used in Changestream veneer client.") static ChangeStreamContinuationToken fromProto( @Nonnull StreamContinuationToken streamContinuationToken) { return new ChangeStreamContinuationToken(streamContinuationToken.toBuilder()); diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java index f49c560c6d15..403705f67617 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java @@ -69,7 +69,6 @@ private void writeObject(ObjectOutputStream output) throws IOException { } /** Wraps the protobuf {@link ReadChangeStreamResponse.CloseStream}. */ - @InternalApi("Used in Changestream veneer client.") static CloseStream fromProto(@Nonnull ReadChangeStreamResponse.CloseStream closeStream) { return new CloseStream(closeStream.getStatus(), closeStream.getContinuationTokensList()); } 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 8ce08e7a6017..73876f887b70 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 @@ -45,7 +45,6 @@ public Timestamp getLowWatermark() { } /** Wraps the protobuf {@link ReadChangeStreamResponse.Heartbeat}. */ - @InternalApi("Used in Changestream veneer client.") static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) { return new Heartbeat( heartbeat.getLowWatermark(), 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 cbf2d5600f81..c82aae733088 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 @@ -94,6 +94,7 @@ public void closeStreamSerializationTest() throws IOException, ClassNotFoundExce CloseStream actual = (CloseStream) ois.readObject(); assertThat(actual.getChangeStreamContinuationTokens()) .isEqualTo(closeStream.getChangeStreamContinuationTokens()); + assertThat(actual.getStatus()).isEqualTo(closeStream.getStatus()); } @Test diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java index 5dc0b95df182..cae2d93926fd 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/ReadChangeStreamQueryTest.java @@ -253,8 +253,8 @@ public void testFromProto() { StreamPartition.newBuilder() .setRowRange( RowRange.newBuilder() - .setStartKeyClosed(ByteString.copyFromUtf8("start")) - .setEndKeyOpen(ByteString.copyFromUtf8("end")) + .setStartKeyClosed(ByteString.copyFromUtf8("")) + .setEndKeyOpen(ByteString.copyFromUtf8("")) .build()) .build()) .setToken("random-token")