This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 42129fd3154 Upgrade GCP Libraries BOM to 26.87.0 (#39887)
42129fd3154 is described below
commit 42129fd31544512d64301d683f57e58bf58ba6cc
Author: Lawrence Qiu <[email protected]>
AuthorDate: Wed Sep 2 12:17:53 2026 +0000
Upgrade GCP Libraries BOM to 26.87.0 (#39887)
* Upgrade GCP Libraries BOM to 26.87.0
- Upgrades com.google.cloud:libraries-bom to 26.87.0
- Updates synchronized versions: gax (2.84.0), grpc (1.82.4), netty
(4.1.133.Final), protobuf (4.33.6), google-api-services-bigquery
(v2-rev20260731-2.0.0), google-cloud-datastore (3.4.0)
- Updates dep_urls_java.yaml license entry
- Aligns Bigtable change streams cell mutation timestamps to milliseconds
to comply with Bigtable timestamp granularity requirements
* chore: Resolve spotless complaint
* test: Align BigtableChangeStreamIT mutation timestamps to milliseconds
---
.../org/apache/beam/gradle/BeamModulePlugin.groovy | 14 ++++-----
.../beam/it/gcp/pubsub/PubsubResourceManager.java | 6 +++-
.../container/license_scripts/dep_urls_java.yaml | 2 +-
.../changestreams/dao/MetadataTableDao.java | 26 +++++++++++++++--
.../changestreams/dao/MetadataTableDaoTest.java | 2 ++
.../changestreams/dofn/InitializeDoFnTest.java | 30 ++++++++++++++++----
.../changestreams/it/BigtableChangeStreamIT.java | 33 ++++++++++++++--------
7 files changed, 85 insertions(+), 28 deletions(-)
diff --git
a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index d0319175a39..da54354f01b 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -620,14 +620,14 @@ class BeamModulePlugin implements Plugin<Project> {
def dbcp2_version = "2.9.0"
def errorprone_version = "2.31.0"
// [bomupgrader] determined by: com.google.api:gax, consistent with:
google_cloud_platform_libraries_bom
- def gax_version = "2.82.0"
+ def gax_version = "2.84.0"
def google_ads_version = "33.0.0"
def google_clients_version = "2.0.0"
def google_cloud_bigdataoss_version = "3.1.16"
def google_code_gson_version = "2.10.1"
def google_oauth_clients_version = "1.34.1"
// [bomupgrader] determined by: io.grpc:grpc-netty, consistent with:
google_cloud_platform_libraries_bom
- def grpc_version = "1.81.0"
+ def grpc_version = "1.82.4"
def guava_version = "33.1.0-jre"
def hadoop_version = "3.4.2"
def hamcrest_version = "2.1"
@@ -643,13 +643,13 @@ class BeamModulePlugin implements Plugin<Project> {
def log4j2_version = "2.25.4"
def nemo_version = "0.1"
// [bomupgrader] determined by: io.grpc:grpc-netty, consistent with:
google_cloud_platform_libraries_bom
- def netty_version = "4.1.132.Final"
+ def netty_version = "4.1.133.Final"
// [bomupgrader] determined by: io.opentelemetry:opentelemetry-sdk,
consistent with: google_cloud_platform_libraries_bom
def opentelemetry_version = "1.62.0"
def opentelemetry_contrib_version = "1.52.0"
def postgres_version = "42.6.2"
// [bomupgrader] determined by: com.google.protobuf:protobuf-java,
consistent with: google_cloud_platform_libraries_bom
- def protobuf_version = "4.33.2"
+ def protobuf_version = "4.33.6"
def qpid_jms_client_version = "0.61.0"
def quickcheck_version = "1.0"
def sbe_tool_version = "1.25.1"
@@ -753,7 +753,7 @@ class BeamModulePlugin implements Plugin<Project> {
google_api_client_gson :
"com.google.api-client:google-api-client-gson:$google_clients_version",
google_api_client_java6 :
"com.google.api-client:google-api-client-java6:$google_clients_version",
google_api_common :
"com.google.api:api-common", // google_cloud_platform_libraries_bom sets version
- google_api_services_bigquery :
"com.google.apis:google-api-services-bigquery:v2-rev20260612-2.0.0", //
[bomupgrader] sets version
+ google_api_services_bigquery :
"com.google.apis:google-api-services-bigquery:v2-rev20260731-2.0.0", //
[bomupgrader] sets version
google_api_services_cloudresourcemanager :
"com.google.apis:google-api-services-cloudresourcemanager:v1-rev20250606-2.0.0",
// [bomupgrader] sets version
google_api_services_dataflow :
"com.google.apis:google-api-services-dataflow:v1b3-rev20260503-$google_clients_version",
google_api_services_healthcare :
"com.google.apis:google-api-services-healthcare:v1-rev20240130-$google_clients_version",
@@ -770,14 +770,14 @@ class BeamModulePlugin implements Plugin<Project> {
google_cloud_core_grpc :
"com.google.cloud:google-cloud-core-grpc", //
google_cloud_platform_libraries_bom sets version
google_cloud_datacatalog_v1beta1 :
"com.google.cloud:google-cloud-datacatalog", //
google_cloud_platform_libraries_bom sets version
google_cloud_dataflow_java_proto_library_all:
"com.google.cloud.dataflow:google-cloud-dataflow-java-proto-library-all:0.5.160304",
- google_cloud_datastore_v1_proto_client :
"com.google.cloud.datastore:datastore-v1-proto-client:3.2.0", //
[bomupgrader] sets version
+ google_cloud_datastore_v1_proto_client :
"com.google.cloud.datastore:datastore-v1-proto-client:3.4.0", //
[bomupgrader] sets version
google_cloud_firestore :
"com.google.cloud:google-cloud-firestore", //
google_cloud_platform_libraries_bom sets version
google_cloud_kms :
"com.google.cloud:google-cloud-kms", // google_cloud_platform_libraries_bom
sets version
google_cloud_logging :
"com.google.cloud:google-cloud-logging", // google_cloud_platform_libraries_bom
sets version
google_cloud_pubsub :
"com.google.cloud:google-cloud-pubsub", // google_cloud_platform_libraries_bom
sets version
// [bomupgrader] the BOM version is set by
scripts/tools/bomupgrader.py. If update manually, also update
// libraries-bom version on
sdks/java/container/license_scripts/dep_urls_java.yaml
- google_cloud_platform_libraries_bom :
"com.google.cloud:libraries-bom:26.85.0",
+ google_cloud_platform_libraries_bom :
"com.google.cloud:libraries-bom:26.87.0",
google_cloud_secret_manager :
"com.google.cloud:google-cloud-secretmanager", //
google_cloud_platform_libraries_bom sets version
google_cloud_spanner :
"com.google.cloud:google-cloud-spanner", // google_cloud_platform_libraries_bom
sets version
google_cloud_storage :
"com.google.cloud:google-cloud-storage", // google_cloud_platform_libraries_bom
sets version
diff --git
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/pubsub/PubsubResourceManager.java
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/pubsub/PubsubResourceManager.java
index f947d70efd9..e38251431c5 100644
---
a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/pubsub/PubsubResourceManager.java
+++
b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/pubsub/PubsubResourceManager.java
@@ -283,7 +283,11 @@ public final class PubsubResourceManager implements
ResourceManager {
ProjectName.newBuilder().setProject(projectId).build(),
Schema.newBuilder().setType(schemaType).setDefinition(schemaDefinition).build(),
"schema-" + testId + "-" + schemaTopic.getTopic());
- createdSchemas.add(SchemaName.parse(schema.getName()));
+ SchemaName schemaName = SchemaName.parse(schema.getName());
+ if (schemaName == null) {
+ throw new IllegalStateException("Failed to parse schema name: " +
schema.getName());
+ }
+ createdSchemas.add(schemaName);
topicAdminClient.updateTopic(
UpdateTopicRequest.newBuilder()
.setUpdateMask(FieldMask.newBuilder().addPaths("schema_settings"))
diff --git a/sdks/java/container/license_scripts/dep_urls_java.yaml
b/sdks/java/container/license_scripts/dep_urls_java.yaml
index 25e438f7854..df64ca0cafc 100644
--- a/sdks/java/container/license_scripts/dep_urls_java.yaml
+++ b/sdks/java/container/license_scripts/dep_urls_java.yaml
@@ -46,7 +46,7 @@ jaxen:
'1.1.6':
type: "3-Clause BSD"
libraries-bom:
- '26.85.0':
+ '26.87.0':
license:
"https://raw.githubusercontent.com/GoogleCloudPlatform/cloud-opensource-java/master/LICENSE";
type: "Apache License 2.0"
paranamer:
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java
index 6a5ac4354ca..83115297764 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDao.java
@@ -308,6 +308,7 @@ public class MetadataTableDao {
* @param newPartition the new partition
*/
public void writeNewPartition(NewPartition newPartition) {
+ long nowMicros = Instant.now().getMillis() * 1000L;
ByteString rowKey =
convertPartitionToNewPartitionRowKey(newPartition.getPartition());
ByteStringRange parentPartition =
newPartition.getChangeStreamContinuationTokens().get(0).getPartition();
@@ -316,10 +317,12 @@ public class MetadataTableDao {
.setCell(
MetadataTableAdminDao.CF_INITIAL_TOKEN,
ByteStringRange.serializeToByteString(parentPartition),
+ nowMicros,
newPartition.getChangeStreamContinuationTokens().get(0).toByteString())
.setCell(
MetadataTableAdminDao.CF_PARENT_LOW_WATERMARKS,
ByteStringRange.serializeToByteString(parentPartition),
+ nowMicros,
newPartition.getLowWatermark().getMillis())
.deleteCells(
MetadataTableAdminDao.CF_SHOULD_DELETE,
@@ -349,13 +352,15 @@ public class MetadataTableDao {
* @param newPartition mark for deletion.
*/
public void markNewPartitionForDeletion(NewPartition newPartition) {
+ long nowMicros = Instant.now().getMillis() * 1000L;
ByteString rowKey =
convertPartitionToNewPartitionRowKey(newPartition.getPartition());
RowMutation rowMutation = RowMutation.create(tableId, rowKey);
for (ChangeStreamContinuationToken token :
newPartition.getChangeStreamContinuationTokens()) {
rowMutation.setCell(
MetadataTableAdminDao.CF_SHOULD_DELETE,
ByteStringRange.serializeToByteString(token.getPartition()),
- 1);
+ nowMicros,
+ 1L);
}
mutateRowWithHardTimeout(rowMutation);
}
@@ -503,16 +508,19 @@ public class MetadataTableDao {
*/
private void writeToMdTableWatermarkHelper(
ByteString rowKey, Instant watermark, @Nullable
ChangeStreamContinuationToken currentToken) {
+ long nowMicros = Instant.now().getMillis() * 1000L;
RowMutation rowMutation =
RowMutation.create(tableId, rowKey)
.setCell(
MetadataTableAdminDao.CF_WATERMARK,
MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
watermark.getMillis());
if (currentToken != null) {
rowMutation.setCell(
MetadataTableAdminDao.CF_CONTINUATION_TOKEN,
MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
currentToken.getToken());
}
mutateRowWithHardTimeout(rowMutation);
@@ -591,6 +599,7 @@ public class MetadataTableDao {
* @return true if releasing the lock was successful.
*/
public boolean releaseStreamPartitionLockForDeletion(ByteStringRange
partition, String uuid) {
+ long nowMicros = Instant.now().getMillis() * 1000L;
ByteString rowKey = convertPartitionToStreamPartitionRowKey(partition);
Filter lockCellFilter =
FILTERS
@@ -602,7 +611,10 @@ public class MetadataTableDao {
Mutation.create()
.deleteCells(MetadataTableAdminDao.CF_LOCK,
MetadataTableAdminDao.QUALIFIER_DEFAULT)
.setCell(
- MetadataTableAdminDao.CF_SHOULD_DELETE,
MetadataTableAdminDao.QUALIFIER_DEFAULT, 1);
+ MetadataTableAdminDao.CF_SHOULD_DELETE,
+ MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
+ 1L);
ConditionalRowMutation rowMutation =
ConditionalRowMutation.create(tableId,
rowKey).condition(lockCellFilter).then(deleteCell);
return dataClient.checkAndMutateRow(rowMutation);
@@ -669,16 +681,19 @@ public class MetadataTableDao {
return true;
}
+ long nowMicros = Instant.now().getMillis() * 1000L;
// Record all the initial metadata.
Mutation mutation =
Mutation.create()
.setCell(
MetadataTableAdminDao.CF_LOCK,
MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
partitionRecord.getUuid())
.setCell(
MetadataTableAdminDao.CF_WATERMARK,
MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
partitionRecord.getParentLowWatermark().getMillis())
.deleteCells(
MetadataTableAdminDao.CF_SHOULD_DELETE,
MetadataTableAdminDao.QUALIFIER_DEFAULT);
@@ -689,6 +704,7 @@ public class MetadataTableDao {
mutation.setCell(
MetadataTableAdminDao.CF_INITIAL_TOKEN,
ByteStringRange.serializeToByteString(token.getPartition()),
+ nowMicros,
token.toByteString());
}
}
@@ -726,12 +742,14 @@ public class MetadataTableDao {
* the existing metadata table is compatible with current beam connector
code.
*/
public void writeDetectNewPartitionVersion() {
+ long nowMicros = Instant.now().getMillis() * 1000L;
RowMutation rowMutation =
RowMutation.create(tableId, getFullDetectNewPartition())
.setCell(
MetadataTableAdminDao.CF_VERSION,
MetadataTableAdminDao.QUALIFIER_DEFAULT,
- MetadataTableAdminDao.CURRENT_METADATA_TABLE_VERSION);
+ nowMicros,
+ (long) MetadataTableAdminDao.CURRENT_METADATA_TABLE_VERSION);
mutateRowWithHardTimeout(rowMutation);
}
@@ -779,12 +797,14 @@ public class MetadataTableDao {
*/
public void writeDetectNewPartitionMissingPartitions(
HashMap<ByteStringRange, Instant> missingPartitionDurations) {
+ long nowMicros = Instant.now().getMillis() * 1000L;
byte[] serializedMissingPartition =
SerializationUtils.serialize(missingPartitionDurations);
RowMutation rowMutation =
RowMutation.create(tableId, getFullDetectNewPartition())
.setCell(
MetadataTableAdminDao.CF_MISSING_PARTITIONS,
ByteString.copyFromUtf8(MetadataTableAdminDao.QUALIFIER_DEFAULT),
+ nowMicros,
ByteString.copyFrom(serializedMissingPartition));
mutateRowWithHardTimeout(rowMutation);
}
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDaoTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDaoTest.java
index f1ae915a012..d53e8b4f551 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDaoTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dao/MetadataTableDaoTest.java
@@ -642,6 +642,7 @@ public class MetadataTableDaoTest {
public void readAndWriteInvalidMissingPartitionsDuration() {
HashMap<ByteStringRange, Instant> missingPartitionsDuration = new
HashMap<>();
+ long nowMicros = Instant.now().getMillis() * 1000L;
RowMutation rowMutation =
RowMutation.create(
metadataTableAdminDao.getTableId(),
@@ -651,6 +652,7 @@ public class MetadataTableDaoTest {
.setCell(
MetadataTableAdminDao.CF_MISSING_PARTITIONS,
ByteString.copyFromUtf8(MetadataTableAdminDao.QUALIFIER_DEFAULT),
+ nowMicros,
ByteString.copyFromUtf8("Invalid serialization"));
dataClient.mutateRow(rowMutation);
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java
index f0337504282..84a3b51f207 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/InitializeDoFnTest.java
@@ -117,6 +117,7 @@ public class InitializeDoFnTest {
@Test
public void testInitializeStopWithoutDNP() throws IOException {
+ long nowMicros = Instant.now().getMillis() * 1000L;
// DNP row doesn't exist, so we don't need to stop the pipeline. But some
random data row with
// the same prefix exists. We want to make sure we clean it up even in
"STOP" option.
dataClient.mutateRow(
@@ -126,7 +127,10 @@ public class InitializeDoFnTest {
.getChangeStreamNamePrefix()
.concat(ByteString.copyFromUtf8("existing_row")))
.setCell(
- MetadataTableAdminDao.CF_WATERMARK,
MetadataTableAdminDao.QUALIFIER_DEFAULT, 123));
+ MetadataTableAdminDao.CF_WATERMARK,
+ MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
+ 123L));
Instant startTime = Instant.now();
InitializeDoFn initializeDoFn =
new InitializeDoFn(
@@ -138,6 +142,7 @@ public class InitializeDoFnTest {
@Test
public void testInitializeResumeWithoutDNP() throws IOException {
+ long nowMicros = Instant.now().getMillis() * 1000L;
dataClient.mutateRow(
RowMutation.create(
tableId,
@@ -145,7 +150,10 @@ public class InitializeDoFnTest {
.getChangeStreamNamePrefix()
.concat(ByteString.copyFromUtf8("existing_row")))
.setCell(
- MetadataTableAdminDao.CF_WATERMARK,
MetadataTableAdminDao.QUALIFIER_DEFAULT, 123));
+ MetadataTableAdminDao.CF_WATERMARK,
+ MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
+ 123L));
Instant startTime = Instant.now();
InitializeDoFn initializeDoFn =
new InitializeDoFn(daoFactory, startTime,
BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW);
@@ -158,6 +166,7 @@ public class InitializeDoFnTest {
public void testInitializeResumeWithDNP() throws IOException {
Instant resumeTime = Instant.now().minus(Duration.standardSeconds(10000));
metadataTableDao.updateDetectNewPartitionWatermark(resumeTime);
+ long nowMicros = Instant.now().getMillis() * 1000L;
dataClient.mutateRow(
RowMutation.create(
tableId,
@@ -165,7 +174,10 @@ public class InitializeDoFnTest {
.getChangeStreamNamePrefix()
.concat(ByteString.copyFromUtf8("existing_row")))
.setCell(
- MetadataTableAdminDao.CF_WATERMARK,
MetadataTableAdminDao.QUALIFIER_DEFAULT, 123));
+ MetadataTableAdminDao.CF_WATERMARK,
+ MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
+ 123L));
Instant startTime = Instant.now();
InitializeDoFn initializeDoFn =
new InitializeDoFn(daoFactory, startTime,
BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW);
@@ -180,10 +192,14 @@ public class InitializeDoFnTest {
metadataTableAdminDao
.getChangeStreamNamePrefix()
.concat(ByteString.copyFromUtf8("existing_row"));
+ long nowMicros = Instant.now().getMillis() * 1000L;
dataClient.mutateRow(
RowMutation.create(tableId, metadataRowKey)
.setCell(
- MetadataTableAdminDao.CF_WATERMARK,
MetadataTableAdminDao.QUALIFIER_DEFAULT, 123));
+ MetadataTableAdminDao.CF_WATERMARK,
+ MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
+ 123L));
Instant startTime = Instant.now();
InitializeDoFn initializeDoFn =
new InitializeDoFn(daoFactory, startTime,
ExistingPipelineOptions.SKIP_CLEANUP);
@@ -202,10 +218,14 @@ public class InitializeDoFnTest {
metadataTableAdminDao
.getChangeStreamNamePrefix()
.concat(ByteString.copyFromUtf8("existing_row"));
+ long nowMicros = Instant.now().getMillis() * 1000L;
dataClient.mutateRow(
RowMutation.create(tableId, metadataRowKey)
.setCell(
- MetadataTableAdminDao.CF_WATERMARK,
MetadataTableAdminDao.QUALIFIER_DEFAULT, 123));
+ MetadataTableAdminDao.CF_WATERMARK,
+ MetadataTableAdminDao.QUALIFIER_DEFAULT,
+ nowMicros,
+ 123L));
Instant startTime = Instant.now();
InitializeDoFn initializeDoFn =
new InitializeDoFn(daoFactory, startTime,
ExistingPipelineOptions.SKIP_CLEANUP);
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/it/BigtableChangeStreamIT.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/it/BigtableChangeStreamIT.java
index d1e0ace7eb8..a83d720be92 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/it/BigtableChangeStreamIT.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/it/BigtableChangeStreamIT.java
@@ -150,8 +150,10 @@ public class BigtableChangeStreamIT {
Instant startTime = Instant.now();
Thread.sleep(500);
String rowKey = "rowKeySetCell";
+ long nowMicros = Instant.now().getMillis() * 1000L;
RowMutationEntry setCellEntry =
- RowMutationEntry.create(rowKey).setCell(COLUMN_FAMILY1,
COLUMN_QUALIFIER, "cell value 1");
+ RowMutationEntry.create(rowKey)
+ .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, nowMicros, "cell value
1");
mutationBatcher.add(setCellEntry);
mutationBatcher.flush();
Instant endTime = Instant.now().plus(Duration.standardSeconds(10));
@@ -168,9 +170,10 @@ public class BigtableChangeStreamIT {
Instant startTime = Instant.now();
Thread.sleep(500);
String rowKeyToDelete = "rowKeyToDelete";
+ long nowMicros = Instant.now().getMillis() * 1000L;
RowMutationEntry setCellMutationToDelete =
RowMutationEntry.create(rowKeyToDelete)
- .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, "cell value 1");
+ .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, nowMicros, "cell value
1");
RowMutationEntry deleteRowMutation =
RowMutationEntry.create(rowKeyToDelete).deleteRow();
mutationBatcher.add(setCellMutationToDelete);
mutationBatcher.flush();
@@ -198,10 +201,11 @@ public class BigtableChangeStreamIT {
Thread.sleep(500);
String cellValue = "cell value 1";
String rowKeyMultiFamily = "rowKeyMultiFamily";
+ long nowMicros = Instant.now().getMillis() * 1000L;
RowMutationEntry setCells =
RowMutationEntry.create(rowKeyMultiFamily)
- .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, cellValue)
- .setCell(COLUMN_FAMILY2, COLUMN_QUALIFIER, cellValue);
+ .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, nowMicros, cellValue)
+ .setCell(COLUMN_FAMILY2, COLUMN_QUALIFIER, nowMicros, cellValue);
mutationBatcher.add(setCells);
mutationBatcher.flush();
RowMutationEntry deleteCF2 =
@@ -223,10 +227,11 @@ public class BigtableChangeStreamIT {
Thread.sleep(500);
String cellValue = "cell value 1";
String rowKeyMultiCell = "rowKeyMultiCell";
+ long nowMicros = Instant.now().getMillis() * 1000L;
RowMutationEntry setCells =
RowMutationEntry.create(rowKeyMultiCell)
- .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, cellValue)
- .setCell(COLUMN_FAMILY1, "CQ2", cellValue);
+ .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, nowMicros, cellValue)
+ .setCell(COLUMN_FAMILY1, "CQ2", nowMicros, cellValue);
mutationBatcher.add(setCells);
mutationBatcher.flush();
RowMutationEntry deleteCQ2 =
@@ -254,15 +259,18 @@ public class BigtableChangeStreamIT {
Instant startTime = Instant.now();
Thread.sleep(500);
String rowKey = "rowKeyComplex";
+ long nowMicros = Instant.now().getMillis() * 1000L;
// We'll delete this in the next mutation
RowMutationEntry setCell =
- RowMutationEntry.create(rowKey).setCell(COLUMN_FAMILY1,
COLUMN_QUALIFIER, "cell value 1");
+ RowMutationEntry.create(rowKey)
+ .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, nowMicros, "cell value
1");
mutationBatcher.add(setCell);
mutationBatcher.flush();
+ long complexNowMicros = Instant.now().getMillis() * 1000L;
RowMutationEntry complexMutation =
RowMutationEntry.create(rowKey)
- .setCell(COLUMN_FAMILY1, "CQ2", "cell value 2")
- .setCell(COLUMN_FAMILY1, "CQ3", "cell value 3")
+ .setCell(COLUMN_FAMILY1, "CQ2", complexNowMicros, "cell value 2")
+ .setCell(COLUMN_FAMILY1, "CQ3", complexNowMicros, "cell value 3")
// need to set timestamp range to make change stream output match
.deleteCells(
COLUMN_FAMILY1,
@@ -290,9 +298,10 @@ public class BigtableChangeStreamIT {
Arrays.fill(chars, '\u200B'); // zero-width space
String largeString = String.valueOf(chars);
String rowKeyLargeCell = "rowKeyLargeCell";
+ long nowMicros = Instant.now().getMillis() * 1000L;
RowMutationEntry setLargeCell =
RowMutationEntry.create(rowKeyLargeCell)
- .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, largeString);
+ .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, nowMicros, largeString);
mutationBatcher.add(setLargeCell);
mutationBatcher.flush();
Instant endTime = Instant.now().plus(Duration.standardSeconds(10));
@@ -316,9 +325,11 @@ public class BigtableChangeStreamIT {
ImmutableList.Builder<RowMutationEntry> originalWrites =
ImmutableList.builder();
for (int i = 0; i < 100; ++i) {
String rowKey = "rowKey" + i;
+ long nowMicros = Instant.now().getMillis() * 1000L;
// SetCell.
RowMutationEntry setLargeCell =
- RowMutationEntry.create(rowKey).setCell(COLUMN_FAMILY1,
COLUMN_QUALIFIER, largeString);
+ RowMutationEntry.create(rowKey)
+ .setCell(COLUMN_FAMILY1, COLUMN_QUALIFIER, nowMicros,
largeString);
// DeleteFamily.
RowMutationEntry deleteFamily =
RowMutationEntry.create(rowKey).deleteFamily(COLUMN_FAMILY1);
// DeleteCells.
