Encapsulate Redis Scan CursorId.

We now retain the raw cursor value without attempting to convert it into a long as Redis uses 64 bit unsigned integers exceeding Long.MAX_VALUE.
Fix broken id parsing for unsigned long value
Update deprecation warnings and method visibility.

See: #2796
Closes: #2802

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

@@ -275,14 +275,14 @@ class JedisClusterHashCommands implements RedisHashCommands {
 		return new ScanCursor<Entry<byte[], byte[]>>(options) {
 
 			@Override
-			protected ScanIteration<Entry<byte[], byte[]>> doScan(long cursorId, ScanOptions options) {
+			protected ScanIteration<Entry<byte[], byte[]>> doScan(CursorId cursorId, ScanOptions options) {
 
 				ScanParams params = JedisConverters.toScanParams(options);
 
 				ScanResult<Entry<byte[], byte[]>> result = connection.getCluster().hscan(key,
-						JedisConverters.toBytes(Long.toUnsignedString(cursorId)),
+						JedisConverters.toBytes(cursorId),
 						params);
-				return new ScanIteration<>(Long.parseUnsignedLong(result.getCursor()), result.getResult());
+				return new ScanIteration<>(CursorId.of(result.getCursor()), result.getResult());
 			}
 		}.open();
 	}

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

@@ -177,11 +177,11 @@ class JedisClusterKeyCommands implements RedisKeyCommands {
 					return new ScanCursor<byte[]>(0, options) {
 
 						@Override
-						protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
+						protected ScanIteration<byte[]> doScan(CursorId cursorId, ScanOptions options) {
 
 							ScanParams params = JedisConverters.toScanParams(options);
-							ScanResult<String> result = client.scan(Long.toUnsignedString(cursorId), params);
-							return new ScanIteration<>(Long.parseUnsignedLong(result.getCursor()),
+							ScanResult<String> result = client.scan(cursorId.getCursorId(), params);
+							return new ScanIteration<>(CursorId.of(result.getCursor()),
 									JedisConverters.stringListToByteList().convert(result.getResult()));
 						}
 					}.open();

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

@@ -394,12 +394,11 @@ class JedisClusterSetCommands implements RedisSetCommands {
 		return new ScanCursor<byte[]>(options) {
 
 			@Override
-			protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
+			protected ScanIteration<byte[]> doScan(CursorId cursorId, ScanOptions options) {
 
 				ScanParams params = JedisConverters.toScanParams(options);
-				ScanResult<byte[]> result = connection.getCluster().sscan(key,
-						JedisConverters.toBytes(Long.toUnsignedString(cursorId)), params);
-				return new ScanIteration<>(Long.parseUnsignedLong(result.getCursor()), result.getResult());
+				ScanResult<byte[]> result = connection.getCluster().sscan(key, JedisConverters.toBytes(cursorId), params);
+				return new ScanIteration<>(CursorId.of(result.getCursor()), result.getResult());
 			}
 		}.open();
 	}

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

@@ -1079,13 +1079,13 @@ class JedisClusterZSetCommands implements RedisZSetCommands {
 		return new ScanCursor<Tuple>(options) {
 
 			@Override
-			protected ScanIteration<Tuple> doScan(long cursorId, ScanOptions options) {
+			protected ScanIteration<Tuple> doScan(CursorId cursorId, ScanOptions options) {
 
 				ScanParams params = JedisConverters.toScanParams(options);
 
 				ScanResult<redis.clients.jedis.resps.Tuple> result = connection.getCluster().zscan(key,
-						JedisConverters.toBytes(Long.toUnsignedString(cursorId)), params);
-				return new ScanIteration<>(Long.parseUnsignedLong(result.getCursor()),
+						JedisConverters.toBytes(cursorId), params);
+				return new ScanIteration<>(CursorId.of(result.getCursor()),
 						JedisConverters.tuplesToTuples().convert(result.getResult()));
 			}
 		}.open();

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

@@ -75,6 +75,7 @@ import org.springframework.data.redis.connection.convert.SetConverter;
 import org.springframework.data.redis.connection.convert.StringToRedisClientInfoConverter;
 import org.springframework.data.redis.connection.zset.DefaultTuple;
 import org.springframework.data.redis.connection.zset.Tuple;
+import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.core.types.Expiration;
 import org.springframework.data.redis.core.types.RedisClientInfo;
@@ -175,6 +176,17 @@ abstract class JedisConverters extends Converters {
 		return toBytes(String.valueOf(source));
 	}
 
+	/**
+	 * Convert the given {@link org.springframework.data.redis.core.Cursor.CursorId} into its binary representation.
+	 *
+	 * @param source must not be {@literal null}.
+	 * @return the binary representation.
+	 * @since 3.3
+	 */
+	static byte[] toBytes(Cursor.CursorId source) {
+		return toBytes(source.getCursorId());
+	}
+
 	@Nullable
 	public static byte[] toBytes(@Nullable String source) {
 		return source == null ? null : SafeEncoder.encode(source);

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

@@ -30,6 +30,7 @@ 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;
+import org.springframework.data.redis.core.Cursor.CursorId;
 import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
@@ -149,8 +150,7 @@ class JedisHashCommands implements RedisHashCommands {
 
 					List<Entry<byte[], byte[]>> convertedMapEntryList = new ArrayList<>(mapEntryList.size());
 
-					mapEntryList.forEach(entry ->
-							convertedMapEntryList.add(Converters.entryOf(entry.getKey(), entry.getValue())));
+					mapEntryList.forEach(entry -> convertedMapEntryList.add(Converters.entryOf(entry.getKey(), entry.getValue())));
 
 					return convertedMapEntryList;
 
@@ -219,24 +219,17 @@ class JedisHashCommands implements RedisHashCommands {
 
 	@Override
 	public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, ScanOptions options) {
-		return hScan(key, 0, options);
+		return hScan(key, CursorId.initial(), options);
 	}
 
-	/**
-	 * @since 1.4
-	 * @param key
-	 * @param cursorId
-	 * @param options
-	 * @return
-	 */
-	public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, long cursorId, ScanOptions options) {
+	public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 		Assert.notNull(key, "Key must not be null");
 
 		return new KeyBoundCursor<Entry<byte[], byte[]>>(key, cursorId, options) {
 
 			@Override
-			protected ScanIteration<Entry<byte[], byte[]>> doScan(byte[] key, long cursorId, ScanOptions options) {
+			protected ScanIteration<Entry<byte[], byte[]>> doScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 				if (isQueueing() || isPipelined()) {
 					throw new InvalidDataAccessApiUsageException("'HSCAN' cannot be called in pipeline / transaction mode");
@@ -245,9 +238,8 @@ class JedisHashCommands implements RedisHashCommands {
 				ScanParams params = JedisConverters.toScanParams(options);
 
 				ScanResult<Entry<byte[], byte[]>> result = connection.getJedis().hscan(key,
-						JedisConverters.toBytes(Long.toUnsignedString(cursorId)),
-						params);
-				return new ScanIteration<>(Long.parseUnsignedLong(result.getCursor()), result.getResult());
+						JedisConverters.toBytes(cursorId), params);
+				return new ScanIteration<>(CursorId.of(result.getCursor()), result.getResult());
 			}
 
 			@Override

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

@@ -36,6 +36,7 @@ import org.springframework.data.redis.connection.ValueEncoding;
 import org.springframework.data.redis.connection.ValueEncoding.RedisValueEncoding;
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.Cursor.CursorId;
 import org.springframework.data.redis.core.KeyScanOptions;
 import org.springframework.data.redis.core.ScanCursor;
 import org.springframework.data.redis.core.ScanIteration;
@@ -131,7 +132,7 @@ class JedisKeyCommands implements RedisKeyCommands {
 
 	@Override
 	public Cursor<byte[]> scan(ScanOptions options) {
-		return scan(0, options != null ? options : ScanOptions.NONE);
+		return scan(CursorId.initial(), options != null ? options : ScanOptions.NONE);
 	}
 
 	/**
@@ -140,12 +141,12 @@ class JedisKeyCommands implements RedisKeyCommands {
 	 * @param options
 	 * @return
 	 */
-	public Cursor<byte[]> scan(long cursorId, ScanOptions options) {
+	public Cursor<byte[]> scan(CursorId cursorId, ScanOptions options) {
 
 		return new ScanCursor<byte[]>(cursorId, options) {
 
 			@Override
-			protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
+			protected ScanIteration<byte[]> doScan(CursorId cursorId, ScanOptions options) {
 
 				if (isQueueing() || isPipelined()) {
 					throw new InvalidDataAccessApiUsageException("'SCAN' cannot be called in pipeline / transaction mode");
@@ -165,12 +166,12 @@ class JedisKeyCommands implements RedisKeyCommands {
 				}
 
 				if (type != null) {
-					result = connection.getJedis().scan(JedisConverters.toBytes(Long.toUnsignedString(cursorId)), params, type);
+					result = connection.getJedis().scan(JedisConverters.toBytes(cursorId), params, type);
 				} else {
-					result = connection.getJedis().scan(JedisConverters.toBytes(Long.toUnsignedString(cursorId)), params);
+					result = connection.getJedis().scan(JedisConverters.toBytes(cursorId), params);
 				}
 
-				return new ScanIteration<>(Long.parseUnsignedLong(result.getCursor()), result.getResult());
+				return new ScanIteration<>(CursorId.of(result.getCursor()), result.getResult());
 			}
 
 			protected void doClose() {

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

@@ -27,6 +27,7 @@ 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.Cursor.CursorId;
 import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
@@ -206,24 +207,24 @@ class JedisSetCommands implements RedisSetCommands {
 
 	@Override
 	public Cursor<byte[]> sScan(byte[] key, ScanOptions options) {
-		return sScan(key, 0, options);
+		return sScan(key, CursorId.initial(), options);
 	}
 
 	/**
-	 * @since 1.4
 	 * @param key
 	 * @param cursorId
 	 * @param options
 	 * @return
+	 * @since 3.2.1
 	 */
-	public Cursor<byte[]> sScan(byte[] key, long cursorId, ScanOptions options) {
+	public Cursor<byte[]> sScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 		Assert.notNull(key, "Key must not be null");
 
 		return new KeyBoundCursor<byte[]>(key, cursorId, options) {
 
 			@Override
-			protected ScanIteration<byte[]> doScan(byte[] key, long cursorId, ScanOptions options) {
+			protected ScanIteration<byte[]> doScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 				if (isQueueing() || isPipelined()) {
 					throw new InvalidDataAccessApiUsageException("'SSCAN' cannot be called in pipeline / transaction mode");
@@ -231,9 +232,8 @@ class JedisSetCommands implements RedisSetCommands {
 
 				ScanParams params = JedisConverters.toScanParams(options);
 
-				ScanResult<byte[]> result = connection.getJedis().sscan(key,
-						JedisConverters.toBytes(Long.toUnsignedString(cursorId)), params);
-				return new ScanIteration<>(Long.parseUnsignedLong(result.getCursor()), result.getResult());
+				ScanResult<byte[]> result = connection.getJedis().sscan(key, JedisConverters.toBytes(cursorId), params);
+				return new ScanIteration<>(CursorId.of(result.getCursor()), result.getResult());
 			}
 
 			protected void doClose() {

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

@@ -35,6 +35,7 @@ import org.springframework.data.redis.connection.zset.Aggregate;
 import org.springframework.data.redis.connection.zset.Tuple;
 import org.springframework.data.redis.connection.zset.Weights;
 import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.Cursor.CursorId;
 import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
@@ -561,24 +562,25 @@ class JedisZSetCommands implements RedisZSetCommands {
 
 	@Override
 	public Cursor<Tuple> zScan(byte[] key, ScanOptions options) {
-		return zScan(key, 0L, options);
+		return zScan(key, CursorId.initial(), options);
 	}
 
+
 	/**
-	 * @since 1.4
 	 * @param key
 	 * @param cursorId
 	 * @param options
 	 * @return
+	 * @since 3.2.1
 	 */
-	public Cursor<Tuple> zScan(byte[] key, Long cursorId, ScanOptions options) {
+	public Cursor<Tuple> zScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 		Assert.notNull(key, "Key must not be null");
 
 		return new KeyBoundCursor<Tuple>(key, cursorId, options) {
 
 			@Override
-			protected ScanIteration<Tuple> doScan(byte[] key, long cursorId, ScanOptions options) {
+			protected ScanIteration<Tuple> doScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 				if (isQueueing() || isPipelined()) {
 					throw new InvalidDataAccessApiUsageException("'ZSCAN' cannot be called in pipeline / transaction mode");
@@ -587,8 +589,8 @@ class JedisZSetCommands implements RedisZSetCommands {
 				ScanParams params = JedisConverters.toScanParams(options);
 
 				ScanResult<redis.clients.jedis.resps.Tuple> result = connection.getJedis().zscan(key,
-						JedisConverters.toBytes(Long.toUnsignedString(cursorId)), params);
-				return new ScanIteration<>(Long.parseUnsignedLong(result.getCursor()),
+						JedisConverters.toBytes(cursorId), params);
+				return new ScanIteration<>(CursorId.of(result.getCursor()),
 						JedisConverters.tuplesToTuples().convert(result.getResult()));
 			}
 

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

@@ -57,6 +57,7 @@ import java.util.function.Supplier;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.BeanUtils;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
@@ -70,6 +71,7 @@ import org.springframework.data.redis.connection.convert.TransactionResultConver
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware;
 import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceResultBuilder;
 import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceStatusResult;
+import org.springframework.data.redis.core.Cursor.CursorId;
 import org.springframework.data.redis.core.RedisCommand;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -1060,8 +1062,8 @@ public class LettuceConnection extends AbstractRedisConnection {
 		}
 	}
 
-	io.lettuce.core.ScanCursor getScanCursor(long cursorId) {
-		return io.lettuce.core.ScanCursor.of(Long.toUnsignedString(cursorId));
+	io.lettuce.core.ScanCursor getScanCursor(CursorId cursorId) {
+		return io.lettuce.core.ScanCursor.of(cursorId.getCursorId());
 	}
 
 	private void validateCommandIfRunningInTransactionMode(ProtocolKeyword cmd, byte[]... args) {

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

@@ -29,6 +29,7 @@ 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;
+import org.springframework.data.redis.core.Cursor.CursorId;
 import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
@@ -204,24 +205,25 @@ class LettuceHashCommands implements RedisHashCommands {
 
 	@Override
 	public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, ScanOptions options) {
-		return hScan(key, 0, options);
+		return hScan(key, CursorId.initial(), options);
 	}
 
+
 	/**
-	 * @since 1.4
 	 * @param key
 	 * @param cursorId
 	 * @param options
 	 * @return
+	 * @since 1.4
 	 */
-	public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, long cursorId, ScanOptions options) {
+	public Cursor<Entry<byte[], byte[]>> hScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 		Assert.notNull(key, "Key must not be null");
 
 		return new KeyBoundCursor<Entry<byte[], byte[]>>(key, cursorId, options) {
 
 			@Override
-			protected ScanIteration<Entry<byte[], byte[]>> doScan(byte[] key, long cursorId, ScanOptions options) {
+			protected ScanIteration<Entry<byte[], byte[]>> doScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 				if (connection.isQueueing() || connection.isPipelined()) {
 					throw new InvalidDataAccessApiUsageException("'HSCAN' cannot be called in pipeline / transaction mode");
@@ -235,7 +237,7 @@ class LettuceHashCommands implements RedisHashCommands {
 				String nextCursorId = mapScanCursor.getCursor();
 
 				Map<byte[], byte[]> values = mapScanCursor.getMap();
-				return new ScanIteration<>(Long.parseUnsignedLong(nextCursorId), values.entrySet());
+				return new ScanIteration<>(CursorId.of(nextCursorId), values.entrySet());
 			}
 
 			@Override

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

@@ -46,9 +46,9 @@ abstract class LettuceScanCursor<T> extends ScanCursor<T> {
 	}
 
 	@Override
-	protected ScanIteration<T> doScan(long cursorId, ScanOptions options) {
+	protected ScanIteration<T> doScan(CursorId cursorId, ScanOptions options) {
 
-		if (state == null && cursorId == 0) {
+		if (state == null && cursorId.isInitial()) {
 			return scanAndProcessState(io.lettuce.core.ScanCursor.INITIAL, options);
 		}
 
@@ -64,7 +64,7 @@ abstract class LettuceScanCursor<T> extends ScanCursor<T> {
 	}
 
 	@Override
-	protected boolean isFinished(long cursorId) {
+	protected boolean isFinished(CursorId cursorId) {
 		return state != null && isMatchingCursor(cursorId) ? state.isFinished() : super.isFinished(cursorId);
 	}
 
@@ -76,8 +76,8 @@ abstract class LettuceScanCursor<T> extends ScanCursor<T> {
 		return iteration;
 	}
 
-	private boolean isMatchingCursor(long cursorId) {
-		return state != null && state.getCursor().equals(Long.toUnsignedString(cursorId));
+	private boolean isMatchingCursor(CursorId cursorId) {
+		return state != null && state.getCursor().equals(cursorId.getCursorId());
 	}
 
 	/**
@@ -101,7 +101,7 @@ abstract class LettuceScanCursor<T> extends ScanCursor<T> {
 
 		LettuceScanIteration(io.lettuce.core.ScanCursor cursor, Collection<T> items) {
 
-			super(Long.parseUnsignedLong(cursor.getCursor()), items);
+			super(CursorId.of(cursor.getCursor()), items);
 			this.cursor = cursor;
 		}
 	}

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

@@ -27,6 +27,7 @@ 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.Cursor.CursorId;
 import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
@@ -201,24 +202,24 @@ class LettuceSetCommands implements RedisSetCommands {
 
 	@Override
 	public Cursor<byte[]> sScan(byte[] key, ScanOptions options) {
-		return sScan(key, 0, options);
+		return sScan(key, CursorId.initial(), options);
 	}
 
 	/**
-	 * @since 1.4
 	 * @param key
 	 * @param cursorId
 	 * @param options
 	 * @return
+	 * @since 1.4
 	 */
-	public Cursor<byte[]> sScan(byte[] key, long cursorId, ScanOptions options) {
+	public Cursor<byte[]> sScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 		Assert.notNull(key, "Key must not be null");
 
 		return new KeyBoundCursor<byte[]>(key, cursorId, options) {
 
 			@Override
-			protected ScanIteration<byte[]> doScan(byte[] key, long cursorId, ScanOptions options) {
+			protected ScanIteration<byte[]> doScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 				if (connection.isQueueing() || connection.isPipelined()) {
 					throw new InvalidDataAccessApiUsageException("'SSCAN' cannot be called in pipeline / transaction mode");
@@ -232,7 +233,7 @@ class LettuceSetCommands implements RedisSetCommands {
 				String nextCursorId = valueScanCursor.getCursor();
 
 				List<byte[]> values = connection.failsafeReadScanValues(valueScanCursor.getValues(), null);
-				return new ScanIteration<>(Long.parseUnsignedLong(nextCursorId), values);
+				return new ScanIteration<>(CursorId.of(nextCursorId), values);
 			}
 
 			protected void doClose() {

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

@@ -35,6 +35,7 @@ import org.springframework.data.redis.connection.zset.Aggregate;
 import org.springframework.data.redis.connection.zset.Tuple;
 import org.springframework.data.redis.connection.zset.Weights;
 import org.springframework.data.redis.core.Cursor;
+import org.springframework.data.redis.core.Cursor.CursorId;
 import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
@@ -530,20 +531,20 @@ class LettuceZSetCommands implements RedisZSetCommands {
 
 	@Override
 	public Cursor<Tuple> zScan(byte[] key, ScanOptions options) {
-		return zScan(key, 0L, options);
+		return zScan(key, CursorId.initial(), options);
 	}
 
 	/**
 	 * @since 1.4
 	 */
-	public Cursor<Tuple> zScan(byte[] key, long cursorId, ScanOptions options) {
+	public Cursor<Tuple> zScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 		Assert.notNull(key, "Key must not be null");
 
 		return new KeyBoundCursor<Tuple>(key, cursorId, options) {
 
 			@Override
-			protected ScanIteration<Tuple> doScan(byte[] key, long cursorId, ScanOptions options) {
+			protected ScanIteration<Tuple> doScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
 				if (connection.isQueueing() || connection.isPipelined()) {
 					throw new InvalidDataAccessApiUsageException("'ZSCAN' cannot be called in pipeline / transaction mode");
@@ -559,7 +560,7 @@ class LettuceZSetCommands implements RedisZSetCommands {
 				List<ScoredValue<byte[]>> result = scoredValueScanCursor.getValues();
 
 				List<Tuple> values = connection.failsafeReadScanValues(result, LettuceConverters.scoredValuesToTupleList());
-				return new ScanIteration<>(Long.parseUnsignedLong(nextCursorId), values);
+				return new ScanIteration<>(CursorId.of(nextCursorId), values);
 			}
 
 			@Override

src/main/java/org/springframework/data/redis/core/ConvertingCursor.java

@@ -69,6 +69,12 @@ public class ConvertingCursor<S, T> implements Cursor<T> {
 	}
 
 	@Override
+	public CursorId getId() {
+		return delegate.getId();
+	}
+
+	@Override
+	@Deprecated
 	public long getCursorId() {
 		return delegate.getCursorId();
 	}

src/main/java/org/springframework/data/redis/core/Cursor.java

@@ -16,6 +16,7 @@
 package org.springframework.data.redis.core;
 
 import org.springframework.data.util.CloseableIterator;
+import org.springframework.util.Assert;
 
 /**
  * Cursor abstraction to scan over the keyspace or elements within a data structure using a variant of a {@code SCAN}
@@ -35,12 +36,22 @@ import org.springframework.data.util.CloseableIterator;
  */
 public interface Cursor<T> extends CloseableIterator<T> {
 
+	/**
+	 * Returns the reference cursor.
+	 *
+	 * @return the reference cursor.
+	 * @since 3.2.1
+	 */
+	CursorId getId();
+
 	/**
 	 * Get the reference cursor. <br>
 	 * <strong>NOTE:</strong> the id might change while iterating items.
 	 *
 	 * @return
+	 * @deprecated since 3.3.0, use {@link #getId()} instead as the cursorId can exceed {@link Long#MAX_VALUE}.
 	 */
+	@Deprecated(since = "3.3.0")
 	long getCursorId();
 
 	/**
@@ -52,4 +63,112 @@ public interface Cursor<T> extends CloseableIterator<T> {
 	 * @return the current position of the cursor.
 	 */
 	long getPosition();
+
+	/**
+	 * Value class representing a cursor identifier.
+	 *
+	 * @since 3.2.1
+	 */
+	abstract class CursorId {
+
+		private final static CursorId INITIAL = new CursorId() {
+			@Override
+			public String getCursorId() {
+				return "0";
+			}
+		};
+
+		/**
+		 * Creates a new initial {@link CursorId}.
+		 *
+		 * @return an initial {@link CursorId}.
+		 */
+		public static CursorId initial() {
+			return INITIAL;
+		}
+
+		/**
+		 * Creates a {@link CursorId} from the given {@code cursorId}.
+		 *
+		 * @param cursorId the provided cursor identifier.
+		 * @return the provided cursor Id.
+		 */
+		public static CursorId of(String cursorId) {
+
+			Assert.notNull(cursorId, "CursorId must not be null");
+
+			if (INITIAL.getCursorId().equals(cursorId)) {
+				return INITIAL;
+			}
+
+			return new CursorId() {
+
+				@Override
+				public String getCursorId() {
+					return cursorId;
+				}
+			};
+		}
+
+		/**
+		 * Creates a {@link CursorId} from the given {@code cursorId}.
+		 *
+		 * @param cursorId the provided cursor identifier.
+		 * @return the provided cursor Id.
+		 */
+		public static CursorId of(long cursorId) {
+
+			if (cursorId == 0) {
+				return INITIAL;
+			}
+			return of(Long.toUnsignedString(cursorId));
+		}
+
+		/**
+		 * Returns whether the given {@code cursorId} represent an initial cursor identifier to indicate an initial/finished
+		 * cursor state.
+		 *
+		 * @param cursorId the cursor identifier to inspect.
+		 * @return {@code true} if the cursorId represents an initial/finished state.
+		 */
+		public static boolean isInitial(String cursorId) {
+			return INITIAL.getCursorId().equals(cursorId);
+		}
+
+		/**
+		 * Returns whether the current cursor identifier represent an initial cursor identifier to indicate an
+		 * initial/finished cursor state.
+		 *
+		 * @return {@code true} if the cursorId represents an initial/finished state.
+		 */
+		public boolean isInitial() {
+			return INITIAL.getCursorId().equals(getCursorId());
+		}
+
+		/**
+		 * @return the raw cursor Id.
+		 */
+		public abstract String getCursorId();
+
+		@Override
+		public int hashCode() {
+			return getCursorId().hashCode();
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+
+			if (obj instanceof CursorId other) {
+				return getCursorId().equals(other.getCursorId());
+			}
+
+			return false;
+		}
+
+		@Override
+		public String toString() {
+			return getCursorId();
+		}
+
+	}
 }

src/main/java/org/springframework/data/redis/core/KeyBoundCursor.java

@@ -31,17 +31,36 @@ public abstract class KeyBoundCursor<T> extends ScanCursor<T> {
 	 *
 	 * @param cursorId
 	 * @param options Defaulted to {@link ScanOptions#NONE} if nulled.
+	 * @deprecated since 3.3.0 - Use {@link KeyBoundCursor#KeyBoundCursor(byte[], CursorId, ScanOptions)} instead.
 	 */
+	@Deprecated(since = "3.3.0")
 	public KeyBoundCursor(byte[] key, long cursorId, @Nullable ScanOptions options) {
 		super(cursorId, options != null ? options : ScanOptions.NONE);
 		this.key = key;
 	}
 
+	/**
+	 * Crates new {@link ScanCursor}
+	 *
+	 * @param cursorId
+	 * @param options Defaulted to {@link ScanOptions#NONE} if nulled.
+	 * @since 3.3.0
+	 */
+	public KeyBoundCursor(byte[] key, CursorId cursorId, @Nullable ScanOptions options) {
+		super(cursorId, options != null ? options : ScanOptions.NONE);
+		this.key = key;
+	}
+
+	@Override
 	protected ScanIteration<T> doScan(long cursorId, ScanOptions options) {
+		return doScan(CursorId.of(cursorId), options);
+	}
+
+	protected ScanIteration<T> doScan(CursorId cursorId, ScanOptions options) {
 		return doScan(this.key, cursorId, options);
 	}
 
-	protected abstract ScanIteration<T> doScan(byte[] key, long cursorId, ScanOptions options);
+	protected abstract ScanIteration<T> doScan(byte[] key, CursorId cursorId, ScanOptions options);
 
 	public byte[] getKey() {
 		return key;

src/main/java/org/springframework/data/redis/core/ScanCursor.java

@@ -40,51 +40,76 @@ import org.springframework.util.CollectionUtils;
 public abstract class ScanCursor<T> implements Cursor<T> {
 
 	private CursorState state;
-	private long cursorId;
+	private CursorId id;
 	private Iterator<T> delegate;
 	private final ScanOptions scanOptions;
 	private long position;
 
 	/**
-	 * Crates new {@link ScanCursor} with {@code id=0} and {@link ScanOptions#NONE}
+	 * Crates new {@link ScanCursor} with an initial cursor and {@link ScanOptions#NONE}
 	 */
 	public ScanCursor() {
 		this(ScanOptions.NONE);
 	}
 
 	/**
-	 * Crates new {@link ScanCursor} with {@code id=0}.
+	 * Crates new {@link ScanCursor} with an initial cursor.
 	 *
 	 * @param options the scan options to apply.
 	 */
 	public ScanCursor(ScanOptions options) {
-		this(0, options);
+		this(CursorId.initial(), options);
 	}
 
 	/**
 	 * Crates new {@link ScanCursor} with {@link ScanOptions#NONE}
 	 *
 	 * @param cursorId the cursor Id.
+	 * @deprecated since 3.3.0 - Use {@link ScanCursor#ScanCursor(CursorId)} instead.
 	 */
+	@Deprecated(since = "3.3.0")
 	public ScanCursor(long cursorId) {
 		this(cursorId, ScanOptions.NONE);
 	}
 
+	/**
+	 * Crates new {@link ScanCursor} with {@link ScanOptions#NONE}
+	 *
+	 * @param cursorId the cursor Id.
+	 * @since 3.3.0
+	 */
+	public ScanCursor(CursorId cursorId) {
+		this(cursorId, ScanOptions.NONE);
+	}
+
 	/**
 	 * Crates new {@link ScanCursor}
 	 *
 	 * @param cursorId the cursor Id.
 	 * @param options Defaulted to {@link ScanOptions#NONE} if {@code null}.
+	 * @deprecated since 3.3.0 - Use {@link ScanCursor#ScanCursor(CursorId, ScanOptions)} instead.
 	 */
+	@Deprecated(since = "3.3.0")
 	public ScanCursor(long cursorId, @Nullable ScanOptions options) {
+		this(CursorId.of(cursorId), options);
+	}
+
+	/**
+	 * Crates new {@link ScanCursor}
+	 *
+	 * @param cursorId the cursor Id.
+	 * @param options Defaulted to {@link ScanOptions#NONE} if {@code null}.
+	 * @since 3.3.0
+	 */
+	public ScanCursor(CursorId cursorId, @Nullable ScanOptions options) {
 
 		this.scanOptions = options != null ? options : ScanOptions.NONE;
-		this.cursorId = cursorId;
+		this.id = cursorId;
 		this.state = CursorState.READY;
 		this.delegate = Collections.emptyIterator();
 	}
 
-	private void scan(long cursorId) {
+	private void scan(CursorId cursorId) {
 
 		try {
 			processScanResult(doScan(cursorId, this.scanOptions));
@@ -105,8 +130,25 @@ public abstract class ScanCursor<T> implements Cursor<T> {
 	 * @param cursorId
 	 * @param options
 	 * @return
+	 * @deprecated since 3.3.0, cursorId, can exceed {@link Long#MAX_VALUE}.
 	 */
-	protected abstract ScanIteration<T> doScan(long cursorId, ScanOptions options);
+	@Deprecated(since = "3.3.0")
+	protected ScanIteration<T> doScan(long cursorId, ScanOptions options) {
+		return doScan(CursorId.of(cursorId), scanOptions);
+	}
+
+	/**
+	 * Performs the actual scan command using the native client implementation. The given {@literal options} are never
+	 * {@code null}.
+	 *
+	 * @param cursorId
+	 * @param options
+	 * @return
+	 * @since 3.3.0
+	 */
+	protected ScanIteration<T> doScan(CursorId cursorId, ScanOptions options) {
+		return doScan(Long.parseLong(cursorId.getCursorId()), scanOptions);
+	}
 
 	/**
 	 * Initialize the {@link Cursor} prior to usage.
@@ -118,7 +160,7 @@ public abstract class ScanCursor<T> implements Cursor<T> {
 		}
 
 		state = CursorState.OPEN;
-		doOpen(cursorId);
+		doOpen(getId());
 
 		return this;
 	}
@@ -127,16 +169,27 @@ public abstract class ScanCursor<T> implements Cursor<T> {
 	 * Customization hook when calling {@link #open()}.
 	 *
 	 * @param cursorId
+	 * @deprecated since 3.3.0, use {@link #doOpen(CursorId)} instead.
 	 */
+	@Deprecated(since = "3.3.0", forRemoval = true)
 	protected void doOpen(long cursorId) {
+		doOpen(CursorId.of(cursorId));
+	}
+
+	/**
+	 * Customization hook when calling {@link #open()}.
+	 *
+	 * @param cursorId
+	 */
+	protected void doOpen(CursorId cursorId) {
 		scan(cursorId);
 	}
 
 	private void processScanResult(ScanIteration<T> result) {
 
-		cursorId = result.getCursorId();
+		id = result.getId();
 
-		if (isFinished(cursorId)) {
+		if (isFinished(id)) {
 			state = CursorState.FINISHED;
 		}
 
@@ -154,17 +207,34 @@ public abstract class ScanCursor<T> implements Cursor<T> {
 	 * @return {@literal true} if the cursor is considered finished, {@literal false} otherwise.s
 	 * @since 2.1
 	 */
+	@Deprecated(since = "3.3.0", forRemoval = true)
 	protected boolean isFinished(long cursorId) {
 		return cursorId == 0;
 	}
 
+	/**
+	 * Check whether {@code cursorId} is finished.
+	 *
+	 * @param cursorId the cursor Id
+	 * @return {@literal true} if the cursor is considered finished, {@literal false} otherwise.s
+	 * @since 3.3.0
+	 */
+	protected boolean isFinished(CursorId cursorId) {
+		return CursorId.isInitial(cursorId.getCursorId());
+	}
+
 	private void resetDelegate() {
 		delegate = Collections.emptyIterator();
 	}
 
+	@Override
+	public CursorId getId() {
+		return id;
+	}
+
 	@Override
 	public long getCursorId() {
-		return cursorId;
+		return Long.parseUnsignedLong(getId().getCursorId());
 	}
 
 	@Override
@@ -173,14 +243,14 @@ public abstract class ScanCursor<T> implements Cursor<T> {
 		assertCursorIsOpen();
 
 		while (!delegate.hasNext() && !CursorState.FINISHED.equals(state)) {
-			scan(cursorId);
+			scan(getId());
 		}
 
 		if (delegate.hasNext()) {
 			return true;
 		}
 
-		return cursorId > 0;
+		return !isFinished(id);
 	}
 
 	private void assertCursorIsOpen() {
@@ -196,7 +266,7 @@ public abstract class ScanCursor<T> implements Cursor<T> {
 		assertCursorIsOpen();
 
 		if (!hasNext()) {
-			throw new NoSuchElementException("No more elements available for cursor " + cursorId);
+			throw new NoSuchElementException("No more elements available for cursor " + id);
 		}
 
 		T next = moveNext(delegate);

src/main/java/org/springframework/data/redis/core/ScanIteration.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.redis.core;
 
+import static org.springframework.data.redis.core.Cursor.*;
+
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -32,14 +34,26 @@ import org.springframework.lang.Nullable;
  */
 public class ScanIteration<T> implements Iterable<T> {
 
-	private final long cursorId;
+	private final CursorId cursorId;
 	private final Collection<T> items;
 
 	/**
 	 * @param cursorId
 	 * @param items
+	 * @deprecated since 3.3.0, use {@link ScanIteration#ScanIteration(CursorId, Collection)} instead as {@code cursorId}
+	 *             can exceed {@link Long#MAX_VALUE}.
 	 */
+	@Deprecated(since = "3.3.0")
 	public ScanIteration(long cursorId, @Nullable Collection<T> items) {
+		this(CursorId.of(cursorId), items);
+	}
+
+	/**
+	 * @param cursorId
+	 * @param items
+	 * @since 3.3.0
+	 */
+	public ScanIteration(CursorId cursorId, @Nullable Collection<T> items) {
 
 		this.cursorId = cursorId;
 		this.items = (items != null ? new ArrayList<>(items) : Collections.emptyList());
@@ -49,8 +63,20 @@ public class ScanIteration<T> implements Iterable<T> {
 	 * The cursor id to be used for subsequent requests.
 	 *
 	 * @return
+	 * @deprecated since 3.3.0, use {@link #getId()} instead as the cursorId can exceed {@link Long#MAX_VALUE}.
 	 */
+	@Deprecated(since="3.3.3")
 	public long getCursorId() {
+		return Long.parseLong(getId().getCursorId());
+	}
+
+	/**
+	 * The cursor id to be used for subsequent requests.
+	 *
+	 * @return
+	 * @since 3.3.0
+	 */
+	public CursorId getId() {
 		return cursorId;
 	}