Fix RedisPipelineException double-wrapping in LettuceConnection.closePipeline().

The catch (Exception ex) block in closePipeline() was too broad, catching intentionally thrown RedisPipelineExceptions and re-wrapping them with the misleading default message "Pipeline contained one or more invalid commands". Introduce a dedicated catch (RedisPipelineException ex) block to propagate the original exception directly.

Closes #3346
Original pull request: #3347

Signed-off-by: JEONGJAEIK <wodlr1207@naver.com>

Author: JEONGJAEIK

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java

@@ -103,6 +103,7 @@
  * @author Tamil Selvan
  * @author ihaohong
  * @author John Blum
+ * @author Jaeik Jeong
  */
 @NullUnmarked
 public class LettuceConnection extends AbstractRedisConnection {
@@ -651,6 +652,8 @@ public void openPipeline() {
 			}
 
 			throw new RedisPipelineException(new QueryTimeoutException("Redis command timed out"));
+		} catch (RedisPipelineException ex) {
+			throw ex;
 		} catch (Exception ex) {
 			throw new RedisPipelineException(ex);
 		}

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTests.java

@@ -39,6 +39,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
@@ -48,8 +49,11 @@
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
+import org.springframework.dao.QueryTimeoutException;
 import org.springframework.data.redis.connection.AbstractConnectionUnitTestBase;
+import org.springframework.data.redis.connection.RedisPipelineException;
 import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption;
 import org.springframework.data.redis.connection.RedisStreamCommands.TrimOptions;
 import org.springframework.data.redis.connection.RedisStreamCommands.XAddOptions;
@@ -428,6 +432,47 @@ protected void setOutput(ByteBuffer bytes) {
 
 	}
 
+	@Nested
+	class ClosePipelineUnitTests {
+
+		@Test // GH-3346
+		void closePipelineShouldNotDoubleWrapTimeoutException() {
+
+			connection.openPipeline();
+			connection.set("foo".getBytes(), "bar".getBytes());
+
+			try (MockedStatic<LettuceFutures> lf = Mockito.mockStatic(LettuceFutures.class)) {
+				lf.when(() -> LettuceFutures.awaitAll(anyLong(), any(TimeUnit.class), any())).thenReturn(false);
+
+				assertThatThrownBy(() -> connection.closePipeline())
+						.isInstanceOf(RedisPipelineException.class)
+						.hasCauseInstanceOf(QueryTimeoutException.class);
+			}
+		}
+
+		@SuppressWarnings({ "rawtypes", "unchecked" })
+		@Test // GH-3346
+		void closePipelineShouldNotDoubleWrapCommandException() {
+
+			Command<?, ?, ?> cmd = new Command<>(CommandType.SET, new StatusOutput<>(ByteArrayCodec.INSTANCE));
+			AsyncCommand<?, ?, ?> future = new AsyncCommand<>(cmd);
+			future.completeExceptionally(new RuntimeException("ERR some error"));
+
+			when(asyncCommandsMock.set(any(byte[].class), any(byte[].class))).thenReturn((RedisFuture) future);
+
+			connection.openPipeline();
+			connection.set("foo".getBytes(), "bar".getBytes());
+
+			try (MockedStatic<LettuceFutures> lf = Mockito.mockStatic(LettuceFutures.class)) {
+				lf.when(() -> LettuceFutures.awaitAll(anyLong(), any(TimeUnit.class), any())).thenReturn(true);
+
+				assertThatThrownBy(() -> connection.closePipeline())
+						.isInstanceOf(RedisPipelineException.class)
+						.hasCauseInstanceOf(InvalidDataAccessApiUsageException.class);
+			}
+		}
+	}
+
 	@Nested
 	class LettucePipelineConnectionUnitTests extends BasicUnitTests {