Fix RedisVectorStore not closing Jedis pipelines

Author: Bragolgirith

vector-stores/spring-ai-redis/src/main/java/org/springframework/ai/vectorstore/RedisVectorStore.java

@@ -303,25 +303,26 @@ public JedisPooled getJedis() {
 
 	@Override
 	public void add(List<Document> documents) {
-		Pipeline pipeline = this.jedis.pipelined();
-		for (Document document : documents) {
-			var embedding = this.embeddingClient.embed(document);
-			document.setEmbedding(embedding);
-
-			var fields = new HashMap<String, Object>();
-			fields.put(this.config.embeddingFieldName, embedding);
-			fields.put(this.config.contentFieldName, document.getContent());
-			fields.putAll(document.getMetadata());
-			pipeline.jsonSetWithEscape(key(document.getId()), JSON_SET_PATH, fields);
-		}
-		List<Object> responses = pipeline.syncAndReturnAll();
-		Optional<Object> errResponse = responses.stream().filter(Predicate.not(RESPONSE_OK)).findAny();
-		if (errResponse.isPresent()) {
-			String message = MessageFormat.format("Could not add document: {0}", errResponse.get());
-			if (logger.isErrorEnabled()) {
-				logger.error(message);
+		try (Pipeline pipeline = this.jedis.pipelined()) {
+			for (Document document : documents) {
+				var embedding = this.embeddingClient.embed(document);
+				document.setEmbedding(embedding);
+
+				var fields = new HashMap<String, Object>();
+				fields.put(this.config.embeddingFieldName, embedding);
+				fields.put(this.config.contentFieldName, document.getContent());
+				fields.putAll(document.getMetadata());
+				pipeline.jsonSetWithEscape(key(document.getId()), JSON_SET_PATH, fields);
+			}
+			List<Object> responses = pipeline.syncAndReturnAll();
+			Optional<Object> errResponse = responses.stream().filter(Predicate.not(RESPONSE_OK)).findAny();
+			if (errResponse.isPresent()) {
+				String message = MessageFormat.format("Could not add document: {0}", errResponse.get());
+				if (logger.isErrorEnabled()) {
+					logger.error(message);
+				}
+				throw new RuntimeException(message);
 			}
-			throw new RuntimeException(message);
 		}
 	}
 
@@ -331,19 +332,20 @@ private String key(String id) {
 
 	@Override
 	public Optional<Boolean> delete(List<String> idList) {
-		Pipeline pipeline = this.jedis.pipelined();
-		for (String id : idList) {
-			pipeline.jsonDel(key(id));
-		}
-		List<Object> responses = pipeline.syncAndReturnAll();
-		Optional<Object> errResponse = responses.stream().filter(Predicate.not(RESPONSE_DEL_OK)).findAny();
-		if (errResponse.isPresent()) {
-			if (logger.isErrorEnabled()) {
-				logger.error("Could not delete document: {}", errResponse.get());
+		try (Pipeline pipeline = this.jedis.pipelined()) {
+			for (String id : idList) {
+				pipeline.jsonDel(key(id));
+			}
+			List<Object> responses = pipeline.syncAndReturnAll();
+			Optional<Object> errResponse = responses.stream().filter(Predicate.not(RESPONSE_DEL_OK)).findAny();
+			if (errResponse.isPresent()) {
+				if (logger.isErrorEnabled()) {
+					logger.error("Could not delete document: {}", errResponse.get());
+				}
+				return Optional.of(false);
 			}
-			return Optional.of(false);
+			return Optional.of(true);
 		}
-		return Optional.of(true);
 	}
 
 	@Override