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 e177becc7a8b..a1343d3698fb 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 { 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 { 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 { 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 { 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 f947d70efd94..e38251431c55 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 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")) diff --git a/sdks/java/container/license_scripts/dep_urls_java.yaml b/sdks/java/container/license_scripts/dep_urls_java.yaml index 25e438f78541..df64ca0cafcd 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 6a5ac4354cae..83115297764b 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 List 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 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 readDetectNewPartitionMissingPartitions */ public void writeDetectNewPartitionMissingPartitions( HashMap 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 f1ae915a012e..d53e8b4f5518 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 void readAndWriteValidMissingPartitionsDuration() { public void readAndWriteInvalidMissingPartitionsDuration() { HashMap 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); 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 f0337504282f..84a3b51f207e 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 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,6 +142,7 @@ public void testInitializeStopWithoutDNP() throws IOException { @Test public void testInitializeResumeWithoutDNP() throws IOException { + long nowMicros = Instant.now().getMillis() * 1000L; dataClient.mutateRow( RowMutation.create( tableId, @@ -145,7 +150,10 @@ public void testInitializeResumeWithoutDNP() 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(daoFactory, startTime, BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW); @@ -158,6 +166,7 @@ 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, @@ -165,7 +174,10 @@ public void testInitializeResumeWithDNP() 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(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);