DATAREDIS-1084 - Add support for XCLAIM.

Original pull request: #512.

src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java

@@ -3651,6 +3651,24 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertAndReturn(delegate.xAdd(record.serialize(serializer)), identityConverter);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#xClaimJustId(java.lang.String, java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions)
+	 */
+	@Override
+	public List<RecordId> xClaimJustId(String key, String group, String consumer, XClaimOptions options) {
+		return convertAndReturn(delegate.xClaimJustId(serialize(key), group, consumer, options), identityConverter);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#xClaim(java.lang.String, java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions)
+	 */
+	@Override
+	public List<StringRecord> xClaim(String key, String group, String consumer, XClaimOptions options) {
+		return convertAndReturn(delegate.xClaim(serialize(key), group, consumer, options), listByteMapRecordToStringMapRecordConverter);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.StringRedisConnection#xDel(java.lang.String, java.lang.String[])
@@ -3803,6 +3821,24 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return delegate.xAdd(record);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xClaimJustId(byte[], java.lang.String, java.lag.String, org.springframework.data.redis.connection.RedisStreamCommands.XCLaimOptions)
+	 */
+	@Override
+	public List<RecordId> xClaimJustId(byte[] key, String group, String newOwner, XClaimOptions options) {
+		return delegate.xClaimJustId(key, group, newOwner, options);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xClaim(byte[], java.lang.String, java.lag.String, org.springframework.data.redis.connection.RedisStreamCommands.XCLaimOptions)
+	 */
+	@Override
+	public List<ByteRecord> xClaim(byte[] key, String group, String newOwner, XClaimOptions options) {
+		return delegate.xClaim(key, group, newOwner, options);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xDel(byte[], RecordId)

src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java

@@ -453,6 +453,20 @@ public interface DefaultedRedisConnection extends RedisConnection {
 		return streamCommands().xAdd(record);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */
+	@Override
+	@Deprecated
+	default List<RecordId> xClaimJustId(byte[] key, String group, String newOwner, XClaimOptions options) {
+		return streamCommands().xClaimJustId(key, group, newOwner, options);
+	}
+
+	/** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */
+	@Override
+	@Deprecated
+	default List<ByteRecord> xClaim(byte[] key, String group, String newOwner, XClaimOptions options) {
+		return streamCommands().xClaim(key, group, newOwner, options);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */
 	@Override
 	@Deprecated

src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java

@@ -19,6 +19,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -30,6 +31,7 @@ import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
+import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions;
 import org.springframework.data.redis.connection.RedisStreamCommands.XPendingOptions;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
 import org.springframework.data.redis.connection.stream.ByteBufferRecord;
@@ -285,6 +287,112 @@ public interface ReactiveStreamCommands {
 	 */
 	Flux<CommandResponse<AddStreamRecord, RecordId>> xAdd(Publisher<AddStreamRecord> commands);
 
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer} without increasing the delivered
+	 * count.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param options must not be {@literal null}.
+	 * @return a {@link Flux} emitting {@link RecordId is} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	default Flux<RecordId> xClaimJustId(ByteBuffer key, String group, String newOwner, XClaimOptions options) {
+
+		return xClaimJustId(Mono.just(new XClaimCommand(key, group, newOwner, options))).next()
+				.flatMapMany(CommandResponse::getOutput);
+	}
+
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer} without increasing the delivered
+	 * count.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return a {@link Flux} emitting {@link RecordId is} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	Flux<CommandResponse<XClaimCommand, Flux<RecordId>>> xClaimJustId(Publisher<XClaimCommand> commands);
+
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param minIdleTime must not be {@literal null}.
+	 * @param recordIds must not be {@literal null}.
+	 * @return a {@link Flux} emitting {@link ByteBufferRecord} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	default Flux<ByteBufferRecord> xClaim(ByteBuffer key, String group, String newOwner, Duration minIdleTime,
+			RecordId... recordIds) {
+
+		return xClaim(key, group, newOwner, XClaimOptions.minIdle(minIdleTime).ids(recordIds));
+	}
+
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param options must not be {@literal null}.
+	 * @return a {@link Flux} emitting {@link ByteBufferRecord} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	default Flux<ByteBufferRecord> xClaim(ByteBuffer key, String group, String newOwner, XClaimOptions options) {
+
+		return xClaim(Mono.just(new XClaimCommand(key, group, newOwner, options))).next()
+				.flatMapMany(CommandResponse::getOutput);
+	}
+
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer}.
+	 * 
+	 * @param commands must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	Flux<CommandResponse<XClaimCommand, Flux<ByteBufferRecord>>> xClaim(Publisher<XClaimCommand> commands);
+
+	/**
+	 * {@code XCLAIM} command parameters.
+	 *
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 */
+	class XClaimCommand extends KeyCommand {
+
+		private final String groupName;
+		private final String consumerName;
+		private final XClaimOptions options;
+
+		private XClaimCommand(@Nullable ByteBuffer key, String groupName, String consumerName, XClaimOptions options) {
+
+			super(key);
+			this.groupName = groupName;
+			this.consumerName = consumerName;
+			this.options = options;
+		}
+
+		public XClaimOptions getOptions() {
+			return options;
+		}
+
+		public String getConsumerName() {
+			return consumerName;
+		}
+
+		public String getGroupName() {
+			return groupName;
+		}
+	}
+
 	/**
 	 * {@code XDEL} command parameters.
 	 *

src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java

@@ -15,9 +15,13 @@
  */
 package org.springframework.data.redis.connection;
 
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
@@ -84,6 +88,249 @@ public interface RedisStreamCommands {
 	 */
 	RecordId xAdd(MapRecord<byte[], byte[], byte[]> record);
 
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer} without increasing the delivered
+	 * count.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param options must not be {@literal null}.
+	 * @return list of {@link RecordId ids} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	List<RecordId> xClaimJustId(byte[] key, String group, String newOwner, XClaimOptions options);
+
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param minIdleTime must not be {@literal null}.
+	 * @param recordIds must not be {@literal null}.
+	 * @return list of {@link ByteRecord} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	default List<ByteRecord> xClaim(byte[] key, String group, String newOwner, Duration minIdleTime,
+			RecordId... recordIds) {
+		return xClaim(key, group, newOwner, XClaimOptions.minIdle(minIdleTime).ids(recordIds));
+	}
+
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer}.
+	 * 
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param options must not be {@literal null}.
+	 * @return list of {@link ByteRecord} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	List<ByteRecord> xClaim(byte[] key, String group, String newOwner, XClaimOptions options);
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 2.3
+	 */
+	class XClaimOptions {
+
+		private final List<RecordId> ids;
+		private final Duration minIdleTime;
+		private final @Nullable Duration idleTime;
+		private final @Nullable Instant unixTime;
+		private final @Nullable Long retryCount;
+		private final boolean force;
+
+		private XClaimOptions(List<RecordId> ids, Duration minIdleTime, Duration idleTime, Instant unixTime,
+				Long retryCount, boolean force) {
+
+			this.ids = new ArrayList<>(ids);
+			this.minIdleTime = minIdleTime;
+			this.idleTime = idleTime;
+			this.unixTime = unixTime;
+			this.retryCount = retryCount;
+			this.force = force;
+		}
+
+		/**
+		 * Set the {@literal min-idle-time} to limit the command to messages that have been idle for at at least the given
+		 * {@link Duration}.
+		 * 
+		 * @param minIdleTime must not be {@literal null}.
+		 * @return new instance of {@link XClaimOptions}.
+		 */
+		public static XClaimOptionsBuilder minIdle(Duration minIdleTime) {
+			return new XClaimOptionsBuilder(minIdleTime);
+		}
+
+		/**
+		 * Set the {@literal min-idle-time} to limit the command to messages that have been idle for at at least the given
+		 * {@literal milliseconds}.
+		 *
+		 * @param millis
+		 * @return new instance of {@link XClaimOptions}.
+		 */
+		public static XClaimOptionsBuilder minIdleMs(long millis) {
+			return minIdle(Duration.ofMillis(millis));
+		}
+
+		/**
+		 * Set the idle time since last delivery of a message. To specify a specific point in time use
+		 * {@link #time(Instant)}.
+		 *
+		 * @param idleTime idle time.
+		 * @return {@code this}.
+		 */
+		public XClaimOptions idle(Duration idleTime) {
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force);
+		}
+
+		/**
+		 * Sets the idle time to a specific unix time (in milliseconds). To define a relative idle time use
+		 * {@link #idle(Duration)}.
+		 *
+		 * @param unixTime idle time.
+		 * @return {@code this}.
+		 */
+		public XClaimOptions time(Instant unixTime) {
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force);
+		}
+
+		/**
+		 * Set the retry counter to the specified value.
+		 *
+		 * @param retryCount can be {@literal null}. If {@literal null} no change to the retry counter will be made.
+		 * @return new instance of {@link XClaimOptions}.
+		 */
+		public XClaimOptions retryCount(@Nullable Long retryCount) {
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force);
+		}
+
+		/**
+		 * Forces creation of a pending message entry in the PEL even if it does not already exist as long a the given
+		 * stream record id is valid.
+		 *
+		 * @return new instance of {@link XClaimOptions}.
+		 */
+		public XClaimOptions force() {
+			return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, true);
+		}
+
+		/**
+		 * Get the {@link List} of {@literal ID}.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public List<RecordId> getIds() {
+			return ids;
+		}
+
+		/**
+		 * Get the {@literal ID} array as {@link String strings}.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public String[] getIdsAsStringArray() {
+			return getIds().stream().map(RecordId::getValue).toArray(String[]::new);
+		}
+
+		/**
+		 * Get the {@literal min-idle-time}.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public Duration getMinIdleTime() {
+			return minIdleTime;
+		}
+
+		/**
+		 * Get the {@literal IDLE ms} time.
+		 * 
+		 * @return can be {@literal null}.
+		 */
+		@Nullable
+		public Duration getIdleTime() {
+			return idleTime;
+		}
+
+		/**
+		 * Get the {@literal TIME ms-unix-time}
+		 * 
+		 * @return
+		 */
+		@Nullable
+		public Instant getUnixTime() {
+			return unixTime;
+		}
+
+		/**
+		 * Get the {@literal RETRYCOUNT count}.
+		 * 
+		 * @return
+		 */
+		@Nullable
+		public Long getRetryCount() {
+			return retryCount;
+		}
+
+		/**
+		 * Get the {@literal FORCE} flag.
+		 * 
+		 * @return
+		 */
+		public boolean isForce() {
+			return force;
+		}
+
+		public static class XClaimOptionsBuilder {
+
+			private Duration minIdleTime;
+
+			XClaimOptionsBuilder(Duration minIdleTime) {
+				this.minIdleTime = minIdleTime;
+			}
+
+			/**
+			 * Set the {@literal ID}s to claim.
+			 *
+			 * @param ids must not be {@literal null}.
+			 * @return
+			 */
+			public XClaimOptions ids(List<?> ids) {
+
+				List<RecordId> idList = ids.stream()
+						.map(it -> it instanceof RecordId ? (RecordId) it : RecordId.of(it.toString()))
+						.collect(Collectors.toList());
+
+				return new XClaimOptions(idList, minIdleTime, null, null, null, false);
+			}
+
+			/**
+			 * Set the {@literal ID}s to claim.
+			 *
+			 * @param ids must not be {@literal null}.
+			 * @return
+			 */
+			public XClaimOptions ids(RecordId... ids) {
+				return ids(Arrays.asList(ids));
+			}
+
+			/**
+			 * Set the {@literal ID}s to claim.
+			 *
+			 * @param ids must not be {@literal null}.
+			 * @return
+			 */
+			public XClaimOptions ids(String... ids) {
+				return ids(Arrays.asList(ids));
+			}
+		}
+	}
+
 	/**
 	 * Removes the records with the given id's from the stream. Returns the number of items deleted, that may be different
 	 * from the number of id's passed in case certain id's do not exist.

src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java

@@ -23,7 +23,6 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
-import org.springframework.data.domain.Range;
 import org.springframework.data.geo.Circle;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResults;
@@ -2013,6 +2012,50 @@ public interface StringRedisConnection extends RedisConnection {
 
 	RecordId xAdd(StringRecord record);
 
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer} without increasing the delivered
+	 * count.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param options must not be {@literal null}.
+	 * @return list of {@link RecordId ids} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	List<RecordId> xClaimJustId(String key, String group, String newOwner, XClaimOptions options);
+
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param minIdleTime must not be {@literal null}.
+	 * @param recordIds must not be {@literal null}.
+	 * @return list of {@link StringRecord} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	default List<StringRecord> xClaim(String key, String group, String newOwner, Duration minIdleTime,
+			RecordId... recordIds) {
+		return xClaim(key, group, newOwner, XClaimOptions.minIdle(minIdleTime).ids(recordIds));
+	}
+
+	/**
+	 * Change the ownership of a pending message to the given new {@literal consumer}.
+	 *
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param group the name of the {@literal consumer group}.
+	 * @param newOwner the name of the new {@literal consumer}.
+	 * @param options must not be {@literal null}.
+	 * @return list of {@link StringRecord} that changed user.
+	 * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+	 * @since 2.3
+	 */
+	List<StringRecord> xClaim(String key, String group, String newOwner, XClaimOptions options);
+
 	/**
 	 * Removes the specified entries from the stream. Returns the number of items deleted, that may be different from the
 	 * number of IDs passed in case certain IDs do not exist.
@@ -2082,15 +2125,15 @@ public interface StringRedisConnection extends RedisConnection {
 	 * @param groupName the name of the {@literal consumer group}. Must not be {@literal null}.
 	 * @return a summary of pending messages within the given {@literal consumer group} or {@literal null} when used in
 	 *         pipeline / transaction.
-	 *         @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+	 * @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
 	 * @since 2.3
 	 */
 	@Nullable
 	PendingMessagesSummary xPending(String key, String groupName);
 
 	/**
-	 * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link org.springframework.data.domain.Range} within a
-	 * {@literal consumer group}.
+	 * Obtain detailed information about pending {@link PendingMessage messages} for a given
+	 * {@link org.springframework.data.domain.Range} within a {@literal consumer group}.
 	 *
 	 * @param key the {@literal key} the stream is stored at. Must not be {@literal null}.
 	 * @param groupName the name of the {@literal consumer group}. Must not be {@literal null}.
@@ -2099,15 +2142,16 @@ public interface StringRedisConnection extends RedisConnection {
 	 * @param count limit the number of results. Must not be {@literal null}.
 	 * @return pending messages for the given {@literal consumer group} or {@literal null} when used in pipeline /
 	 *         transaction.
-	 *         @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+	 * @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
 	 * @since 2.3
 	 */
 	@Nullable
