feat: use AbstractPipeline for Redis Cluster compatibility

Port of Python RedisVL fix for issue #365. Ensures storage classes use
AbstractPipeline instead of Pipeline to support both regular Pipeline
and MultiClusterPipeline objects.

Changes:
- BaseStorage: Changed Pipeline to AbstractPipeline in write() and get()
  methods, removed unsafe casts to Pipeline
- HashStorage: Updated set() and getResponse() signatures to accept
  AbstractPipeline, updated get() to use AbstractPipeline
- JsonStorage: Updated set() and getResponse() signatures to accept
  AbstractPipeline
- EmbeddingsCache: Updated mset(), mget(), and mexists() to use
  AbstractPipeline
- Add AbstractPipelineTest with 4 tests verifying compatibility

This ensures compatibility with both standalone Redis and Redis Cluster
deployments, matching Python's graceful handling of both Pipeline and
ClusterPipeline types via isinstance() checks.

In Java/Jedis:
- Pipeline extends PipelineBase (deprecated)
- MultiClusterPipeline extends PipelineBase
- Both extend AbstractPipeline as common base
- UnifiedJedis.pipelined() returns AbstractPipeline (may be Pipeline or
  MultiClusterPipeline depending on provider)

Without this fix, code would fail with ClassCastException when using
MultiClusterPipeline in cluster deployments.

Author: bsbodden

core/src/main/java/com/redis/vl/extensions/cache/EmbeddingsCache.java

@@ -8,7 +8,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import redis.clients.jedis.Pipeline;
+import redis.clients.jedis.AbstractPipeline;
 import redis.clients.jedis.Response;
 import redis.clients.jedis.UnifiedJedis;
 
@@ -160,7 +160,7 @@ public void mset(Map<String, float[]> embeddings, String modelName) {
       return;
     }
 
