Instantiate Jedis client lazily and only once per JVM process

khorshuheng · khorshuheng · commit 501e27b2f474 · 2022-06-21T13:01:40.000+08:00
Signed-off-by: khorshuheng &lt;khor.heng@gojek.com&gt;
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala
@@ -67,6 +67,12 @@ class RedisSinkRelation(override val sqlContext: SQLContext, config: SparkRedisC
 
   lazy val isClusterMode: Boolean = checkIfInClusterMode(endpoint)
 
+  lazy val pipelineProvider: PipelineProvider = if (isClusterMode) {
+    ClusterPipelineProvider(endpoint)
+  } else {
+    SingleNodePipelineProvider(newJedisClient(endpoint))
+  }
+
   def newJedisClient(endpoint: RedisEndpoint): Jedis = {
     val jedis = new Jedis(endpoint.host, endpoint.port)
     if (endpoint.password.nonEmpty) {
@@ -95,12 +101,6 @@ class RedisSinkRelation(override val sqlContext: SQLContext, config: SparkRedisC
       java.security.Security.setProperty("networkaddress.cache.ttl", "3");
       java.security.Security.setProperty("networkaddress.cache.negative.ttl", "0");
 
-      val pipelineProvider = if (isClusterMode) {
-        ClusterPipelineProvider(endpoint)
-      } else {
-        SingleNodePipelineProvider(newJedisClient(endpoint))
-      }
-
       // grouped iterator to only allocate memory for a portion of rows
       partition.grouped(properties.pipelineSize).foreach { batch =>
         // group by key and keep only latest row per each key
@@ -146,7 +146,6 @@ class RedisSinkRelation(override val sqlContext: SQLContext, config: SparkRedisC
         }
         writePipeline.close()
       }
-      pipelineProvider.close()
     }
     dataToStore.unpersist()
   }
