diff --git a/yaml/README_Kafka_To_Bigtable_Yaml.md b/yaml/README_Kafka_To_Bigtable_Yaml.md index e00d2839fb..2776cc2428 100644 --- a/yaml/README_Kafka_To_Bigtable_Yaml.md +++ b/yaml/README_Kafka_To_Bigtable_Yaml.md @@ -28,11 +28,13 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters +* **autoOffsetResetConfig**: What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest For example, `earliest`. Defaults to: earliest. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. * **fileDescriptorPath**: The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb For example, `gs://bucket/path/to/descriptor.pb`. * **format**: The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON For example, `JSON`. Defaults to: JSON. +* **maxReadTimeSeconds**: Upper bound of how long to read from Kafka in seconds. For example: 60 For example, `60`. * **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. * **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. * **language**: The language used to define (and execute) the expressions and/or callables in fields. Defaults to generic. For example, `python`. @@ -137,11 +139,13 @@ export TABLE_ID= export WINDOWING= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export LANGUAGE=generic @@ -152,11 +156,13 @@ gcloud dataflow flex-template run "kafka-to-bigtable-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ + --parameters "autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ --parameters "fileDescriptorPath=$FILE_DESCRIPTOR_PATH" \ --parameters "format=$FORMAT" \ + --parameters "maxReadTimeSeconds=$MAX_READ_TIME_SECONDS" \ --parameters "messageName=$MESSAGE_NAME" \ --parameters "schema=$SCHEMA" \ --parameters "language=$LANGUAGE" \ @@ -192,11 +198,13 @@ export TABLE_ID= export WINDOWING= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export LANGUAGE=generic @@ -208,7 +216,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-bigtable-yaml-job" \ -DtemplateName="Kafka_To_Bigtable_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,messageName=$MESSAGE_NAME,schema=$SCHEMA,language=$LANGUAGE,fields=$FIELDS,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,tableId=$TABLE_ID,windowing=$WINDOWING" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,language=$LANGUAGE,fields=$FIELDS,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,tableId=$TABLE_ID,windowing=$WINDOWING" \ -f yaml ``` @@ -260,11 +268,13 @@ resource "google_dataflow_flex_template_job" "kafka_to_bigtable_yaml" { instanceId = "" tableId = "" windowing = "" + # autoOffsetResetConfig = "earliest" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" # fileDescriptorPath = "" # format = "JSON" + # maxReadTimeSeconds = "" # messageName = "" # schema = "" # language = "generic" diff --git a/yaml/README_Kafka_To_Iceberg_Yaml.md b/yaml/README_Kafka_To_Iceberg_Yaml.md index 375cc95dbc..9cf6ec244e 100644 --- a/yaml/README_Kafka_To_Iceberg_Yaml.md +++ b/yaml/README_Kafka_To_Iceberg_Yaml.md @@ -23,11 +23,13 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters +* **autoOffsetResetConfig**: What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest For example, `earliest`. Defaults to: earliest. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. * **fileDescriptorPath**: The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb For example, `gs://bucket/path/to/descriptor.pb`. * **format**: The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON For example, `JSON`. Defaults to: JSON. +* **maxReadTimeSeconds**: Upper bound of how long to read from Kafka in seconds. For example: 60 For example, `60`. * **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. * **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. * **configProperties**: A map of properties to pass to the Hadoop Configuration. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. @@ -138,11 +140,13 @@ export CATALOG_PROPERTIES= export TRIGGERING_FREQUENCY_SECONDS= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export CONFIG_PROPERTIES= @@ -160,11 +164,13 @@ gcloud dataflow flex-template run "kafka-to-iceberg-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ + --parameters "autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ --parameters "fileDescriptorPath=$FILE_DESCRIPTOR_PATH" \ --parameters "format=$FORMAT" \ + --parameters "maxReadTimeSeconds=$MAX_READ_TIME_SECONDS" \ --parameters "messageName=$MESSAGE_NAME" \ --parameters "schema=$SCHEMA" \ --parameters "table=$TABLE" \ @@ -205,11 +211,13 @@ export CATALOG_PROPERTIES= export TRIGGERING_FREQUENCY_SECONDS= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export CONFIG_PROPERTIES= @@ -228,7 +236,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-iceberg-yaml-job" \ -DtemplateName="Kafka_To_Iceberg_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,catalogName=$CATALOG_NAME,catalogProperties=$CATALOG_PROPERTIES,configProperties=$CONFIG_PROPERTIES,drop=$DROP,keep=$KEEP,only=$ONLY,partitionFields=$PARTITION_FIELDS,tableProperties=$TABLE_PROPERTIES,triggeringFrequencySeconds=$TRIGGERING_FREQUENCY_SECONDS,sdfCheckpointAfterDuration=$SDF_CHECKPOINT_AFTER_DURATION,sdfCheckpointAfterOutputBytes=$SDF_CHECKPOINT_AFTER_OUTPUT_BYTES" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,catalogName=$CATALOG_NAME,catalogProperties=$CATALOG_PROPERTIES,configProperties=$CONFIG_PROPERTIES,drop=$DROP,keep=$KEEP,only=$ONLY,partitionFields=$PARTITION_FIELDS,tableProperties=$TABLE_PROPERTIES,triggeringFrequencySeconds=$TRIGGERING_FREQUENCY_SECONDS,sdfCheckpointAfterDuration=$SDF_CHECKPOINT_AFTER_DURATION,sdfCheckpointAfterOutputBytes=$SDF_CHECKPOINT_AFTER_OUTPUT_BYTES" \ -f yaml ``` @@ -279,11 +287,13 @@ resource "google_dataflow_flex_template_job" "kafka_to_iceberg_yaml" { catalogName = "" catalogProperties = "" triggeringFrequencySeconds = "" + # autoOffsetResetConfig = "earliest" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" # fileDescriptorPath = "" # format = "JSON" + # maxReadTimeSeconds = "" # messageName = "" # schema = "" # configProperties = "" diff --git a/yaml/README_Kafka_To_Spanner_Yaml.md b/yaml/README_Kafka_To_Spanner_Yaml.md new file mode 100644 index 0000000000..508d62c4bc --- /dev/null +++ b/yaml/README_Kafka_To_Spanner_Yaml.md @@ -0,0 +1,282 @@ + +Kafka to Spanner (YAML) template +--- +The Kafka to Spanner template is a streaming pipeline which ingests data from an +Apache Kafka topic and writes the resulting records to Cloud Spanner. + + +:memo: This is a Google-provided template! Please +check [Provided templates documentation](https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-spanner) +on how to use it without having to build from sources using [Create job from template](https://console.cloud.google.com/dataflow/createjob?template=Kafka_To_Spanner_Yaml). + +:bulb: This is a generated documentation based +on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/contributor-docs/code-contributions.md#metadata-annotations) +. Do not change this file directly. + +## Parameters + +### Required parameters + +* **bootstrapServers**: A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. For example: host1:port1,host2:port2 For example, `host1:port1,host2:port2,localhost:9092,127.0.0.1:9093`. +* **topic**: Kafka topic to read from. For example: my_topic For example, `my_topic`. +* **instanceId**: The Cloud Spanner instance ID. For example, `your-instance-id`. +* **databaseId**: The Cloud Spanner database ID. For example, `your-database-id`. +* **tableId**: The Cloud Spanner table ID. For example, `your-table-id`. + +### Optional parameters + +* **autoOffsetResetConfig**: What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest For example, `earliest`. Defaults to: earliest. +* **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. +* **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. +* **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. +* **fileDescriptorPath**: The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb For example, `gs://bucket/path/to/descriptor.pb`. +* **format**: The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON For example, `JSON`. Defaults to: JSON. +* **maxReadTimeSeconds**: Upper bound of how long to read from Kafka in seconds. For example: 60 For example, `60`. +* **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. +* **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. +* **projectId**: The Google Cloud project ID of the Cloud Spanner instance. For example, `your-project-id`. +* **type**: The type of windowing to apply (e.g., fixed, sliding, sessions, global). Defaults to fixed. For example, `fixed`. +* **size**: The window size in seconds. Defaults to 60. For example, `60`. + + + +## Getting Started + +### Requirements + +* Java 17 +* Maven +* [gcloud CLI](https://cloud.google.com/sdk/gcloud), and execution of the + following commands: + * `gcloud auth login` + * `gcloud auth application-default login` + +:star2: Those dependencies are pre-installed if you use Google Cloud Shell! + +[![Open in Cloud Shell](http://gstatic.com/cloudssh/images/open-btn.svg)](https://console.cloud.google.com/cloudshell/editor?cloudshell_git_repo=https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2FDataflowTemplates.git&cloudshell_open_in_editor=yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java) + +### Templates Plugin + +This README provides instructions using +the [Templates Plugin](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/contributor-docs/code-contributions.md#templates-plugin). + +#### Validating the Template + +This template has a validation command that is used to check code quality. + +```shell +mvn clean install -PtemplatesValidate \ +-DskipTests -am \ +-pl yaml +``` + +### Building Template + +This template is a Flex Template, meaning that the pipeline code will be +containerized and the container will be executed on Dataflow. Please +check [Use Flex Templates](https://cloud.google.com/dataflow/docs/guides/templates/using-flex-templates) +and [Configure Flex Templates](https://cloud.google.com/dataflow/docs/guides/templates/configuring-flex-templates) +for more information. + +#### Staging the Template + +If the plan is to just stage the template (i.e., make it available to use) by +the `gcloud` command or Dataflow "Create job from template" UI, +the `-PtemplatesStage` profile should be used: + +```shell +export PROJECT= +export BUCKET_NAME= +export ARTIFACT_REGISTRY_REPO=-docker.pkg.dev/$PROJECT/ + +mvn clean package -PtemplatesStage \ +-DskipTests \ +-DprojectId="$PROJECT" \ +-DbucketName="$BUCKET_NAME" \ +-DartifactRegistry="$ARTIFACT_REGISTRY_REPO" \ +-DstagePrefix="templates" \ +-DtemplateName="Kafka_To_Spanner_Yaml" \ +-f yaml +``` + +The `-DartifactRegistry` parameter can be specified to set the artifact registry repository of the Flex Templates image. +If not provided, it defaults to `gcr.io/`. + +The command should build and save the template to Google Cloud, and then print +the complete location on Cloud Storage: + +``` +Flex Template was staged! gs:///templates/flex/Kafka_To_Spanner_Yaml +``` + +The specific path should be copied as it will be used in the following steps. + +#### Running the Template + +**Using the staged template**: + +You can use the path above run the template (or share with others for execution). + +To start a job with the template at any time using `gcloud`, you are going to +need valid resources for the required parameters. + +Provided that, the following command line can be used: + +```shell +export PROJECT= +export BUCKET_NAME= +export REGION=us-central1 +export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/Kafka_To_Spanner_Yaml" + +### Required +export BOOTSTRAP_SERVERS= +export TOPIC= +export INSTANCE_ID= +export DATABASE_ID= +export TABLE_ID= + +### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest +export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= +export CONFLUENT_SCHEMA_REGISTRY_URL= +export CONSUMER_CONFIG_UPDATES= +export FILE_DESCRIPTOR_PATH= +export FORMAT=JSON +export MAX_READ_TIME_SECONDS= +export MESSAGE_NAME= +export SCHEMA= +export PROJECT_ID= +export TYPE=fixed +export SIZE=60 + +gcloud dataflow flex-template run "kafka-to-spanner-yaml-job" \ + --project "$PROJECT" \ + --region "$REGION" \ + --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ + --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ + --parameters "topic=$TOPIC" \ + --parameters "autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG" \ + --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ + --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ + --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ + --parameters "fileDescriptorPath=$FILE_DESCRIPTOR_PATH" \ + --parameters "format=$FORMAT" \ + --parameters "maxReadTimeSeconds=$MAX_READ_TIME_SECONDS" \ + --parameters "messageName=$MESSAGE_NAME" \ + --parameters "schema=$SCHEMA" \ + --parameters "projectId=$PROJECT_ID" \ + --parameters "instanceId=$INSTANCE_ID" \ + --parameters "databaseId=$DATABASE_ID" \ + --parameters "tableId=$TABLE_ID" \ + --parameters "type=$TYPE" \ + --parameters "size=$SIZE" +``` + +For more information about the command, please check: +https://cloud.google.com/sdk/gcloud/reference/dataflow/flex-template/run + + +**Using the plugin**: + +Instead of just generating the template in the folder, it is possible to stage +and run the template in a single command. This may be useful for testing when +changing the templates. + +```shell +export PROJECT= +export BUCKET_NAME= +export REGION=us-central1 + +### Required +export BOOTSTRAP_SERVERS= +export TOPIC= +export INSTANCE_ID= +export DATABASE_ID= +export TABLE_ID= + +### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest +export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= +export CONFLUENT_SCHEMA_REGISTRY_URL= +export CONSUMER_CONFIG_UPDATES= +export FILE_DESCRIPTOR_PATH= +export FORMAT=JSON +export MAX_READ_TIME_SECONDS= +export MESSAGE_NAME= +export SCHEMA= +export PROJECT_ID= +export TYPE=fixed +export SIZE=60 + +mvn clean package -PtemplatesRun \ +-DskipTests \ +-DprojectId="$PROJECT" \ +-DbucketName="$BUCKET_NAME" \ +-Dregion="$REGION" \ +-DjobName="kafka-to-spanner-yaml-job" \ +-DtemplateName="Kafka_To_Spanner_Yaml" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,databaseId=$DATABASE_ID,tableId=$TABLE_ID,type=$TYPE,size=$SIZE" \ +-f yaml +``` + +## Terraform + +Dataflow supports the utilization of Terraform to manage template jobs, +see [dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job). + +Terraform modules have been generated for most templates in this repository. This includes the relevant parameters +specific to the template. If available, they may be used instead of +[dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job) +directly. + +To use the autogenerated module, execute the standard +[terraform workflow](https://developer.hashicorp.com/terraform/intro/core-workflow): + +```shell +cd yaml/terraform/Kafka_To_Spanner_Yaml +terraform init +terraform apply +``` + +To use +[dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job) +directly: + +```terraform +provider "google-beta" { + project = var.project +} +variable "project" { + default = "" +} +variable "region" { + default = "us-central1" +} + +resource "google_dataflow_flex_template_job" "kafka_to_spanner_yaml" { + + provider = google-beta + container_spec_gcs_path = "gs://dataflow-templates-${var.region}/latest/flex/Kafka_To_Spanner_Yaml" + name = "kafka-to-spanner-yaml" + region = var.region + parameters = { + bootstrapServers = "" + topic = "" + instanceId = "" + databaseId = "" + tableId = "" + # autoOffsetResetConfig = "earliest" + # confluentSchemaRegistrySubject = "" + # confluentSchemaRegistryUrl = "" + # consumerConfigUpdates = "" + # fileDescriptorPath = "" + # format = "JSON" + # maxReadTimeSeconds = "" + # messageName = "" + # schema = "" + # projectId = "" + # type = "fixed" + # size = "60" + } +} +``` diff --git a/yaml/README_Kafka_to_BigQuery_Yaml.md b/yaml/README_Kafka_to_BigQuery_Yaml.md index 909d793eac..a39d98de15 100644 --- a/yaml/README_Kafka_to_BigQuery_Yaml.md +++ b/yaml/README_Kafka_to_BigQuery_Yaml.md @@ -28,11 +28,13 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters +* **autoOffsetResetConfig**: What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest For example, `earliest`. Defaults to: earliest. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. * **fileDescriptorPath**: The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb For example, `gs://bucket/path/to/descriptor.pb`. * **format**: The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON For example, `JSON`. Defaults to: JSON. +* **maxReadTimeSeconds**: Upper bound of how long to read from Kafka in seconds. For example: 60 For example, `60`. * **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. * **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. * **createDisposition**: Specifies whether a table should be created if it does not exist. Valid inputs are 'CREATE_NEVER' and 'CREATE_IF_NEEDED'. For example, `CREATE_NEVER`. Defaults to: CREATE_NEVER. @@ -136,11 +138,13 @@ export TABLE= export OUTPUT_DEADLETTER_TABLE= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export CREATE_DISPOSITION=CREATE_NEVER @@ -153,11 +157,13 @@ gcloud dataflow flex-template run "kafka-to-bigquery-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ + --parameters "autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ --parameters "fileDescriptorPath=$FILE_DESCRIPTOR_PATH" \ --parameters "format=$FORMAT" \ + --parameters "maxReadTimeSeconds=$MAX_READ_TIME_SECONDS" \ --parameters "messageName=$MESSAGE_NAME" \ --parameters "schema=$SCHEMA" \ --parameters "table=$TABLE" \ @@ -189,11 +195,13 @@ export TABLE=
export OUTPUT_DEADLETTER_TABLE= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export CREATE_DISPOSITION=CREATE_NEVER @@ -207,7 +215,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-bigquery-yaml-job" \ -DtemplateName="Kafka_to_BigQuery_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,createDisposition=$CREATE_DISPOSITION,writeDisposition=$WRITE_DISPOSITION,numStreams=$NUM_STREAMS,outputDeadletterTable=$OUTPUT_DEADLETTER_TABLE" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,createDisposition=$CREATE_DISPOSITION,writeDisposition=$WRITE_DISPOSITION,numStreams=$NUM_STREAMS,outputDeadletterTable=$OUTPUT_DEADLETTER_TABLE" \ -f yaml ``` @@ -256,11 +264,13 @@ resource "google_dataflow_flex_template_job" "kafka_to_bigquery_yaml" { topic = "" table = "
" outputDeadletterTable = "" + # autoOffsetResetConfig = "earliest" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" # fileDescriptorPath = "" # format = "JSON" + # maxReadTimeSeconds = "" # messageName = "" # schema = "" # createDisposition = "CREATE_NEVER" diff --git a/yaml/README_Oracle_To_Iceberg_Yaml.md b/yaml/README_Oracle_To_Iceberg_Yaml.md index 5b22934472..3759e334d2 100644 --- a/yaml/README_Oracle_To_Iceberg_Yaml.md +++ b/yaml/README_Oracle_To_Iceberg_Yaml.md @@ -1,9 +1,8 @@ Oracle to Iceberg (YAML) template --- -The Oracle to Iceberg template is a batch pipeline executes the user provided -SQL query to read data from Oracle table and outputs the records to Iceberg -table. +The Oracle to Iceberg template is a batch pipeline executes the user provided SQL +query to read data from Oracle table and outputs the records to Iceberg table. @@ -15,7 +14,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Required parameters -* **jdbcUrl**: The JDBC connection URL. For example, `jdbc:oracle:thin:@your-host:1521:your-sid`. +* **jdbcUrl**: The JDBC connection URL. For example, `jdbc:oracle://your-host:1521/serviceName`. * **table**: A fully-qualified table identifier, e.g., my_dataset.my_table. For example, `my_dataset.my_table`. * **catalogName**: The name of the Iceberg catalog that contains the table. For example, `my_hadoop_catalog`. * **catalogProperties**: A map of properties for setting up the Iceberg catalog. For example, `{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}`. diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java index d7d5c60345..29d3e07e54 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java @@ -67,6 +67,18 @@ public interface KafkaToBigQueryYaml { @TemplateParameter.Text( order = 3, + name = "autoOffsetResetConfig", + optional = true, + description = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server.", + helpText = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest", + example = "earliest") + @Default.String("earliest") + String getAutoOffsetResetConfig(); + + @TemplateParameter.Text( + order = 4, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -75,7 +87,7 @@ public interface KafkaToBigQueryYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 4, + order = 5, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -85,7 +97,7 @@ public interface KafkaToBigQueryYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 5, + order = 6, name = "consumerConfigUpdates", optional = true, description = @@ -96,7 +108,7 @@ public interface KafkaToBigQueryYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 6, + order = 7, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -106,7 +118,7 @@ public interface KafkaToBigQueryYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 7, + order = 8, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -116,8 +128,17 @@ public interface KafkaToBigQueryYaml { @Default.String("JSON") String getFormat(); + @TemplateParameter.Integer( + order = 9, + name = "maxReadTimeSeconds", + optional = true, + description = "Upper bound of how long to read from Kafka.", + helpText = "Upper bound of how long to read from Kafka in seconds. For example: 60", + example = "60") + Integer getMaxReadTimeSeconds(); + @TemplateParameter.Text( - order = 8, + order = 10, name = "messageName", optional = true, description = @@ -128,7 +149,7 @@ public interface KafkaToBigQueryYaml { String getMessageName(); @TemplateParameter.Text( - order = 9, + order = 11, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -139,7 +160,7 @@ public interface KafkaToBigQueryYaml { String getSchema(); @TemplateParameter.Text( - order = 10, + order = 12, name = "table", optional = false, description = "BigQuery table", @@ -150,7 +171,7 @@ public interface KafkaToBigQueryYaml { String getTable(); @TemplateParameter.Text( - order = 11, + order = 13, name = "createDisposition", optional = true, description = "How to create", @@ -161,7 +182,7 @@ public interface KafkaToBigQueryYaml { String getCreateDisposition(); @TemplateParameter.Text( - order = 12, + order = 14, name = "writeDisposition", optional = true, description = "How to write", @@ -172,7 +193,7 @@ public interface KafkaToBigQueryYaml { String getWriteDisposition(); @TemplateParameter.Integer( - order = 13, + order = 15, name = "numStreams", optional = true, description = "Number of streams for BigQuery Storage Write API", @@ -183,7 +204,7 @@ public interface KafkaToBigQueryYaml { Integer getNumStreams(); @TemplateParameter.Text( - order = 14, + order = 16, name = "outputDeadletterTable", optional = false, description = "The dead-letter table name to output failed messages to BigQuery", diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java index 7a241663a6..9f06355b52 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java @@ -67,6 +67,18 @@ public interface KafkaToBigtableYaml { @TemplateParameter.Text( order = 3, + name = "autoOffsetResetConfig", + optional = true, + description = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server.", + helpText = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest", + example = "earliest") + @Default.String("earliest") + String getAutoOffsetResetConfig(); + + @TemplateParameter.Text( + order = 4, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -75,7 +87,7 @@ public interface KafkaToBigtableYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 4, + order = 5, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -85,7 +97,7 @@ public interface KafkaToBigtableYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 5, + order = 6, name = "consumerConfigUpdates", optional = true, description = @@ -96,7 +108,7 @@ public interface KafkaToBigtableYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 6, + order = 7, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -106,7 +118,7 @@ public interface KafkaToBigtableYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 7, + order = 8, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -116,8 +128,17 @@ public interface KafkaToBigtableYaml { @Default.String("JSON") String getFormat(); + @TemplateParameter.Integer( + order = 9, + name = "maxReadTimeSeconds", + optional = true, + description = "Upper bound of how long to read from Kafka.", + helpText = "Upper bound of how long to read from Kafka in seconds. For example: 60", + example = "60") + Integer getMaxReadTimeSeconds(); + @TemplateParameter.Text( - order = 8, + order = 10, name = "messageName", optional = true, description = @@ -128,7 +149,7 @@ public interface KafkaToBigtableYaml { String getMessageName(); @TemplateParameter.Text( - order = 9, + order = 11, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -139,7 +160,7 @@ public interface KafkaToBigtableYaml { String getSchema(); @TemplateParameter.Text( - order = 10, + order = 12, name = "language", optional = true, description = "Language used to define the expressions.", @@ -150,7 +171,7 @@ public interface KafkaToBigtableYaml { String getLanguage(); @TemplateParameter.Text( - order = 11, + order = 13, name = "fields", optional = false, description = "Field mapping configuration", @@ -161,7 +182,7 @@ public interface KafkaToBigtableYaml { String getFields(); @TemplateParameter.Text( - order = 12, + order = 14, name = "projectId", optional = false, description = "Bigtable project ID", @@ -171,7 +192,7 @@ public interface KafkaToBigtableYaml { String getProjectId(); @TemplateParameter.Text( - order = 13, + order = 15, name = "instanceId", optional = false, description = "Bigtable instance ID", @@ -181,7 +202,7 @@ public interface KafkaToBigtableYaml { String getInstanceId(); @TemplateParameter.Text( - order = 14, + order = 16, name = "tableId", optional = false, description = "Bigtable output table", @@ -191,7 +212,7 @@ public interface KafkaToBigtableYaml { String getTableId(); @TemplateParameter.Text( - order = 15, + order = 17, name = "windowing", optional = false, description = "Windowing options", diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java index de7b718932..07a9d9c89c 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java @@ -70,6 +70,18 @@ public interface KafkaToIcebergYaml { @TemplateParameter.Text( order = 3, + name = "autoOffsetResetConfig", + optional = true, + description = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server.", + helpText = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest", + example = "earliest") + @Default.String("earliest") + String getAutoOffsetResetConfig(); + + @TemplateParameter.Text( + order = 4, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -78,7 +90,7 @@ public interface KafkaToIcebergYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 4, + order = 5, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -88,7 +100,7 @@ public interface KafkaToIcebergYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 5, + order = 6, name = "consumerConfigUpdates", optional = true, description = @@ -99,7 +111,7 @@ public interface KafkaToIcebergYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 6, + order = 7, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -109,7 +121,7 @@ public interface KafkaToIcebergYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 7, + order = 8, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -119,8 +131,17 @@ public interface KafkaToIcebergYaml { @Default.String("JSON") String getFormat(); + @TemplateParameter.Integer( + order = 9, + name = "maxReadTimeSeconds", + optional = true, + description = "Upper bound of how long to read from Kafka.", + helpText = "Upper bound of how long to read from Kafka in seconds. For example: 60", + example = "60") + Integer getMaxReadTimeSeconds(); + @TemplateParameter.Text( - order = 8, + order = 10, name = "messageName", optional = true, description = @@ -131,7 +152,7 @@ public interface KafkaToIcebergYaml { String getMessageName(); @TemplateParameter.Text( - order = 9, + order = 11, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -142,7 +163,7 @@ public interface KafkaToIcebergYaml { String getSchema(); @TemplateParameter.Text( - order = 10, + order = 12, name = "table", optional = false, description = "A fully-qualified table identifier.", @@ -152,7 +173,7 @@ public interface KafkaToIcebergYaml { String getTable(); @TemplateParameter.Text( - order = 11, + order = 13, name = "catalogName", optional = false, description = "Name of the catalog containing the table.", @@ -162,7 +183,7 @@ public interface KafkaToIcebergYaml { String getCatalogName(); @TemplateParameter.Text( - order = 12, + order = 14, name = "catalogProperties", optional = false, description = "Properties used to set up the Iceberg catalog.", @@ -172,7 +193,7 @@ public interface KafkaToIcebergYaml { String getCatalogProperties(); @TemplateParameter.Text( - order = 13, + order = 15, name = "configProperties", optional = true, description = "Properties passed to the Hadoop Configuration.", @@ -181,7 +202,7 @@ public interface KafkaToIcebergYaml { String getConfigProperties(); @TemplateParameter.Text( - order = 14, + order = 16, name = "drop", optional = true, description = "A list of field names to drop from the input record before writing.", @@ -190,7 +211,7 @@ public interface KafkaToIcebergYaml { String getDrop(); @TemplateParameter.Text( - order = 15, + order = 17, name = "keep", optional = true, description = "A list of field names to keep in the input record.", @@ -199,7 +220,7 @@ public interface KafkaToIcebergYaml { String getKeep(); @TemplateParameter.Text( - order = 16, + order = 18, name = "only", optional = true, description = "The name of a single record field that should be written.", @@ -208,7 +229,7 @@ public interface KafkaToIcebergYaml { String getOnly(); @TemplateParameter.Text( - order = 17, + order = 19, name = "partitionFields", optional = true, description = "Fields used to create a partition spec for new tables.", @@ -217,7 +238,7 @@ public interface KafkaToIcebergYaml { String getPartitionFields(); @TemplateParameter.Text( - order = 18, + order = 20, name = "tableProperties", optional = true, description = "Iceberg table properties to be set on table creation.", @@ -226,7 +247,7 @@ public interface KafkaToIcebergYaml { String getTableProperties(); @TemplateParameter.Integer( - order = 19, + order = 21, name = "triggeringFrequencySeconds", optional = false, description = "For a streaming pipeline, the frequency at which snapshots are produced.", @@ -236,7 +257,7 @@ public interface KafkaToIcebergYaml { Integer getTriggeringFrequencySeconds(); @TemplateParameter.Text( - order = 20, + order = 22, name = "sdfCheckpointAfterDuration", optional = true, description = "Dataflow Pipeline Option: Duration after which to checkpoint stateful DoFns.", @@ -247,7 +268,7 @@ public interface KafkaToIcebergYaml { String getSdfCheckpointAfterDuration(); @TemplateParameter.Integer( - order = 21, + order = 23, name = "sdfCheckpointAfterOutputBytes", optional = true, description = diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java new file mode 100644 index 0000000000..e71c517b1d --- /dev/null +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java @@ -0,0 +1,221 @@ +/* + * 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.templates.yaml; + +import com.google.cloud.teleport.metadata.Template; +import com.google.cloud.teleport.metadata.TemplateCategory; +import com.google.cloud.teleport.metadata.TemplateParameter; +import org.apache.beam.sdk.options.Default; +import org.apache.beam.sdk.options.Validation; + +@Template( + name = "Kafka_To_Spanner_Yaml", + category = TemplateCategory.STREAMING, + type = Template.TemplateType.YAML, + displayName = "Kafka to Spanner (YAML)", + description = + "The Kafka to Spanner template is a streaming pipeline which ingests data from an Apache Kafka topic and writes the resulting records to Cloud Spanner.", + flexContainerName = "pipeline-yaml", + yamlTemplateFile = "KafkaToSpanner.yaml", + filesToCopy = {"main.py", "requirements.txt"}, + documentation = + "https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-spanner", + contactInformation = "https://cloud.google.com/support", + requirements = { + "The input Apache Kafka topic must exist.", + "The Apache Kafka broker server must be running and be reachable from the Dataflow worker machines.", + "The output Cloud Spanner instance, database, and table must exist." + }, + streaming = true, + hidden = false) +public interface KafkaToSpannerYaml { + + @TemplateParameter.Text( + order = 1, + name = "bootstrapServers", + optional = false, + description = + "A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.", + helpText = + "A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. For example: host1:port1,host2:port2", + example = "host1:port1,host2:port2,localhost:9092,127.0.0.1:9093") + @Validation.Required + String getBootstrapServers(); + + @TemplateParameter.Text( + order = 2, + name = "topic", + optional = false, + description = "Kafka topic to read from.", + helpText = "Kafka topic to read from. For example: my_topic", + example = "my_topic") + @Validation.Required + String getTopic(); + + @TemplateParameter.Text( + order = 3, + name = "autoOffsetResetConfig", + optional = true, + description = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server.", + helpText = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest", + example = "earliest") + @Default.String("earliest") + String getAutoOffsetResetConfig(); + + @TemplateParameter.Text( + order = 4, + name = "confluentSchemaRegistrySubject", + optional = true, + description = "The subject name for the Confluent Schema Registry.", + helpText = "The subject name for the Confluent Schema Registry. For example: my_subject", + example = "my_subject") + String getConfluentSchemaRegistrySubject(); + + @TemplateParameter.Text( + order = 5, + name = "confluentSchemaRegistryUrl", + optional = true, + description = "The URL for the Confluent Schema Registry.", + helpText = + "The URL for the Confluent Schema Registry. For example: http://schema-registry:8081", + example = "http://schema-registry:8081") + String getConfluentSchemaRegistryUrl(); + + @TemplateParameter.Text( + order = 6, + name = "consumerConfigUpdates", + optional = true, + description = + "A list of key-value pairs that act as configuration parameters for Kafka consumers.", + helpText = + "A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'}", + example = "{\"group.id\": \"my_group\"}") + String getConsumerConfigUpdates(); + + @TemplateParameter.Text( + order = 7, + name = "fileDescriptorPath", + optional = true, + description = "The path to the Protocol Buffer File Descriptor Set file.", + helpText = + "The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb", + example = "gs://bucket/path/to/descriptor.pb") + String getFileDescriptorPath(); + + @TemplateParameter.Text( + order = 8, + name = "format", + optional = true, + description = "The encoding format for the data stored in Kafka.", + helpText = + "The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON", + example = "JSON") + @Default.String("JSON") + String getFormat(); + + @TemplateParameter.Integer( + order = 9, + name = "maxReadTimeSeconds", + optional = true, + description = "Upper bound of how long to read from Kafka.", + helpText = "Upper bound of how long to read from Kafka in seconds. For example: 60", + example = "60") + Integer getMaxReadTimeSeconds(); + + @TemplateParameter.Text( + order = 10, + name = "messageName", + optional = true, + description = + "The name of the Protocol Buffer message to be used for schema extraction and data conversion.", + helpText = + "The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage", + example = "MyMessage") + String getMessageName(); + + @TemplateParameter.Text( + order = 11, + name = "schema", + optional = true, + description = "The schema in which the data is encoded in the Kafka topic.", + helpText = + "The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO.", + example = + "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"name\", \"type\": \"string\"}]}") + String getSchema(); + + @TemplateParameter.Text( + order = 12, + name = "projectId", + optional = true, + description = "Cloud Spanner Project ID.", + helpText = "The Google Cloud project ID of the Cloud Spanner instance.", + example = "your-project-id") + String getProjectId(); + + @TemplateParameter.Text( + order = 13, + name = "instanceId", + optional = false, + description = "Cloud Spanner Instance ID.", + helpText = "The Cloud Spanner instance ID.", + example = "your-instance-id") + @Validation.Required + String getInstanceId(); + + @TemplateParameter.Text( + order = 14, + name = "databaseId", + optional = false, + description = "Cloud Spanner Database ID.", + helpText = "The Cloud Spanner database ID.", + example = "your-database-id") + @Validation.Required + String getDatabaseId(); + + @TemplateParameter.Text( + order = 15, + name = "tableId", + optional = false, + description = "Cloud Spanner Table ID.", + helpText = "The Cloud Spanner table ID.", + example = "your-table-id") + @Validation.Required + String getTableId(); + + @TemplateParameter.Text( + order = 16, + name = "type", + optional = true, + description = "The type of windowing to apply.", + helpText = + "The type of windowing to apply (e.g., fixed, sliding, sessions, global). Defaults to fixed.", + example = "fixed") + @Default.String("fixed") + String getType(); + + @TemplateParameter.Integer( + order = 17, + name = "size", + optional = true, + description = "The window size in seconds.", + helpText = "The window size in seconds. Defaults to 60.", + example = "60") + @Default.Integer(60) + Integer getSize(); +} diff --git a/yaml/src/main/python/options/kafka_options.yaml b/yaml/src/main/python/options/kafka_options.yaml index 1ec99aae67..fdead46144 100644 --- a/yaml/src/main/python/options/kafka_options.yaml +++ b/yaml/src/main/python/options/kafka_options.yaml @@ -16,34 +16,42 @@ options: required: true type: text - order: 3 + name: "autoOffsetResetConfig" + description: "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server." + help: "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest" + example: "earliest" + required: false + default: "earliest" + type: text + - order: 4 name: "confluentSchemaRegistrySubject" description: "The subject name for the Confluent Schema Registry." help: "The subject name for the Confluent Schema Registry. For example: my_subject" example: "my_subject" required: false type: text - - order: 4 + - order: 5 name: "confluentSchemaRegistryUrl" description: "The URL for the Confluent Schema Registry." help: "The URL for the Confluent Schema Registry. For example: http://schema-registry:8081" example: "http://schema-registry:8081" required: false type: text - - order: 5 + - order: 6 name: "consumerConfigUpdates" description: "A list of key-value pairs that act as configuration parameters for Kafka consumers." help: "A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'}" example: '{"group.id": "my_group"}' required: false type: map - - order: 6 + - order: 7 name: "fileDescriptorPath" description: "The path to the Protocol Buffer File Descriptor Set file." help: "The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb" example: "gs://bucket/path/to/descriptor.pb" required: false type: text - - order: 7 + - order: 8 name: "format" description: "The encoding format for the data stored in Kafka." help: "The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON" @@ -51,14 +59,21 @@ options: required: false default: "JSON" type: text - - order: 8 + - order: 9 + name: "maxReadTimeSeconds" + description: "Upper bound of how long to read from Kafka." + help: "Upper bound of how long to read from Kafka in seconds. For example: 60" + example: "60" + required: false + type: integer + - order: 10 name: "messageName" description: "The name of the Protocol Buffer message to be used for schema extraction and data conversion." help: "The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage" example: "MyMessage" required: false type: text - - order: 9 + - order: 11 name: "schema" description: "The schema in which the data is encoded in the Kafka topic." help: > diff --git a/yaml/src/main/python/options/spanner_options.yaml b/yaml/src/main/python/options/spanner_options.yaml new file mode 100644 index 0000000000..9f46860f9e --- /dev/null +++ b/yaml/src/main/python/options/spanner_options.yaml @@ -0,0 +1,31 @@ +options: + - name: "spanner_common_options" + parameters: + - order: 1 + name: "projectId" + description: "Cloud Spanner Project ID." + help: "The Google Cloud project ID of the Cloud Spanner instance." + example: "your-project-id" + required: false + type: text + - order: 2 + name: "instanceId" + description: "Cloud Spanner Instance ID." + help: "The Cloud Spanner instance ID." + example: "your-instance-id" + required: true + type: text + - order: 3 + name: "databaseId" + description: "Cloud Spanner Database ID." + help: "The Cloud Spanner database ID." + example: "your-database-id" + required: true + type: text + - order: 4 + name: "tableId" + description: "Cloud Spanner Table ID." + help: "The Cloud Spanner table ID." + example: "your-table-id" + required: true + type: text diff --git a/yaml/src/main/python/options/windowing_options.yaml b/yaml/src/main/python/options/windowing_options.yaml new file mode 100644 index 0000000000..b695914ede --- /dev/null +++ b/yaml/src/main/python/options/windowing_options.yaml @@ -0,0 +1,19 @@ +options: + - name: "windowing_common_options" + parameters: + - order: 1 + name: "type" + description: "The type of windowing to apply." + help: "The type of windowing to apply (e.g., fixed, sliding, sessions, global). Defaults to fixed." + example: "fixed" + required: false + default: "fixed" + type: text + - order: 2 + name: "size" + description: "The window size in seconds." + help: "The window size in seconds. Defaults to 60." + example: "60" + required: false + default: 60 + type: integer diff --git a/yaml/src/main/yaml/KafkaToBigQuery.yaml b/yaml/src/main/yaml/KafkaToBigQuery.yaml index d06fa6ad5a..ecd22e5148 100644 --- a/yaml/src/main/yaml/KafkaToBigQuery.yaml +++ b/yaml/src/main/yaml/KafkaToBigQuery.yaml @@ -62,7 +62,8 @@ pipeline: format: {{ format | default('JSON') }} topic: {{ topic }} bootstrap_servers: {{ bootstrapServers }} - auto_offset_reset_config: 'earliest' + max_read_time_seconds: {{ maxReadTimeSeconds }} + auto_offset_reset_config: {{ autoOffsetResetConfig | default('earliest') }} error_handling: output: errors - type: WriteToBigQuery diff --git a/yaml/src/main/yaml/KafkaToBigtable.yaml b/yaml/src/main/yaml/KafkaToBigtable.yaml index 0dceaa22f7..d7884223b1 100644 --- a/yaml/src/main/yaml/KafkaToBigtable.yaml +++ b/yaml/src/main/yaml/KafkaToBigtable.yaml @@ -49,12 +49,13 @@ pipeline: consumer_config: {{ consumerConfigUpdates }} file_descriptor_path: {{ fileDescriptorPath }} format: {{ format | default('JSON') }} + max_read_time_seconds: {{ maxReadTimeSeconds }} message_name: {{ messageName }} {% if schema %} schema: | {{ schema }} {% endif %} - auto_offset_reset_config: 'earliest' + auto_offset_reset_config: {{ autoOffsetResetConfig | default('earliest') }} - type: MapToFields name: ConvertStringsToBytes input: ReadFromKafka diff --git a/yaml/src/main/yaml/KafkaToIceberg.yaml b/yaml/src/main/yaml/KafkaToIceberg.yaml index 749a4ceee3..f511681740 100644 --- a/yaml/src/main/yaml/KafkaToIceberg.yaml +++ b/yaml/src/main/yaml/KafkaToIceberg.yaml @@ -57,12 +57,13 @@ pipeline: consumer_config: {{ consumerConfigUpdates }} file_descriptor_path: {{ fileDescriptorPath }} format: {{ format | default('JSON') }} + max_read_time_seconds: {{ maxReadTimeSeconds }} message_name: {{ messageName }} {% if schema %} schema: | {{ schema }} {% endif %} - auto_offset_reset_config: 'earliest' + auto_offset_reset_config: {{ autoOffsetResetConfig | default('earliest') }} - type: WriteToIceberg name: WriteToIceberg diff --git a/yaml/src/main/yaml/KafkaToSpanner.yaml b/yaml/src/main/yaml/KafkaToSpanner.yaml new file mode 100644 index 0000000000..a640124503 --- /dev/null +++ b/yaml/src/main/yaml/KafkaToSpanner.yaml @@ -0,0 +1,68 @@ +template: + + name: "Kafka_To_Spanner_Yaml" + category: "STREAMING" + type: "YAML" + display_name: "Kafka to Spanner (YAML)" + description: > + The Kafka to Spanner template is a streaming pipeline which ingests + data from an Apache Kafka topic and writes the resulting records to + Cloud Spanner. + flex_container_name: "pipeline-yaml" + yamlTemplateFile: "KafkaToSpanner.yaml" + filesToCopy: > + {"main.py", "requirements.txt"} + documentation: > + https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-spanner + contactInformation: "https://cloud.google.com/support" + requirements: { + "The input Apache Kafka topic must exist.", + "The Apache Kafka broker server must be running and be reachable from the Dataflow worker machines.", + "The output Cloud Spanner instance, database, and table must exist." + } + streaming: true + hidden: false + + options_file: + - "kafka_options" + - "spanner_options" + - "windowing_options" + + parameters: + - kafka_read_options + - spanner_common_options + - windowing_common_options + +pipeline: + type: chain + transforms: + - type: ReadFromKafka + name: ReadFromKafka + config: + bootstrap_servers: {{ bootstrapServers }} + topic: {{ topic }} + confluent_schema_registry_subject: {{ confluentSchemaRegistrySubject }} + confluent_schema_registry_url: {{ confluentSchemaRegistryUrl }} + consumer_config: {{ consumerConfigUpdates }} + file_descriptor_path: {{ fileDescriptorPath }} + format: {{ format | default('JSON') }} + max_read_time_seconds: {{ maxReadTimeSeconds }} + message_name: {{ messageName }} + {% if schema %} + schema: | + {{ schema }} + {% endif %} + auto_offset_reset_config: {{ autoOffsetResetConfig | default('earliest') }} + - type: WriteToSpanner + name: WriteToSpanner + config: + project_id: {{ projectId }} + instance_id: {{ instanceId }} + database_id: {{ databaseId }} + table: {{ tableId }} + windowing: + type: {{ type | default('fixed') }} + size: {{ size | default(60) }} + +options: + streaming: true diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYamlIT.java new file mode 100644 index 0000000000..6a5e23640e --- /dev/null +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYamlIT.java @@ -0,0 +1,183 @@ +/* + * 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.templates.yaml; + +import static com.google.common.truth.Truth.assertThat; +import static org.apache.beam.it.gcp.spanner.matchers.SpannerAsserts.assertThatStructs; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; + +import com.google.cloud.spanner.Struct; +import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; +import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import org.apache.beam.it.common.PipelineLauncher; +import org.apache.beam.it.common.PipelineOperator; +import org.apache.beam.it.common.TestProperties; +import org.apache.beam.it.common.utils.ResourceManagerUtils; +import org.apache.beam.it.gcp.TemplateTestBase; +import org.apache.beam.it.gcp.spanner.SpannerResourceManager; +import org.apache.beam.it.gcp.spanner.conditions.SpannerRowsCheck; +import org.apache.beam.it.kafka.KafkaResourceManager; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Integration test for {@link KafkaToSpannerYaml}. */ +@Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) +@TemplateIntegrationTest(KafkaToSpannerYaml.class) +@RunWith(JUnit4.class) +public final class KafkaToSpannerYamlIT extends TemplateTestBase { + + private static final Logger LOG = LoggerFactory.getLogger(KafkaToSpannerYamlIT.class); + + private KafkaResourceManager kafkaResourceManager; + private SpannerResourceManager spannerResourceManager; + + @Before + public void setUp() throws IOException { + kafkaResourceManager = + KafkaResourceManager.builder(testName).setHost(TestProperties.hostIp()).build(); + spannerResourceManager = + SpannerResourceManager.builder(testName, PROJECT, REGION) + .maybeUseStaticInstance() + .setCredentials(credentials) + .build(); + } + + @After + public void tearDown() { + ResourceManagerUtils.cleanResources(kafkaResourceManager, spannerResourceManager); + } + + @Test + public void testKafkaToSpanner() throws IOException { + kafkaToSpanner(Function.identity()); + } + + public void kafkaToSpanner( + Function + paramsAdder) + throws IOException { + + LOG.info("Starting kafkaToSpanner test. Test name: {}. Spec path: {}", testName, specPath); + + // Arrange + LOG.info("Creating Kafka topic..."); + String topicName = kafkaResourceManager.createTopic(testName, 5); + + LOG.info("Creating Spanner table..."); + String tableId = testName; + spannerResourceManager.executeDdlStatement( + String.format( + "CREATE TABLE `%s` (\n" + + " id INT64 NOT NULL,\n" + + " name STRING(1024)\n" + + ") PRIMARY KEY (id)", + tableId)); + + String bootstrapServers = + kafkaResourceManager.getBootstrapServers().replace("PLAINTEXT://", ""); + LOG.info("Using Kafka Bootstrap Servers: {}", bootstrapServers); + + LOG.info("Creating launch config with yaml pipeline parameters..."); + PipelineLauncher.LaunchConfig.Builder options = + paramsAdder.apply( + PipelineLauncher.LaunchConfig.builder(testName, specPath) + .addParameter("bootstrapServers", bootstrapServers) + .addParameter("topic", topicName) + .addParameter("format", "JSON") + .addParameter( + "schema", + "{\"type\":\"object\",\"properties\":{\"id\":{\"type\":\"integer\"},\"name\":{\"type\":\"string\"}}}") + .addParameter("projectId", PROJECT) + .addParameter("instanceId", spannerResourceManager.getInstanceId()) + .addParameter("databaseId", spannerResourceManager.getDatabaseId()) + .addParameter("tableId", tableId)); + + // Act + LOG.info("Launching template with options..."); + PipelineLauncher.LaunchInfo info = launchTemplate(options); + LOG.info("Template launched. LaunchInfo: {}", info); + assertThatPipeline(info).isRunning(); + + LOG.info("Publishing messages into the Kafka topic..."); + List> expectedRecords = new ArrayList<>(); + try (KafkaProducer kafkaProducer = + kafkaResourceManager.buildProducer(new StringSerializer(), new StringSerializer())) { + for (int i = 1; i <= 10; i++) { + long id1 = Long.parseLong(i + "1"); + long id2 = Long.parseLong(i + "2"); + publish( + kafkaProducer, + topicName, + String.valueOf(id1), + "{\"id\": " + id1 + ", \"name\": \"Dataflow\"}"); + publish( + kafkaProducer, + topicName, + String.valueOf(id2), + "{\"id\": " + id2 + ", \"name\": \"Spanner\"}"); + expectedRecords.add(Map.of("id", id1, "name", "Dataflow")); + expectedRecords.add(Map.of("id", id2, "name", "Spanner")); + } + } + + LOG.info("Waiting for pipeline condition..."); + PipelineOperator.Result result = + pipelineOperator() + .waitForConditionAndFinish( + createConfig(info), + SpannerRowsCheck.builder(spannerResourceManager, tableId).setMinRows(20).build()); + + // Assert + assertThatResult(result).meetsConditions(); + + LOG.info("Verifying 20 rows in Spanner table..."); + List records = spannerResourceManager.readTableRecords(tableId, List.of("id", "name")); + assertThat(records).hasSize(20); + assertThatStructs(records).hasRecordsUnorderedCaseInsensitiveColumns(expectedRecords); + } + + private void publish( + KafkaProducer producer, String topicName, String key, String value) { + try { + RecordMetadata recordMetadata = + producer.send(new ProducerRecord<>(topicName, key, value)).get(); + LOG.info( + "Published record {}, partition {} - offset: {}", + recordMetadata.topic(), + recordMetadata.partition(), + recordMetadata.offset()); + LOG.info("Published record with key: {}, value: {}", key, value); + } catch (Exception e) { + throw new RuntimeException("Error publishing record to Kafka", e); + } + } +}