Polishing.

Consistently use InvalidDataAccessApiUsageException to signal that an operation is not allowed/supported in the current mode of operation.
Replace converter objects with conversion within the actual method to remove indirections.

Original Pull Request: #2287

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java

@@ -653,17 +653,17 @@ public class JedisClusterConnection implements RedisClusterConnection {
 
 	@Override
 	public void openPipeline() {
-		throw new UnsupportedOperationException("Pipeline is currently not supported for JedisClusterConnection.");
+		throw new InvalidDataAccessApiUsageException("Pipeline is not supported for JedisClusterConnection.");
 	}
 
 	@Override
 	public List<Object> closePipeline() throws RedisPipelineException {
-		throw new UnsupportedOperationException("Pipeline is currently not supported for JedisClusterConnection.");
+		throw new InvalidDataAccessApiUsageException("Pipeline is not supported for JedisClusterConnection.");
 	}
 
 	@Override
 	public RedisSentinelConnection getSentinelConnection() {
-		throw new UnsupportedOperationException("Sentinel is currently not supported for JedisClusterConnection.");
+		throw new InvalidDataAccessApiUsageException("Sentinel is not supported for JedisClusterConnection.");
 	}
 
 	@Override

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterKeyCommands.java

@@ -335,7 +335,7 @@ class JedisClusterKeyCommands implements RedisKeyCommands {
 
 	@Override
 	public Boolean move(byte[] key, int dbIndex) {
-		throw new UnsupportedOperationException("Cluster mode does not allow moving keys.");
+		throw new InvalidDataAccessApiUsageException("Cluster mode does not allow moving keys.");
 	}
 
 	@Override

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -553,7 +553,7 @@ public class JedisConnection extends AbstractRedisConnection {
 	@Override
 	public void watch(byte[]... keys) {
 		if (isQueueing()) {
-			throw new UnsupportedOperationException();
+			throw new InvalidDataAccessApiUsageException("WATCH is not supported when a transaction is active");
 		}
 		doWithJedis(it -> {
 
@@ -591,7 +591,7 @@ public class JedisConnection extends AbstractRedisConnection {
 		}
 
 		if (isQueueing() || isPipelined()) {
-			throw new UnsupportedOperationException();
+			throw new InvalidDataAccessApiUsageException("Cannot subscribe in pipeline / transaction mode");
 		}
 
 		doWithJedis(it -> {
@@ -612,7 +612,7 @@ public class JedisConnection extends AbstractRedisConnection {
 		}
 
 		if (isQueueing() || isPipelined()) {
-			throw new UnsupportedOperationException();
+			throw new InvalidDataAccessApiUsageException("Cannot subscribe in pipeline / transaction mode");
 		}
 
 		doWithJedis(it -> {

src/main/java/org/springframework/data/redis/connection/jedis/JedisInvoker.java

@@ -34,6 +34,7 @@ import java.util.function.Supplier;
 
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.core.convert.converter.Converter;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -82,7 +83,7 @@ class JedisInvoker {
 		Assert.notNull(function, "ConnectionFunction must not be null!");
 
 		return synchronizer.invoke(function::apply, it -> {
-			throw new UnsupportedOperationException("Operation not supported by Jedis in pipelining/transaction mode");
+			throw new InvalidDataAccessApiUsageException("Operation not supported by Jedis in pipelining/transaction mode");
 		}, Converters.identityConverter(), () -> null);
 	}
 
@@ -231,7 +232,7 @@ class JedisInvoker {
 		Assert.notNull(function, "ConnectionFunction must not be null!");
 
 		return from(function, connection -> {
-			throw new UnsupportedOperationException("Operation not supported in pipelining/transaction mode");
+			throw new InvalidDataAccessApiUsageException("Operation not supported in pipelining/transaction mode");
 		});
 	}
 
@@ -379,7 +380,7 @@ class JedisInvoker {
 		Assert.notNull(function, "ConnectionFunction must not be null!");
 
 		return fromMany(function, connection -> {
-			throw new UnsupportedOperationException("Operation not supported in pipelining/transaction mode");
+			throw new InvalidDataAccessApiUsageException("Operation not supported in pipelining/transaction mode");
 		});
 	}
 

src/main/java/org/springframework/data/redis/connection/jedis/JedisScriptingCommands.java

@@ -19,6 +19,7 @@ import redis.clients.jedis.Jedis;
 
 import java.util.List;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.RedisScriptingCommands;
 import org.springframework.data.redis.connection.ReturnType;
 import org.springframework.util.Assert;
@@ -101,7 +102,7 @@ class JedisScriptingCommands implements RedisScriptingCommands {
 
 	private void assertDirectMode() {
 		if (connection.isQueueing() || connection.isPipelined()) {
-			throw new UnsupportedOperationException("Scripting commands not supported in pipelining/transaction mode");
+			throw new InvalidDataAccessApiUsageException("Scripting commands not supported in pipelining/transaction mode");
 		}
 	}
 

src/main/java/org/springframework/data/redis/connection/jedis/JedisStringCommands.java

@@ -283,7 +283,7 @@ class JedisStringCommands implements RedisStringCommands {
 		Assert.notNull(destination, "Destination key must not be null!");
 
 		if (op == BitOperation.NOT && keys.length > 1) {
-			throw new UnsupportedOperationException("Bitop NOT should only be performed against one key");
+			throw new IllegalArgumentException("Bitop NOT should only be performed against one key");
 		}
 
 		return connection.invoke().just(Jedis::bitop, PipelineBinaryCommands::bitop, JedisConverters.toBitOp(op),

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

@@ -26,6 +26,7 @@ import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -110,7 +111,7 @@ class ClusterConnectionProvider implements LettuceConnectionProvider, RedisClien
 		}
 
 		return LettuceFutureUtils
-				.failed(new UnsupportedOperationException("Connection type " + connectionType + " not supported!"));
+				.failed(new InvalidDataAccessApiUsageException("Connection type " + connectionType + " not supported!"));
 	}
 
 	@Override

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

@@ -24,6 +24,7 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ThreadLocalRandom;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.ClusterSlotHashUtil;
 import org.springframework.data.redis.connection.RedisClusterNode;
 import org.springframework.data.redis.connection.SortParameters;
@@ -134,7 +135,7 @@ class LettuceClusterKeyCommands extends LettuceKeyCommands {
 
 	@Override
 	public Boolean move(byte[] key, int dbIndex) {
-		throw new UnsupportedOperationException("MOVE not supported in CLUSTER mode!");
+		throw new InvalidDataAccessApiUsageException("MOVE not supported in CLUSTER mode!");
 	}
 
 	@Nullable

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

@@ -547,7 +547,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 	public void select(int dbIndex) {
 
 		if (asyncSharedConn != null) {
-			throw new UnsupportedOperationException("Selecting a new database not supported due to shared connection. "
+			throw new InvalidDataAccessApiUsageException("Selecting a new database not supported due to shared connection. "
 					+ "Use separate ConnectionFactorys to work with multiple databases");
 		}
 
@@ -578,7 +578,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 	@Override
 	public void watch(byte[]... keys) {
 		if (isQueueing()) {
-			throw new UnsupportedOperationException();
+			throw new InvalidDataAccessApiUsageException("WATCH is not supported when a transaction is active");
 		}
 		try {
 			if (isPipelined()) {
@@ -620,7 +620,8 @@ public class LettuceConnection extends AbstractRedisConnection {
 		checkSubscription();
 
 		if (isQueueing() || isPipelined()) {
-			throw new UnsupportedOperationException("Transaction/Pipelining is not supported for Pub/Sub subscriptions!");
+			throw new InvalidDataAccessApiUsageException(
+					"Transaction/Pipelining is not supported for Pub/Sub subscriptions!");
 		}
 
 		try {
@@ -637,7 +638,8 @@ public class LettuceConnection extends AbstractRedisConnection {
 		checkSubscription();
 
 		if (isQueueing() || isPipelined()) {
-			throw new UnsupportedOperationException("Transaction/Pipelining is not supported for Pub/Sub subscriptions!");
+			throw new InvalidDataAccessApiUsageException(
+					"Transaction/Pipelining is not supported for Pub/Sub subscriptions!");
 		}
 
 		try {

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

@@ -25,6 +25,7 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.RedisHashCommands;
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.core.Cursor;
@@ -223,7 +224,7 @@ class LettuceHashCommands implements RedisHashCommands {
 			protected ScanIteration<Entry<byte[], byte[]>> doScan(byte[] key, long cursorId, ScanOptions options) {
 
 				if (connection.isQueueing() || connection.isPipelined()) {
-					throw new UnsupportedOperationException("'HSCAN' cannot be called in pipeline / transaction mode.");
+					throw new InvalidDataAccessApiUsageException("'HSCAN' cannot be called in pipeline / transaction mode.");
 				}
 
 				io.lettuce.core.ScanCursor scanCursor = connection.getScanCursor(cursorId);

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

@@ -28,6 +28,7 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.RedisKeyCommands;
 import org.springframework.data.redis.connection.SortParameters;
@@ -149,7 +150,7 @@ class LettuceKeyCommands implements RedisKeyCommands {
 			protected LettuceScanIteration<byte[]> doScan(ScanCursor cursor, ScanOptions options) {
 
 				if (connection.isQueueing() || connection.isPipelined()) {
-					throw new UnsupportedOperationException("'SCAN' cannot be called in pipeline / transaction mode.");
+					throw new InvalidDataAccessApiUsageException("'SCAN' cannot be called in pipeline / transaction mode.");
 				}
 
 				ScanArgs scanArgs = LettuceConverters.toScanArgs(options);

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

@@ -24,6 +24,8 @@ import java.nio.ByteBuffer;
 import java.util.List;
 
 import org.reactivestreams.Publisher;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.ClusterSlotHashUtil;
 import org.springframework.data.redis.connection.ReactiveClusterKeyCommands;
@@ -122,6 +124,6 @@ class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommands imple
 
 	@Override
 	public Flux<BooleanResponse<MoveCommand>> move(Publisher<MoveCommand> commands) {
-		throw new UnsupportedOperationException("MOVE not supported in CLUSTER mode!");
+		throw new InvalidDataAccessApiUsageException("MOVE not supported in CLUSTER mode!");
 	}
 }

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

@@ -21,6 +21,7 @@ import java.util.Arrays;
 import java.util.List;
 
 import org.springframework.core.convert.converter.Converter;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.RedisScriptingCommands;
 import org.springframework.data.redis.connection.ReturnType;
 import org.springframework.util.Assert;
@@ -46,7 +47,7 @@ class LettuceScriptingCommands implements RedisScriptingCommands {
 	public void scriptKill() {
 
 		if (connection.isQueueing()) {
-			throw new UnsupportedOperationException("Script kill not permitted in a transaction");
+			throw new InvalidDataAccessApiUsageException("Script kill not permitted in a transaction");
 		}
 
 		connection.invoke().just(RedisScriptingAsyncCommands::scriptKill);

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

@@ -26,6 +26,7 @@ import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.RedisNode;
 import org.springframework.data.redis.connection.RedisServerCommands;
 import org.springframework.data.redis.core.types.RedisClientInfo;
@@ -192,11 +193,6 @@ class LettuceServerCommands implements RedisServerCommands {
 
 	@Override
 	public List<RedisClientInfo> getClientList() {
-
-		if (connection.isPipelined()) {
-			throw new UnsupportedOperationException("Cannot be called in pipeline mode.");
-		}
-
 		return connection.invoke().from(RedisServerAsyncCommands::clientList)
 				.get(LettuceConverters.stringToRedisClientListConverter());
 	}

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

@@ -24,6 +24,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.RedisSetCommands;
 import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.KeyBoundCursor;
@@ -220,7 +221,7 @@ class LettuceSetCommands implements RedisSetCommands {
 			protected ScanIteration<byte[]> doScan(byte[] key, long cursorId, ScanOptions options) {
 
 				if (connection.isQueueing() || connection.isPipelined()) {
-					throw new UnsupportedOperationException("'SSCAN' cannot be called in pipeline / transaction mode.");
+					throw new InvalidDataAccessApiUsageException("'SSCAN' cannot be called in pipeline / transaction mode.");
 				}
 
 				io.lettuce.core.ScanCursor scanCursor = connection.getScanCursor(cursorId);

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

@@ -21,6 +21,7 @@ import io.lettuce.core.api.async.RedisStringAsyncCommands;
 import java.util.List;
 import java.util.Map;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.BitFieldSubCommands;
 import org.springframework.data.redis.connection.RedisStringCommands;
@@ -275,7 +276,7 @@ class LettuceStringCommands implements RedisStringCommands {
 		Assert.notNull(destination, "Destination key must not be null!");
 
 		if (op == BitOperation.NOT && keys.length > 1) {
-			throw new UnsupportedOperationException("Bitop NOT should only be performed against one key");
+			throw new IllegalArgumentException("Bitop NOT should only be performed against one key");
 		}
 
 		return connection.invoke().just(it -> {
@@ -289,7 +290,7 @@ class LettuceStringCommands implements RedisStringCommands {
 					return it.bitopXor(destination, keys);
 				case NOT:
 					if (keys.length != 1) {
-						throw new UnsupportedOperationException("Bitop NOT should only be performed against one key");
+						throw new IllegalArgumentException("Bitop NOT should only be performed against one key");
 					}
 					return it.bitopNot(destination, keys[0]);
 				default:

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

@@ -27,6 +27,7 @@ import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.RedisZSetCommands;
 import org.springframework.data.redis.connection.RedisZSetCommands.ZAddArgs.Flag;
 import org.springframework.data.redis.connection.convert.Converters;
@@ -546,7 +547,7 @@ class LettuceZSetCommands implements RedisZSetCommands {
 			protected ScanIteration<Tuple> doScan(byte[] key, long cursorId, ScanOptions options) {
 
 				if (connection.isQueueing() || connection.isPipelined()) {
-					throw new UnsupportedOperationException("'ZSCAN' cannot be called in pipeline / transaction mode.");
+					throw new InvalidDataAccessApiUsageException("'ZSCAN' cannot be called in pipeline / transaction mode.");
 				}
 
 				io.lettuce.core.ScanCursor scanCursor = connection.getScanCursor(cursorId);

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

@@ -29,6 +29,7 @@ import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.lang.Nullable;
 
 /**

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

@@ -25,11 +25,9 @@ import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.function.BiFunction;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions;
-import org.springframework.data.redis.connection.convert.ListConverter;
 import org.springframework.data.redis.connection.stream.ByteRecord;
 import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
@@ -53,39 +51,6 @@ import org.springframework.util.NumberUtils;
 @SuppressWarnings({ "rawtypes" })
 class StreamConverters {
 
-	private static final Converter<List<StreamMessage<byte[], byte[]>>, List<RecordId>> MESSAGEs_TO_IDs = new ListConverter<>(
-			messageToIdConverter());
-
-	private static final BiFunction<List<PendingMessage>, String, org.springframework.data.redis.connection.stream.PendingMessages> PENDING_MESSAGES_CONVERTER = (
-			source, groupName) -> {
-
-
-		List<org.springframework.data.redis.connection.stream.PendingMessage> messages = source.stream()
-				.map(it -> {
-
-					RecordId id = RecordId.of(it.getId());
-					Consumer consumer = Consumer.from(groupName, it.getConsumer());
-
-					return new org.springframework.data.redis.connection.stream.PendingMessage(id, consumer,
-							Duration.ofMillis(it.getMsSinceLastDelivery()), it.getRedeliveryCount());
-
-				}).toList();
-
-		return new org.springframework.data.redis.connection.stream.PendingMessages(groupName, messages);
-
-	};
-
-	private static final BiFunction<PendingMessages, String, PendingMessagesSummary> PENDING_MESSAGES_SUMMARY_CONVERTER = (
-			source, groupName) -> {
-
-		org.springframework.data.domain.Range<String> range = source.getMessageIds().isUnbounded()
-				? org.springframework.data.domain.Range.unbounded()
-				: org.springframework.data.domain.Range.open(source.getMessageIds().getLower().getValue(),
-						source.getMessageIds().getUpper().getValue());
-
-		return new PendingMessagesSummary(groupName, source.getCount(), range, source.getConsumerMessageCount());
-	};
-
 	/**
 	 * Convert {@link StreamReadOptions} to Lettuce's {@link XReadArgs}.
 	 *
@@ -126,7 +91,18 @@ class StreamConverters {
 	 */
 	static org.springframework.data.redis.connection.stream.PendingMessages toPendingMessages(String groupName,
 			org.springframework.data.domain.Range<?> range, List<PendingMessage> source) {
-		return PENDING_MESSAGES_CONVERTER.apply(source, groupName).withinRange(range);
+
+		List<org.springframework.data.redis.connection.stream.PendingMessage> messages = source.stream().map(it -> {
+
+			RecordId id = RecordId.of(it.getId());
+			Consumer consumer = Consumer.from(groupName, it.getConsumer());
+
+			return new org.springframework.data.redis.connection.stream.PendingMessage(id, consumer,
+					Duration.ofMillis(it.getMsSinceLastDelivery()), it.getRedeliveryCount());
+
+		}).toList();
+
+		return new org.springframework.data.redis.connection.stream.PendingMessages(groupName, messages).withinRange(range);
 	}
 
 	/**
@@ -138,7 +114,13 @@ class StreamConverters {
 	 * @since 2.3
 	 */
 	static PendingMessagesSummary toPendingMessagesInfo(String groupName, PendingMessages source) {
-		return PENDING_MESSAGES_SUMMARY_CONVERTER.apply(source, groupName);
+
+		org.springframework.data.domain.Range<String> range = source.getMessageIds().isUnbounded()
+				? org.springframework.data.domain.Range.unbounded()
+				: org.springframework.data.domain.Range.open(source.getMessageIds().getLower().getValue(),
+						source.getMessageIds().getUpper().getValue());
+
+		return new PendingMessagesSummary(groupName, source.getCount(), range, source.getConsumerMessageCount());
 	}
 
 	/**