diff --git a/src/main/asciidoc/new-features.adoc b/src/main/asciidoc/new-features.adoc
index 61653e755..28d9cc7f1 100644
--- a/src/main/asciidoc/new-features.adoc
+++ b/src/main/asciidoc/new-features.adoc
@@ -7,7 +7,7 @@ This section briefly covers items that are new and noteworthy in the latest rele
 == New in Spring Data Redis 2.6
 
 * Support for `SubscriptionListener` when using `MessageListener` for subscription confirmation callbacks. `ReactiveRedisMessageListenerContainer` and `ReactiveRedisOperations` provide `receiveLater(…)` and `listenToLater(…)` methods to await until Redis acknowledges the subscription.
-* Support Redis 6.2 commands (`LPOP`/`RPOP` with `count`, `COPY`, `GETEX`, `GETDEL`, `ZMSCORE`).
+* Support Redis 6.2 commands (`LPOP`/`RPOP` with `count`, `COPY`, `GETEX`, `GETDEL`, `ZPOPMIN`, `BZPOPMIN`, `ZPOPMAX`, `BZPOPMAX`, `ZMSCORE`).
 
 [[new-in-2.5.0]]
 == New in Spring Data Redis 2.5
diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
index aeb5ba896..063975165 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
+++ b/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)
diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultTuple.java b/src/main/java/org/springframework/data/redis/connection/DefaultTuple.java
index 52dd42629..640c5bab5 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultTuple.java
+++ b/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();
+	}
 }
diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java
index e4df1a5ab..cf51c4141 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java
+++ b/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
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java
index 393846310..7be989992 100644
--- a/src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java
+++ b/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
diff --git a/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java
index 08eaf9739..ddb208a89 100644
--- a/src/main/java/org/springframework/data/redis/connection/RedisZSetCommands.java
+++ b/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}.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java
index df3eee406..469b2074d 100644
--- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java
+++ b/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}.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterZSetCommands.java
index 15480d38a..3334a45dc 100644
--- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterZSetCommands.java
+++ b/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))));
+	}
+
 }
diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java
index 54a129eb5..a90cac7f2 100644
--- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConverters.java
+++ b/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.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java
index f4a401cbb..338b5c173 100644
--- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java
+++ b/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))));
+	}
+
 }
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java
index aa86703ee..f986127d2 100644
--- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java
+++ b/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) {
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java
index af24ccf5a..6ffb24897 100644
--- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java
+++ b/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() {
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java
index 40b9178c9..287a81b25 100644
--- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceZSetCommands.java
+++ b/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[])
diff --git a/src/main/java/org/springframework/data/redis/core/AbstractOperations.java b/src/main/java/org/springframework/data/redis/core/AbstractOperations.java
index 53048d9b4..4cd8a942d 100644
--- a/src/main/java/org/springframework/data/redis/core/AbstractOperations.java
+++ b/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());
diff --git a/src/main/java/org/springframework/data/redis/core/BoundZSetOperations.java b/src/main/java/org/springframework/data/redis/core/BoundZSetOperations.java
index 213aa53cd..82c694af5 100644
--- a/src/main/java/org/springframework/data/redis/core/BoundZSetOperations.java
+++ b/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.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultBoundZSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultBoundZSetOperations.java
index ede279583..848c4318f 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultBoundZSetOperations.java
+++ b/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()
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java
index b7f03020f..234320b1e 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveZSetOperations.java
+++ b/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)
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultTypedTuple.java b/src/main/java/org/springframework/data/redis/core/DefaultTypedTuple.java
index bf3ecda2c..4bf8a9778 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultTypedTuple.java
+++ b/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();
+	}
 }
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultZSetOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultZSetOperations.java
index 28e3ee82d..c905088b6 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultZSetOperations.java
+++ b/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)
diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java
index a476ffe1c..32d994618 100644
--- a/src/main/java/org/springframework/data/redis/core/ReactiveZSetOperations.java
+++ b/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}.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/core/ZSetOperations.java b/src/main/java/org/springframework/data/redis/core/ZSetOperations.java
index 0c209e41e..1414ea78c 100644
--- a/src/main/java/org/springframework/data/redis/core/ZSetOperations.java
+++ b/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}.
 	 *
diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java
index 541e90c58..94abb4a10 100644
--- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java
+++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java
@@ -1830,6 +1830,52 @@ public abstract class AbstractConnectionIntegrationTests {
 		assertThat((Long) results.get(11)).isEqualTo(3);
 	}
 
+	@Test // GH-2007
+	@EnabledOnCommand("ZPOPMIN")
+	void zPopMin() {
+
+		actual.add(connection.zAdd("myzset", 1, "a"));
+		actual.add(connection.zAdd("myzset", 2, "b"));
+		actual.add(connection.zAdd("myzset", 3, "c"));
+		actual.add(connection.zAdd("myzset", 4, "d"));
+
+		actual.add(connection.zPopMin("myzset"));
+		actual.add(connection.bZPopMin("myzset", 1, TimeUnit.SECONDS));
+		actual.add(connection.zPopMin("myzset", 2));
+		actual.add(connection.zPopMin("myzset"));
+
+		List<Object> results = getResults();
+
+		assertThat(results.get(4)).isEqualTo(new DefaultStringTuple("a".getBytes(), "a", 1D));
+		assertThat(results.get(5)).isEqualTo(new DefaultStringTuple("b".getBytes(), "b", 2D));
+		assertThat((Collection) results.get(6)).containsExactly(new DefaultStringTuple("c".getBytes(), "c", 3D),
+				new DefaultStringTuple("d".getBytes(), "d", 4D));
+		assertThat(results.get(7)).isNull();
+	}
+
+	@Test // GH-2007
+	@EnabledOnCommand("ZPOPMAX")
+	void zPopMax() {
+
+		actual.add(connection.zAdd("myzset", 1, "a"));
+		actual.add(connection.zAdd("myzset", 2, "b"));
+		actual.add(connection.zAdd("myzset", 3, "c"));
+		actual.add(connection.zAdd("myzset", 4, "d"));
+
+		actual.add(connection.zPopMax("myzset"));
+		actual.add(connection.bZPopMax("myzset", 1, TimeUnit.SECONDS));
+		actual.add(connection.zPopMax("myzset", 2));
+		actual.add(connection.zPopMax("myzset"));
+
+		List<Object> results = getResults();
+
+		assertThat(results.get(4)).isEqualTo(new DefaultStringTuple("d".getBytes(), "d", 4D));
+		assertThat(results.get(5)).isEqualTo(new DefaultStringTuple("c".getBytes(), "c", 3D));
+		assertThat((Collection) results.get(6)).containsExactly(new DefaultStringTuple("b".getBytes(), "b", 2D),
+				new DefaultStringTuple("a".getBytes(), "a", 1D));
+		assertThat(results.get(7)).isNull();
+	}
+
 	@Test
 	void testZIncrBy() {
 		actual.add(connection.zAdd("myset", 2, "Bob"));
diff --git a/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java
index 08a0bd848..661fc4fd4 100644
--- a/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java
+++ b/src/test/java/org/springframework/data/redis/connection/ClusterConnectionTests.java
@@ -610,6 +610,18 @@ public interface ClusterConnectionTests {
 	// DATAREDIS-315
 	void zInterStoreShouldWorkForSameSlotKeys();
 
+	// GH-2007
+	void zPopMinShouldWorkCorrectly();
+
+	// GH-2007
+	void bzPopMinShouldWorkCorrectly();
+
+	// GH-2007
+	void zPopMaxShouldWorkCorrectly();
+
+	// GH-2007
+	void bzPopMaxShouldWorkCorrectly();
+
 	// DATAREDIS-315
 	void zRangeByLexShouldReturnResultCorrectly();
 
diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java
index f4d625eb3..b14727033 100644
--- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java
+++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisClusterConnectionTests.java
@@ -2026,6 +2026,56 @@ public class JedisClusterConnectionTests implements ClusterConnectionTests {
 		assertThat(nativeConnection.zrange(SAME_SLOT_KEY_3_BYTES, 0, -1)).contains(VALUE_2_BYTES);
 	}
 
+	@Test // GH-2007
+	@EnabledOnCommand("ZPOPMIN")
+	public void zPopMinShouldWorkCorrectly() {
+
+		nativeConnection.zadd(KEY_1_BYTES, 10D, VALUE_1_BYTES);
+		nativeConnection.zadd(KEY_1_BYTES, 20D, VALUE_2_BYTES);
+		nativeConnection.zadd(KEY_1_BYTES, 30D, VALUE_3_BYTES);
+
+		assertThat(clusterConnection.zPopMin(KEY_1_BYTES)).isEqualTo(new DefaultTuple(VALUE_1_BYTES, 10D));
+		assertThat(clusterConnection.zPopMin(KEY_1_BYTES, 2)).containsExactly(new DefaultTuple(VALUE_2_BYTES, 20D),
+				new DefaultTuple(VALUE_3_BYTES, 30D));
+	}
+
+	@Test // GH-2007
+	@EnabledOnCommand("BZPOPMIN")
+	public void bzPopMinShouldWorkCorrectly() {
+
+		nativeConnection.zadd(KEY_1_BYTES, 10D, VALUE_1_BYTES);
+		nativeConnection.zadd(KEY_1_BYTES, 20D, VALUE_2_BYTES);
+		nativeConnection.zadd(KEY_1_BYTES, 30D, VALUE_3_BYTES);
+
+		assertThat(clusterConnection.bZPopMin(KEY_1_BYTES, 1, TimeUnit.SECONDS))
+				.isEqualTo(new DefaultTuple(VALUE_1_BYTES, 10D));
+	}
+
+	@Test // GH-2007
+	@EnabledOnCommand("ZPOPMAX")
+	public void zPopMaxShouldWorkCorrectly() {
+
+		nativeConnection.zadd(KEY_1_BYTES, 10D, VALUE_1_BYTES);
+		nativeConnection.zadd(KEY_1_BYTES, 20D, VALUE_2_BYTES);
+		nativeConnection.zadd(KEY_1_BYTES, 30D, VALUE_3_BYTES);
+
+		assertThat(clusterConnection.zPopMax(KEY_1_BYTES)).isEqualTo(new DefaultTuple(VALUE_3_BYTES, 30D));
+		assertThat(clusterConnection.zPopMax(KEY_1_BYTES, 2)).containsExactly(new DefaultTuple(VALUE_2_BYTES, 20D),
+				new DefaultTuple(VALUE_1_BYTES, 10D));
+	}
+
+	@Test // GH-2007
+	@EnabledOnCommand("BZPOPMAX")
+	public void bzPopMaxShouldWorkCorrectly() {
+
+		nativeConnection.zadd(KEY_1_BYTES, 10D, VALUE_1_BYTES);
+		nativeConnection.zadd(KEY_1_BYTES, 20D, VALUE_2_BYTES);
+		nativeConnection.zadd(KEY_1_BYTES, 30D, VALUE_3_BYTES);
+
+		assertThat(clusterConnection.bZPopMax(KEY_1_BYTES, 1, TimeUnit.SECONDS))
+				.isEqualTo(new DefaultTuple(VALUE_3_BYTES, 30D));
+	}
+
 	@Test // DATAREDIS-315
 	public void zRangeByLexShouldReturnResultCorrectly() {
 
diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java
index 103b835a6..d6c098fea 100644
--- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java
+++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java
@@ -2067,6 +2067,56 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
 		assertThat(nativeConnection.zrange(SAME_SLOT_KEY_3, 0, -1)).contains(VALUE_2);
 	}
 
+	@Test // GH-2007
+	@EnabledOnCommand("ZPOPMIN")
+	public void zPopMinShouldWorkCorrectly() {
+
+		nativeConnection.zadd(KEY_1, 10D, VALUE_1);
+		nativeConnection.zadd(KEY_1, 20D, VALUE_2);
+		nativeConnection.zadd(KEY_1, 30D, VALUE_3);
+
+		assertThat(clusterConnection.zPopMin(KEY_1_BYTES)).isEqualTo(new DefaultTuple(VALUE_1_BYTES, 10D));
+		assertThat(clusterConnection.zPopMin(KEY_1_BYTES, 2)).containsExactly(new DefaultTuple(VALUE_2_BYTES, 20D),
+				new DefaultTuple(VALUE_3_BYTES, 30D));
+	}
+
+	@Test // GH-2007
+	@EnabledOnCommand("BZPOPMIN")
+	public void bzPopMinShouldWorkCorrectly() {
+
+		nativeConnection.zadd(KEY_1, 10D, VALUE_1);
+		nativeConnection.zadd(KEY_1, 20D, VALUE_2);
+		nativeConnection.zadd(KEY_1, 30D, VALUE_3);
+
+		assertThat(clusterConnection.bZPopMin(KEY_1_BYTES, 1, TimeUnit.SECONDS))
+				.isEqualTo(new DefaultTuple(VALUE_1_BYTES, 10D));
+	}
+
+	@Test // GH-2007
+	@EnabledOnCommand("ZPOPMAX")
+	public void zPopMaxShouldWorkCorrectly() {
+
+		nativeConnection.zadd(KEY_1, 10D, VALUE_1);
+		nativeConnection.zadd(KEY_1, 20D, VALUE_2);
+		nativeConnection.zadd(KEY_1, 30D, VALUE_3);
+
+		assertThat(clusterConnection.zPopMax(KEY_1_BYTES)).isEqualTo(new DefaultTuple(VALUE_3_BYTES, 30D));
+		assertThat(clusterConnection.zPopMax(KEY_1_BYTES, 2)).containsExactly(new DefaultTuple(VALUE_2_BYTES, 20D),
+				new DefaultTuple(VALUE_1_BYTES, 10D));
+	}
+
+	@Test // GH-2007
+	@EnabledOnCommand("BZPOPMAX")
+	public void bzPopMaxShouldWorkCorrectly() {
+
+		nativeConnection.zadd(KEY_1, 10D, VALUE_1);
+		nativeConnection.zadd(KEY_1, 20D, VALUE_2);
+		nativeConnection.zadd(KEY_1, 30D, VALUE_3);
+
+		assertThat(clusterConnection.bZPopMax(KEY_1_BYTES, 1, TimeUnit.SECONDS))
+				.isEqualTo(new DefaultTuple(VALUE_3_BYTES, 30D));
+	}
+
 	@Test // DATAREDIS-315
 	public void zRangeByLexShouldReturnResultCorrectly() {
 
diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsIntegrationTests.java
index 3ef98d713..9609c9870 100644
--- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsIntegrationTests.java
+++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsIntegrationTests.java
@@ -22,6 +22,7 @@ import static org.springframework.data.domain.Range.Bound.*;
 import reactor.test.StepVerifier;
 
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.Arrays;
 
 import org.springframework.data.domain.Range;
@@ -385,6 +386,62 @@ public class LettuceReactiveZSetCommandsIntegrationTests extends LettuceReactive
 				.isEqualTo(2L);
 	}
 
+	@ParameterizedRedisTest // GH-2007
+	@EnabledOnCommand("ZPOPMIN")
+	void zPopMinShouldReturnCorrectly() {
+
+		nativeCommands.zadd(KEY_1, 1D, VALUE_1);
+		nativeCommands.zadd(KEY_1, 2D, VALUE_2);
+		nativeCommands.zadd(KEY_1, 3D, VALUE_3);
+
+		connection.zSetCommands().zPopMin(KEY_1_BBUFFER).as(StepVerifier::create)
+				.expectNext(new DefaultTuple(VALUE_1_BYTES, 1D)).verifyComplete();
+
+		connection.zSetCommands().zPopMin(KEY_1_BBUFFER, 2).as(StepVerifier::create)
+				.expectNext(new DefaultTuple(VALUE_2_BYTES, 2D)).expectNext(new DefaultTuple(VALUE_3_BYTES, 3D))
+				.verifyComplete();
+	}
+
+	@ParameterizedRedisTest // GH-2007
+	@EnabledOnCommand("BZPOPMIN")
+	void bzPopMinShouldReturnCorrectly() {
+
+		nativeCommands.zadd(KEY_1, 1D, VALUE_1);
+		nativeCommands.zadd(KEY_1, 2D, VALUE_2);
+		nativeCommands.zadd(KEY_1, 3D, VALUE_3);
+
+		connection.zSetCommands().bZPopMin(KEY_1_BBUFFER, Duration.ofSeconds(1)).as(StepVerifier::create)
+				.expectNext(new DefaultTuple(VALUE_1_BYTES, 1D)).verifyComplete();
+	}
+
+	@ParameterizedRedisTest // GH-2007
+	@EnabledOnCommand("ZPOPMAX")
+	void zPopMaxShouldReturnCorrectly() {
+
+		nativeCommands.zadd(KEY_1, 1D, VALUE_1);
+		nativeCommands.zadd(KEY_1, 2D, VALUE_2);
+		nativeCommands.zadd(KEY_1, 3D, VALUE_3);
+
+		connection.zSetCommands().zPopMax(KEY_1_BBUFFER).as(StepVerifier::create)
+				.expectNext(new DefaultTuple(VALUE_3_BYTES, 3D)).verifyComplete();
+
+		connection.zSetCommands().zPopMax(KEY_1_BBUFFER, 2).as(StepVerifier::create)
+				.expectNext(new DefaultTuple(VALUE_2_BYTES, 2D)).expectNext(new DefaultTuple(VALUE_1_BYTES, 1D))
+				.verifyComplete();
+	}
+
+	@ParameterizedRedisTest // GH-2007
+	@EnabledOnCommand("BZPOPMAX")
+	void bzPopMaxShouldReturnCorrectly() {
+
+		nativeCommands.zadd(KEY_1, 1D, VALUE_1);
+		nativeCommands.zadd(KEY_1, 2D, VALUE_2);
+		nativeCommands.zadd(KEY_1, 3D, VALUE_3);
+
+		connection.zSetCommands().bZPopMax(KEY_1_BBUFFER, Duration.ofSeconds(1)).as(StepVerifier::create)
+				.expectNext(new DefaultTuple(VALUE_3_BYTES, 3D)).verifyComplete();
+	}
+
 	@ParameterizedRedisTest // DATAREDIS-525
 	void zCardShouldReturnSizeCorrectly() {
 
diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java
index 34ffea7f7..6c2d00e87 100644
--- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java
+++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveZSetOperationsIntegrationTests.java
@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assumptions.*;
 
 import reactor.test.StepVerifier;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -434,6 +435,52 @@ public class DefaultReactiveZSetOperationsIntegrationTests<K, V> {
 		zSetOperations.lexCount(key, Range.rightOpen("b", "f")).as(StepVerifier::create).expectNext(4L).verifyComplete();
 	}
 
+	@ParameterizedRedisTest // GH-2007
+	@EnabledOnCommand("ZPOPMIN")
+	void popMin() {
+
+		K key = keyFactory.instance();
+		V value1 = valueFactory.instance();
+		V value2 = valueFactory.instance();
+		V value3 = valueFactory.instance();
+		V value4 = valueFactory.instance();
+
+		zSetOperations.add(key, value1, 1).as(StepVerifier::create).expectNext(true).verifyComplete();
+		zSetOperations.add(key, value2, 2).as(StepVerifier::create).expectNext(true).verifyComplete();
+		zSetOperations.add(key, value3, 3).as(StepVerifier::create).expectNext(true).verifyComplete();
+		zSetOperations.add(key, value4, 4).as(StepVerifier::create).expectNext(true).verifyComplete();
+
+		zSetOperations.popMin(key).as(StepVerifier::create).expectNext(new DefaultTypedTuple<>(value1, 1D))
+				.verifyComplete();
+		zSetOperations.popMin(key, Duration.ofSeconds(1)).as(StepVerifier::create)
+				.expectNext(new DefaultTypedTuple<>(value2, 2D)).verifyComplete();
+		zSetOperations.popMin(key, 2).as(StepVerifier::create).expectNext(new DefaultTypedTuple<>(value3, 3D))
+				.expectNext(new DefaultTypedTuple<>(value4, 4D)).verifyComplete();
+	}
+
+	@ParameterizedRedisTest // GH-2007
+	@EnabledOnCommand("ZPOPMAX")
+	void popMax() {
+
+		K key = keyFactory.instance();
+		V value1 = valueFactory.instance();
+		V value2 = valueFactory.instance();
+		V value3 = valueFactory.instance();
+		V value4 = valueFactory.instance();
+
+		zSetOperations.add(key, value1, 1).as(StepVerifier::create).expectNext(true).verifyComplete();
+		zSetOperations.add(key, value2, 2).as(StepVerifier::create).expectNext(true).verifyComplete();
+		zSetOperations.add(key, value3, 3).as(StepVerifier::create).expectNext(true).verifyComplete();
+		zSetOperations.add(key, value4, 4).as(StepVerifier::create).expectNext(true).verifyComplete();
+
+		zSetOperations.popMax(key).as(StepVerifier::create).expectNext(new DefaultTypedTuple<>(value4, 4D))
+				.verifyComplete();
+		zSetOperations.popMax(key, Duration.ofSeconds(1)).as(StepVerifier::create)
+				.expectNext(new DefaultTypedTuple<>(value3, 3D)).verifyComplete();
+		zSetOperations.popMax(key, 2).as(StepVerifier::create).expectNext(new DefaultTypedTuple<>(value2, 2D))
+				.expectNext(new DefaultTypedTuple<>(value1, 1D)).verifyComplete();
+	}
+
 	@ParameterizedRedisTest // DATAREDIS-602
 	void size() {
 
diff --git a/src/test/java/org/springframework/data/redis/core/DefaultZSetOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultZSetOperationsIntegrationTests.java
index 5aa157eaa..62f33ef7b 100644
--- a/src/test/java/org/springframework/data/redis/core/DefaultZSetOperationsIntegrationTests.java
+++ b/src/test/java/org/springframework/data/redis/core/DefaultZSetOperationsIntegrationTests.java
@@ -24,6 +24,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.BeforeEach;
 
@@ -35,6 +36,7 @@ import org.springframework.data.redis.ObjectFactory;
 import org.springframework.data.redis.connection.RedisZSetCommands;
 import org.springframework.data.redis.connection.RedisZSetCommands.Weights;
 import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
+import org.springframework.data.redis.test.condition.EnabledOnCommand;
 import org.springframework.data.redis.test.extension.parametrized.MethodSource;
 import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest;
 
@@ -128,6 +130,48 @@ public class DefaultZSetOperationsIntegrationTests<K, V> {
 		assertThat(zSetOps.lexCount(key, RedisZSetCommands.Range.range().gt(value1))).isEqualTo(2);
 	}
 
+	@ParameterizedRedisTest // GH-2007
+	@EnabledOnCommand("ZPOPMIN")
+	void testPopMin() {
+
+		K key = keyFactory.instance();
+		V value1 = valueFactory.instance();
+		V value2 = valueFactory.instance();
+		V value3 = valueFactory.instance();
+		V value4 = valueFactory.instance();
+
+		zSetOps.add(key, value1, 1);
+		zSetOps.add(key, value2, 2);
+		zSetOps.add(key, value3, 3);
+		zSetOps.add(key, value4, 4);
+
+		assertThat(zSetOps.popMin(key)).isEqualTo(new DefaultTypedTuple<>(value1, 1d));
+		assertThat(zSetOps.popMin(key, 2)).containsExactly(new DefaultTypedTuple<>(value2, 2d),
+				new DefaultTypedTuple<>(value3, 3d));
+		assertThat(zSetOps.popMin(key, 1, TimeUnit.SECONDS)).isEqualTo(new DefaultTypedTuple<>(value4, 4d));
+	}
+
+	@ParameterizedRedisTest // GH-2007
+	@EnabledOnCommand("ZPOPMAX")
+	void testPopMax() {
+
+		K key = keyFactory.instance();
+		V value1 = valueFactory.instance();
+		V value2 = valueFactory.instance();
+		V value3 = valueFactory.instance();
+		V value4 = valueFactory.instance();
+
+		zSetOps.add(key, value1, 1);
+		zSetOps.add(key, value2, 2);
+		zSetOps.add(key, value3, 3);
+		zSetOps.add(key, value4, 4);
+
+		assertThat(zSetOps.popMax(key)).isEqualTo(new DefaultTypedTuple<>(value4, 4d));
+		assertThat(zSetOps.popMax(key, 2)).containsExactly(new DefaultTypedTuple<>(value3, 3d),
+				new DefaultTypedTuple<>(value2, 2d));
+		assertThat(zSetOps.popMax(key, 1, TimeUnit.SECONDS)).isEqualTo(new DefaultTypedTuple<>(value1, 1d));
+	}
+
 	@ParameterizedRedisTest
 	void testIncrementScore() {