diff --git a/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/DataStreamToSpannerOptions.java b/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/DataStreamToSpannerOptions.java new file mode 100644 index 0000000000..a69ce87a92 --- /dev/null +++ b/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/DataStreamToSpannerOptions.java @@ -0,0 +1,459 @@ +/* + * Copyright (C) 2026 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 + * + * http://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.teleport.v2.options; + +import com.google.cloud.spanner.Options.RpcPriority; +import com.google.cloud.teleport.metadata.TemplateParameter; +import com.google.cloud.teleport.metadata.TemplateParameter.TemplateEnumOption; +import com.google.cloud.teleport.v2.spanner.migrations.constants.Constants; +import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions; +import org.apache.beam.sdk.options.Default; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.StreamingOptions; + +/** + * Options supported by the pipeline. + * + *

Inherits standard configuration options. + */ +public interface DataStreamToSpannerOptions + extends PipelineOptions, StreamingOptions, DataflowPipelineWorkerPoolOptions { + @TemplateParameter.GcsReadFile( + order = 1, + groupName = "Source", + optional = true, + description = + "File location for Datastream file output in Cloud Storage. Support for this feature has been disabled.", + helpText = + "The Cloud Storage file location that contains the Datastream files to replicate. Typically, " + + "this is the root path for a stream. Support for this feature has been disabled." + + " Please use this feature only for retrying entries that land in severe DLQ.") + String getInputFilePattern(); + + void setInputFilePattern(String value); + + @TemplateParameter.Enum( + order = 2, + enumOptions = {@TemplateEnumOption("avro"), @TemplateEnumOption("json")}, + optional = true, + description = "Datastream output file format (avro/json).", + helpText = + "The format of the output file produced by Datastream. For example `avro,json`. Defaults to `avro`.") + @Default.String("avro") + String getInputFileFormat(); + + void setInputFileFormat(String value); + + @TemplateParameter.GcsReadFile( + order = 3, + optional = true, + description = "Session File Path in Cloud Storage", + helpText = + "Session file path in Cloud Storage that contains mapping information from" + + " HarbourBridge") + String getSessionFilePath(); + + void setSessionFilePath(String value); + + @TemplateParameter.Text( + order = 4, + groupName = "Target", + description = "Cloud Spanner Instance Id.", + helpText = "The Spanner instance where the changes are replicated.") + String getInstanceId(); + + void setInstanceId(String value); + + @TemplateParameter.Text( + order = 5, + groupName = "Target", + description = "Cloud Spanner Database Id.", + helpText = "The Spanner database where the changes are replicated.") + String getDatabaseId(); + + void setDatabaseId(String value); + + @TemplateParameter.ProjectId( + order = 6, + groupName = "Target", + optional = true, + description = "Cloud Spanner Project Id.", + helpText = "The Spanner project ID.") + String getProjectId(); + + void setProjectId(String projectId); + + @TemplateParameter.Text( + order = 7, + groupName = "Target", + optional = true, + description = "The Cloud Spanner Endpoint to call", + helpText = "The Cloud Spanner endpoint to call in the template.", + example = "https://batch-spanner.googleapis.com") + @Default.String("https://batch-spanner.googleapis.com") + String getSpannerHost(); + + void setSpannerHost(String value); + + @TemplateParameter.PubsubSubscription( + order = 8, + optional = true, + description = "The Pub/Sub subscription being used in a Cloud Storage notification policy.", + helpText = + "The Pub/Sub subscription being used in a Cloud Storage notification policy. For the name," + + " use the format `projects//subscriptions/`.") + String getGcsPubSubSubscription(); + + void setGcsPubSubSubscription(String value); + + @TemplateParameter.Text( + order = 9, + groupName = "Source", + optional = true, + description = "Datastream stream name.", + helpText = + "The name or template for the stream to poll for schema information and source type.") + String getStreamName(); + + void setStreamName(String value); + + @TemplateParameter.Text( + order = 10, + optional = true, + description = "Cloud Spanner shadow table prefix.", + helpText = "The prefix used to name shadow tables. Default: `shadow_`.") + @Default.String("shadow_") + String getShadowTablePrefix(); + + void setShadowTablePrefix(String value); + + @TemplateParameter.Boolean( + order = 11, + optional = true, + description = "If true, create shadow tables in Cloud Spanner.", + helpText = + "This flag indicates whether shadow tables must be created in Cloud Spanner database.") + @Default.Boolean(true) + Boolean getShouldCreateShadowTables(); + + void setShouldCreateShadowTables(Boolean value); + + @TemplateParameter.DateTime( + order = 12, + optional = true, + description = + "The starting DateTime used to fetch from Cloud Storage " + + "(https://tools.ietf.org/html/rfc3339).", + helpText = + "The starting DateTime used to fetch from Cloud Storage " + + "(https://tools.ietf.org/html/rfc3339).") + @Default.String("1970-01-01T00:00:00.00Z") + String getRfcStartDateTime(); + + void setRfcStartDateTime(String value); + + @TemplateParameter.Integer( + order = 13, + optional = true, + description = "File read concurrency", + helpText = "The number of concurrent DataStream files to read.") + @Default.Integer(30) + Integer getFileReadConcurrency(); + + void setFileReadConcurrency(Integer value); + + @TemplateParameter.Text( + order = 14, + optional = true, + description = "Dead letter queue directory.", + helpText = + "The file path used when storing the error queue output. " + + "The default file path is a directory under the Dataflow job's temp location.") + @Default.String("") + String getDeadLetterQueueDirectory(); + + void setDeadLetterQueueDirectory(String value); + + @TemplateParameter.Integer( + order = 15, + optional = true, + description = "Dead letter queue retry minutes", + helpText = "The number of minutes between dead letter queue retries. Defaults to `10`.") + @Default.Integer(10) + Integer getDlqRetryMinutes(); + + void setDlqRetryMinutes(Integer value); + + @TemplateParameter.Integer( + order = 16, + optional = true, + description = "Dead letter queue maximum retry count", + helpText = + "The max number of times temporary errors can be retried through DLQ. Defaults to `500`.") + @Default.Integer(500) + Integer getDlqMaxRetryCount(); + + void setDlqMaxRetryCount(Integer value); + + // DataStream API Root Url (only used for testing) + @TemplateParameter.Text( + order = 17, + optional = true, + description = "Datastream API Root URL (only required for testing)", + helpText = "Datastream API Root URL.") + @Default.String("https://datastream.googleapis.com/") + String getDataStreamRootUrl(); + + void setDataStreamRootUrl(String value); + + @TemplateParameter.Text( + order = 18, + optional = true, + description = "Datastream source type (only required for testing)", + helpText = + "This is the type of source database that Datastream connects to. Example -" + + " mysql/oracle. Need to be set when testing without an actual running" + + " Datastream.") + String getDatastreamSourceType(); + + void setDatastreamSourceType(String value); + + @TemplateParameter.Boolean( + order = 19, + optional = true, + description = + "If true, rounds the decimal values in json columns to a number that can be stored" + + " without loss of precision.", + helpText = + "This flag if set, rounds the decimal values in json columns to a number that can be" + + " stored without loss of precision.") + @Default.Boolean(false) + Boolean getRoundJsonDecimals(); + + void setRoundJsonDecimals(Boolean value); + + @TemplateParameter.Enum( + order = 20, + optional = true, + description = "Run mode - currently supported are : regular, retryDLQ, or retryAllDLQ", + enumOptions = { + @TemplateEnumOption(Constants.RUN_MODE_REGULAR), + @TemplateEnumOption(Constants.RUN_MODE_RETRY_DLQ), + @TemplateEnumOption(Constants.RUN_MODE_RETRY_ALL_DLQ) + }, + helpText = + "This is the run mode type. Default is regular. Use `retryDLQ` mode to process exclusively severe error files concurrently with your live migration pipeline. Use `retryAllDLQ` mode only when the regular pipeline is stopped. This mode processes both retry and severe directories. Do NOT run `retryAllDLQ` concurrently with any active pipeline as it will cause conflicts.") + @Default.String(Constants.RUN_MODE_REGULAR) + String getRunMode(); + + void setRunMode(String value); + + @TemplateParameter.GcsReadFile( + order = 21, + optional = true, + helpText = + "Transformation context file path in cloud storage used to populate data used in" + + " transformations performed during migrations Eg: The shard id to db name to" + + " identify the db from which a row was migrated", + description = "Transformation context file path in cloud storage") + String getTransformationContextFilePath(); + + void setTransformationContextFilePath(String value); + + @TemplateParameter.Integer( + order = 22, + optional = true, + description = "Directory watch duration in minutes. Default: 10 minutes", + helpText = + "The Duration for which the pipeline should keep polling a directory in GCS. Datastream" + + "output files are arranged in a directory structure which depicts the timestamp " + + "of the event grouped by minutes. This parameter should be approximately equal to" + + "maximum delay which could occur between event occurring in source database and " + + "the same event being written to GCS by Datastream. 99.9 percentile = 10 minutes") + @Default.Integer(10) + Integer getDirectoryWatchDurationInMinutes(); + + void setDirectoryWatchDurationInMinutes(Integer value); + + @TemplateParameter.Enum( + order = 23, + enumOptions = { + @TemplateEnumOption("LOW"), + @TemplateEnumOption("MEDIUM"), + @TemplateEnumOption("HIGH") + }, + optional = true, + description = "Priority for Spanner RPC invocations", + helpText = + "The request priority for Cloud Spanner calls. The value must be one of:" + + " [`HIGH`,`MEDIUM`,`LOW`]. Defaults to `HIGH`.") + @Default.Enum("HIGH") + RpcPriority getSpannerPriority(); + + void setSpannerPriority(RpcPriority value); + + @TemplateParameter.PubsubSubscription( + order = 24, + optional = true, + description = + "The Pub/Sub subscription being used in a Cloud Storage notification policy for DLQ" + + " retry directory when running in regular mode.", + helpText = + "The Pub/Sub subscription being used in a Cloud Storage notification policy for DLQ" + + " retry directory when running in regular mode. For the name, use the format" + + " `projects//subscriptions/`. When set, the" + + " deadLetterQueueDirectory and dlqRetryMinutes are ignored.") + String getDlqGcsPubSubSubscription(); + + void setDlqGcsPubSubSubscription(String value); + + @TemplateParameter.GcsReadFile( + order = 25, + optional = true, + description = "Custom jar location in Cloud Storage", + helpText = + "Custom JAR file location in Cloud Storage for the file that contains the custom transformation logic for processing records" + + " in forward migration.") + @Default.String("") + String getTransformationJarPath(); + + void setTransformationJarPath(String value); + + @TemplateParameter.Text( + order = 26, + optional = true, + description = "Custom class name", + helpText = + "Fully qualified class name having the custom transformation logic. It is a" + + " mandatory field in case transformationJarPath is specified") + @Default.String("") + String getTransformationClassName(); + + void setTransformationClassName(String value); + + @TemplateParameter.Text( + order = 27, + optional = true, + description = "Custom parameters for transformation", + helpText = + "String containing any custom parameters to be passed to the custom transformation class.") + @Default.String("") + String getTransformationCustomParameters(); + + void setTransformationCustomParameters(String value); + + @TemplateParameter.Text( + order = 28, + optional = true, + description = "Filtered events directory", + helpText = + "This is the file path to store the events filtered via custom transformation. Default is a directory" + + " under the Dataflow job's temp location. The default value is enough under most" + + " conditions.") + @Default.String("") + String getFilteredEventsDirectory(); + + void setFilteredEventsDirectory(String value); + + @TemplateParameter.GcsReadFile( + order = 29, + optional = true, + description = "Source Config URL", + helpText = + "Cloud Storage path to a shard config file for sharded migrations. It expects a HOCON or JSON file. For a sample file, please refer to v2/datastream-to-spanner/src/test/resources/DatastreamToSpannerSingleDFShardedMigrationIT/sharding-config.conf in the repository. For example, `gs://my-bucket/my-shard-config.conf`.", + example = "gs://my-bucket/my-shard-config.conf") + String getSourceConfigURL(); + + void setSourceConfigURL(String value); + + @TemplateParameter.Text( + order = 30, + optional = true, + description = "Table name overrides from source to spanner", + regexes = + "^\\[([[:space:]]*\\{[[:space:]]*[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", + example = "[{Singers, Vocalists}, {Albums, Records}]", + helpText = + "These are the table name overrides from source to spanner. They are written in the" + + "following format: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]" + + "This example shows mapping Singers table to Vocalists and Albums table to Records.") + @Default.String("") + String getTableOverrides(); + + void setTableOverrides(String value); + + @TemplateParameter.Text( + order = 31, + optional = true, + regexes = + "^\\[([[:space:]]*\\{[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", + description = "Column name overrides from source to spanner", + example = "[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]", + helpText = + "These are the column name overrides from source to spanner. They are written in the" + + "following format: [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]" + + "Note that the SourceTableName should remain the same in both the source and spanner pair. To override table names, use tableOverrides." + + "The example shows mapping SingerName to TalentName and AlbumName to RecordName in Singers and Albums table respectively.") + @Default.String("") + String getColumnOverrides(); + + void setColumnOverrides(String value); + + @TemplateParameter.Text( + order = 32, + optional = true, + description = "File based overrides from source to spanner", + helpText = + "A file which specifies the table and the column name overrides from source to spanner.") + @Default.String("") + String getSchemaOverridesFilePath(); + + void setSchemaOverridesFilePath(String value); + + @TemplateParameter.Text( + order = 33, + optional = true, + groupName = "Target", + description = "Cloud Spanner Shadow Table Instance Id.", + helpText = + "Optional separate instance for shadow tables. If not specified, shadow tables will be created in the main instance. If specified, ensure shadowTableSpannerDatabaseId is specified as well.") + @Default.String("") + String getShadowTableSpannerInstanceId(); + + void setShadowTableSpannerInstanceId(String value); + + @TemplateParameter.Text( + order = 34, + optional = true, + groupName = "Target", + description = "Cloud Spanner Shadow Table Database Id.", + helpText = + "Optional separate database for shadow tables. If not specified, shadow tables will be created in the main database. If specified, ensure shadowTableSpannerInstanceId is specified as well.") + @Default.String("") + String getShadowTableSpannerDatabaseId(); + + void setShadowTableSpannerDatabaseId(String value); + + @TemplateParameter.Text( + order = 35, + optional = true, + description = "Failure injection parameter", + helpText = "Failure injection parameter. Only used for testing.") + @Default.String("") + String getFailureInjectionParameter(); + + void setFailureInjectionParameter(String value); +} diff --git a/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/package-info.java b/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/package-info.java new file mode 100644 index 0000000000..dbdf77b046 --- /dev/null +++ b/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/options/package-info.java @@ -0,0 +1,18 @@ +/* + * Copyright (C) 2026 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 + * + * http://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. + */ + +/** Options classes. */ +package com.google.cloud.teleport.v2.options; diff --git a/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToSpanner.java b/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToSpanner.java index 5abb19cc63..d3568bfe37 100644 --- a/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToSpanner.java +++ b/v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToSpanner.java @@ -17,11 +17,8 @@ import com.google.api.gax.retrying.RetrySettings; import com.google.api.services.datastream.v1.model.SourceConfig; -import com.google.cloud.spanner.Options.RpcPriority; import com.google.cloud.teleport.metadata.Template; import com.google.cloud.teleport.metadata.TemplateCategory; -import com.google.cloud.teleport.metadata.TemplateParameter; -import com.google.cloud.teleport.metadata.TemplateParameter.TemplateEnumOption; import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager; import com.google.cloud.teleport.v2.cdc.dlq.PubSubNotifiedDlqIO; import com.google.cloud.teleport.v2.cdc.dlq.StringDeadLetterQueueSanitizer; @@ -29,6 +26,7 @@ import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger; import com.google.cloud.teleport.v2.datastream.sources.DataStreamIO; import com.google.cloud.teleport.v2.datastream.utils.DataStreamClient; +import com.google.cloud.teleport.v2.options.DataStreamToSpannerOptions; import com.google.cloud.teleport.v2.spanner.ddl.Ddl; import com.google.cloud.teleport.v2.spanner.migrations.constants.Constants; import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaOverridesParser; @@ -47,7 +45,6 @@ import com.google.cloud.teleport.v2.spanner.migrations.utils.SecretManagerAccessorImpl; import com.google.cloud.teleport.v2.spanner.migrations.utils.SessionFileReader; import com.google.cloud.teleport.v2.spanner.migrations.utils.TransformationContextReader; -import com.google.cloud.teleport.v2.templates.DataStreamToSpanner.Options; import com.google.cloud.teleport.v2.templates.constants.DatastreamToSpannerConstants; import com.google.cloud.teleport.v2.templates.source.DatastreamToSpannerSourceConnectorRegistry; import com.google.cloud.teleport.v2.templates.spanner.ProcessInformationSchema; @@ -71,10 +68,7 @@ import org.apache.beam.sdk.io.TextIO; import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig; import org.apache.beam.sdk.io.gcp.spanner.SpannerServiceFactoryImpl; -import org.apache.beam.sdk.options.Default; -import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; -import org.apache.beam.sdk.options.StreamingOptions; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.transforms.Flatten; import org.apache.beam.sdk.transforms.MapElements; @@ -136,7 +130,7 @@ + " The `retryAllDLQ` mode consumes errors from both the retry and severe buckets. Do NOT run `retryAllDLQ` concurrently" + " with the regular pipeline as they will conflict." }, - optionsClass = Options.class, + optionsClass = DataStreamToSpannerOptions.class, flexContainerName = "datastream-to-spanner", documentation = "https://cloud.google.com/dataflow/docs/guides/templates/provided/datastream-to-cloud-spanner", @@ -153,442 +147,7 @@ public class DataStreamToSpanner { private static final String AVRO_SUFFIX = "avro"; private static final String JSON_SUFFIX = "json"; - /** - * Options supported by the pipeline. - * - *

Inherits standard configuration options. - */ - public interface Options - extends PipelineOptions, StreamingOptions, DataflowPipelineWorkerPoolOptions { - @TemplateParameter.GcsReadFile( - order = 1, - groupName = "Source", - optional = true, - description = - "File location for Datastream file output in Cloud Storage. Support for this feature has been disabled.", - helpText = - "The Cloud Storage file location that contains the Datastream files to replicate. Typically, " - + "this is the root path for a stream. Support for this feature has been disabled." - + " Please use this feature only for retrying entries that land in severe DLQ.") - String getInputFilePattern(); - - void setInputFilePattern(String value); - - @TemplateParameter.Enum( - order = 2, - enumOptions = {@TemplateEnumOption("avro"), @TemplateEnumOption("json")}, - optional = true, - description = "Datastream output file format (avro/json).", - helpText = - "The format of the output file produced by Datastream. For example `avro,json`. Defaults to `avro`.") - @Default.String("avro") - String getInputFileFormat(); - - void setInputFileFormat(String value); - - @TemplateParameter.GcsReadFile( - order = 3, - optional = true, - description = "Session File Path in Cloud Storage", - helpText = - "Session file path in Cloud Storage that contains mapping information from" - + " HarbourBridge") - String getSessionFilePath(); - - void setSessionFilePath(String value); - - @TemplateParameter.Text( - order = 4, - groupName = "Target", - description = "Cloud Spanner Instance Id.", - helpText = "The Spanner instance where the changes are replicated.") - String getInstanceId(); - - void setInstanceId(String value); - - @TemplateParameter.Text( - order = 5, - groupName = "Target", - description = "Cloud Spanner Database Id.", - helpText = "The Spanner database where the changes are replicated.") - String getDatabaseId(); - - void setDatabaseId(String value); - - @TemplateParameter.ProjectId( - order = 6, - groupName = "Target", - optional = true, - description = "Cloud Spanner Project Id.", - helpText = "The Spanner project ID.") - String getProjectId(); - - void setProjectId(String projectId); - - @TemplateParameter.Text( - order = 7, - groupName = "Target", - optional = true, - description = "The Cloud Spanner Endpoint to call", - helpText = "The Cloud Spanner endpoint to call in the template.", - example = "https://batch-spanner.googleapis.com") - @Default.String("https://batch-spanner.googleapis.com") - String getSpannerHost(); - - void setSpannerHost(String value); - - @TemplateParameter.PubsubSubscription( - order = 8, - optional = true, - description = "The Pub/Sub subscription being used in a Cloud Storage notification policy.", - helpText = - "The Pub/Sub subscription being used in a Cloud Storage notification policy. For the name," - + " use the format `projects//subscriptions/`.") - String getGcsPubSubSubscription(); - - void setGcsPubSubSubscription(String value); - - @TemplateParameter.Text( - order = 9, - groupName = "Source", - optional = true, - description = "Datastream stream name.", - helpText = - "The name or template for the stream to poll for schema information and source type.") - String getStreamName(); - - void setStreamName(String value); - - @TemplateParameter.Text( - order = 10, - optional = true, - description = "Cloud Spanner shadow table prefix.", - helpText = "The prefix used to name shadow tables. Default: `shadow_`.") - @Default.String("shadow_") - String getShadowTablePrefix(); - - void setShadowTablePrefix(String value); - - @TemplateParameter.Boolean( - order = 11, - optional = true, - description = "If true, create shadow tables in Cloud Spanner.", - helpText = - "This flag indicates whether shadow tables must be created in Cloud Spanner database.") - @Default.Boolean(true) - Boolean getShouldCreateShadowTables(); - - void setShouldCreateShadowTables(Boolean value); - - @TemplateParameter.DateTime( - order = 12, - optional = true, - description = - "The starting DateTime used to fetch from Cloud Storage " - + "(https://tools.ietf.org/html/rfc3339).", - helpText = - "The starting DateTime used to fetch from Cloud Storage " - + "(https://tools.ietf.org/html/rfc3339).") - @Default.String("1970-01-01T00:00:00.00Z") - String getRfcStartDateTime(); - - void setRfcStartDateTime(String value); - - @TemplateParameter.Integer( - order = 13, - optional = true, - description = "File read concurrency", - helpText = "The number of concurrent DataStream files to read.") - @Default.Integer(30) - Integer getFileReadConcurrency(); - - void setFileReadConcurrency(Integer value); - - @TemplateParameter.Text( - order = 14, - optional = true, - description = "Dead letter queue directory.", - helpText = - "The file path used when storing the error queue output. " - + "The default file path is a directory under the Dataflow job's temp location.") - @Default.String("") - String getDeadLetterQueueDirectory(); - - void setDeadLetterQueueDirectory(String value); - - @TemplateParameter.Integer( - order = 15, - optional = true, - description = "Dead letter queue retry minutes", - helpText = "The number of minutes between dead letter queue retries. Defaults to `10`.") - @Default.Integer(10) - Integer getDlqRetryMinutes(); - - void setDlqRetryMinutes(Integer value); - - @TemplateParameter.Integer( - order = 16, - optional = true, - description = "Dead letter queue maximum retry count", - helpText = - "The max number of times temporary errors can be retried through DLQ. Defaults to `500`.") - @Default.Integer(500) - Integer getDlqMaxRetryCount(); - - void setDlqMaxRetryCount(Integer value); - - // DataStream API Root Url (only used for testing) - @TemplateParameter.Text( - order = 17, - optional = true, - description = "Datastream API Root URL (only required for testing)", - helpText = "Datastream API Root URL.") - @Default.String("https://datastream.googleapis.com/") - String getDataStreamRootUrl(); - - void setDataStreamRootUrl(String value); - - @TemplateParameter.Text( - order = 18, - optional = true, - description = "Datastream source type (only required for testing)", - helpText = - "This is the type of source database that Datastream connects to. Example -" - + " mysql/oracle. Need to be set when testing without an actual running" - + " Datastream.") - String getDatastreamSourceType(); - - void setDatastreamSourceType(String value); - - @TemplateParameter.Boolean( - order = 19, - optional = true, - description = - "If true, rounds the decimal values in json columns to a number that can be stored" - + " without loss of precision.", - helpText = - "This flag if set, rounds the decimal values in json columns to a number that can be" - + " stored without loss of precision.") - @Default.Boolean(false) - Boolean getRoundJsonDecimals(); - - void setRoundJsonDecimals(Boolean value); - - @TemplateParameter.Enum( - order = 20, - optional = true, - description = "Run mode - currently supported are : regular, retryDLQ, or retryAllDLQ", - enumOptions = { - @TemplateEnumOption(Constants.RUN_MODE_REGULAR), - @TemplateEnumOption(Constants.RUN_MODE_RETRY_DLQ), - @TemplateEnumOption(Constants.RUN_MODE_RETRY_ALL_DLQ) - }, - helpText = - "This is the run mode type. Default is regular. Use `retryDLQ` mode to process exclusively severe error files concurrently with your live migration pipeline. Use `retryAllDLQ` mode only when the regular pipeline is stopped. This mode processes both retry and severe directories. Do NOT run `retryAllDLQ` concurrently with any active pipeline as it will cause conflicts.") - @Default.String(Constants.RUN_MODE_REGULAR) - String getRunMode(); - - void setRunMode(String value); - - @TemplateParameter.GcsReadFile( - order = 21, - optional = true, - helpText = - "Transformation context file path in cloud storage used to populate data used in" - + " transformations performed during migrations Eg: The shard id to db name to" - + " identify the db from which a row was migrated", - description = "Transformation context file path in cloud storage") - String getTransformationContextFilePath(); - - void setTransformationContextFilePath(String value); - - @TemplateParameter.Integer( - order = 22, - optional = true, - description = "Directory watch duration in minutes. Default: 10 minutes", - helpText = - "The Duration for which the pipeline should keep polling a directory in GCS. Datastream" - + "output files are arranged in a directory structure which depicts the timestamp " - + "of the event grouped by minutes. This parameter should be approximately equal to" - + "maximum delay which could occur between event occurring in source database and " - + "the same event being written to GCS by Datastream. 99.9 percentile = 10 minutes") - @Default.Integer(10) - Integer getDirectoryWatchDurationInMinutes(); - - void setDirectoryWatchDurationInMinutes(Integer value); - - @TemplateParameter.Enum( - order = 23, - enumOptions = { - @TemplateEnumOption("LOW"), - @TemplateEnumOption("MEDIUM"), - @TemplateEnumOption("HIGH") - }, - optional = true, - description = "Priority for Spanner RPC invocations", - helpText = - "The request priority for Cloud Spanner calls. The value must be one of:" - + " [`HIGH`,`MEDIUM`,`LOW`]. Defaults to `HIGH`.") - @Default.Enum("HIGH") - RpcPriority getSpannerPriority(); - - void setSpannerPriority(RpcPriority value); - - @TemplateParameter.PubsubSubscription( - order = 24, - optional = true, - description = - "The Pub/Sub subscription being used in a Cloud Storage notification policy for DLQ" - + " retry directory when running in regular mode.", - helpText = - "The Pub/Sub subscription being used in a Cloud Storage notification policy for DLQ" - + " retry directory when running in regular mode. For the name, use the format" - + " `projects//subscriptions/`. When set, the" - + " deadLetterQueueDirectory and dlqRetryMinutes are ignored.") - String getDlqGcsPubSubSubscription(); - - void setDlqGcsPubSubSubscription(String value); - - @TemplateParameter.GcsReadFile( - order = 25, - optional = true, - description = "Custom jar location in Cloud Storage", - helpText = - "Custom JAR file location in Cloud Storage for the file that contains the custom transformation logic for processing records" - + " in forward migration.") - @Default.String("") - String getTransformationJarPath(); - - void setTransformationJarPath(String value); - - @TemplateParameter.Text( - order = 26, - optional = true, - description = "Custom class name", - helpText = - "Fully qualified class name having the custom transformation logic. It is a" - + " mandatory field in case transformationJarPath is specified") - @Default.String("") - String getTransformationClassName(); - - void setTransformationClassName(String value); - - @TemplateParameter.Text( - order = 27, - optional = true, - description = "Custom parameters for transformation", - helpText = - "String containing any custom parameters to be passed to the custom transformation class.") - @Default.String("") - String getTransformationCustomParameters(); - - void setTransformationCustomParameters(String value); - - @TemplateParameter.Text( - order = 28, - optional = true, - description = "Filtered events directory", - helpText = - "This is the file path to store the events filtered via custom transformation. Default is a directory" - + " under the Dataflow job's temp location. The default value is enough under most" - + " conditions.") - @Default.String("") - String getFilteredEventsDirectory(); - - void setFilteredEventsDirectory(String value); - - @TemplateParameter.GcsReadFile( - order = 29, - optional = true, - description = "Source Config URL", - helpText = - "Cloud Storage path to a shard config file for sharded migrations. It expects a HOCON or JSON file. For a sample file, please refer to v2/datastream-to-spanner/src/test/resources/DatastreamToSpannerSingleDFShardedMigrationIT/sharding-config.conf in the repository. For example, `gs://my-bucket/my-shard-config.conf`.", - example = "gs://my-bucket/my-shard-config.conf") - String getSourceConfigURL(); - - void setSourceConfigURL(String value); - - @TemplateParameter.Text( - order = 30, - optional = true, - description = "Table name overrides from source to spanner", - regexes = - "^\\[([[:space:]]*\\{[[:space:]]*[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", - example = "[{Singers, Vocalists}, {Albums, Records}]", - helpText = - "These are the table name overrides from source to spanner. They are written in the" - + "following format: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]" - + "This example shows mapping Singers table to Vocalists and Albums table to Records.") - @Default.String("") - String getTableOverrides(); - - void setTableOverrides(String value); - - @TemplateParameter.Text( - order = 31, - optional = true, - regexes = - "^\\[([[:space:]]*\\{[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", - description = "Column name overrides from source to spanner", - example = - "[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]", - helpText = - "These are the column name overrides from source to spanner. They are written in the" - + "following format: [{SourceTableName1.SourceColumnName1, SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1, SourceTableName2.SpannerColumnName1}]" - + "Note that the SourceTableName should remain the same in both the source and spanner pair. To override table names, use tableOverrides." - + "The example shows mapping SingerName to TalentName and AlbumName to RecordName in Singers and Albums table respectively.") - @Default.String("") - String getColumnOverrides(); - - void setColumnOverrides(String value); - - @TemplateParameter.Text( - order = 32, - optional = true, - description = "File based overrides from source to spanner", - helpText = - "A file which specifies the table and the column name overrides from source to spanner.") - @Default.String("") - String getSchemaOverridesFilePath(); - - void setSchemaOverridesFilePath(String value); - - @TemplateParameter.Text( - order = 33, - optional = true, - groupName = "Target", - description = "Cloud Spanner Shadow Table Instance Id.", - helpText = - "Optional separate instance for shadow tables. If not specified, shadow tables will be created in the main instance. If specified, ensure shadowTableSpannerDatabaseId is specified as well.") - @Default.String("") - String getShadowTableSpannerInstanceId(); - - void setShadowTableSpannerInstanceId(String value); - - @TemplateParameter.Text( - order = 33, - optional = true, - groupName = "Target", - description = "Cloud Spanner Shadow Table Database Id.", - helpText = - "Optional separate database for shadow tables. If not specified, shadow tables will be created in the main database. If specified, ensure shadowTableSpannerInstanceId is specified as well.") - @Default.String("") - String getShadowTableSpannerDatabaseId(); - - void setShadowTableSpannerDatabaseId(String value); - - @TemplateParameter.Text( - order = 34, - optional = true, - description = "Failure injection parameter", - helpText = "Failure injection parameter. Only used for testing.") - @Default.String("") - String getFailureInjectionParameter(); - - void setFailureInjectionParameter(String value); - } - - static void validateSourceType(Options options) { + static void validateSourceType(DataStreamToSpannerOptions options) { boolean isRetryMode = Constants.RUN_MODE_RETRY_DLQ.equals(options.getRunMode()); if (isRetryMode) { // retry mode does not read from Datastream @@ -606,7 +165,7 @@ static void validateSourceType(Options options) { options.setDatastreamSourceType(sourceType); } - static String getSourceType(Options options) { + static String getSourceType(DataStreamToSpannerOptions options) { if (options.getDatastreamSourceType() != null) { return options.getDatastreamSourceType(); } @@ -643,7 +202,8 @@ static String getSourceTypeFromConfig(SourceConfig sourceConfig) { public static void main(String[] args) { UncaughtExceptionLogger.register(); LOG.info("Starting DataStream to Cloud Spanner"); - Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(DataStreamToSpannerOptions.class); boolean isRetryDLQMode = Constants.RUN_MODE_RETRY_DLQ.equals(options.getRunMode()); options.setStreaming(!isRetryDLQMode); validateSourceType(options); @@ -656,11 +216,11 @@ public static void main(String[] args) { * @param options The execution parameters to the pipeline. * @return The result of the pipeline execution. */ - public static PipelineResult run(Options options) { + public static PipelineResult run(DataStreamToSpannerOptions options) { return buildPipeline(options).run(); } - static Pipeline buildPipeline(Options options) { + static Pipeline buildPipeline(DataStreamToSpannerOptions options) { long startTime = System.currentTimeMillis(); /* * Stages: @@ -958,7 +518,7 @@ static Pipeline buildPipeline(Options options) { return pipeline; } - static ShardingContext getShardingContext(Options options) { + static ShardingContext getShardingContext(DataStreamToSpannerOptions options) { // Ingest sharding context file into memory. ShardingContext shardingContext = new ShardingContext(); if (options.getSourceConfigURL() != null && !options.getSourceConfigURL().isEmpty()) { @@ -986,7 +546,7 @@ static ShardingContext getShardingContext(Options options) { return shardingContext; } - static SpannerConfig getShadowTableSpannerConfig(Options options) { + static SpannerConfig getShadowTableSpannerConfig(DataStreamToSpannerOptions options) { // Validate shadow table Spanner config - both instance and database must be specified together String shadowTableSpannerInstanceId = options.getShadowTableSpannerInstanceId(); String shadowTableSpannerDatabaseId = options.getShadowTableSpannerDatabaseId(); @@ -1033,7 +593,7 @@ static SpannerConfig getShadowTableSpannerConfig(Options options) { .build()); } - static DeadLetterQueueManager buildDlqManager(Options options) { + static DeadLetterQueueManager buildDlqManager(DataStreamToSpannerOptions options) { String tempLocation = options.as(DataflowPipelineOptions.class).getTempLocation().endsWith("/") ? options.as(DataflowPipelineOptions.class).getTempLocation() @@ -1047,7 +607,7 @@ static DeadLetterQueueManager buildDlqManager(Options options) { return DeadLetterQueueManager.create(dlqDirectory, options.getDlqMaxRetryCount(), true); } - static ISchemaOverridesParser configureSchemaOverrides(Options options) { + static ISchemaOverridesParser configureSchemaOverrides(DataStreamToSpannerOptions options) { // incorrect configuration if (!options.getSchemaOverridesFilePath().isEmpty() && (!options.getTableOverrides().isEmpty() || !options.getColumnOverrides().isEmpty())) { diff --git a/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerTest.java b/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerTest.java index 818e3d8925..df22416f69 100644 --- a/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerTest.java +++ b/v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerTest.java @@ -28,6 +28,7 @@ import com.google.api.services.datastream.v1.model.OracleSourceConfig; import com.google.api.services.datastream.v1.model.PostgresqlSourceConfig; import com.google.api.services.datastream.v1.model.SourceConfig; +import com.google.cloud.teleport.v2.options.DataStreamToSpannerOptions; import com.google.cloud.teleport.v2.spanner.migrations.constants.Constants; import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaOverridesParser; import com.google.cloud.teleport.v2.spanner.migrations.schema.NoopSchemaOverridesParser; @@ -50,10 +51,8 @@ public class DataStreamToSpannerTest { @Test public void testGetSourceTypeWithDatastreamSourceType() { String[] args = new String[] {"--datastreamSourceType=mysql"}; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args) - .withValidation() - .as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(DataStreamToSpannerOptions.class); String result = DataStreamToSpanner.getSourceType(options); assertEquals("mysql", result); @@ -63,10 +62,8 @@ public void testGetSourceTypeWithDatastreamSourceType() { public void testGetSourceTypeWithDatastreamInputFilePattern() { String[] args = new String[] {"--inputFilePattern=gs://test-bkt/", "--directoryWatchDurationInMinutes=42"}; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args) - .withValidation() - .as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(DataStreamToSpannerOptions.class); String inputFilePattern = options.getInputFilePattern(); Integer directoryWatchDurationInMinutes = options.getDirectoryWatchDurationInMinutes(); Integer expectedWatchDuration = 42; @@ -80,10 +77,8 @@ public void testGetSourceTypeWithEmptyStreamName() { expectedEx.expect(IllegalArgumentException.class); expectedEx.expectMessage("Stream name cannot be empty."); String[] args = new String[] {""}; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args) - .withValidation() - .as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(DataStreamToSpannerOptions.class); String result = DataStreamToSpanner.getSourceType(options); } @@ -95,16 +90,21 @@ public void testGetSourceTypeWithGcpCredentialsMissing() { new String[] { "--streamName=projects/sample-project/locations/sample-location/streams/sample-stream" }; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args) - .withValidation() - .as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(DataStreamToSpannerOptions.class); String result = DataStreamToSpanner.getSourceType(options); } + @Test + public void testGetShardingContext_Empty() { + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); + when(options.getSourceConfigURL()).thenReturn(""); + assertNotNull(DataStreamToSpanner.getShardingContext(options)); + } + @Test public void testConfigureSchemaOverrides_fileBased() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getSchemaOverridesFilePath()) .thenReturn( Resources.getResource("DataStreamToSpannerFileOverridesIT/override.json").getPath()); @@ -125,7 +125,7 @@ public void testConfigureSchemaOverrides_fileBased() { @Test public void testConfigureSchemaOverrides_stringBased() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getSchemaOverridesFilePath()).thenReturn(""); when(options.getTableOverrides()).thenReturn("[{person1, human1}]"); when(options.getColumnOverrides()).thenReturn("[{person1.first_name1, person1.name1}]"); @@ -144,7 +144,7 @@ public void testConfigureSchemaOverrides_stringBased() { @Test public void testConfigureSchemaOverrides_noOverrides() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getSchemaOverridesFilePath()).thenReturn(""); when(options.getTableOverrides()).thenReturn(""); when(options.getColumnOverrides()).thenReturn(""); @@ -156,7 +156,7 @@ public void testConfigureSchemaOverrides_noOverrides() { @Test public void testConfigureSchemaOverrides_incorrectConfiguration() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getSchemaOverridesFilePath()).thenReturn("/path/to/overrides.json"); when(options.getTableOverrides()).thenReturn("table1=schema1"); @@ -167,7 +167,7 @@ public void testConfigureSchemaOverrides_incorrectConfiguration() { @Test public void testConfigureSchemaOverrides_onlyTableOverrides() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getSchemaOverridesFilePath()).thenReturn(""); when(options.getTableOverrides()).thenReturn("[{person1, human1}]"); when(options.getColumnOverrides()).thenReturn(""); @@ -179,7 +179,7 @@ public void testConfigureSchemaOverrides_onlyTableOverrides() { @Test public void testConfigureSchemaOverrides_onlyColumnOverrides() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getSchemaOverridesFilePath()).thenReturn(""); when(options.getTableOverrides()).thenReturn(""); when(options.getColumnOverrides()).thenReturn("[{person1.first_name1, person1.name1}]"); @@ -191,7 +191,7 @@ public void testConfigureSchemaOverrides_onlyColumnOverrides() { @Test public void testConfigureSchemaOverrides_incorrectConfiguration_columnOverrides() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getSchemaOverridesFilePath()).thenReturn("/path/to/overrides.json"); when(options.getTableOverrides()).thenReturn(""); when(options.getColumnOverrides()).thenReturn("col1=col2"); @@ -203,7 +203,7 @@ public void testConfigureSchemaOverrides_incorrectConfiguration_columnOverrides( @Test public void testValidateSourceType_validSource() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getRunMode()).thenReturn(""); when(options.getDatastreamSourceType()).thenReturn("mysql"); @@ -214,7 +214,7 @@ public void testValidateSourceType_validSource() { @Test public void testValidateSourceType_invalidSource() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getRunMode()).thenReturn(""); when(options.getDatastreamSourceType()).thenReturn("invalid_source"); @@ -224,7 +224,7 @@ public void testValidateSourceType_invalidSource() { @Test public void testValidateSourceType_retryMode() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getRunMode()).thenReturn(Constants.RUN_MODE_RETRY_DLQ); DataStreamToSpanner.validateSourceType(options); @@ -234,7 +234,7 @@ public void testValidateSourceType_retryMode() { @Test public void testGetShadowTableSpannerConfig_validInput() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getShadowTableSpannerInstanceId()).thenReturn("shadow-instance-id"); when(options.getShadowTableSpannerDatabaseId()).thenReturn("shadow-database-id"); when(options.getProjectId()).thenReturn("project-id"); @@ -248,7 +248,7 @@ public void testGetShadowTableSpannerConfig_validInput() { @Test public void testGetShadowTableSpannerConfig_missingInstanceId() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getShadowTableSpannerInstanceId()).thenReturn(""); when(options.getShadowTableSpannerDatabaseId()).thenReturn("shadow-database-id"); @@ -259,7 +259,7 @@ public void testGetShadowTableSpannerConfig_missingInstanceId() { @Test public void testGetShadowTableSpannerConfig_missingDatabaseId() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getShadowTableSpannerInstanceId()).thenReturn("shadow-instance-id"); when(options.getShadowTableSpannerDatabaseId()).thenReturn(""); @@ -270,7 +270,7 @@ public void testGetShadowTableSpannerConfig_missingDatabaseId() { @Test public void testGetShadowTableSpannerConfig_defaultValues() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); when(options.getShadowTableSpannerInstanceId()).thenReturn(""); when(options.getShadowTableSpannerDatabaseId()).thenReturn(""); when(options.getInstanceId()).thenReturn("main-instance-id"); @@ -286,7 +286,7 @@ public void testGetShadowTableSpannerConfig_defaultValues() { @Test public void testBuildDlqManager_defaultTempLocation() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); DataflowPipelineOptions dfOptions = mock(DataflowPipelineOptions.class); when(options.as(DataflowPipelineOptions.class)).thenReturn(dfOptions); when(dfOptions.getTempLocation()).thenReturn("/tmp/test-bucket/temp"); @@ -300,7 +300,7 @@ public void testBuildDlqManager_defaultTempLocation() { @Test public void testBuildDlqManager_tempLocationWithTrailingSlash() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); DataflowPipelineOptions dfOptions = mock(DataflowPipelineOptions.class); when(options.as(DataflowPipelineOptions.class)).thenReturn(dfOptions); when(dfOptions.getTempLocation()).thenReturn("/tmp/test-bucket/temp/"); @@ -314,7 +314,7 @@ public void testBuildDlqManager_tempLocationWithTrailingSlash() { @Test public void testBuildDlqManager_withDlqDirectory() { - DataStreamToSpanner.Options options = mock(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = mock(DataStreamToSpannerOptions.class); DataflowPipelineOptions dfOptions = mock(DataflowPipelineOptions.class); when(options.as(DataflowPipelineOptions.class)).thenReturn(dfOptions); when(dfOptions.getTempLocation()).thenReturn("/tmp/test-bucket/temp"); @@ -349,8 +349,8 @@ public void testBuildPipeline() throws Exception { "--workerMachineType=n1-standard-4" }; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpannerOptions.class); Pipeline pipeline = DataStreamToSpanner.buildPipeline(options); @@ -380,8 +380,8 @@ public void testBuildPipeline_retryDLQ() throws Exception { "--workerMachineType=n1-standard-4" }; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpannerOptions.class); Pipeline pipeline = DataStreamToSpanner.buildPipeline(options); @@ -411,8 +411,8 @@ public void testBuildPipeline_retryAllDLQ() throws Exception { "--workerMachineType=n1-standard-4" }; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpannerOptions.class); Pipeline pipeline = DataStreamToSpanner.buildPipeline(options); @@ -443,8 +443,8 @@ public void testBuildPipeline_withDlqGcsPubSubSubscription() throws Exception { "--dlqGcsPubSubSubscription=projects/project-id/subscriptions/sub-id" }; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpannerOptions.class); Pipeline pipeline = DataStreamToSpanner.buildPipeline(options); @@ -517,8 +517,8 @@ public void testBuildPipeline_withAllOptions() throws Exception { "--shadowTableSpannerDatabaseId=shadow-database-id" // Set shadow database }; - DataStreamToSpanner.Options options = - PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpanner.Options.class); + DataStreamToSpannerOptions options = + PipelineOptionsFactory.fromArgs(args).as(DataStreamToSpannerOptions.class); Pipeline pipeline = DataStreamToSpanner.buildPipeline(options); diff --git a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/options/GCSSpannerDVOptions.java b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/options/GCSSpannerDVOptions.java new file mode 100644 index 0000000000..7181f9ecd7 --- /dev/null +++ b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/options/GCSSpannerDVOptions.java @@ -0,0 +1,214 @@ +/* + * Copyright (C) 2026 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 + * + * http://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.teleport.v2.options; + +import com.google.cloud.spanner.Options.RpcPriority; +import com.google.cloud.teleport.metadata.TemplateParameter; +import org.apache.beam.sdk.options.Default; +import org.apache.beam.sdk.options.PipelineOptions; + +/** + * Options supported by the pipeline. + * + *

Inherits standard configuration options. + */ +public interface GCSSpannerDVOptions extends PipelineOptions { + + @TemplateParameter.GcsReadFolder( + order = 1, + optional = true, + description = "GCS directory for AVRO files", + helpText = "This directory is used to read the AVRO files of the records read from source.", + example = "gs://your-bucket/your-path") + String getGcsInputDirectory(); + + void setGcsInputDirectory(String value); + + @TemplateParameter.ProjectId( + order = 2, + optional = true, + description = "Cloud Spanner Project Id.", + helpText = "This is the name of the Cloud Spanner project.") + String getProjectId(); + + void setProjectId(String projectId); + + @TemplateParameter.Text( + order = 3, + optional = true, + description = "Cloud Spanner Endpoint to call", + helpText = "The Cloud Spanner endpoint to call in the template.", + example = "https://batch-spanner.googleapis.com") + @Default.String("https://batch-spanner.googleapis.com") + String getSpannerHost(); + + void setSpannerHost(String value); + + @TemplateParameter.Text( + order = 4, + groupName = "Target", + description = "Cloud Spanner Instance Id.", + helpText = "The destination Cloud Spanner instance.") + String getInstanceId(); + + void setInstanceId(String value); + + @TemplateParameter.Text( + order = 5, + regexes = {"^[a-z]([a-z0-9_-]{0,28})[a-z0-9]$"}, + description = "Cloud Spanner Database Id.", + helpText = "The destination Cloud Spanner database.") + String getDatabaseId(); + + void setDatabaseId(String value); + + @TemplateParameter.Enum( + order = 6, + enumOptions = { + @TemplateParameter.TemplateEnumOption("LOW"), + @TemplateParameter.TemplateEnumOption("MEDIUM"), + @TemplateParameter.TemplateEnumOption("HIGH") + }, + optional = true, + description = "Priority for Spanner RPC invocations", + helpText = + "The request priority for Cloud Spanner calls. The value must be one of:" + + " [`HIGH`,`MEDIUM`,`LOW`]. Defaults to `HIGH`.") + @Default.Enum("HIGH") + RpcPriority getSpannerPriority(); + + void setSpannerPriority(RpcPriority value); + + @TemplateParameter.GcsReadFile( + order = 7, + optional = true, + description = + "Session File Path in Cloud Storage, to provide mapping information in the form of a session file", + helpText = + "Session file path in Cloud Storage that contains mapping information from" + + " Spanner Migration Tool") + @Default.String("") + String getSessionFilePath(); + + void setSessionFilePath(String value); + + @TemplateParameter.GcsReadFile( + order = 8, + optional = true, + description = "File based overrides from source to spanner", + helpText = + "A file which specifies the table and the column name overrides from source to spanner.") + @Default.String("") + String getSchemaOverridesFilePath(); + + void setSchemaOverridesFilePath(String value); + + @TemplateParameter.Text( + order = 9, + optional = true, + description = "Table name overrides from source to spanner", + regexes = + "^\\[([[:space:]]*\\{[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", + example = "[{Singers, Vocalists}, {Albums, Records}]", + helpText = + "These are the table name overrides from source to spanner. They are written in the" + + " following format: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]" + + " This example shows mapping Singers table to Vocalists and Albums table to Records.") + @Default.String("") + String getTableOverrides(); + + void setTableOverrides(String value); + + @TemplateParameter.Text( + order = 10, + optional = true, + regexes = + "^\\[([[:space:]]*\\{[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", + description = "Column name overrides from source to spanner", + example = "[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]", + helpText = + "These are the column name overrides from source to spanner. They are written in" + + " the following format: [{SourceTableName1.SourceColumnName1," + + " SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1," + + " SourceTableName2.SpannerColumnName1}]Note that the SourceTableName should" + + " remain the same in both the source and spanner pair. To override table names," + + " use tableOverrides.The example shows mapping SingerName to TalentName and" + + " AlbumName to RecordName in Singers and Albums table respectively.") + @Default.String("") + String getColumnOverrides(); + + void setColumnOverrides(String value); + + @TemplateParameter.Text( + order = 11, + optional = false, + regexes = {"^[^ ;]*$"}, + description = "BigQuery dataset for reporting", + helpText = "The BigQuery dataset ID where the validation results will be stored.", + example = "validation_report_dataset") + String getBigQueryDataset(); + + void setBigQueryDataset(String value); + + @TemplateParameter.Text( + order = 12, + optional = true, + regexes = {"^[^ ;]*$"}, + description = "Run ID for the validation job", + helpText = + "A unique identifier for the validation run. If not provided, the Dataflow Job Name" + + " will be used.", + example = "run_20230101_120000") + String getRunId(); + + void setRunId(String value); + + @TemplateParameter.GcsReadFile( + order = 13, + optional = true, + description = "Custom jar location in Cloud Storage", + helpText = + "Custom jar location in Cloud Storage that contains the custom transformation logic for" + + " processing records.") + @Default.String("") + String getTransformationJarPath(); + + void setTransformationJarPath(String value); + + @TemplateParameter.Text( + order = 14, + optional = true, + description = "Custom class name", + helpText = + "Fully qualified class name having the custom transformation logic. It is a" + + " mandatory field in case transformationJarPath is specified") + @Default.String("") + String getTransformationClassName(); + + void setTransformationClassName(String value); + + @TemplateParameter.Text( + order = 15, + optional = true, + description = "Custom parameters for transformation", + helpText = + "String containing any custom parameters to be passed to the custom transformation" + + " class.") + @Default.String("") + String getTransformationCustomParameters(); + + void setTransformationCustomParameters(String value); +} diff --git a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/options/package-info.java b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/options/package-info.java new file mode 100644 index 0000000000..dbdf77b046 --- /dev/null +++ b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/options/package-info.java @@ -0,0 +1,18 @@ +/* + * Copyright (C) 2026 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 + * + * http://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. + */ + +/** Options classes. */ +package com.google.cloud.teleport.v2.options; diff --git a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/templates/GCSSpannerDV.java b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/templates/GCSSpannerDV.java index 35a6011291..801446aeaf 100644 --- a/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/templates/GCSSpannerDV.java +++ b/v2/gcs-spanner-dv/src/main/java/com/google/cloud/teleport/v2/templates/GCSSpannerDV.java @@ -18,13 +18,12 @@ import static com.google.cloud.teleport.v2.constants.GCSSpannerDVConstants.SOURCE_TAG; import static com.google.cloud.teleport.v2.constants.GCSSpannerDVConstants.SPANNER_TAG; -import com.google.cloud.spanner.Options.RpcPriority; import com.google.cloud.teleport.metadata.Template; import com.google.cloud.teleport.metadata.TemplateCategory; -import com.google.cloud.teleport.metadata.TemplateParameter; import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger; import com.google.cloud.teleport.v2.dto.ComparisonRecord; import com.google.cloud.teleport.v2.fn.SchemaMapperProviderFn; +import com.google.cloud.teleport.v2.options.GCSSpannerDVOptions; import com.google.cloud.teleport.v2.spanner.ddl.Ddl; import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper; import com.google.cloud.teleport.v2.spanner.migrations.transformation.CustomTransformation; @@ -37,8 +36,6 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig; -import org.apache.beam.sdk.options.Default; -import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.transforms.SerializableFunction; @@ -54,7 +51,7 @@ description = "Batch pipeline that reads data from GCS and Spanner compares them to validate migration" + " correctness.", - optionsClass = GCSSpannerDV.Options.class, + optionsClass = GCSSpannerDVOptions.class, flexContainerName = "gcs-spanner-dv", documentation = "https://cloud.google.com/dataflow/docs/guides/templates/provided/gcs-spanner-dv", @@ -67,203 +64,15 @@ }) public class GCSSpannerDV { - public interface Options extends PipelineOptions { - - @TemplateParameter.GcsReadFolder( - order = 1, - optional = true, - description = "GCS directory for AVRO files", - helpText = "This directory is used to read the AVRO files of the records read from source.", - example = "gs://your-bucket/your-path") - String getGcsInputDirectory(); - - void setGcsInputDirectory(String value); - - @TemplateParameter.ProjectId( - order = 2, - optional = true, - description = "Cloud Spanner Project Id.", - helpText = "This is the name of the Cloud Spanner project.") - String getProjectId(); - - void setProjectId(String projectId); - - @TemplateParameter.Text( - order = 3, - optional = true, - description = "Cloud Spanner Endpoint to call", - helpText = "The Cloud Spanner endpoint to call in the template.", - example = "https://batch-spanner.googleapis.com") - @Default.String("https://batch-spanner.googleapis.com") - String getSpannerHost(); - - void setSpannerHost(String value); - - @TemplateParameter.Text( - order = 4, - groupName = "Target", - description = "Cloud Spanner Instance Id.", - helpText = "The destination Cloud Spanner instance.") - String getInstanceId(); - - void setInstanceId(String value); - - @TemplateParameter.Text( - order = 5, - regexes = {"^[a-z]([a-z0-9_-]{0,28})[a-z0-9]$"}, - description = "Cloud Spanner Database Id.", - helpText = "The destination Cloud Spanner database.") - String getDatabaseId(); - - void setDatabaseId(String value); - - @TemplateParameter.Enum( - order = 6, - enumOptions = { - @TemplateParameter.TemplateEnumOption("LOW"), - @TemplateParameter.TemplateEnumOption("MEDIUM"), - @TemplateParameter.TemplateEnumOption("HIGH") - }, - optional = true, - description = "Priority for Spanner RPC invocations", - helpText = - "The request priority for Cloud Spanner calls. The value must be one of:" - + " [`HIGH`,`MEDIUM`,`LOW`]. Defaults to `HIGH`.") - @Default.Enum("HIGH") - RpcPriority getSpannerPriority(); - - void setSpannerPriority(RpcPriority value); - - @TemplateParameter.GcsReadFile( - order = 7, - optional = true, - description = - "Session File Path in Cloud Storage, to provide mapping information in the form of a session file", - helpText = - "Session file path in Cloud Storage that contains mapping information from" - + " Spanner Migration Tool") - @Default.String("") - String getSessionFilePath(); - - void setSessionFilePath(String value); - - @TemplateParameter.GcsReadFile( - order = 8, - optional = true, - description = "File based overrides from source to spanner", - helpText = - "A file which specifies the table and the column name overrides from source to spanner.") - @Default.String("") - String getSchemaOverridesFilePath(); - - void setSchemaOverridesFilePath(String value); - - @TemplateParameter.Text( - order = 9, - optional = true, - description = "Table name overrides from source to spanner", - regexes = - "^\\[([[:space:]]*\\{[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", - example = "[{Singers, Vocalists}, {Albums, Records}]", - helpText = - "These are the table name overrides from source to spanner. They are written in the" - + " following format: [{SourceTableName1, SpannerTableName1}, {SourceTableName2, SpannerTableName2}]" - + " This example shows mapping Singers table to Vocalists and Albums table to Records.") - @Default.String("") - String getTableOverrides(); - - void setTableOverrides(String value); - - @TemplateParameter.Text( - order = 10, - optional = true, - regexes = - "^\\[([[:space:]]*\\{[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", - description = "Column name overrides from source to spanner", - example = - "[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]", - helpText = - "These are the column name overrides from source to spanner. They are written in" - + " the following format: [{SourceTableName1.SourceColumnName1," - + " SourceTableName1.SpannerColumnName1}, {SourceTableName2.SourceColumnName1," - + " SourceTableName2.SpannerColumnName1}]Note that the SourceTableName should" - + " remain the same in both the source and spanner pair. To override table names," - + " use tableOverrides.The example shows mapping SingerName to TalentName and" - + " AlbumName to RecordName in Singers and Albums table respectively.") - @Default.String("") - String getColumnOverrides(); - - void setColumnOverrides(String value); - - @TemplateParameter.Text( - order = 11, - optional = false, - regexes = {"^[^ ;]*$"}, - description = "BigQuery dataset for reporting", - helpText = "The BigQuery dataset ID where the validation results will be stored.", - example = "validation_report_dataset") - String getBigQueryDataset(); - - void setBigQueryDataset(String value); - - @TemplateParameter.Text( - order = 12, - optional = true, - regexes = {"^[^ ;]*$"}, - description = "Run ID for the validation job", - helpText = - "A unique identifier for the validation run. If not provided, the Dataflow Job Name" - + " will be used.", - example = "run_20230101_120000") - String getRunId(); - - void setRunId(String value); - - @TemplateParameter.GcsReadFile( - order = 13, - optional = true, - description = "Custom jar location in Cloud Storage", - helpText = - "Custom jar location in Cloud Storage that contains the custom transformation logic for" - + " processing records.") - @Default.String("") - String getTransformationJarPath(); - - void setTransformationJarPath(String value); - - @TemplateParameter.Text( - order = 14, - optional = true, - description = "Custom class name", - helpText = - "Fully qualified class name having the custom transformation logic. It is a" - + " mandatory field in case transformationJarPath is specified") - @Default.String("") - String getTransformationClassName(); - - void setTransformationClassName(String value); - - @TemplateParameter.Text( - order = 15, - optional = true, - description = "Custom parameters for transformation", - helpText = - "String containing any custom parameters to be passed to the custom transformation" - + " class.") - @Default.String("") - String getTransformationCustomParameters(); - - void setTransformationCustomParameters(String value); - } - public static void main(String[] args) { UncaughtExceptionLogger.register(); - Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class); + GCSSpannerDVOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(GCSSpannerDVOptions.class); run(options); } - public static PipelineResult run(Options options) { + public static PipelineResult run(GCSSpannerDVOptions options) { Pipeline pipeline = Pipeline.create(options); SpannerConfig spannerConfig = createSpannerConfig(options); @@ -326,7 +135,7 @@ public static PipelineResult run(Options options) { } @VisibleForTesting - static SpannerConfig createSpannerConfig(Options options) { + static SpannerConfig createSpannerConfig(GCSSpannerDVOptions options) { return SpannerConfig.create() .withProjectId(ValueProvider.StaticValueProvider.of(options.getProjectId())) .withHost(ValueProvider.StaticValueProvider.of(options.getSpannerHost())) diff --git a/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVTest.java b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVTest.java new file mode 100644 index 0000000000..dbebe3a033 --- /dev/null +++ b/v2/gcs-spanner-dv/src/test/java/com/google/cloud/teleport/v2/templates/GCSSpannerDVTest.java @@ -0,0 +1,39 @@ +/* + * Copyright (C) 2026 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 + * + * http://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.teleport.v2.templates; + +import static org.junit.Assert.assertNotNull; + +import com.google.cloud.teleport.v2.options.GCSSpannerDVOptions; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class GCSSpannerDVTest { + + @Test + public void testCreateSpannerConfig() { + String[] args = + new String[] { + "--projectId=test-project", "--instanceId=test-instance", "--databaseId=test-database" + }; + GCSSpannerDVOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(GCSSpannerDVOptions.class); + assertNotNull(GCSSpannerDV.createSpannerConfig(options)); + } +} diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/options/SpannerToSourceDbOptions.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/options/SpannerToSourceDbOptions.java new file mode 100644 index 0000000000..0a820bd436 --- /dev/null +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/options/SpannerToSourceDbOptions.java @@ -0,0 +1,462 @@ +/* + * Copyright (C) 2026 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 + * + * http://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.teleport.v2.options; + +import static com.google.cloud.teleport.v2.spanner.migrations.constants.Constants.RUN_MODE_REGULAR; +import static com.google.cloud.teleport.v2.spanner.migrations.constants.Constants.RUN_MODE_RETRY_ALL_DLQ; +import static com.google.cloud.teleport.v2.spanner.migrations.constants.Constants.RUN_MODE_RETRY_DLQ; + +import com.google.cloud.spanner.Options.RpcPriority; +import com.google.cloud.teleport.metadata.TemplateParameter; +import com.google.cloud.teleport.metadata.TemplateParameter.TemplateEnumOption; +import org.apache.beam.sdk.options.Default; +import org.apache.beam.sdk.options.StreamingOptions; + +/** + * Options supported by the pipeline. + * + *

Inherits standard configuration options. + */ +public interface SpannerToSourceDbOptions extends CommonTemplateOptions, StreamingOptions { + + @TemplateParameter.Text( + order = 1, + optional = false, + description = "Name of the change stream to read from", + helpText = "This is the name of the Spanner change stream that the pipeline will read from.") + String getChangeStreamName(); + + void setChangeStreamName(String value); + + @TemplateParameter.Text( + order = 2, + optional = false, + description = "Cloud Spanner Instance Id.", + helpText = + "This is the name of the Cloud Spanner instance where the changestream is present.") + String getInstanceId(); + + void setInstanceId(String value); + + @TemplateParameter.Text( + order = 3, + optional = false, + description = "Cloud Spanner Database Id.", + helpText = + "This is the name of the Cloud Spanner database that the changestream is monitoring") + String getDatabaseId(); + + void setDatabaseId(String value); + + @TemplateParameter.ProjectId( + order = 4, + optional = false, + description = "Cloud Spanner Project Id.", + helpText = "This is the name of the Cloud Spanner project.") + String getSpannerProjectId(); + + void setSpannerProjectId(String projectId); + + @TemplateParameter.Text( + order = 5, + optional = false, + description = "Cloud Spanner Instance to store metadata when reading from changestreams", + helpText = + "This is the instance to store the metadata used by the connector to control the" + + " consumption of the change stream API data.") + String getMetadataInstance(); + + void setMetadataInstance(String value); + + @TemplateParameter.Text( + order = 6, + optional = false, + description = "Cloud Spanner Database to store metadata when reading from changestreams", + helpText = + "This is the database to store the metadata used by the connector to control the" + + " consumption of the change stream API data.") + String getMetadataDatabase(); + + void setMetadataDatabase(String value); + + @TemplateParameter.Text( + order = 36, + optional = true, + description = "Cloud Spanner Database to store change stream connector metadata", + helpText = + "This is the database to store the metadata used by the change stream connector. " + + "If not provided, it defaults to the metadata database.") + String getChangeStreamMetadataDatabase(); + + void setChangeStreamMetadataDatabase(String value); + + @TemplateParameter.Text( + order = 7, + optional = true, + description = "Cloud Spanner metadata table name", + helpText = + "The Spanner change streams connector metadata table name to use. If not provided," + + " Spanner automatically creates the streams connector metadata table during the pipeline flow" + + " change. You must provide this parameter when updating an existing pipeline to ensure" + + " that the metadata table from the original job is carried over.") + String getSpannerMetadataTableName(); + + void setSpannerMetadataTableName(String value); + + @TemplateParameter.Text( + order = 8, + optional = true, + description = "Changes are read from the given timestamp", + helpText = "Read changes from the given timestamp.") + @Default.String("") + String getStartTimestamp(); + + void setStartTimestamp(String value); + + @TemplateParameter.Text( + order = 9, + optional = true, + description = "Changes are read until the given timestamp", + helpText = + "Read changes until the given timestamp. If no timestamp provided, reads indefinitely.") + @Default.String("") + String getEndTimestamp(); + + void setEndTimestamp(String value); + + @TemplateParameter.Text( + order = 10, + optional = true, + description = "Cloud Spanner shadow table prefix.", + helpText = "The prefix used to name shadow tables. Default: `shadow_`.") + @Default.String("rev_shadow_") + String getShadowTablePrefix(); + + void setShadowTablePrefix(String value); + + @TemplateParameter.GcsReadFile( + order = 11, + optional = false, + description = "Path to GCS file containing the the Source shard details", + helpText = "Path to GCS file containing connection profile info for source shards.") + String getSourceShardsFilePath(); + + void setSourceShardsFilePath(String value); + + @TemplateParameter.GcsReadFile( + order = 12, + optional = true, + description = "Session File Path in Cloud Storage", + helpText = + "Session file path in Cloud Storage that contains mapping information from" + + " HarbourBridge") + String getSessionFilePath(); + + void setSessionFilePath(String value); + + @TemplateParameter.Enum( + order = 13, + optional = true, + enumOptions = {@TemplateEnumOption("none"), @TemplateEnumOption("forward_migration")}, + description = "Filtration mode", + helpText = + "Mode of Filtration, decides how to drop certain records based on a criteria. Currently" + + " supported modes are: none (filter nothing), forward_migration (filter records" + + " written via the forward migration pipeline). Defaults to forward_migration.") + @Default.String("forward_migration") + String getFiltrationMode(); + + void setFiltrationMode(String value); + + @TemplateParameter.GcsReadFile( + order = 14, + optional = true, + description = "Custom jar location in Cloud Storage", + helpText = + "Custom jar location in Cloud Storage that contains the customization logic" + + " for fetching shard id.") + @Default.String("") + String getShardingCustomJarPath(); + + void setShardingCustomJarPath(String value); + + @TemplateParameter.Text( + order = 15, + optional = true, + description = "Custom class name", + helpText = + "Fully qualified class name having the custom shard id implementation. It is a" + + " mandatory field in case shardingCustomJarPath is specified") + @Default.String("") + String getShardingCustomClassName(); + + void setShardingCustomClassName(String value); + + @TemplateParameter.Text( + order = 16, + optional = true, + description = "Custom sharding logic parameters", + helpText = + "String containing any custom parameters to be passed to the custom sharding class.") + @Default.String("") + String getShardingCustomParameters(); + + void setShardingCustomParameters(String value); + + @TemplateParameter.Text( + order = 17, + optional = true, + description = "SourceDB timezone offset", + helpText = + "This is the timezone offset from UTC for the source database. Example value: +10:00") + @Default.String("+00:00") + String getSourceDbTimezoneOffset(); + + void setSourceDbTimezoneOffset(String value); + + @TemplateParameter.PubsubSubscription( + order = 18, + optional = true, + description = + "The Pub/Sub subscription being used in a Cloud Storage notification policy for DLQ" + + " retry directory when running in regular mode.", + helpText = + "The Pub/Sub subscription being used in a Cloud Storage notification policy for DLQ" + + " retry directory when running in regular mode. The name should be in the format" + + " of projects//subscriptions/. When set, the" + + " deadLetterQueueDirectory and dlqRetryMinutes are ignored.") + String getDlqGcsPubSubSubscription(); + + void setDlqGcsPubSubSubscription(String value); + + @TemplateParameter.Text( + order = 19, + optional = true, + description = "Directory name for holding skipped records", + helpText = + "Records skipped from reverse replication are written to this directory. Default" + + " directory name is skip.") + @Default.String("skip") + String getSkipDirectoryName(); + + void setSkipDirectoryName(String value); + + @TemplateParameter.Long( + order = 20, + optional = true, + description = "Maximum connections per shard.", + helpText = "This will come from shard file eventually.") + @Default.Long(10000) + Long getMaxShardConnections(); + + void setMaxShardConnections(Long value); + + @TemplateParameter.Text( + order = 21, + optional = true, + description = "Dead letter queue directory.", + helpText = + "The file path used when storing the error queue output. " + + "The default file path is a directory under the Dataflow job's temp location.") + @Default.String("") + String getDeadLetterQueueDirectory(); + + void setDeadLetterQueueDirectory(String value); + + @TemplateParameter.Integer( + order = 22, + optional = true, + description = "Dead letter queue maximum retry count", + helpText = + "The max number of times temporary errors can be retried through DLQ. Defaults to 500.") + @Default.Integer(500) + Integer getDlqMaxRetryCount(); + + void setDlqMaxRetryCount(Integer value); + + @TemplateParameter.Enum( + order = 23, + optional = true, + description = "Run mode - currently supported are : regular, retryDLQ, or retryAllDLQ", + enumOptions = { + @TemplateEnumOption(RUN_MODE_REGULAR), + @TemplateEnumOption(RUN_MODE_RETRY_DLQ), + @TemplateEnumOption(RUN_MODE_RETRY_ALL_DLQ) + }, + helpText = + "This is the run mode type. Default is regular. Use `retryDLQ` mode to process exclusively severe error files concurrently with your reverse migration pipeline. Use `retryAllDLQ` mode only when the regular pipeline is stopped. This mode processes both retry and severe directories. Do NOT run `retryAllDLQ` concurrently with any active pipeline as it will cause conflicts.") + @Default.String(RUN_MODE_REGULAR) + String getRunMode(); + + void setRunMode(String value); + + @TemplateParameter.Integer( + order = 24, + optional = true, + description = "Dead letter queue retry minutes", + helpText = "The number of minutes between dead letter queue retries. Defaults to 10.") + @Default.Integer(10) + Integer getDlqRetryMinutes(); + + void setDlqRetryMinutes(Integer value); + + @TemplateParameter.Enum( + order = 25, + optional = true, + description = "Source database type, ex: mysql", + enumOptions = { + @TemplateEnumOption("mysql"), + @TemplateEnumOption("cassandra"), + @TemplateEnumOption("postgresql"), + @TemplateEnumOption("spanner"), + @TemplateEnumOption("oracle") + }, + helpText = "The type of source database to reverse replicate to.") + @Default.String("mysql") + String getSourceType(); + + void setSourceType(String value); + + @TemplateParameter.GcsReadFile( + order = 26, + optional = true, + description = "Custom transformation jar location in Cloud Storage", + helpText = + "Custom jar location in Cloud Storage that contains the custom transformation logic for processing records" + + " in reverse replication.") + @Default.String("") + String getTransformationJarPath(); + + void setTransformationJarPath(String value); + + @TemplateParameter.Text( + order = 27, + optional = true, + description = "Custom class name for transformation", + helpText = + "Fully qualified class name having the custom transformation logic. It is a" + + " mandatory field in case transformationJarPath is specified") + @Default.String("") + String getTransformationClassName(); + + void setTransformationClassName(String value); + + @TemplateParameter.Text( + order = 28, + optional = true, + description = "Custom parameters for transformation", + helpText = + "String containing any custom parameters to be passed to the custom transformation class.") + @Default.String("") + String getTransformationCustomParameters(); + + void setTransformationCustomParameters(String value); + + @TemplateParameter.Text( + order = 29, + optional = true, + description = "Table name overrides from spanner to source", + regexes = + "^\\[([[:space:]]*\\{[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", + example = "[{Singers, Vocalists}, {Albums, Records}]", + helpText = + "These are the table name overrides from spanner to source. They are written in the" + + "following format: [{SpannerTableName1, SourceTableName1}, {SpannerTableName2, SourceTableName2}]" + + "This example shows mapping Singers table to Vocalists and Albums table to Records.") + @Default.String("") + String getTableOverrides(); + + void setTableOverrides(String value); + + @TemplateParameter.Text( + order = 30, + optional = true, + description = "Column name overrides from spanner to source", + regexes = + "^\\[([[:space:]]*\\{[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", + example = "[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]", + helpText = + "These are the column name overrides from spanner to source. They are written in the" + + "following format: [{SpannerTableName1.SpannerColumnName1, SpannerTableName1.SourceColumnName1}, {SpannerTableName2.SpannerColumnName1, SpannerTableName2.SourceColumnName1}]" + + "Note that the SpannerTableName should remain the same in both the spanner and source pair. To override table names, use tableOverrides." + + "The example shows mapping SingerName to TalentName and AlbumName to RecordName in Singers and Albums table respectively.") + @Default.String("") + String getColumnOverrides(); + + void setColumnOverrides(String value); + + @TemplateParameter.GcsReadFile( + order = 31, + optional = true, + description = "File based overrides from spanner to source", + helpText = + "A file which specifies the table and the column name overrides from spanner to source.") + @Default.String("") + String getSchemaOverridesFilePath(); + + void setSchemaOverridesFilePath(String value); + + @TemplateParameter.Text( + order = 32, + optional = true, + description = "Directory name for holding filtered records", + helpText = + "Records skipped from reverse replication are written to this directory. Default" + + " directory name is skip.") + @Default.String("filteredEvents") + String getFilterEventsDirectoryName(); + + void setFilterEventsDirectoryName(String value); + + @TemplateParameter.Boolean( + order = 33, + optional = true, + description = "Boolean setting if reverse migration is sharded", + helpText = + "Sets the template to a sharded migration. If source shard template contains more" + + " than one shard, the value will be set to true. This value defaults to false.") + @Default.Boolean(false) + Boolean getIsShardedMigration(); + + void setIsShardedMigration(Boolean value); + + @TemplateParameter.Text( + order = 34, + optional = true, + description = "Failure injection parameter", + helpText = "Failure injection parameter. Only used for testing.") + @Default.String("") + String getFailureInjectionParameter(); + + void setFailureInjectionParameter(String value); + + @TemplateParameter.Enum( + order = 35, + enumOptions = { + @TemplateEnumOption("LOW"), + @TemplateEnumOption("MEDIUM"), + @TemplateEnumOption("HIGH") + }, + optional = true, + description = "Priority for Spanner RPC invocations", + helpText = + "The request priority for Cloud Spanner calls. The value must be one of:" + + " [`HIGH`,`MEDIUM`,`LOW`]. Defaults to `HIGH`.") + @Default.Enum("HIGH") + RpcPriority getSpannerPriority(); + + void setSpannerPriority(RpcPriority value); +} diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/options/package-info.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/options/package-info.java new file mode 100644 index 0000000000..dbdf77b046 --- /dev/null +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/options/package-info.java @@ -0,0 +1,18 @@ +/* + * Copyright (C) 2026 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 + * + * http://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. + */ + +/** Options classes. */ +package com.google.cloud.teleport.v2.options; diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDb.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDb.java index 30fb8300ec..e0e98814ea 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDb.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDb.java @@ -16,28 +16,23 @@ package com.google.cloud.teleport.v2.templates; import static com.google.cloud.teleport.v2.spanner.migrations.constants.Constants.RUN_MODE_REGULAR; -import static com.google.cloud.teleport.v2.spanner.migrations.constants.Constants.RUN_MODE_RETRY_ALL_DLQ; import static com.google.cloud.teleport.v2.spanner.migrations.constants.Constants.RUN_MODE_RETRY_DLQ; import com.google.cloud.Timestamp; -import com.google.cloud.spanner.Options.RpcPriority; import com.google.cloud.teleport.metadata.Template; import com.google.cloud.teleport.metadata.TemplateCategory; -import com.google.cloud.teleport.metadata.TemplateParameter; -import com.google.cloud.teleport.metadata.TemplateParameter.TemplateEnumOption; import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager; import com.google.cloud.teleport.v2.cdc.dlq.PubSubNotifiedDlqIO; import com.google.cloud.teleport.v2.cdc.dlq.StringDeadLetterQueueSanitizer; import com.google.cloud.teleport.v2.coders.FailsafeElementCoder; import com.google.cloud.teleport.v2.common.CommonTemplateJvmInitializer; import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger; -import com.google.cloud.teleport.v2.options.CommonTemplateOptions; +import com.google.cloud.teleport.v2.options.SpannerToSourceDbOptions; import com.google.cloud.teleport.v2.spanner.ddl.Ddl; import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; import com.google.cloud.teleport.v2.spanner.migrations.transformation.CustomTransformation; import com.google.cloud.teleport.v2.spanner.migrations.utils.DataflowWorkerMachineTypeUtils; import com.google.cloud.teleport.v2.spanner.sourceddl.SourceSchema; -import com.google.cloud.teleport.v2.templates.SpannerToSourceDb.Options; import com.google.cloud.teleport.v2.templates.changestream.TrimmedShardedDataChangeRecord; import com.google.cloud.teleport.v2.templates.constants.Constants; import com.google.cloud.teleport.v2.templates.dbutils.processor.ISpToSrcSourceConnector; @@ -70,9 +65,7 @@ import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig; import org.apache.beam.sdk.io.gcp.spanner.SpannerIO; import org.apache.beam.sdk.io.gcp.spanner.SpannerServiceFactoryImpl; -import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.PipelineOptionsFactory; -import org.apache.beam.sdk.options.StreamingOptions; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.transforms.Flatten; import org.apache.beam.sdk.transforms.MapElements; @@ -94,7 +87,7 @@ description = "Streaming pipeline. Reads data from Spanner Change Streams and" + " writes them to a source.", - optionsClass = Options.class, + optionsClass = SpannerToSourceDbOptions.class, flexContainerName = "spanner-to-sourcedb", contactInformation = "https://cloud.google.com/support", hidden = false, @@ -111,444 +104,6 @@ public class SpannerToSourceDb { private static final String MYSQL_JDBC_PREFIX = "jdbc:mysql://"; private static final String POSTGRESQL_JDBC_PREFIX = "jdbc:postgresql://"; - /** - * Options supported by the pipeline. - * - *

Inherits standard configuration options. - */ - public interface Options extends CommonTemplateOptions, StreamingOptions { - - @TemplateParameter.Text( - order = 1, - optional = false, - description = "Name of the change stream to read from", - helpText = - "This is the name of the Spanner change stream that the pipeline will read from.") - String getChangeStreamName(); - - void setChangeStreamName(String value); - - @TemplateParameter.Text( - order = 2, - optional = false, - description = "Cloud Spanner Instance Id.", - helpText = - "This is the name of the Cloud Spanner instance where the changestream is present.") - String getInstanceId(); - - void setInstanceId(String value); - - @TemplateParameter.Text( - order = 3, - optional = false, - description = "Cloud Spanner Database Id.", - helpText = - "This is the name of the Cloud Spanner database that the changestream is monitoring") - String getDatabaseId(); - - void setDatabaseId(String value); - - @TemplateParameter.ProjectId( - order = 4, - optional = false, - description = "Cloud Spanner Project Id.", - helpText = "This is the name of the Cloud Spanner project.") - String getSpannerProjectId(); - - void setSpannerProjectId(String projectId); - - @TemplateParameter.Text( - order = 5, - optional = false, - description = "Cloud Spanner Instance to store metadata when reading from changestreams", - helpText = - "This is the instance to store the metadata used by the connector to control the" - + " consumption of the change stream API data.") - String getMetadataInstance(); - - void setMetadataInstance(String value); - - @TemplateParameter.Text( - order = 6, - optional = false, - description = "Cloud Spanner Database to store metadata when reading from changestreams", - helpText = - "This is the database to store the metadata used by the connector to control the" - + " consumption of the change stream API data.") - String getMetadataDatabase(); - - void setMetadataDatabase(String value); - - @TemplateParameter.Text( - order = 36, - optional = true, - description = "Cloud Spanner Database to store change stream connector metadata", - helpText = - "This is the database to store the metadata used by the change stream connector. " - + "If not provided, it defaults to the metadata database.") - String getChangeStreamMetadataDatabase(); - - void setChangeStreamMetadataDatabase(String value); - - @TemplateParameter.Text( - order = 7, - optional = true, - description = "Cloud Spanner metadata table name", - helpText = - "The Spanner change streams connector metadata table name to use. If not provided," - + " Spanner automatically creates the streams connector metadata table during the pipeline flow" - + " change. You must provide this parameter when updating an existing pipeline to ensure" - + " that the metadata table from the original job is carried over.") - String getSpannerMetadataTableName(); - - void setSpannerMetadataTableName(String value); - - @TemplateParameter.Text( - order = 8, - optional = true, - description = "Changes are read from the given timestamp", - helpText = "Read changes from the given timestamp.") - @Default.String("") - String getStartTimestamp(); - - void setStartTimestamp(String value); - - @TemplateParameter.Text( - order = 9, - optional = true, - description = "Changes are read until the given timestamp", - helpText = - "Read changes until the given timestamp. If no timestamp provided, reads indefinitely.") - @Default.String("") - String getEndTimestamp(); - - void setEndTimestamp(String value); - - @TemplateParameter.Text( - order = 10, - optional = true, - description = "Cloud Spanner shadow table prefix.", - helpText = "The prefix used to name shadow tables. Default: `shadow_`.") - @Default.String("rev_shadow_") - String getShadowTablePrefix(); - - void setShadowTablePrefix(String value); - - @TemplateParameter.GcsReadFile( - order = 11, - optional = false, - description = "Path to GCS file containing the the Source shard details", - helpText = "Path to GCS file containing connection profile info for source shards.") - String getSourceShardsFilePath(); - - void setSourceShardsFilePath(String value); - - @TemplateParameter.GcsReadFile( - order = 12, - optional = true, - description = "Session File Path in Cloud Storage", - helpText = - "Session file path in Cloud Storage that contains mapping information from" - + " HarbourBridge") - String getSessionFilePath(); - - void setSessionFilePath(String value); - - @TemplateParameter.Enum( - order = 13, - optional = true, - enumOptions = {@TemplateEnumOption("none"), @TemplateEnumOption("forward_migration")}, - description = "Filtration mode", - helpText = - "Mode of Filtration, decides how to drop certain records based on a criteria. Currently" - + " supported modes are: none (filter nothing), forward_migration (filter records" - + " written via the forward migration pipeline). Defaults to forward_migration.") - @Default.String("forward_migration") - String getFiltrationMode(); - - void setFiltrationMode(String value); - - @TemplateParameter.GcsReadFile( - order = 14, - optional = true, - description = "Custom jar location in Cloud Storage", - helpText = - "Custom jar location in Cloud Storage that contains the customization logic" - + " for fetching shard id.") - @Default.String("") - String getShardingCustomJarPath(); - - void setShardingCustomJarPath(String value); - - @TemplateParameter.Text( - order = 15, - optional = true, - description = "Custom class name", - helpText = - "Fully qualified class name having the custom shard id implementation. It is a" - + " mandatory field in case shardingCustomJarPath is specified") - @Default.String("") - String getShardingCustomClassName(); - - void setShardingCustomClassName(String value); - - @TemplateParameter.Text( - order = 16, - optional = true, - description = "Custom sharding logic parameters", - helpText = - "String containing any custom parameters to be passed to the custom sharding class.") - @Default.String("") - String getShardingCustomParameters(); - - void setShardingCustomParameters(String value); - - @TemplateParameter.Text( - order = 17, - optional = true, - description = "SourceDB timezone offset", - helpText = - "This is the timezone offset from UTC for the source database. Example value: +10:00") - @Default.String("+00:00") - String getSourceDbTimezoneOffset(); - - void setSourceDbTimezoneOffset(String value); - - @TemplateParameter.PubsubSubscription( - order = 18, - optional = true, - description = - "The Pub/Sub subscription being used in a Cloud Storage notification policy for DLQ" - + " retry directory when running in regular mode.", - helpText = - "The Pub/Sub subscription being used in a Cloud Storage notification policy for DLQ" - + " retry directory when running in regular mode. The name should be in the format" - + " of projects//subscriptions/. When set, the" - + " deadLetterQueueDirectory and dlqRetryMinutes are ignored.") - String getDlqGcsPubSubSubscription(); - - void setDlqGcsPubSubSubscription(String value); - - @TemplateParameter.Text( - order = 19, - optional = true, - description = "Directory name for holding skipped records", - helpText = - "Records skipped from reverse replication are written to this directory. Default" - + " directory name is skip.") - @Default.String("skip") - String getSkipDirectoryName(); - - void setSkipDirectoryName(String value); - - @TemplateParameter.Long( - order = 20, - optional = true, - description = "Maximum connections per shard.", - helpText = "This will come from shard file eventually.") - @Default.Long(10000) - Long getMaxShardConnections(); - - void setMaxShardConnections(Long value); - - @TemplateParameter.Text( - order = 21, - optional = true, - description = "Dead letter queue directory.", - helpText = - "The file path used when storing the error queue output. " - + "The default file path is a directory under the Dataflow job's temp location.") - @Default.String("") - String getDeadLetterQueueDirectory(); - - void setDeadLetterQueueDirectory(String value); - - @TemplateParameter.Integer( - order = 22, - optional = true, - description = "Dead letter queue maximum retry count", - helpText = - "The max number of times temporary errors can be retried through DLQ. Defaults to 500.") - @Default.Integer(500) - Integer getDlqMaxRetryCount(); - - void setDlqMaxRetryCount(Integer value); - - @TemplateParameter.Enum( - order = 23, - optional = true, - description = "Run mode - currently supported are : regular, retryDLQ, or retryAllDLQ", - enumOptions = { - @TemplateEnumOption(RUN_MODE_REGULAR), - @TemplateEnumOption(RUN_MODE_RETRY_DLQ), - @TemplateEnumOption(RUN_MODE_RETRY_ALL_DLQ) - }, - helpText = - "This is the run mode type. Default is regular. Use `retryDLQ` mode to process exclusively severe error files concurrently with your reverse migration pipeline. Use `retryAllDLQ` mode only when the regular pipeline is stopped. This mode processes both retry and severe directories. Do NOT run `retryAllDLQ` concurrently with any active pipeline as it will cause conflicts.") - @Default.String(RUN_MODE_REGULAR) - String getRunMode(); - - void setRunMode(String value); - - @TemplateParameter.Integer( - order = 24, - optional = true, - description = "Dead letter queue retry minutes", - helpText = "The number of minutes between dead letter queue retries. Defaults to 10.") - @Default.Integer(10) - Integer getDlqRetryMinutes(); - - void setDlqRetryMinutes(Integer value); - - @TemplateParameter.Enum( - order = 25, - optional = true, - description = "Source database type, ex: mysql", - enumOptions = { - @TemplateEnumOption("mysql"), - @TemplateEnumOption("cassandra"), - @TemplateEnumOption("postgresql"), - @TemplateEnumOption("spanner"), - @TemplateEnumOption("oracle") - }, - helpText = "The type of source database to reverse replicate to.") - @Default.String("mysql") - String getSourceType(); - - void setSourceType(String value); - - @TemplateParameter.GcsReadFile( - order = 26, - optional = true, - description = "Custom transformation jar location in Cloud Storage", - helpText = - "Custom jar location in Cloud Storage that contains the custom transformation logic for processing records" - + " in reverse replication.") - @Default.String("") - String getTransformationJarPath(); - - void setTransformationJarPath(String value); - - @TemplateParameter.Text( - order = 27, - optional = true, - description = "Custom class name for transformation", - helpText = - "Fully qualified class name having the custom transformation logic. It is a" - + " mandatory field in case transformationJarPath is specified") - @Default.String("") - String getTransformationClassName(); - - void setTransformationClassName(String value); - - @TemplateParameter.Text( - order = 28, - optional = true, - description = "Custom parameters for transformation", - helpText = - "String containing any custom parameters to be passed to the custom transformation class.") - @Default.String("") - String getTransformationCustomParameters(); - - void setTransformationCustomParameters(String value); - - @TemplateParameter.Text( - order = 29, - optional = true, - description = "Table name overrides from spanner to source", - regexes = - "^\\[([[:space:]]*\\{[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", - example = "[{Singers, Vocalists}, {Albums, Records}]", - helpText = - "These are the table name overrides from spanner to source. They are written in the" - + "following format: [{SpannerTableName1, SourceTableName1}, {SpannerTableName2, SourceTableName2}]" - + "This example shows mapping Singers table to Vocalists and Albums table to Records.") - @Default.String("") - String getTableOverrides(); - - void setTableOverrides(String value); - - @TemplateParameter.Text( - order = 30, - optional = true, - description = "Column name overrides from spanner to source", - regexes = - "^\\[([[:space:]]*\\{[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*,[[:space:]]*[[:graph:]]+\\.[[:graph:]]+[[:space:]]*\\}[[:space:]]*(,[[:space:]]*)*)*\\]$", - example = - "[{Singers.SingerName, Singers.TalentName}, {Albums.AlbumName, Albums.RecordName}]", - helpText = - "These are the column name overrides from spanner to source. They are written in the" - + "following format: [{SpannerTableName1.SpannerColumnName1, SpannerTableName1.SourceColumnName1}, {SpannerTableName2.SpannerColumnName1, SpannerTableName2.SourceColumnName1}]" - + "Note that the SpannerTableName should remain the same in both the spanner and source pair. To override table names, use tableOverrides." - + "The example shows mapping SingerName to TalentName and AlbumName to RecordName in Singers and Albums table respectively.") - @Default.String("") - String getColumnOverrides(); - - void setColumnOverrides(String value); - - @TemplateParameter.GcsReadFile( - order = 31, - optional = true, - description = "File based overrides from spanner to source", - helpText = - "A file which specifies the table and the column name overrides from spanner to source.") - @Default.String("") - String getSchemaOverridesFilePath(); - - void setSchemaOverridesFilePath(String value); - - @TemplateParameter.Text( - order = 32, - optional = true, - description = "Directory name for holding filtered records", - helpText = - "Records skipped from reverse replication are written to this directory. Default" - + " directory name is skip.") - @Default.String("filteredEvents") - String getFilterEventsDirectoryName(); - - void setFilterEventsDirectoryName(String value); - - @TemplateParameter.Boolean( - order = 33, - optional = true, - description = "Boolean setting if reverse migration is sharded", - helpText = - "Sets the template to a sharded migration. If source shard template contains more" - + " than one shard, the value will be set to true. This value defaults to false.") - @Default.Boolean(false) - Boolean getIsShardedMigration(); - - void setIsShardedMigration(Boolean value); - - @TemplateParameter.Text( - order = 34, - optional = true, - description = "Failure injection parameter", - helpText = "Failure injection parameter. Only used for testing.") - @Default.String("") - String getFailureInjectionParameter(); - - void setFailureInjectionParameter(String value); - - @TemplateParameter.Enum( - order = 35, - enumOptions = { - @TemplateEnumOption("LOW"), - @TemplateEnumOption("MEDIUM"), - @TemplateEnumOption("HIGH") - }, - optional = true, - description = "Priority for Spanner RPC invocations", - helpText = - "The request priority for Cloud Spanner calls. The value must be one of:" - + " [`HIGH`,`MEDIUM`,`LOW`]. Defaults to `HIGH`.") - @Default.Enum("HIGH") - RpcPriority getSpannerPriority(); - - void setSpannerPriority(RpcPriority value); - } - /** * Main entry point for executing the pipeline. * @@ -559,7 +114,8 @@ public static void main(String[] args) { LOG.info("Starting Spanner change streams to sink"); - Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class); + SpannerToSourceDbOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(SpannerToSourceDbOptions.class); // Stage SSL certificates to extraFiles if required as per the pipeline options. // Ref https://cloud.google.com/dataflow/docs/guides/templates/ssl-certificates @@ -577,7 +133,7 @@ public static void main(String[] args) { * @param options The execution parameters to the pipeline. * @return The result of the pipeline execution. */ - public static PipelineResult run(Options options) { + public static PipelineResult run(SpannerToSourceDbOptions options) { long startTime = System.currentTimeMillis(); Pipeline pipeline = Pipeline.create(options); pipeline @@ -702,7 +258,7 @@ public static PipelineResult run(Options options) { static void buildPipeline( Pipeline pipeline, - Options options, + SpannerToSourceDbOptions options, SourceSchema sourceSchema, List shards, PCollectionView ddlView, @@ -956,7 +512,7 @@ static void buildPipeline( } public static SpannerIO.ReadChangeStream getReadChangeStreamDoFn( - Options options, SpannerConfig spannerConfig) { + SpannerToSourceDbOptions options, SpannerConfig spannerConfig) { Timestamp startTime = Timestamp.now(); if (!options.getStartTimestamp().equals("")) { @@ -989,7 +545,7 @@ public static SpannerIO.ReadChangeStream getReadChangeStreamDoFn( return readChangeStreamDoFn; } - static DeadLetterQueueManager buildDlqManager(Options options) { + static DeadLetterQueueManager buildDlqManager(SpannerToSourceDbOptions options) { String tempLocation = options.as(DataflowPipelineOptions.class).getTempLocation().endsWith("/") ? options.as(DataflowPipelineOptions.class).getTempLocation() diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbTest.java index 81af88469b..ca9c8c4a14 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbTest.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbTest.java @@ -17,9 +17,9 @@ import com.google.cloud.spanner.Options.RpcPriority; import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager; +import com.google.cloud.teleport.v2.options.SpannerToSourceDbOptions; import com.google.cloud.teleport.v2.spanner.ddl.Ddl; import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; -import com.google.cloud.teleport.v2.templates.SpannerToSourceDb.Options; import java.util.Collections; import java.util.List; import org.apache.beam.runners.dataflow.options.DataflowPipelineDebugOptions; @@ -45,7 +45,7 @@ public class SpannerToSourceDbTest { @Rule public final transient TestPipeline pipeline = TestPipeline.create(); - private Options options; + private SpannerToSourceDbOptions options; @Mock private PCollectionView mockDdlView; @Mock private PCollectionView mockShadowTableDdlView; @Mock private SpannerConfig mockSpannerConfig; @@ -68,7 +68,7 @@ public void setUp() { dummyDdl = Ddl.builder().createTable("shadow_T").column("c1").string().endColumn().endTable().build(); - options = PipelineOptionsFactory.as(Options.class); + options = PipelineOptionsFactory.as(SpannerToSourceDbOptions.class); options.setRunMode("regular"); options.setDeadLetterQueueDirectory("gs://test/dlq"); options.setSkipDirectoryName("skip"); diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/transforms/AssignShardIdFnTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/transforms/AssignShardIdFnTest.java index edc924323a..9236d267ca 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/transforms/AssignShardIdFnTest.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/transforms/AssignShardIdFnTest.java @@ -43,6 +43,7 @@ import com.google.cloud.spanner.Struct; import com.google.cloud.spanner.TimestampBound; import com.google.cloud.spanner.Value; +import com.google.cloud.teleport.v2.options.SpannerToSourceDbOptions; import com.google.cloud.teleport.v2.spanner.ddl.Ddl; import com.google.cloud.teleport.v2.spanner.migrations.schema.ISchemaMapper; import com.google.cloud.teleport.v2.spanner.migrations.schema.SessionBasedMapper; @@ -54,7 +55,6 @@ import com.google.cloud.teleport.v2.spanner.type.Type; import com.google.cloud.teleport.v2.spanner.utils.IShardIdFetcher; import com.google.cloud.teleport.v2.spanner.utils.ShardIdResponse; -import com.google.cloud.teleport.v2.templates.SpannerToSourceDb.Options; import com.google.cloud.teleport.v2.templates.changestream.TrimmedShardedDataChangeRecord; import com.google.cloud.teleport.v2.templates.constants.Constants; import com.google.cloud.teleport.v2.templates.utils.SchemaMapperUtils; @@ -99,7 +99,7 @@ public class AssignShardIdFnTest { @Mock private DoFn.ProcessContext processContext; - @Mock private Options mockOptions; + @Mock private SpannerToSourceDbOptions mockOptions; @Mock private PCollectionView mockDdlView; diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/transforms/SourceWriterFnTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/transforms/SourceWriterFnTest.java index ae4c67f570..c8f278444a 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/transforms/SourceWriterFnTest.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/transforms/SourceWriterFnTest.java @@ -35,6 +35,7 @@ import com.google.cloud.spanner.Options.RpcPriority; import com.google.cloud.spanner.SpannerExceptionFactory; import com.google.cloud.spanner.TransactionRunner; +import com.google.cloud.teleport.v2.options.SpannerToSourceDbOptions; import com.google.cloud.teleport.v2.spanner.ddl.Ddl; import com.google.cloud.teleport.v2.spanner.exceptions.InvalidTransformationException; import com.google.cloud.teleport.v2.spanner.migrations.exceptions.ChangeEventConvertorException; @@ -47,7 +48,6 @@ import com.google.cloud.teleport.v2.spanner.sourceddl.SourceSchema; import com.google.cloud.teleport.v2.spanner.utils.ISpannerMigrationTransformer; import com.google.cloud.teleport.v2.spanner.utils.MigrationTransformationResponse; -import com.google.cloud.teleport.v2.templates.SpannerToSourceDb.Options; import com.google.cloud.teleport.v2.templates.changestream.ChangeStreamErrorRecord; import com.google.cloud.teleport.v2.templates.changestream.TrimmedShardedDataChangeRecord; import com.google.cloud.teleport.v2.templates.constants.Constants; @@ -107,7 +107,7 @@ public class SourceWriterFnTest { @Mock private DoFn.ProcessContext processContext; @Mock private ISpannerMigrationTransformer mockSpannerMigrationTransformer; @Mock private SourceProcessor mockSourceProcessor; - @Mock private Options mockOptions; + @Mock private SpannerToSourceDbOptions mockOptions; @Mock private PCollectionView mockDdlView; @Mock private PCollectionView mockShadowTableDdlView; @Mock private IDMLGenerator mockDMLGenerator; @@ -245,7 +245,7 @@ public Void answer(InvocationOnMock invocation) throws Throwable { // Mock side input access in ProcessContext when(processContext.sideInput(mockDdlView)).thenReturn(testDdl); when(processContext.sideInput(mockShadowTableDdlView)).thenReturn(shadowTableDdl); - when(mockOptions.as(Options.class)).thenReturn(mockOptions); + when(mockOptions.as(SpannerToSourceDbOptions.class)).thenReturn(mockOptions); } @Test