Update RedisHashIO to use AutoValue and move UDF comment in PubSubToRedis

Author: Ulada Butsenka

v2/pubsub-to-redis/src/main/java/com/google/cloud/teleport/v2/templates/PubSubToRedis.java

@@ -390,12 +390,11 @@ public static PipelineResult run(PubSubToRedisOptions options) {
    *
    * <p>If no UDF is configured, returns the input messages unchanged.
    *
-   * <p>This follows the same pattern as {@code PubsubProtoToBigQuery.runUdf}.
-   *
    * @param messages the input PubSub messages
    * @param options the pipeline options
    * @return the (possibly transformed) PubSub messages
    */
+  // This follows the same pattern as PubsubProtoToBigQuery.runUdf
   static PCollection<PubsubMessage> applyUdf(
       PCollection<PubsubMessage> messages, PubSubToRedisOptions options) {
 

v2/pubsub-to-redis/src/main/java/com/google/cloud/teleport/v2/templates/io/RedisHashIO.java

@@ -15,6 +15,7 @@
  */
 package com.google.cloud.teleport.v2.templates.io;
 
+import com.google.auto.value.AutoValue;
 import com.google.common.base.Preconditions;
 import java.util.HashMap;
 import java.util.Map;
@@ -27,43 +28,41 @@
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PDone;
 import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.Pipeline;
 import redis.clients.jedis.Transaction;
 
 public abstract class RedisHashIO {
 
   public static WriteHash write() {
-    return new WriteHash(RedisConnectionConfiguration.create(), null);
+
+    return (new AutoValue_RedisHashIO_WriteHash.Builder())
+        .setConnectionConfiguration(RedisConnectionConfiguration.create())
+        .build();
   }
 
-  public static class WriteHash
+  @AutoValue
+  public abstract static class WriteHash
       extends PTransform<@NonNull PCollection<KV<String, KV<String, String>>>, @NonNull PDone> {
+    public WriteHash() {}
 
-    private final RedisConnectionConfiguration connectionConfiguration;
-    private final Long expireTime;
+    @Nullable
+    abstract RedisConnectionConfiguration connectionConfiguration();
 
-    private WriteHash(RedisConnectionConfiguration connectionConfiguration, Long expireTime) {
-      this.connectionConfiguration = connectionConfiguration;
-      this.expireTime = expireTime;
-    }
+    @Nullable
+    abstract Long expireTime();
 
-    public RedisConnectionConfiguration connectionConfiguration() {
-      return connectionConfiguration;
-    }
-
-    public Long expireTime() {
-      return expireTime;
-    }
+    abstract RedisHashIO.WriteHash.Builder builder();
 
     public RedisHashIO.WriteHash withConnectionConfiguration(
         RedisConnectionConfiguration connectionConfiguration) {
       Preconditions.checkArgument(connectionConfiguration != null, "connection cannot be null");
-      return new WriteHash(connectionConfiguration, this.expireTime);
+      return this.builder().setConnectionConfiguration(connectionConfiguration).build();
     }
 
     public RedisHashIO.WriteHash withTtl(Long expireTimeMillis) {
-      return new WriteHash(this.connectionConfiguration, expireTimeMillis);
+      return this.builder().setExpireTime(expireTimeMillis).build();
     }
 
     @NonNull
@@ -149,5 +148,17 @@ public void teardown() {
         this.jedis.close();
       }
     }
+
+    @AutoValue.Builder
+    abstract static class Builder {
+      Builder() {}
+
+      abstract WriteHash.Builder setExpireTime(Long expireTimeMillis);
+
+      abstract WriteHash.Builder setConnectionConfiguration(
+          RedisConnectionConfiguration connectionConfiguration);
+
+      abstract WriteHash build();
+    }
   }
 }