-    try (Pipeline pipeline = (Pipeline) redisClient.pipelined()) {
+    try (AbstractPipeline pipeline = redisClient.pipelined()) {
       for (Map.Entry<String, float[]> entry : embeddings.entrySet()) {
         String key = generateKey(entry.getKey(), modelName);
         byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
@@ -192,7 +192,7 @@ public Map<String, float[]> mget(List<String> texts, String modelName) {
     Map<String, float[]> results = new HashMap<>();
     Map<String, Response<byte[]>> responses = new HashMap<>();
 
-    try (Pipeline pipeline = (Pipeline) redisClient.pipelined()) {
+    try (AbstractPipeline pipeline = redisClient.pipelined()) {
       for (String text : texts) {
         String key = generateKey(text, modelName);
         byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
@@ -227,7 +227,7 @@ public Map<String, Boolean> mexists(List<String> texts, String modelName) {
     Map<String, Boolean> results = new HashMap<>();
     Map<String, Response<Boolean>> responses = new HashMap<>();
 
-    try (Pipeline pipeline = (Pipeline) redisClient.pipelined()) {
+    try (AbstractPipeline pipeline = redisClient.pipelined()) {
       for (String text : texts) {
         String key = generateKey(text, modelName);
         responses.put(text, pipeline.exists(key));

core/src/main/java/com/redis/vl/storage/BaseStorage.java

@@ -7,7 +7,7 @@
 import java.util.function.Function;
 import lombok.AccessLevel;
 import lombok.Getter;
-import redis.clients.jedis.Pipeline;
+import redis.clients.jedis.AbstractPipeline;
 import redis.clients.jedis.Response;
 import redis.clients.jedis.UnifiedJedis;
 
@@ -325,7 +325,7 @@ public List<String> write(
     // Pass 2: Write all valid objects in batches
     List<String> addedKeys = new ArrayList<>();
 
-    try (Pipeline pipeline = (Pipeline) redisClient.pipelined()) {
+    try (AbstractPipeline pipeline = redisClient.pipelined()) {
       for (int i = 0; i < preparedObjects.size(); i++) {
         KeyValuePair kvp = preparedObjects.get(i);
         set(pipeline, kvp.key, kvp.value);
@@ -465,7 +465,7 @@ public List<Map<String, Object>> get(
     // Use a pipeline to batch the retrieval
     List<Response<Map<String, Object>>> responses = new ArrayList<>();
 
-    try (Pipeline pipeline = (Pipeline) redisClient.pipelined()) {
+    try (AbstractPipeline pipeline = redisClient.pipelined()) {
       for (String key : keys) {
         Response<Map<String, Object>> response = getResponse(pipeline, key);
         responses.add(response);
@@ -517,20 +517,29 @@ protected Map<String, Object> convertBytes(Map<String, Object> map) {
   /**
    * Set a key-value pair in Redis using a pipeline.
    *
+   * <p>Python port: Uses AbstractPipeline to support both regular Pipeline and
+   * MultiClusterPipeline, matching Python's handling of both Pipeline and ClusterPipeline types
+   * (issue #365).
+   *
    * @param pipeline The Redis pipeline to use
    * @param key The Redis key
    * @param obj The object to store
    */
-  protected abstract void set(Pipeline pipeline, String key, Map<String, Object> obj);
+  protected abstract void set(AbstractPipeline pipeline, String key, Map<String, Object> obj);
 
   /**
    * Get a response for retrieving a value from Redis using a pipeline.
    *
+   * <p>Python port: Uses AbstractPipeline to support both regular Pipeline and
+   * MultiClusterPipeline, matching Python's handling of both Pipeline and ClusterPipeline types
+   * (issue #365).
+   *
    * @param pipeline The Redis pipeline to use
    * @param key The Redis key
    * @return Response containing the retrieved object
    */
-  protected abstract Response<Map<String, Object>> getResponse(Pipeline pipeline, String key);
+  protected abstract Response<Map<String, Object>> getResponse(
+      AbstractPipeline pipeline, String key);
 
   /** Helper class for key-value pairs used during preprocessing and validation. */
   protected static class KeyValuePair {

core/src/main/java/com/redis/vl/storage/HashStorage.java

@@ -9,7 +9,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import redis.clients.jedis.Pipeline;
+import redis.clients.jedis.AbstractPipeline;
 import redis.clients.jedis.Response;
 
 /**
@@ -29,7 +29,7 @@ public HashStorage(IndexSchema indexSchema) {
   }
 
   @Override
-  protected void set(Pipeline pipeline, String key, Map<String, Object> obj) {
+  protected void set(AbstractPipeline pipeline, String key, Map<String, Object> obj) {
     Map<byte[], byte[]> binaryFields = new HashMap<>();
     Map<String, String> stringFields = new HashMap<>();
 
@@ -75,7 +75,7 @@ protected void set(Pipeline pipeline, String key, Map<String, Object> obj) {
 
   @Override
   @SuppressWarnings("unchecked")
-  protected Response<Map<String, Object>> getResponse(Pipeline pipeline, String key) {
+  protected Response<Map<String, Object>> getResponse(AbstractPipeline pipeline, String key) {
     // For hash, we use hgetAll to get all fields
     Response<Map<String, String>> response = pipeline.hgetAll(key);
     // We need to return Response<Map<String, Object>> so cast it
@@ -101,7 +101,7 @@ public List<Map<String, Object>> get(
     List<Response<Map<String, String>>> stringResponses = new ArrayList<>();
     Map<String, List<Response<byte[]>>> vectorResponses = new HashMap<>();
 
-    try (Pipeline pipeline = (Pipeline) redisClient.pipelined()) {
+    try (AbstractPipeline pipeline = redisClient.pipelined()) {
       // Get all string fields and identify vector fields
       for (String key : keys) {
         Response<Map<String, String>> response = pipeline.hgetAll(key);

core/src/main/java/com/redis/vl/storage/JsonStorage.java

@@ -7,7 +7,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import redis.clients.jedis.Pipeline;
+import redis.clients.jedis.AbstractPipeline;
 import redis.clients.jedis.Response;
 import redis.clients.jedis.json.Path2;
 
@@ -28,7 +28,7 @@ public JsonStorage(IndexSchema indexSchema) {
   }
 
   @Override
-  protected void set(Pipeline pipeline, String key, Map<String, Object> obj) {
+  protected void set(AbstractPipeline pipeline, String key, Map<String, Object> obj) {
     // For JSON storage, vectors are stored as JSON arrays
     Map<String, Object> jsonDocument = new HashMap<>();
 
@@ -83,7 +83,7 @@ protected void set(Pipeline pipeline, String key, Map<String, Object> obj) {
 
   @Override
   @SuppressWarnings("unchecked")
-  protected Response<Map<String, Object>> getResponse(Pipeline pipeline, String key) {
+  protected Response<Map<String, Object>> getResponse(AbstractPipeline pipeline, String key) {
     // For JSON, we get the entire document
     Response<Object> response = pipeline.jsonGet(key);
     // We need to return Response<Map<String, Object>> so cast it

core/src/test/java/com/redis/vl/storage/AbstractPipelineIntegrationTest.java

@@ -0,0 +1,126 @@
+package com.redis.vl.storage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.redis.vl.BaseIntegrationTest;
+import com.redis.vl.schema.IndexSchema;
+import java.util.*;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import redis.clients.jedis.AbstractPipeline;
+
+/**
+ * Integration tests for AbstractPipeline compatibility in storage classes.
+ *
+ * <p>Tests the fix for issue #365: Ensures BaseStorage uses AbstractPipeline instead of Pipeline to
+ * support both regular Pipeline and MultiClusterPipeline objects, matching Python's graceful
+ * handling of both Pipeline and ClusterPipeline types.
+ *
+ * <p>Python port: Matches Python's isinstance() checks for (AsyncPipeline, AsyncClusterPipeline) by
+ * using the common base class AbstractPipeline in Java.
+ */
+@DisplayName("AbstractPipeline Compatibility Tests")
+class AbstractPipelineIntegrationTest extends BaseIntegrationTest {
+
+  @Test
+  @DisplayName("Should accept AbstractPipeline in HashStorage.set()")
+  void testHashStorageAcceptsAbstractPipeline() {
+    // Create a simple schema
+    Map<String, Object> schemaDict =
+        Map.of(
+            "index",
+            Map.of("name", "test-index", "prefix", "test", "storage_type", "hash"),
+            "fields",
+            List.of(Map.of("name", "field1", "type", "text")));
+
+    IndexSchema schema = IndexSchema.fromDict(schemaDict);
+    HashStorage storage = new HashStorage(schema);
+
+    // Verify the set() method signature accepts AbstractPipeline
+    Map<String, Object> testData = Map.of("field1", "value1");
+
+    // Use the unifiedJedis from BaseIntegrationTest (Testcontainers)
+    try (AbstractPipeline pipeline = unifiedJedis.pipelined()) {
+      // This should compile without ClassCastException
+      storage.set(pipeline, "test:key", testData);
+      pipeline.sync();
+    }
+
+    assertThat(true).as("Method signature accepts AbstractPipeline").isTrue();
+  }
+
+  @Test
+  @DisplayName("Should accept AbstractPipeline in JsonStorage.set()")
+  void testJsonStorageAcceptsAbstractPipeline() {
+    // Create a simple schema
+    Map<String, Object> schemaDict =
+        Map.of(
+            "index",
+            Map.of("name", "test-index", "prefix", "test", "storage_type", "json"),
+            "fields",
+            List.of(Map.of("name", "field1", "type", "text", "path", "$.field1")));
+
+    IndexSchema schema = IndexSchema.fromDict(schemaDict);
+    JsonStorage storage = new JsonStorage(schema);
+
+    // Verify the set() method signature accepts AbstractPipeline
+    Map<String, Object> testData = Map.of("field1", "value1");
+
+    // Use the unifiedJedis from BaseIntegrationTest (Testcontainers)
+    try (AbstractPipeline pipeline = unifiedJedis.pipelined()) {
+      // This should compile without ClassCastException
+      storage.set(pipeline, "test:key", testData);
+      pipeline.sync();
+    }
+
+    assertThat(true).as("Method signature accepts AbstractPipeline").isTrue();
+  }
+
+  @Test
+  @DisplayName("Should accept AbstractPipeline in HashStorage.getResponse()")
+  void testHashStorageGetResponseAcceptsAbstractPipeline() {
+    // Create a simple schema
+    Map<String, Object> schemaDict =
+        Map.of(
+            "index",
+            Map.of("name", "test-index", "prefix", "test", "storage_type", "hash"),
+            "fields",
+            List.of(Map.of("name", "field1", "type", "text")));
+
+    IndexSchema schema = IndexSchema.fromDict(schemaDict);
+    HashStorage storage = new HashStorage(schema);
+
+    // Use the unifiedJedis from BaseIntegrationTest (Testcontainers)
+    try (AbstractPipeline pipeline = unifiedJedis.pipelined()) {
+      // This should compile without ClassCastException
+      storage.getResponse(pipeline, "test:key");
+      pipeline.sync();
+    }
+
+    assertThat(true).as("Method signature accepts AbstractPipeline").isTrue();
+  }
+
+  @Test
+  @DisplayName("Should accept AbstractPipeline in JsonStorage.getResponse()")
+  void testJsonStorageGetResponseAcceptsAbstractPipeline() {
+    // Create a simple schema
+    Map<String, Object> schemaDict =
+        Map.of(
+            "index",
+            Map.of("name", "test-index", "prefix", "test", "storage_type", "json"),
+            "fields",
+            List.of(Map.of("name", "field1", "type", "text", "path", "$.field1")));
+
+    IndexSchema schema = IndexSchema.fromDict(schemaDict);
+    JsonStorage storage = new JsonStorage(schema);
+
+    // Use the unifiedJedis from BaseIntegrationTest (Testcontainers)
+    try (AbstractPipeline pipeline = unifiedJedis.pipelined()) {
+      // This should compile without ClassCastException
+      storage.getResponse(pipeline, "test:key");
+      pipeline.sync();
+    }
+
+    assertThat(true).as("Method signature accepts AbstractPipeline").isTrue();
+  }
+}