Upgrade GCP Libraries BOM to 26.87.0

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

it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/pubsub/PubsubResourceManager.java

@@ -283,7 +283,11 @@ public String createSchema(
             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"))

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:

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 List<NewPartition> readNewPartitions() throws InvalidProtocolBufferExcept
    * @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 void writeNewPartition(NewPartition newPartition) {
             .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 void writeNewPartition(NewPartition newPartition) {
    * @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 List<PartitionRecord> readAllStreamPartitions() throws InvalidProtocolBuf
    */
   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 void updateWatermark(
    * @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 boolean releaseStreamPartitionLockForDeletion(ByteStringRange partition,
         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 boolean lockAndRecordPartition(PartitionRecord partitionRecord) {
       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 boolean lockAndRecordPartition(PartitionRecord partitionRecord) {
         mutation.setCell(
             MetadataTableAdminDao.CF_INITIAL_TOKEN,
             ByteStringRange.serializeToByteString(token.getPartition()),
+            nowMicros,
             token.toByteString());
       }
     }
@@ -726,12 +742,14 @@ tableId, convertPartitionToStreamPartitionRowKey(partitionRecord.getPartition())
    * 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 HashMap<ByteStringRange, Instant> readDetectNewPartitionMissingPartitions
    */
   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);
   }

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 void readAndWriteValidMissingPartitionsDuration() {
   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 void readAndWriteInvalidMissingPartitionsDuration() {
             .setCell(
                 MetadataTableAdminDao.CF_MISSING_PARTITIONS,
                 ByteString.copyFromUtf8(MetadataTableAdminDao.QUALIFIER_DEFAULT),
+                nowMicros,
                 ByteString.copyFromUtf8("Invalid serialization"));
     dataClient.mutateRow(rowMutation);
 

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 void testInitializeStopWithExistingPipeline() throws IOException {
 
   @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 void testInitializeStopWithoutDNP() throws IOException {
                     .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,14 +142,18 @@ public void testInitializeStopWithoutDNP() throws IOException {
 
   @Test
   public void testInitializeResumeWithoutDNP() throws IOException {
+    long nowMicros = Instant.now().getMillis() * 1000L;
     dataClient.mutateRow(
         RowMutation.create(
                 tableId,
                 metadataTableAdminDao
                     .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,14 +166,18 @@ public void testInitializeResumeWithoutDNP() throws IOException {
   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,
                 metadataTableAdminDao
                     .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 void testInitializeSkipCleanupWithoutDNP() throws IOException {
         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 void testInitializeSkipCleanupWithDNP() throws IOException {
         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);