-	PendingMessages xPending(String key, String groupName, String consumerName, org.springframework.data.domain.Range<String> range, Long count);
+	PendingMessages xPending(String key, String groupName, String consumerName,
+			org.springframework.data.domain.Range<String> range, Long count);
 
 	/**
-	 * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link org.springframework.data.domain.Range} within a
-	 * {@literal consumer group}.
+	 * Obtain detailed information about pending {@link PendingMessage messages} for a given
+	 * {@link org.springframework.data.domain.Range} within a {@literal consumer group}.
 	 *
 	 * @param key the {@literal key} the stream is stored at. Must not be {@literal null}.
 	 * @param groupName the name of the {@literal consumer group}. Must not be {@literal null}.
@@ -2115,12 +2159,12 @@ public interface StringRedisConnection extends RedisConnection {
 	 * @param count limit the number of results. Must not be {@literal null}.
 	 * @return pending messages for the given {@literal consumer group} or {@literal null} when used in pipeline /
 	 *         transaction.
-	 *         @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+	 * @see <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
 	 * @since 2.3
 	 */
 	@Nullable
-	PendingMessages xPending(String key, String groupName, org.springframework.data.domain.Range<String> range, Long count);
-
+	PendingMessages xPending(String key, String groupName, org.springframework.data.domain.Range<String> range,
+			Long count);
 
 	/**
 	 * Obtain detailed information about pending {@link PendingMessage messages} applying given {@link XPendingOptions

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

@@ -16,6 +16,7 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.XAddArgs;
+import io.lettuce.core.XClaimArgs;
 import io.lettuce.core.XReadArgs;
 import io.lettuce.core.XReadArgs.StreamOffset;
 import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
@@ -105,6 +106,50 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
 		}));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xClaimJustId(byte[], java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions)
+	 */
+	@Override
+	public Flux<CommandResponse<XClaimCommand, Flux<RecordId>>> xClaimJustId(Publisher<XClaimCommand> commands) {
+
+		if (true /* TODO: set the JUSTID flag */ ) {
+			throw new UnsupportedOperationException("Lettuce does not support XCLAIM with JUSTID. (Ref: lettuce-io#1233)");
+		}
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			String[] ids = command.getOptions().getIdsAsStringArray();
+			io.lettuce.core.Consumer<ByteBuffer> from = io.lettuce.core.Consumer
+					.from(ByteUtils.getByteBuffer(command.getGroupName()), ByteUtils.getByteBuffer(command.getConsumerName()));
+			XClaimArgs args = StreamConverters.toXClaimArgs(command.getOptions());
+
+			Flux<RecordId> result = cmd.xclaim(command.getKey(), from, args, ids).map(it -> RecordId.of(it.getId()));
+			return new CommandResponse<>(command, result);
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xClaim(byte[], java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions)
+	 */
+	@Override
+	public Flux<CommandResponse<XClaimCommand, Flux<ByteBufferRecord>>> xClaim(Publisher<XClaimCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			String[] ids = command.getOptions().getIdsAsStringArray();
+			io.lettuce.core.Consumer<ByteBuffer> from = io.lettuce.core.Consumer
+					.from(ByteUtils.getByteBuffer(command.getGroupName()), ByteUtils.getByteBuffer(command.getConsumerName()));
+			XClaimArgs args = StreamConverters.toXClaimArgs(command.getOptions());
+
+			Flux<ByteBufferRecord> result = cmd.xclaim(command.getKey(), from, args, ids)
+					.map(it -> StreamRecords.newRecord().in(it.getStream()).withId(it.getId()).ofBuffer(it.getBody()));
+			return new CommandResponse<>(command, result);
+
+		}));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xDel(org.reactivestreams.Publisher)

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

