Add support for UDF function transformations in PubSubToRedis template (#3338)

* Merge branch 'udb_redis'

* Update RedisHashIO to use AutoValue and move UDF comment in PubSubToRedis

* Specify standard charset as UTF-8

* Add assertions to RedisStreamsSink test

* Update deadletter Pub/Sub topic description

* Relax validation so DLQ is required when UDF is configured

* Apply formatting with spotless

---------

Co-authored-by: Ulada Butsenka <uladabutsenka@google.com>

Author: wladabutenko

metadata/src/main/java/com/google/cloud/teleport/metadata/util/MetadataUtils.java

@@ -184,10 +184,10 @@ public static List<String> getRegexes(Annotation parameterAnnotation) {
         return List.of("^(true|false)$");
       case "Integer":
         TemplateParameter.Integer integerParam = (TemplateParameter.Integer) parameterAnnotation;
-        return List.of("^[0-9]+$");
+        return List.of("^-?[0-9]+$");
       case "Long":
         TemplateParameter.Long longParam = (TemplateParameter.Long) parameterAnnotation;
-        return List.of("^[0-9]+$");
+        return List.of("^-?[0-9]+$");
       case "Float":
         TemplateParameter.Float floatParam = (TemplateParameter.Float) parameterAnnotation;
         return List.of("^-?(0|[1-9][0-9]*)(\\.[0-9]+)?([eE][-+]?[0-9]+)?$");

plugins/templates-maven-plugin/pom.xml

@@ -50,6 +50,11 @@
       <artifactId>core-plugin</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.google.cloud.teleport.metadata</groupId>
+      <artifactId>metadata</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-dataflow</artifactId>

v2/pubsub-to-redis/README_Cloud_PubSub_to_Redis.md

@@ -6,6 +6,10 @@ Pub/Sub subscription and writes the message payload to Redis. The most common us
 case of this template is to export logs to Redis Enterprise for advanced
 search-based log analysis in real time.
 
+Before writing to Redis, you can apply a JavaScript user-defined function to the
+message payload. Any messages that experience processing failures are forwarded
+to a Pub/Sub unprocessed topic for further troubleshooting and reprocessing.
+
 For added security, enable an SSL connection when setting up your database
 endpoint connection.
 
@@ -33,7 +37,21 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat
 * **redisSinkType**: The Redis sink. Supported values are `STRING_SINK, HASH_SINK, STREAMS_SINK, and LOGGING_SINK`. For example, `STRING_SINK`. Defaults to: STRING_SINK.
 * **connectionTimeout**: The Redis connection timeout in milliseconds.  For example, `2000`. Defaults to: 2000.
 * **ttl**: The key expiration time in seconds. The `ttl` default for `HASH_SINK` is -1, which means it never expires.
+* **outputDeadletterTopic**: The Pub/Sub topic to forward unprocessable messages to. Messages that fail UDF transformation are forwarded here, Required if using a JavaScript UDF. For example, `projects/<PROJECT_ID>/topics/<TOPIC_NAME>`.
+* **javascriptTextTransformGcsPath**: The Cloud Storage URI of the .js file that defines the JavaScript user-defined function (UDF) to use. For example, `gs://my-bucket/my-udfs/my_file.js`.
+* **javascriptTextTransformFunctionName**: The name of the JavaScript user-defined function (UDF) to use. For example, if your JavaScript function code is `myTransform(inJson) { /*...do stuff...*/ }`, then the function name is `myTransform`. For sample JavaScript UDFs, see UDF Examples (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).
+* **javascriptTextTransformReloadIntervalMinutes**: Specifies how frequently to reload the UDF, in minutes. If the value is greater than 0, Dataflow periodically checks the UDF file in Cloud Storage, and reloads the UDF if the file is modified. This parameter allows you to update the UDF while the pipeline is running, without needing to restart the job. If the value is `0`, UDF reloading is disabled. The default value is `0`.
+
+
+## User-Defined functions (UDFs)
+
+The Pub/Sub to Redis Template supports User-Defined functions (UDFs).
+UDFs allow you to customize functionality by providing a JavaScript function
+without having to maintain or build the entire template code.
 
+Check [Create user-defined functions for Dataflow templates](https://cloud.google.com/dataflow/docs/guides/templates/create-template-udf)
+and [Using UDFs](https://github.com/GoogleCloudPlatform/DataflowTemplates#using-udfs)
+for more information about how to create and test those functions.
 
 
 ## Getting Started
@@ -135,6 +153,10 @@ export SSL_ENABLED=false
 export REDIS_SINK_TYPE=STRING_SINK
 export CONNECTION_TIMEOUT=2000
 export TTL=-1
+export OUTPUT_DEADLETTER_TOPIC=<outputDeadletterTopic>
+export JAVASCRIPT_TEXT_TRANSFORM_GCS_PATH=<javascriptTextTransformGcsPath>
+export JAVASCRIPT_TEXT_TRANSFORM_FUNCTION_NAME=<javascriptTextTransformFunctionName>
+export JAVASCRIPT_TEXT_TRANSFORM_RELOAD_INTERVAL_MINUTES=0
 
 gcloud dataflow flex-template run "cloud-pubsub-to-redis-job" \
   --project "$PROJECT" \
@@ -147,7 +169,11 @@ gcloud dataflow flex-template run "cloud-pubsub-to-redis-job" \
   --parameters "sslEnabled=$SSL_ENABLED" \
   --parameters "redisSinkType=$REDIS_SINK_TYPE" \
   --parameters "connectionTimeout=$CONNECTION_TIMEOUT" \
-  --parameters "ttl=$TTL"
+  --parameters "ttl=$TTL" \
+  --parameters "outputDeadletterTopic=$OUTPUT_DEADLETTER_TOPIC" \
+  --parameters "javascriptTextTransformGcsPath=$JAVASCRIPT_TEXT_TRANSFORM_GCS_PATH" \
+  --parameters "javascriptTextTransformFunctionName=$JAVASCRIPT_TEXT_TRANSFORM_FUNCTION_NAME" \
+  --parameters "javascriptTextTransformReloadIntervalMinutes=$JAVASCRIPT_TEXT_TRANSFORM_RELOAD_INTERVAL_MINUTES"
 ```
 
 For more information about the command, please check:
@@ -176,6 +202,10 @@ export SSL_ENABLED=false
 export REDIS_SINK_TYPE=STRING_SINK
 export CONNECTION_TIMEOUT=2000
 export TTL=-1
+export OUTPUT_DEADLETTER_TOPIC=<outputDeadletterTopic>
+export JAVASCRIPT_TEXT_TRANSFORM_GCS_PATH=<javascriptTextTransformGcsPath>
+export JAVASCRIPT_TEXT_TRANSFORM_FUNCTION_NAME=<javascriptTextTransformFunctionName>
+export JAVASCRIPT_TEXT_TRANSFORM_RELOAD_INTERVAL_MINUTES=0
 
 mvn clean package -PtemplatesRun \
 -DskipTests \
@@ -184,7 +214,7 @@ mvn clean package -PtemplatesRun \
 -Dregion="$REGION" \
 -DjobName="cloud-pubsub-to-redis-job" \
 -DtemplateName="Cloud_PubSub_to_Redis" \
--Dparameters="inputSubscription=$INPUT_SUBSCRIPTION,redisHost=$REDIS_HOST,redisPort=$REDIS_PORT,redisPassword=$REDIS_PASSWORD,sslEnabled=$SSL_ENABLED,redisSinkType=$REDIS_SINK_TYPE,connectionTimeout=$CONNECTION_TIMEOUT,ttl=$TTL" \
+-Dparameters="inputSubscription=$INPUT_SUBSCRIPTION,redisHost=$REDIS_HOST,redisPort=$REDIS_PORT,redisPassword=$REDIS_PASSWORD,sslEnabled=$SSL_ENABLED,redisSinkType=$REDIS_SINK_TYPE,connectionTimeout=$CONNECTION_TIMEOUT,ttl=$TTL,outputDeadletterTopic=$OUTPUT_DEADLETTER_TOPIC,javascriptTextTransformGcsPath=$JAVASCRIPT_TEXT_TRANSFORM_GCS_PATH,javascriptTextTransformFunctionName=$JAVASCRIPT_TEXT_TRANSFORM_FUNCTION_NAME,javascriptTextTransformReloadIntervalMinutes=$JAVASCRIPT_TEXT_TRANSFORM_RELOAD_INTERVAL_MINUTES" \
 -f v2/pubsub-to-redis
 ```
 
@@ -237,6 +267,10 @@ resource "google_dataflow_flex_template_job" "cloud_pubsub_to_redis" {
     # redisSinkType = "STRING_SINK"
     # connectionTimeout = "2000"
     # ttl = "-1"
+    # outputDeadletterTopic = "<outputDeadletterTopic>"
+    # javascriptTextTransformGcsPath = "<javascriptTextTransformGcsPath>"
+    # javascriptTextTransformFunctionName = "<javascriptTextTransformFunctionName>"
+    # javascriptTextTransformReloadIntervalMinutes = "0"
   }
 }
 ```

v2/pubsub-to-redis/pom.xml

@@ -62,6 +62,41 @@
             <version>${ptr.jedis-mock.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <!-- Integration Test Dependencies -->
+        <dependency>
+            <groupId>com.google.cloud.teleport</groupId>
+            <artifactId>it-google-cloud-platform</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.google.cloud.teleport.metadata</groupId>
+            <artifactId>metadata</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 
+    <build>
+        <plugins>
+            <!-- Configure Surefire to run integration tests sequentially.
+                 This is required because all tests share the same fixed Redis port (6379). -->
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>${surefire.version}</version>
+                <configuration>
+                    <!-- Run tests sequentially to avoid port conflicts on fixed port 6379 -->
+                    <parallel>none</parallel>
+                    <threadCount>1</threadCount>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
 </project>