feast-dev
diff --git a/‎ingestion/src/main/java/feast/store/serving/redis/RedisCustomIO.java‎
Lines changed: 7 additions & 9 deletions b/‎ingestion/src/main/java/feast/store/serving/redis/RedisCustomIO.java‎
Lines changed: 7 additions & 9 deletions
@@ -138,12 +138,16 @@ public static class WriteDoFn extends DoFn<RedisMutation, Void> {
       }
 
       public WriteDoFn withBatchSize(int batchSize) {
-        if (batchSize > 0) this.batchSize = batchSize;
+        if (batchSize > 0) {
+          this.batchSize = batchSize;
+        }
         return this;
       }
 
       public WriteDoFn withTimeout(int timeout) {
-        if (timeout > 0) this.timeout = timeout;
+        if (timeout > 0) {
+          this.timeout = timeout;
+        }
         return this;
       }
 
@@ -155,7 +159,6 @@ public void setup() {
       @StartBundle
       public void startBundle() {
         pipeline = jedis.pipelined();
-        pipeline.multi();
         batchCount = 0;
       }
 
@@ -168,9 +171,7 @@ public void processElement(ProcessContext context) {
         }
         batchCount++;
         if (batchCount >= batchSize) {
-          pipeline.exec();
           pipeline.sync();
-          pipeline.multi();
           batchCount = 0;
         }
       }
@@ -197,10 +198,7 @@ private Response<?> writeRecord(RedisMutation mutation) {
 
       @FinishBundle
       public void finishBundle() {
-        if (pipeline.isInMulti()) {
-          pipeline.exec();
-          pipeline.sync();
-        }
+        pipeline.sync();
         batchCount = 0;
       }
Original file line number	Diff line number	Diff line change
`@@ -138,12 +138,16 @@ public static class WriteDoFn extends DoFn<RedisMutation, Void> {`
`138`	`138`	`}`
`139`	`139`
`140`	`140`	`public WriteDoFn withBatchSize(int batchSize) {`
`141`		`- if (batchSize > 0) this.batchSize = batchSize;`
	`141`	`+ if (batchSize > 0) {`
	`142`	`+ this.batchSize = batchSize;`
	`143`	`+ }`
`142`	`144`	`return this;`
`143`	`145`	`}`
`144`	`146`
`145`	`147`	`public WriteDoFn withTimeout(int timeout) {`
`146`		`- if (timeout > 0) this.timeout = timeout;`
	`148`	`+ if (timeout > 0) {`
	`149`	`+ this.timeout = timeout;`
	`150`	`+ }`
`147`	`151`	`return this;`
`148`	`152`	`}`
`149`	`153`
`@@ -155,7 +159,6 @@ public void setup() {`
`155`	`159`	`@StartBundle`
`156`	`160`	`public void startBundle() {`
`157`	`161`	`pipeline = jedis.pipelined();`
`158`		`- pipeline.multi();`
`159`	`162`	`batchCount = 0;`
`160`	`163`	`}`
`161`	`164`
`@@ -168,9 +171,7 @@ public void processElement(ProcessContext context) {`
`168`	`171`	`}`
`169`	`172`	`batchCount++;`
`170`	`173`	`if (batchCount >= batchSize) {`
`171`		`- pipeline.exec();`
`172`	`174`	`pipeline.sync();`
`173`		`- pipeline.multi();`
`174`	`175`	`batchCount = 0;`
`175`	`176`	`}`
`176`	`177`	`}`
`@@ -197,10 +198,7 @@ private Response<?> writeRecord(RedisMutation mutation) {`
`197`	`198`
`198`	`199`	`@FinishBundle`
`199`	`200`	`public void finishBundle() {`
`200`		`- if (pipeline.isInMulti()) {`
`201`		`- pipeline.exec();`
`202`		`- pipeline.sync();`
`203`		`- }`
	`201`	`+ pipeline.sync();`
`204`	`202`	`batchCount = 0;`
`205`	`203`	`}`
`206`	`204`