@@ -16,6 +16,7 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.XAddArgs;
+import io.lettuce.core.XClaimArgs;
 import io.lettuce.core.XReadArgs;
 import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
 import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
@@ -111,6 +112,72 @@ class LettuceStreamCommands implements RedisStreamCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xClaimJustId(byte[], java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions)
+	 */
+	@Override
+	public List<RecordId> xClaimJustId(byte[] key, String group, String newOwner, XClaimOptions options) {
+
+		String[] ids = options.getIdsAsStringArray();
+		io.lettuce.core.Consumer<byte[]> from = io.lettuce.core.Consumer.from(LettuceConverters.toBytes(group),
+				LettuceConverters.toBytes(newOwner));
+		XClaimArgs args = StreamConverters.toXClaimArgs(options);
+
+		if (true /* TODO: set the JUSTID flag */ ) {
+			throw new UnsupportedOperationException("Lettuce does not support XCLAIM with JUSTID. (Ref: lettuce-io#1233)");
+		}
+
+		try {
+			if (isPipelined()) {
+
+				pipeline(connection.newLettuceResult(getAsyncConnection().xclaim(key, from, args, ids),
+						StreamConverters.messagesToIds()));
+				return null;
+			}
+			if (isQueueing()) {
+
+				transaction(connection.newLettuceResult(getAsyncConnection().xclaim(key, from, args, ids),
+						StreamConverters.messagesToIds()));
+				return null;
+			}
+
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+
+		return StreamConverters.messagesToIds().convert(getConnection().xclaim(key, from, args, ids));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xClaim(byte[], java.lang.String, java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions)
+	 */
+	@Override
+	public List<ByteRecord> xClaim(byte[] key, String group, String newOwner, XClaimOptions options) {
+
+		String[] ids = options.getIdsAsStringArray();
+		io.lettuce.core.Consumer<byte[]> from = io.lettuce.core.Consumer.from(LettuceConverters.toBytes(group),
+				LettuceConverters.toBytes(newOwner));
+		XClaimArgs args = StreamConverters.toXClaimArgs(options);
+
+		try {
+			if (isPipelined()) {
+				pipeline(connection.newLettuceResult(getAsyncConnection().xclaim(key, from, args, ids),
+						StreamConverters.byteRecordListConverter()));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(connection.newLettuceResult(getAsyncConnection().xclaim(key, from, args, ids),
+						StreamConverters.byteRecordListConverter()));
+				return null;
+			}
+			return StreamConverters.byteRecordListConverter().convert(getConnection().xclaim(key, from, args, ids));
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisStreamCommands#xDel(byte[], java.lang.String[])

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

@@ -16,15 +16,19 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.StreamMessage;
+import io.lettuce.core.XClaimArgs;
 import io.lettuce.core.XReadArgs;
 
 import java.util.List;
 
 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.RecordId;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
 import org.springframework.data.redis.connection.stream.StreamRecords;
-import org.springframework.data.redis.connection.convert.ListConverter;
+import org.springframework.lang.Nullable;
 
 /**
  * Converters for Redis Stream-specific types.
@@ -39,6 +43,9 @@ import org.springframework.data.redis.connection.convert.ListConverter;
 @SuppressWarnings({ "unchecked", "rawtypes" })
 class StreamConverters {
 
+	private static final Converter<List<StreamMessage<byte[], byte[]>>, List<RecordId>> MESSAGEs_TO_IDs = new ListConverter<>(
+			messageToIdConverter());
+
 	/**
 	 * Convert {@link StreamReadOptions} to Lettuce's {@link XReadArgs}.
 	 *
@@ -49,14 +56,33 @@ class StreamConverters {
 		return StreamReadOptionsToXReadArgsConverter.INSTANCE.convert(readOptions);
 	}
 
-	public static Converter<StreamMessage<byte[], byte[]>, ByteRecord> byteRecordConverter() {
+	/**
+	 * Convert {@link XClaimOptions} to Lettuce's {@link XClaimArgs}.
+	 *
+	 * @param options must not be {@literal null}.
+	 * @return the converted {@link XClaimArgs}.
+	 * @since 2.3
+	 */
+	static XClaimArgs toXClaimArgs(XClaimOptions options) {
+		return XClaimOptionsToXClaimArgsConverter.INSTANCE.convert(options);
+	}
+
+	static Converter<StreamMessage<byte[], byte[]>, ByteRecord> byteRecordConverter() {
 		return (it) -> StreamRecords.newRecord().in(it.getStream()).withId(it.getId()).ofBytes(it.getBody());
 	}
 
-	public static Converter<List<StreamMessage<byte[], byte[]>>, List<ByteRecord>> byteRecordListConverter() {
+	static Converter<List<StreamMessage<byte[], byte[]>>, List<ByteRecord>> byteRecordListConverter() {
 		return new ListConverter<>(byteRecordConverter());
 	}
 
+	static Converter<StreamMessage<byte[], byte[]>, RecordId> messageToIdConverter() {
+		return (it) -> RecordId.of(it.getId());
+	}
+
+	static Converter<List<StreamMessage<byte[], byte[]>>, List<RecordId>> messagesToIds() {
+		return MESSAGEs_TO_IDs;
+	}
+
 	/**
 	 * {@link Converter} to convert {@link StreamReadOptions} to Lettuce's {@link XReadArgs}.
 	 */
@@ -87,4 +113,35 @@ class StreamConverters {
 			return args;
 		}
 	}
+
+	/**
+	 * {@link Converter} to convert {@link XClaimOptions} to Lettuce's {@link XClaimArgs}.
+	 * 
+	 * @since 2.3
+	 */
+	enum XClaimOptionsToXClaimArgsConverter implements Converter<XClaimOptions, XClaimArgs> {
+		INSTANCE;
+
+		@Nullable
+		@Override
+		public XClaimArgs convert(XClaimOptions source) {
+
+			XClaimArgs args = XClaimArgs.Builder.minIdleTime(source.getMinIdleTime());
+			args.minIdleTime(source.getMinIdleTime());
+			args.force(source.isForce());
+
+			if (source.getIdleTime() != null) {
+				args.idle(source.getIdleTime());
+			}
+			if (source.getRetryCount() != null) {
+				args.retryCount(source.getRetryCount());
+			}
+			if (source.getUnixTime() != null) {
+				args.time(source.getUnixTime());
+			}
+
+			return args;
+
+		}
+	}
 }