Add support for ZPOPMIN, ZPOPMAX and their blocking variants.

We now support ZPOPMIN, BZPOPMIN, ZPOPMAX and BZPOPMAX through the Template API for all drivers.

Closes #2007
Original Pull Request: #2088

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

@@ -81,7 +81,8 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 	private final RedisSerializer<String> serializer;
 	private Converter<byte[], String> bytesToString = new DeserializingConverter();
 	private Converter<String, byte[]> stringToBytes = new SerializingConverter();
-	private SetConverter<Tuple, StringTuple> tupleToStringTuple = new SetConverter<>(new TupleConverter());
+	private final TupleConverter tupleConverter = new TupleConverter();
+	private SetConverter<Tuple, StringTuple> tupleToStringTuple = new SetConverter<>(tupleConverter);
 	private SetConverter<StringTuple, Tuple> stringTupleToTuple = new SetConverter<>(new StringTupleConverter());
 	private ListConverter<byte[], String> byteListToStringList = new ListConverter<>(bytesToString);
 	private MapConverter<byte[], String> byteMapToStringMap = new MapConverter<>(bytesToString);
@@ -2861,6 +2862,126 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return delegate.zLexCount(key, range);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(byte[])
+	 */
+	@Nullable
+	@Override
+	public Tuple zPopMin(byte[] key) {
+		return delegate.zPopMin(key);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(String)
+	 */
+	@Nullable
+	@Override
+	public StringTuple zPopMin(String key) {
+		return convertAndReturn(delegate.zPopMin(serialize(key)), tupleConverter);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMinMin(byte[], count)
+	 */
+	@Nullable
+	@Override
+	public Set<Tuple> zPopMin(byte[] key, long count) {
+		return delegate.zPopMin(key, count);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(String, long)
+	 */
+	@Nullable
+	@Override
+	public Set<StringTuple> zPopMin(String key, long count) {
+		return convertAndReturn(delegate.zPopMin(serialize(key), count), tupleToStringTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMin(byte[], long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public Tuple bZPopMin(byte[] key, long timeout, TimeUnit unit) {
+		return delegate.bZPopMin(key, timeout, unit);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMin(String, long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public StringTuple bZPopMin(String key, long timeout, TimeUnit unit) {
+		return convertAndReturn(delegate.bZPopMin(serialize(key), timeout, unit), tupleConverter);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(byte[])
+	 */
+	@Nullable
+	@Override
+	public Tuple zPopMax(byte[] key) {
+		return delegate.zPopMax(key);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(String)
+	 */
+	@Nullable
+	@Override
+	public StringTuple zPopMax(String key) {
+		return convertAndReturn(delegate.zPopMax(serialize(key)), tupleConverter);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(byte[], long)
+	 */
+	@Nullable
+	@Override
+	public Set<Tuple> zPopMax(byte[] key, long count) {
+		return delegate.zPopMax(key, count);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(String, long)
+	 */
+	@Nullable
+	@Override
+	public Set<StringTuple> zPopMax(String key, long count) {
+		return convertAndReturn(delegate.zPopMax(serialize(key), count), tupleToStringTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMax(byte[], long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public Tuple bZPopMax(byte[] key, long timeout, TimeUnit unit) {
+		return delegate.bZPopMax(key, timeout, unit);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMax(String, long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public StringTuple bZPopMax(String key, long timeout, TimeUnit unit) {
+		return convertAndReturn(delegate.bZPopMax(serialize(key), timeout, unit), tupleConverter);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.StringRedisConnection#zIncrBy(java.lang.String, double, java.lang.String)

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

@@ -89,4 +89,14 @@ public class DefaultTuple implements Tuple {
 		Double a = (o == null ? Double.valueOf(0.0d) : o);
 		return d.compareTo(a);
 	}
+
+	@Override
+	public String toString() {
+		StringBuffer sb = new StringBuffer();
+		sb.append(getClass().getSimpleName());
+		sb.append(" [score=").append(score);
+		sb.append(", value=").append(value == null ? "null" : new String(value));
+		sb.append(']');
+		return sb.toString();
+	}
 }

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

@@ -942,6 +942,48 @@ public interface DefaultedRedisConnection extends RedisConnection {
 		return zSetCommands().zLexCount(key, range);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */
+	@Override
+	@Deprecated
+	default Tuple zPopMin(byte[] key) {
+		return zSetCommands().zPopMin(key);
+	}
+
+	/** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */
+	@Override
+	@Deprecated
+	default Set<Tuple> zPopMin(byte[] key, long count) {
+		return zSetCommands().zPopMin(key, count);
+	}
+
+	/** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */
+	@Override
+	@Deprecated
+	default Tuple bZPopMin(byte[] key, long timeout, TimeUnit unit) {
+		return zSetCommands().bZPopMin(key, timeout, unit);
+	}
+
+	/** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */
+	@Override
+	@Deprecated
+	default Tuple zPopMax(byte[] key) {
+		return zSetCommands().zPopMax(key);
+	}
+
+	/** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */
+	@Override
+	@Deprecated
+	default Set<Tuple> zPopMax(byte[] key, long count) {
+		return zSetCommands().zPopMax(key, count);
+	}
+
+	/** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */
+	@Override
+	@Deprecated
+	default Tuple bZPopMax(byte[] key, long timeout, TimeUnit unit) {
+		return zSetCommands().bZPopMax(key, timeout, unit);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#zSetCommands()}}. */
 	@Override
 	@Deprecated

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

@@ -19,13 +19,16 @@ 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.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.domain.Range;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
@@ -1241,6 +1244,293 @@ public interface ReactiveZSetCommands {
 	 */
 	Flux<NumericResponse<ZLexCountCommand, Long>> zLexCount(Publisher<ZLexCountCommand> commands);
 
+	/**
+	 * @author Mark Paluch
+	 */
+	enum PopDirection {
+		MIN, MAX
+	}
+
+	/**
+	 * {@code ZPOPMIN}/{@literal ZPOPMAX} command parameters.
+	 *
+	 * @author Mark Paluch
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	class ZPopCommand extends KeyCommand {
+
+		private final PopDirection direction;
+
+		private final long count;
+
+		private ZPopCommand(PopDirection direction, @Nullable ByteBuffer key, long count) {
+
+			super(key);
+			this.count = count;
+			this.direction = direction;
+		}
+
+		/**
+		 * Creates a new {@link ZPopCommand} for min pop ({@literal ZPOPMIN}).
+		 *
+		 * @return a new {@link ZPopCommand} for min pop ({@literal ZPOPMIN}).
+		 */
+		public static ZPopCommand min() {
+			return new ZPopCommand(PopDirection.MIN, null, 1);
+		}
+
+		/**
+		 * Creates a new {@link ZPopCommand} for max pop ({@literal ZPOPMAX}).
+		 *
+		 * @return a new {@link ZPopCommand} for max pop ({@literal ZPOPMAX}).
+		 */
+		public static ZPopCommand max() {
+			return new ZPopCommand(PopDirection.MAX, null, 1);
+		}
+
+		/**
+		 * Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
+		 *
+		 * @param key must not be {@literal null}.
+		 * @return a new {@link ZPopCommand} with {@literal value} applied.
+		 */
+		public ZPopCommand from(ByteBuffer key) {
+
+			Assert.notNull(key, "Key must not be null!");
+
+			return new ZPopCommand(direction, key, count);
+		}
+
+		/**
+		 * Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
+		 *
+		 * @param count
+		 * @return a new {@link ZPopCommand} with {@literal value} applied.
+		 */
+		public ZPopCommand count(long count) {
+			return new ZPopCommand(direction, getKey(), count);
+		}
+
+		/**
+		 * @return never {@literal null}.
+		 */
+		public PopDirection getDirection() {
+			return direction;
+		}
+
+		public long getCount() {
+			return count;
+		}
+	}
+
+	/**
+	 * {@code BZPOPMIN}/{@literal BZPOPMAX} command parameters.
+	 *
+	 * @author Mark Paluch
+	 * @see <a href="https://redis.io/commands/bzpopmin">Redis Documentation: BZPOPMIN</a>
+	 * @see <a href="https://redis.io/commands/bzpopmax">Redis Documentation: BZPOPMAX</a>
+	 * @since 2.6
+	 */
+	class BZPopCommand extends KeyCommand {
+
+		private final PopDirection direction;
+
+		private final Duration timeout;
+
+		private final long count;
+
+		private BZPopCommand(@Nullable ByteBuffer key, Duration timeout, long count, PopDirection direction) {
+
+			super(key);
+			this.count = count;
+			this.timeout = timeout;
+			this.direction = direction;
+		}
+
+		/**
+		 * Creates a new {@link BZPopCommand} for min pop ({@literal ZPOPMIN}).
+		 *
+		 * @return a new {@link BZPopCommand} for min pop ({@literal ZPOPMIN}).
+		 */
+		public static BZPopCommand min() {
+			return new BZPopCommand(null, null, 0, PopDirection.MIN);
+		}
+
+		/**
+		 * Creates a new {@link BZPopCommand} for max pop ({@literal ZPOPMAX}).
+		 *
+		 * @return a new {@link BZPopCommand} for max pop ({@literal ZPOPMAX}).
+		 */
+		public static BZPopCommand max() {
+			return new BZPopCommand(null, null, 0, PopDirection.MAX);
+		}
+
+		/**
+		 * Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
+		 *
+		 * @param key must not be {@literal null}.
+		 * @return a new {@link BZPopCommand} with {@literal value} applied.
+		 */
+		public BZPopCommand from(ByteBuffer key) {
+
+			Assert.notNull(key, "Key must not be null!");
+
+			return new BZPopCommand(key, timeout, count, direction);
+		}
+
+		/**
+		 * Applies the {@literal key}. Constructs a new command instance with all previously configured properties.
+		 *
+		 * @param count
+		 * @return a new {@link BZPopCommand} with {@literal value} applied.
+		 */
+		public BZPopCommand count(long count) {
+			return new BZPopCommand(getKey(), timeout, count, direction);
+		}
+
+		/**
+		 * Applies a {@link Duration timeout}. Constructs a new command instance with all previously configured properties.
+		 *
+		 * @param timeout must not be {@literal null}.
+		 * @return a new {@link BZPopCommand} with {@link Duration timeout} applied.
+		 */
+		public BZPopCommand blockingFor(Duration timeout) {
+
+			Assert.notNull(timeout, "Timeout must not be null!");
+
+			return new BZPopCommand(getKey(), timeout, count, direction);
+		}
+
+		/**
+		 * @return never {@literal null}.
+		 */
+		public PopDirection getDirection() {
+			return direction;
+		}
+
+		public Duration getTimeout() {
+			return timeout;
+		}
+
+		public long getCount() {
+			return count;
+		}
+	}
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	default Mono<Tuple> zPopMin(ByteBuffer key) {
+		return zPop(Mono.just(ZPopCommand.min().from(key))).map(CommandResponse::getOutput).flatMap(Flux::next).next();
+	}
+
+	/**
+	 * Remove and return {@code count} values with their score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	default Flux<Tuple> zPopMin(ByteBuffer key, long count) {
+		return zPop(Mono.just(ZPopCommand.min().from(key).count(count))).map(CommandResponse::getOutput)
+				.flatMap(Function.identity());
+	}
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout must not be {@literal null}.
+	 * @return
+	 * @throws IllegalArgumentException if the timeout is {@literal null} or negative.
+	 * @see <a href="https://redis.io/commands/bzpopmin">Redis Documentation: BZPOPMIN</a>
+	 * @since 2.6
+	 */
+	default Mono<Tuple> bZPopMin(ByteBuffer key, Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return bZPop(Mono.just(BZPopCommand.min().from(key).blockingFor(timeout))).map(CommandResponse::getOutput)
+				.flatMap(Flux::next).next();
+	}
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	default Mono<Tuple> zPopMax(ByteBuffer key) {
+		return zPop(Mono.just(ZPopCommand.max().from(key))).map(CommandResponse::getOutput).flatMap(Flux::next).next();
+	}
+
+	/**
+	 * Remove and return {@code count} values with their score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	default Flux<Tuple> zPopMax(ByteBuffer key, long count) {
+		return zPop(Mono.just(ZPopCommand.max().from(key).count(count))).map(CommandResponse::getOutput)
+				.flatMap(Function.identity());
+	}
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout must not be {@literal null}.
+	 * @return
+	 * @throws IllegalArgumentException if the timeout is {@literal null} or negative.
+	 * @see <a href="https://redis.io/commands/bzpopmax">Redis Documentation: BZPOPMAX</a>
+	 * @since 2.6
+	 */
+	default Mono<Tuple> bZPopMax(ByteBuffer key, Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return bZPop(Mono.just(BZPopCommand.max().from(key).blockingFor(timeout))).map(CommandResponse::getOutput)
+				.flatMap(Flux::next).next();
+	}
+
+	/**
+	 * Remove and return elements from sorted set at {@link ByteBuffer keyCommand#getKey()}.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 */
+	Flux<CommandResponse<ZPopCommand, Flux<Tuple>>> zPop(Publisher<ZPopCommand> commands);
+
+	/**
+	 * Remove and return elements from sorted set at {@link ByteBuffer keyCommand#getKey()}.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 */
+	Flux<CommandResponse<BZPopCommand, Flux<Tuple>>> bZPop(Publisher<BZPopCommand> commands);
+
 	/**
 	 * Get the size of sorted set with {@literal key}.
 	 *
@@ -1256,7 +1546,7 @@ public interface ReactiveZSetCommands {
 	}
 
 	/**
-	 * Get the size of sorted set with {@link KeyCommand#getKey()}.
+	 * Get the size of sorted set with {@linByteBuffer keyCommand#getKey()}.
 	 *
 	 * @param commands must not be {@literal null}.
 	 * @return

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

@@ -20,6 +20,7 @@ import java.util.Collections;
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.function.DoubleUnaryOperator;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -972,6 +973,80 @@ public interface RedisZSetCommands {
 	@Nullable
 	Long zLexCount(byte[] key, Range range);
 
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Tuple zPopMin(byte[] key);
+
+	/**
+	 * Remove and return {@code count} values with their score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Set<Tuple> zPopMin(byte[] key, long count);
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/bzpopmin">Redis Documentation: BZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Tuple bZPopMin(byte[] key, long timeout, TimeUnit unit);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Tuple zPopMax(byte[] key);
+
+	/**
+	 * Remove and return {@code count} values with their score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Set<Tuple> zPopMax(byte[] key, long count);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/bzpopmax">Redis Documentation: BZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Tuple bZPopMax(byte[] key, long timeout, TimeUnit unit);
+
 	/**
 	 * Get the size of sorted set with {@code key}.
 	 *

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

@@ -1434,6 +1434,80 @@ public interface StringRedisConnection extends RedisConnection {
 	@Nullable
 	Long zLexCount(String key, Range range);
 
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Tuple zPopMin(String key);
+
+	/**
+	 * Remove and return {@code count} values with their score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Set<StringTuple> zPopMin(String key, long count);
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/bzpopmin">Redis Documentation: BZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	StringTuple bZPopMin(String key, long timeout, TimeUnit unit);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	StringTuple zPopMax(String key);
+
+	/**
+	 * Remove and return {@code count} values with their score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Set<StringTuple> zPopMax(String key, long count);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/bzpopmax">Redis Documentation: BZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	StringTuple bZPopMax(String key, long timeout, TimeUnit unit);
+
 	/**
 	 * Get the size of sorted set with {@code key}.
 	 *

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

@@ -20,10 +20,12 @@ import redis.clients.jedis.ZParams;
 
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.ClusterSlotHashUtil;
+import org.springframework.data.redis.connection.DefaultTuple;
 import org.springframework.data.redis.connection.RedisZSetCommands;
 import org.springframework.data.redis.connection.convert.SetConverter;
 import org.springframework.data.redis.core.Cursor;
@@ -31,6 +33,7 @@ import org.springframework.data.redis.core.ScanCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.data.redis.util.ByteUtils;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -281,6 +284,112 @@ class JedisClusterZSetCommands implements RedisZSetCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(byte[])
+	 */
+	@Nullable
+	@Override
+	public Tuple zPopMin(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		try {
+			redis.clients.jedis.Tuple tuple = connection.getCluster().zpopmin(key);
+			return tuple != null ? JedisConverters.toTuple(tuple) : null;
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(byte[], long)
+	 */
+	@Nullable
+	@Override
+	public Set<Tuple> zPopMin(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		try {
+			return toTupleSet(connection.getCluster().zpopmin(key, Math.toIntExact(count)));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMin(byte[], long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public Tuple bZPopMin(byte[] key, long timeout, TimeUnit unit) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(unit, "TimeUnit must not be null!");
+
+		try {
+			return toTuple(connection.getCluster().bzpopmin(JedisConverters.toSeconds(timeout, unit), key));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(byte[])
+	 */
+	@Nullable
+	@Override
+	public Tuple zPopMax(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		try {
+			redis.clients.jedis.Tuple tuple = connection.getCluster().zpopmax(key);
+			return tuple != null ? JedisConverters.toTuple(tuple) : null;
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(byte[], long)
+	 */
+	@Nullable
+	@Override
+	public Set<Tuple> zPopMax(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		try {
+			return toTupleSet(connection.getCluster().zpopmax(key, Math.toIntExact(count)));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMax(byte[], long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public Tuple bZPopMax(byte[] key, long timeout, TimeUnit unit) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(unit, "TimeUnit must not be null!");
+
+		try {
+			return toTuple(connection.getCluster().bzpopmax(JedisConverters.toSeconds(timeout, unit), key));
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zRemRangeByScore(byte[], org.springframework.data.redis.connection.RedisZSetCommands.Range)
@@ -864,4 +973,21 @@ class JedisClusterZSetCommands implements RedisZSetCommands {
 		return TUPLE_SET_CONVERTER.convert(source);
 	}
 
+	/**
+	 * Workaround for broken Jedis BZPOP signature.
+	 *
+	 * @param bytes
+	 * @return
+	 */
+	@Nullable
+	@SuppressWarnings("unchecked")
+	private static Tuple toTuple(List<?> bytes) {
+
+		if (bytes.isEmpty()) {
+			return null;
+		}
+
+		return new DefaultTuple((byte[]) bytes.get(1), Double.parseDouble(new String((byte[]) bytes.get(2))));
+	}
+
 }

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

@@ -758,6 +758,26 @@ public abstract class JedisConverters extends Converters {
 		return param;
 	}
 
+	/**
+	 * Convert a timeout to seconds using {@code double} representation including fraction of seconds.
+	 *
+	 * @param timeout
+	 * @param unit
+	 * @return
+	 * @since 2.6
+	 */
+	static double toSeconds(long timeout, TimeUnit unit) {
+
+		switch (unit) {
+			case MILLISECONDS:
+			case MICROSECONDS:
+			case NANOSECONDS:
+				return unit.toMillis(timeout) / 1000d;
+			default:
+				return unit.toSeconds(timeout);
+		}
+	}
+
 	/**
 	 * Convert given {@link BitFieldSubCommands} into argument array.
 	 *

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

@@ -26,12 +26,15 @@ import java.nio.charset.StandardCharsets;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
+import org.springframework.data.redis.connection.DefaultTuple;
 import org.springframework.data.redis.connection.RedisZSetCommands;
 import org.springframework.data.redis.core.Cursor;
 import org.springframework.data.redis.core.KeyBoundCursor;
 import org.springframework.data.redis.core.ScanIteration;
 import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -297,6 +300,96 @@ class JedisZSetCommands implements RedisZSetCommands {
 		return connection.invoke().just(BinaryJedis::zlexcount, MultiKeyPipelineBase::zlexcount, key, min, max);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(byte[])
+	 */
+	@Nullable
+	@Override
+	public Tuple zPopMin(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().from(BinaryJedis::zpopmin, MultiKeyPipelineBase::zpopmin, key)
+				.get(JedisConverters::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(byte[], long)
+	 */
+	@Nullable
+	@Override
+	public Set<Tuple> zPopMin(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke()
+				.fromMany(BinaryJedis::zpopmin, MultiKeyPipelineBase::zpopmin, key, Math.toIntExact(count))
+				.toSet(JedisConverters::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMin(byte[], long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public Tuple bZPopMin(byte[] key, long timeout, TimeUnit unit) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(unit, "TimeUnit must not be null!");
+
+		return connection.invoke()
+				.from(BinaryJedis::bzpopmin, MultiKeyPipelineBase::bzpopmin, JedisConverters.toSeconds(timeout, unit), key)
+				.get(JedisZSetCommands::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(byte[])
+	 */
+	@Nullable
+	@Override
+	public Tuple zPopMax(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().from(BinaryJedis::zpopmax, MultiKeyPipelineBase::zpopmax, key)
+				.get(JedisConverters::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(byte[], long)
+	 */
+	@Nullable
+	@Override
+	public Set<Tuple> zPopMax(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke()
+				.fromMany(BinaryJedis::zpopmax, MultiKeyPipelineBase::zpopmax, key, Math.toIntExact(count))
+				.toSet(JedisConverters::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMax(byte[], long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public Tuple bZPopMax(byte[] key, long timeout, TimeUnit unit) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(unit, "TimeUnit must not be null!");
+
+		return connection.invoke()
+				.from(BinaryJedis::bzpopmax, MultiKeyPipelineBase::bzpopmax, JedisConverters.toSeconds(timeout, unit), key)
+				.get(JedisZSetCommands::toTuple);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zCard(byte[])
@@ -602,4 +695,21 @@ class JedisZSetCommands implements RedisZSetCommands {
 		return connection.isQueueing();
 	}
 
+	/**
+	 * Workaround for broken Jedis BZPOP signature.
+	 *
+	 * @param bytes
+	 * @return
+	 */
+	@Nullable
+	@SuppressWarnings("unchecked")
+	private static Tuple toTuple(List<?> bytes) {
+
+		if (bytes.isEmpty()) {
+			return null;
+		}
+
+		return new DefaultTuple((byte[]) bytes.get(1), Double.parseDouble(new String((byte[]) bytes.get(2))));
+	}
+
 }

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

@@ -247,7 +247,8 @@ public abstract class LettuceConverters extends Converters {
 	}
 
 	public static Tuple toTuple(@Nullable ScoredValue<byte[]> source) {
-		return source != null ? new DefaultTuple(source.getValue(), Double.valueOf(source.getScore())) : null;
+		return source != null && source.hasValue() ? new DefaultTuple(source.getValue(), Double.valueOf(source.getScore()))
+				: null;
 	}
 
 	public static String toString(@Nullable byte[] source) {

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

@@ -18,12 +18,14 @@ package org.springframework.data.redis.connection.lettuce;
 import io.lettuce.core.Range;
 import io.lettuce.core.ScanStream;
 import io.lettuce.core.ScoredValue;
+import io.lettuce.core.Value;
 import io.lettuce.core.ZAddArgs;
 import io.lettuce.core.ZStoreArgs;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.nio.ByteBuffer;
+import java.time.temporal.ChronoUnit;
 import java.util.List;
 
 import org.reactivestreams.Publisher;
@@ -199,21 +201,21 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 				if (command.isWithScores()) {
 
 					result = cmd.zrangeWithScores(command.getKey(), start, stop)
-							.map(sc -> new DefaultTuple(getBytes(sc), sc.getScore()));
+							.map(this::toTuple);
 				} else {
 
 					result = cmd.zrange(command.getKey(), start, stop)
-							.map(value -> new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
+							.map(value -> toTuple(value, Double.NaN));
 				}
 			} else {
 				if (command.isWithScores()) {
 
 					result = cmd.zrevrangeWithScores(command.getKey(), start, stop)
-							.map(sc -> new DefaultTuple(getBytes(sc), sc.getScore()));
+							.map(this::toTuple);
 				} else {
 
 					result = cmd.zrevrange(command.getKey(), start, stop)
-							.map(value -> new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
+							.map(value -> toTuple(value, Double.NaN));
 				}
 			}
 
@@ -246,21 +248,21 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 
 					if (!isLimited) {
 						result = cmd.zrangebyscoreWithScores(command.getKey(), range)
-								.map(sc -> new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore()));
+								.map(this::toTuple);
 					} else {
 						result = cmd
 								.zrangebyscoreWithScores(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
-								.map(sc -> new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore()));
+								.map(this::toTuple);
 					}
 				} else {
 
 					if (!isLimited) {
 						result = cmd.zrangebyscore(command.getKey(), range)
-								.map(value -> new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
+								.map(value -> toTuple(value, Double.NaN));
 					} else {
 
 						result = cmd.zrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
-								.map(value -> new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
+								.map(value -> toTuple(value, Double.NaN));
 					}
 				}
 			} else {
@@ -271,23 +273,23 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 
 					if (!isLimited) {
 						result = cmd.zrevrangebyscoreWithScores(command.getKey(), range)
-								.map(sc -> new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore()));
+								.map(this::toTuple);
 					} else {
 
 						result = cmd
 								.zrevrangebyscoreWithScores(command.getKey(), range,
 										LettuceConverters.toLimit(command.getLimit().get()))
-								.map(sc -> new DefaultTuple(ByteUtils.getBytes(sc.getValue()), sc.getScore()));
+								.map(this::toTuple);
 					}
 				} else {
 
 					if (!isLimited) {
 						result = cmd.zrevrangebyscore(command.getKey(), range)
-								.map(value -> new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
+								.map(value -> toTuple(value, Double.NaN));
 					} else {
 
 						result = cmd.zrevrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
-								.map(value -> new DefaultTuple(ByteUtils.getBytes(value), Double.NaN));
+								.map(value -> toTuple(value, Double.NaN));
 					}
 				}
 			}
@@ -309,7 +311,7 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 			Assert.notNull(command.getOptions(), "ScanOptions must not be null!");
 
 			Flux<Tuple> result = ScanStream.zscan(cmd, command.getKey(), LettuceConverters.toScanArgs(command.getOptions()))
-					.map(it -> new DefaultTuple(ByteUtils.getBytes(it.getValue()), it.getScore()));
+					.map(this::toTuple);
 
 			return Mono.just(new CommandResponse<>(command, result));
 		}));
@@ -352,6 +354,52 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 		}));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zPop(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<ZPopCommand, Flux<Tuple>>> zPop(Publisher<ZPopCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null!");
+
+			Flux<ScoredValue<ByteBuffer>> result;
+			if (command.getCount() > 1) {
+				result = command.getDirection() == PopDirection.MIN ? cmd.zpopmin(command.getKey(), command.getCount())
+						: cmd.zpopmax(command.getKey(), command.getCount());
+			} else {
+				result = (command.getDirection() == PopDirection.MIN ? cmd.zpopmin(command.getKey())
+						: cmd.zpopmax(command.getKey())).flux();
+			}
+
+			return new CommandResponse<>(command, result.filter(Value::hasValue).map(this::toTuple));
+		}));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#bZPop(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<CommandResponse<BZPopCommand, Flux<Tuple>>> bZPop(Publisher<BZPopCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).map(command -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null!");
+			Assert.notNull(command.getTimeout(), "Timeout must not be null!");
+
+			long timeout = command.getTimeout().get(ChronoUnit.SECONDS);
+
+			Mono<ScoredValue<ByteBuffer>> result = (command.getDirection() == PopDirection.MIN
+					? cmd.bzpopmin(timeout, command.getKey())
+					: cmd.bzpopmax(timeout, command.getKey())).filter(Value::hasValue).map(Value::getValue);
+
+			return new CommandResponse<>(command, result.filter(Value::hasValue).map(this::toTuple).flux());
+		}));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveZSetCommands#zCard(org.reactivestreams.Publisher)
@@ -568,8 +616,12 @@ class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
 		return args;
 	}
 
-	private static byte[] getBytes(ScoredValue<ByteBuffer> scoredValue) {
-		return scoredValue.optional().map(ByteUtils::getBytes).orElse(new byte[0]);
+	private Tuple toTuple(ScoredValue<ByteBuffer> scoredValue) {
+		return scoredValue.map(it -> new DefaultTuple(ByteUtils.getBytes(it), scoredValue.getScore())).getValue();
+	}
+
+	private Tuple toTuple(ByteBuffer value, double score) {
+		return new DefaultTuple(ByteUtils.getBytes(value), score);
 	}
 
 	protected LettuceReactiveRedisConnection getConnection() {

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

@@ -24,6 +24,7 @@ import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
 
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.data.redis.connection.RedisZSetCommands;
 import org.springframework.data.redis.connection.RedisZSetCommands.ZAddArgs.Flag;
@@ -277,6 +278,92 @@ class LettuceZSetCommands implements RedisZSetCommands {
 				LettuceConverters.<byte[]> toRange(range, true));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(byte[])
+	 */
+	@Nullable
+	@Override
+	public Tuple zPopMin(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().from(RedisSortedSetAsyncCommands::zpopmin, key).get(LettuceConverters::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMin(byte[], long)
+	 */
+	@Nullable
+	@Override
+	public Set<Tuple> zPopMin(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().fromMany(RedisSortedSetAsyncCommands::zpopmin, key, count)
+				.toSet(LettuceConverters::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMin(byte[], long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public Tuple bZPopMin(byte[] key, long timeout, TimeUnit unit) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(unit, "TimeUnit must not be null!");
+
+		return connection.invoke(connection.getAsyncDedicatedConnection())
+				.from(RedisSortedSetAsyncCommands::bzpopmin, unit.toSeconds(timeout), key)
+				.get(it -> it.map(LettuceConverters::toTuple).getValueOrElse(null));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(byte[])
+	 */
+	@Nullable
+	@Override
+	public Tuple zPopMax(byte[] key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().from(RedisSortedSetAsyncCommands::zpopmax, key).get(LettuceConverters::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zPopMax(byte[], long)
+	 */
+	@Nullable
+	@Override
+	public Set<Tuple> zPopMax(byte[] key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return connection.invoke().fromMany(RedisSortedSetAsyncCommands::zpopmax, key, count)
+				.toSet(LettuceConverters::toTuple);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisZSetCommands#bZPopMax(byte[], long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public Tuple bZPopMax(byte[] key, long timeout, TimeUnit unit) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(unit, "TimeUnit must not be null!");
+
+		return connection.invoke(connection.getAsyncDedicatedConnection())
+				.from(RedisSortedSetAsyncCommands::bzpopmax, unit.toSeconds(timeout), key)
+				.get(it -> it.map(LettuceConverters::toTuple).getValueOrElse(null));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisZSetCommands#zCard(byte[])

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

@@ -223,7 +223,8 @@ abstract class AbstractOperations<K, V> {
 		return SerializationUtils.deserialize(rawValues, valueSerializer());
 	}
 
-	Set<TypedTuple<V>> deserializeTupleValues(Collection<Tuple> rawValues) {
+	@Nullable
+	Set<TypedTuple<V>> deserializeTupleValues(@Nullable Collection<Tuple> rawValues) {
 		if (rawValues == null) {
 			return null;
 		}
@@ -235,7 +236,11 @@ abstract class AbstractOperations<K, V> {
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	TypedTuple<V> deserializeTuple(Tuple tuple) {
+	@Nullable
+	TypedTuple<V> deserializeTuple(@Nullable Tuple tuple) {
+		if (tuple == null) {
+			return null;
+		}
 		Object value = tuple.getValue();
 		if (valueSerializer() != null) {
 			value = valueSerializer().deserialize(tuple.getValue());

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

@@ -15,9 +15,11 @@
  */
 package org.springframework.data.redis.core;
 
+import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
@@ -26,6 +28,7 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 import org.springframework.data.redis.connection.RedisZSetCommands.Weights;
 import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * ZSet (or SortedSet) operations bound to a certain key.
@@ -235,6 +238,112 @@ public interface BoundZSetOperations<K, V> extends BoundKeyOperations<K> {
 	@Nullable
 	Long lexCount(Range range);
 
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at the bound key.
+	 *
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	TypedTuple<V> popMin();
+
+	/**
+	 * Remove and return {@code count} values with their score having the lowest score from sorted set at the bound key.
+	 *
+	 * @param count number of elements to pop.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Set<TypedTuple<V>> popMin(long count);
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at the bound key. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/bzpopmin">Redis Documentation: BZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	TypedTuple<V> popMin(long timeout, TimeUnit unit);
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at the bound key. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param timeout must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @throws IllegalArgumentException if the timeout is {@literal null} or negative.
+	 * @see <a href="https://redis.io/commands/bzpopmin">Redis Documentation: BZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	default TypedTuple<V> popMin(Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return popMin(TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at the bound key.
+	 *
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	TypedTuple<V> popMax();
+
+	/**
+	 * Remove and return {@code count} values with their score having the highest score from sorted set at the bound key.
+	 *
+	 * @param count number of elements to pop.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Set<TypedTuple<V>> popMax(long count);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at the bound key. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/bzpopmax">Redis Documentation: BZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	TypedTuple<V> popMax(long timeout, TimeUnit unit);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at the bound key. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param timeout must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @throws IllegalArgumentException if the timeout is {@literal null} or negative.
+	 * @see <a href="https://redis.io/commands/bzpopmax">Redis Documentation: BZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	default TypedTuple<V> popMax(Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return popMax(TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
 	/**
 	 * Returns the number of elements of the sorted set stored with given the bound key.
 	 *

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

@@ -19,6 +19,7 @@ package org.springframework.data.redis.core;
 import java.util.Collection;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
@@ -26,6 +27,7 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
 import org.springframework.data.redis.connection.RedisZSetCommands.Range;
 import org.springframework.data.redis.connection.RedisZSetCommands.Weights;
 import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
+import org.springframework.lang.Nullable;
 
 /**
  * Default implementation for {@link BoundZSetOperations}.
@@ -322,6 +324,66 @@ class DefaultBoundZSetOperations<K, V> extends DefaultBoundKeyOperations<K> impl
 		return ops.lexCount(getKey(), range);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.BoundZSetOperations#popMin()
+	 */
+	@Nullable
+	@Override
+	public TypedTuple<V> popMin() {
+		return ops.popMin(getKey());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.BoundZSetOperations#popMin(long)
+	 */
+	@Nullable
+	@Override
+	public Set<TypedTuple<V>> popMin(long count) {
+		return ops.popMin(getKey(), count);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.BoundZSetOperations#popMin(long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public TypedTuple<V> popMin(long timeout, TimeUnit unit) {
+		return ops.popMin(getKey(), timeout, unit);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.BoundZSetOperations#popMax()
+	 */
+	@Nullable
+	@Override
+	public TypedTuple<V> popMax() {
+		return ops.popMax(getKey());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.BoundZSetOperations#popMax(long)
+	 */
+	@Nullable
+	@Override
+	public Set<TypedTuple<V>> popMax(long count) {
+		return ops.popMax(getKey(), count);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.BoundZSetOperations#popMax(long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public TypedTuple<V> popMax(long timeout, TimeUnit unit) {
+		return ops.popMax(getKey(), timeout, unit);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.BoundZSetOperations#size()

src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.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.Collection;
 import java.util.Collections;
@@ -26,6 +27,7 @@ import java.util.List;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+
 import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.DefaultTuple;
 import org.springframework.data.redis.connection.ReactiveZSetCommands;
@@ -345,6 +347,80 @@ class DefaultReactiveZSetOperations<K, V> implements ReactiveZSetOperations<K, V
 		return createMono(connection -> connection.zLexCount(rawKey(key), range));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#popMin(java.lang.Object)
+	 */
+	@Override
+	public Mono<TypedTuple<V>> popMin(K key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return createMono(connection -> connection.zPopMin(rawKey(key)).map(this::readTypedTuple));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#popMin(java.lang.Object, long)
+	 */
+	@Override
+	public Flux<TypedTuple<V>> popMin(K key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return createFlux(connection -> connection.zPopMin(rawKey(key), count).map(this::readTypedTuple));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#popMin(java.lang.Object, java.time.Duration)
+	 */
+	@Override
+	public Mono<TypedTuple<V>> popMin(K key, Duration timeout) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(timeout, "Timeout must not be null!");
+
+		return createMono(connection -> connection.bZPopMin(rawKey(key), timeout).map(this::readTypedTuple));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#popMax(java.lang.Object)
+	 */
+	@Override
+	public Mono<TypedTuple<V>> popMax(K key) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return createMono(connection -> connection.zPopMax(rawKey(key)).map(this::readTypedTuple));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#popMax(java.lang.Object, long)
+	 */
+	@Override
+	public Flux<TypedTuple<V>> popMax(K key, long count) {
+
+		Assert.notNull(key, "Key must not be null!");
+
+		return createFlux(connection -> connection.zPopMax(rawKey(key), count).map(this::readTypedTuple));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#popMax(java.lang.Object, java.time.Duration)
+	 */
+	@Override
+	public Mono<TypedTuple<V>> popMax(K key, Duration timeout) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(timeout, "Timeout must not be null!");
+
+		return createMono(connection -> connection.bZPopMax(rawKey(key), timeout).map(this::readTypedTuple));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveZSetOperations#size(java.lang.Object)

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

@@ -102,4 +102,14 @@ public class DefaultTypedTuple<V> implements TypedTuple<V> {
 
 		return compareTo(o.getScore());
 	}
+
+	@Override
+	public String toString() {
+		StringBuffer sb = new StringBuffer();
+		sb.append(getClass().getSimpleName());
+		sb.append(" [score=").append(score);
+		sb.append(", value=").append(value);
+		sb.append(']');
+		return sb.toString();
+	}
 }

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

@@ -19,6 +19,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
@@ -472,6 +473,78 @@ class DefaultZSetOperations<K, V> extends AbstractOperations<K, V> implements ZS
 		return execute(connection -> connection.zLexCount(rawKey, range), true);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ZSetOperations#popMin(java.lang.Object)
+	 */
+	@Nullable
+	@Override
+	public TypedTuple<V> popMin(K key) {
+
+		byte[] rawKey = rawKey(key);
+		return deserializeTuple(execute(connection -> connection.zPopMin(rawKey), true));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ZSetOperations#popMin(java.lang.Object, long)
+	 */
+	@Nullable
+	@Override
+	public Set<TypedTuple<V>> popMin(K key, long count) {
+
+		byte[] rawKey = rawKey(key);
+		return deserializeTupleValues(execute(connection -> connection.zPopMin(rawKey, count), true));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ZSetOperations#popMin(java.lang.Object, long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public TypedTuple<V> popMin(K key, long timeout, TimeUnit unit) {
+
+		byte[] rawKey = rawKey(key);
+		return deserializeTuple(execute(connection -> connection.bZPopMin(rawKey, timeout, unit), true));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ZSetOperations#popMax(java.lang.Object)
+	 */
+	@Nullable
+	@Override
+	public TypedTuple<V> popMax(K key) {
+
+		byte[] rawKey = rawKey(key);
+		return deserializeTuple(execute(connection -> connection.zPopMax(rawKey), true));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ZSetOperations#popMax(java.lang.Object, long)
+	 */
+	@Nullable
+	@Override
+	public Set<TypedTuple<V>> popMax(K key, long count) {
+
+		byte[] rawKey = rawKey(key);
+		return deserializeTupleValues(execute(connection -> connection.zPopMax(rawKey, count), true));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ZSetOperations#popMax(java.lang.Object, long, java.util.concurrent.TimeUnit)
+	 */
+	@Nullable
+	@Override
+	public TypedTuple<V> popMax(K key, long timeout, TimeUnit unit) {
+
+		byte[] rawKey = rawKey(key);
+		return deserializeTuple(execute(connection -> connection.bZPopMax(rawKey, timeout, unit), true));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ZSetOperations#size(java.lang.Object)

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

@@ -18,6 +18,7 @@ package org.springframework.data.redis.core;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
 
@@ -282,6 +283,74 @@ public interface ReactiveZSetOperations<K, V> {
 	 */
 	Mono<Long> lexCount(K key, Range<String> range);
 
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	Mono<TypedTuple<V>> popMin(K key);
+
+	/**
+	 * Remove and return {@code count} values with their score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	Flux<TypedTuple<V>> popMin(K key, long count);
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout maximal duration to wait until an entry in the list at {@code key} is available. Must be either
+	 *          {@link Duration#ZERO} or greater {@link 1 second}, must not be {@literal null}. A timeout of zero can be
+	 *          used to wait indefinitely. Durations between zero and one second are not supported.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	Mono<TypedTuple<V>> popMin(K key, Duration timeout);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	Mono<TypedTuple<V>> popMax(K key);
+
+	/**
+	 * Remove and return {@code count} values with their score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	Flux<TypedTuple<V>> popMax(K key, long count);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout maximal duration to wait until an entry in the list at {@code key} is available. Must be either
+	 *          {@link Duration#ZERO} or greater {@link 1 second}, must not be {@literal null}. A timeout of zero can be
+	 *          used to wait indefinitely. Durations between zero and one second are not supported.
+	 * @return
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	Mono<TypedTuple<V>> popMax(K key, Duration timeout);
+
 	/**
 	 * Returns the number of elements of the sorted set stored with given {@code key}.
 	 *

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

@@ -15,9 +15,11 @@
  */
 package org.springframework.data.redis.core;
 
+import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
@@ -25,6 +27,7 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Range;
 import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
 import org.springframework.data.redis.connection.RedisZSetCommands.Weights;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * Redis ZSet/sorted set specific operations.
@@ -338,6 +341,120 @@ public interface ZSetOperations<K, V> {
 	@Nullable
 	Long lexCount(K key, Range range);
 
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	TypedTuple<V> popMin(K key);
+
+	/**
+	 * Remove and return {@code count} values with their score having the lowest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmin">Redis Documentation: ZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Set<TypedTuple<V>> popMin(K key, long count);
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/bzpopmin">Redis Documentation: BZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	TypedTuple<V> popMin(K key, long timeout, TimeUnit unit);
+
+	/**
+	 * Remove and return the value with its score having the lowest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @throws IllegalArgumentException if the timeout is {@literal null} or negative.
+	 * @see <a href="https://redis.io/commands/bzpopmin">Redis Documentation: BZPOPMIN</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	default TypedTuple<V> popMin(K key, Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return popMin(key, TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	TypedTuple<V> popMax(K key);
+
+	/**
+	 * Remove and return {@code count} values with their score having the highest score from sorted set at {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param count number of elements to pop.
+	 * @return {@literal null} when the sorted set is empty or used in pipeline / transaction.
+	 * @see <a href="https://redis.io/commands/zpopmax">Redis Documentation: ZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	Set<TypedTuple<V>> popMax(K key, long count);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @see <a href="https://redis.io/commands/bzpopmax">Redis Documentation: BZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	TypedTuple<V> popMax(K key, long timeout, TimeUnit unit);
+
+	/**
+	 * Remove and return the value with its score having the highest score from sorted set at {@code key}. <b>Blocks
+	 * connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @throws IllegalArgumentException if the timeout is {@literal null} or negative.
+	 * @see <a href="https://redis.io/commands/bzpopmax">Redis Documentation: BZPOPMAX</a>
+	 * @since 2.6
+	 */
+	@Nullable
+	default TypedTuple<V> popMax(K key, Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return popMin(key, TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
 	/**
 	 * Returns the number of elements of the sorted set stored with given {@code key}.
 	 *