feast-dev
diff --git a/‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisCustomIO.java‎
Lines changed: 2 additions & 5 deletions b/‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisCustomIO.java‎
Lines changed: 2 additions & 5 deletions
diff --git a/‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java‎
Lines changed: 15 additions & 15 deletions b/‎storage/connectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java‎
Lines changed: 15 additions & 15 deletions
@@ -46,9 +46,6 @@
 
 public class RedisCustomIO {
 
-  private static final int DEFAULT_BATCH_SIZE = 10000;
-  private static final int DEFAULT_FREQUENCY_SECONDS = 30;
-
   private static TupleTag<FeatureRow> successfulInsertsTag =
       new TupleTag<FeatureRow>("successfulInserts") {};
   private static TupleTag<FailedElement> failedInsertsTupleTag =
@@ -69,8 +66,8 @@ public static class Write extends PTransform<PCollection<FeatureRow>, WriteResul
 
     private PCollectionView<Map<String, Iterable<FeatureSetSpec>>> featureSetSpecs;
     private RedisIngestionClient redisIngestionClient;
-    private int batchSize = DEFAULT_BATCH_SIZE;
-    private Duration flushFrequency = Duration.standardSeconds(DEFAULT_FREQUENCY_SECONDS);
+    private int batchSize;
+    private Duration flushFrequency;
 
     public Write(
         RedisIngestionClient redisIngestionClient,
 
@@ -39,6 +39,8 @@
 
 @AutoValue
 public abstract class RedisFeatureSink implements FeatureSink {
+  private static final int DEFAULT_BATCH_SIZE = 10000;
+  private static final int DEFAULT_FREQUENCY_SECONDS = 30;
 
   /**
    * Initialize a {@link RedisFeatureSink.Builder} from a {@link StoreProto.Store.RedisConfig}.
@@ -113,30 +115,28 @@ public PCollection<FeatureSetReference> prepareWrite(
 
   @Override
   public PTransform<PCollection<FeatureRow>, WriteResult> writer() {
+    int flushFrequencySeconds = DEFAULT_FREQUENCY_SECONDS;
+
     if (getRedisClusterConfig() != null) {
-      RedisCustomIO.Write writer =
-          new RedisCustomIO.Write(
-              new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView());
 
       if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) {
-        writer =
-            writer.withFlushFrequency(
-                Duration.standardSeconds(getRedisClusterConfig().getFlushFrequencySeconds()));
+        flushFrequencySeconds = getRedisClusterConfig().getFlushFrequencySeconds();
       }
 
-      return writer;
-    } else if (getRedisConfig() != null) {
-      RedisCustomIO.Write writer =
-          new RedisCustomIO.Write(
-              new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView());
+      return new RedisCustomIO.Write(
+              new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView())
+          .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds))
+          .withBatchSize(DEFAULT_BATCH_SIZE);
 
+    } else if (getRedisConfig() != null) {
       if (getRedisConfig().getFlushFrequencySeconds() > 0) {
-        writer =
-            writer.withFlushFrequency(
-                Duration.standardSeconds(getRedisConfig().getFlushFrequencySeconds()));
+        flushFrequencySeconds = getRedisConfig().getFlushFrequencySeconds();
       }
 
-      return writer;
+      return new RedisCustomIO.Write(
+              new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView())
+          .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds))
+          .withBatchSize(DEFAULT_BATCH_SIZE);
     } else {
       throw new RuntimeException(
           "At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink");