diff --git a/src/main/java/org/springframework/data/redis/core/BoundKeyOperations.java b/src/main/java/org/springframework/data/redis/core/BoundKeyOperations.java
index 438546530..bf0cb1a99 100644
--- a/src/main/java/org/springframework/data/redis/core/BoundKeyOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/BoundKeyOperations.java
@@ -15,11 +15,14 @@
  */
 package org.springframework.data.redis.core;
 
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Date;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * Operations over a Redis key. Useful for executing common key-'bound' operations to all implementations.
@@ -57,6 +60,22 @@ public interface BoundKeyOperations<K> {
 	@Nullable
 	Long getExpire();
 
+	/**
+	 * Sets the key time-to-live/expiration.
+	 *
+	 * @param timeout must not be {@literal null}.
+	 * @return true if expiration was set, false otherwise. {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	@Nullable
+	default Boolean expire(Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return expire(timeout.toMillis(), TimeUnit.MILLISECONDS);
+	}
+
 	/**
 	 * Sets the key time-to-live/expiration.
 	 *
@@ -76,6 +95,21 @@ public interface BoundKeyOperations<K> {
 	@Nullable
 	Boolean expireAt(Date date);
 
+	/**
+	 * Sets the key time-to-live/expiration.
+	 *
+	 * @param time expiration time.
+	 * @return true if expiration was set, false otherwise. {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	@Nullable
+	default Boolean expireAt(Instant expireAt) {
+
+		Assert.notNull(expireAt, "Timestamp must not be null");
+
+		return expireAt(Date.from(expireAt));
+	}
+
 	/**
 	 * Removes the expiration (if any) of the key.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/core/BoundListOperations.java b/src/main/java/org/springframework/data/redis/core/BoundListOperations.java
index 89f6c7c08..1ec131dbb 100644
--- a/src/main/java/org/springframework/data/redis/core/BoundListOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/BoundListOperations.java
@@ -15,10 +15,12 @@
  */
 package org.springframework.data.redis.core;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * List operations bound to a certain key.
@@ -188,6 +190,24 @@ public interface BoundListOperations<K, V> extends BoundKeyOperations<K> {
 	@Nullable
 	V leftPop(long timeout, TimeUnit unit);
 
+	/**
+	 * Removes and returns first element from lists stored at the bound key . <br>
+	 * <b>Blocks connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param timeout must not be {@literal null}.
+	 * @return {@literal null} when timeout reached or used in pipeline / transaction.
+	 * @since 2.3
+	 * @see <a href="https://redis.io/commands/blpop">Redis Documentation: BLPOP</a>
+	 */
+	@Nullable
+	default V leftPop(Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return leftPop(TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
 	/**
 	 * Removes and returns last element in list stored at the bound key.
 	 *
@@ -209,5 +229,23 @@ public interface BoundListOperations<K, V> extends BoundKeyOperations<K> {
 	@Nullable
 	V rightPop(long timeout, TimeUnit unit);
 
+	/**
+	 * Removes and returns last element from lists stored at the bound key. <br>
+	 * <b>Blocks connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param timeout must not be {@literal null}.
+	 * @return {@literal null} when timeout reached or used in pipeline / transaction.
+	 * @since 2.3
+	 * @see <a href="https://redis.io/commands/brpop">Redis Documentation: BRPOP</a>
+	 */
+	@Nullable
+	default V rightPop(Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return rightPop(TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
 	RedisOperations<K, V> getOperations();
 }
diff --git a/src/main/java/org/springframework/data/redis/core/ListOperations.java b/src/main/java/org/springframework/data/redis/core/ListOperations.java
index 94d5eb833..4e0a7b166 100644
--- a/src/main/java/org/springframework/data/redis/core/ListOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/ListOperations.java
@@ -15,11 +15,13 @@
  */
 package org.springframework.data.redis.core;
 
+import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * Redis list specific operations.
@@ -234,6 +236,25 @@ public interface ListOperations<K, V> {
 	@Nullable
 	V leftPop(K key, long timeout, TimeUnit unit);
 
+	/**
+	 * Removes and returns first element from lists stored at {@code key} . <br>
+	 * <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}.
+	 * @since 2.3
+	 * @see <a href="https://redis.io/commands/blpop">Redis Documentation: BLPOP</a>
+	 */
+	@Nullable
+	default V leftPop(K key, Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return leftPop(key, TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
 	/**
 	 * Removes and returns last element in list stored at {@code key}.
 	 *
@@ -257,6 +278,25 @@ public interface ListOperations<K, V> {
 	@Nullable
 	V rightPop(K key, long timeout, TimeUnit unit);
 
+	/**
+	 * Removes and returns last element from lists stored at {@code key}. <br>
+	 * <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}.
+	 * @since 2.3
+	 * @see <a href="https://redis.io/commands/brpop">Redis Documentation: BRPOP</a>
+	 */
+	@Nullable
+	default V rightPop(K key, Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return rightPop(key, TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
 	/**
 	 * Remove the last element from list at {@code sourceKey}, append it to {@code destinationKey} and return its value.
 	 *
@@ -282,5 +322,25 @@ public interface ListOperations<K, V> {
 	@Nullable
 	V rightPopAndLeftPush(K sourceKey, K destinationKey, long timeout, TimeUnit unit);
 
+	/**
+	 * Remove the last element from list at {@code srcKey}, append it to {@code dstKey} and return its value.<br>
+	 * <b>Blocks connection</b> until element available or {@code timeout} reached.
+	 *
+	 * @param sourceKey must not be {@literal null}.
+	 * @param destinationKey must not be {@literal null}.
+	 * @param timeout must not be {@literal null}.
+	 * @return can be {@literal null}.
+	 * @since 2.3
+	 * @see <a href="https://redis.io/commands/brpoplpush">Redis Documentation: BRPOPLPUSH</a>
+	 */
+	@Nullable
+	default V rightPopAndLeftPush(K sourceKey, K destinationKey, Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return rightPopAndLeftPush(sourceKey, destinationKey, TimeoutUtils.toSeconds(timeout), TimeUnit.SECONDS);
+	}
+
 	RedisOperations<K, V> getOperations();
 }
diff --git a/src/main/java/org/springframework/data/redis/core/RedisOperations.java b/src/main/java/org/springframework/data/redis/core/RedisOperations.java
index c3a26786a..af2d3256b 100644
--- a/src/main/java/org/springframework/data/redis/core/RedisOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/RedisOperations.java
@@ -16,6 +16,8 @@
 package org.springframework.data.redis.core;
 
 import java.io.Closeable;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Collection;
 import java.util.Date;
 import java.util.List;
@@ -31,6 +33,7 @@ import org.springframework.data.redis.core.types.RedisClientInfo;
 import org.springframework.data.redis.hash.HashMapper;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * Interface that specified a basic set of Redis operations, implemented by {@link RedisTemplate}. Not often used but a
@@ -270,7 +273,7 @@ public interface RedisOperations<K, V> {
 	Boolean renameIfAbsent(K oldKey, K newKey);
 
 	/**
-	 * Set time to live for given {@code key}..
+	 * Set time to live for given {@code key}.
 	 *
 	 * @param key must not be {@literal null}.
 	 * @param timeout
@@ -280,6 +283,24 @@ public interface RedisOperations<K, V> {
 	@Nullable
 	Boolean expire(K key, long timeout, TimeUnit unit);
 
+	/**
+	 * Set time to live for given {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param timeout must not be {@literal null}.
+	 * @return {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	@Nullable
+	default Boolean expire(K key, Duration timeout) {
+
+		Assert.notNull(timeout, "Timeout must not be null");
+		Assert.isTrue(!timeout.isNegative(), "Timeout must not be negative");
+
+		return TimeoutUtils.hasMillis(timeout) ? expire(key, timeout.toMillis(), TimeUnit.MILLISECONDS)
+				: expire(key, timeout.getSeconds(), TimeUnit.SECONDS);
+	}
+
 	/**
 	 * Set the expiration for given {@code key} as a {@literal date} timestamp.
 	 *
@@ -290,6 +311,22 @@ public interface RedisOperations<K, V> {
 	@Nullable
 	Boolean expireAt(K key, Date date);
 
+	/**
+	 * Set the expiration for given {@code key} as a {@literal date} timestamp.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param expireAt must not be {@literal null}.
+	 * @return {@literal null} when used in pipeline / transaction.
+	 * @since 2.3
+	 */
+	@Nullable
+	default Boolean expireAt(K key, Instant expireAt) {
+
+		Assert.notNull(expireAt, "Timestamp must not be null");
+
+		return expireAt(key, Date.from(expireAt));
+	}
+
 	/**
 	 * Remove the expiration from given {@code key}.
 	 *
@@ -574,7 +611,7 @@ public interface RedisOperations<K, V> {
 
 	/**
 	 * Returns the operations performed on hash values bound to the given key. * @param <HK> hash key (or field) type
-	 * 
+	 *
 	 * @param <HV> hash value type
 	 * @param key Redis key
 	 * @return hash operations bound to the given key.
diff --git a/src/main/java/org/springframework/data/redis/core/TimeoutUtils.java b/src/main/java/org/springframework/data/redis/core/TimeoutUtils.java
index 5f29d504f..dd5463390 100644
--- a/src/main/java/org/springframework/data/redis/core/TimeoutUtils.java
+++ b/src/main/java/org/springframework/data/redis/core/TimeoutUtils.java
@@ -25,7 +25,7 @@ import java.util.concurrent.TimeUnit;
  * @author Mark Paluch
  * @author Christoph Strobl
  */
-abstract public class TimeoutUtils {
+public abstract class TimeoutUtils {
 
 	/**
 	 * Check if a given Duration can be represented in {@code sec} or requires {@code msec} representation.
@@ -38,6 +38,20 @@ abstract public class TimeoutUtils {
 		return duration.toMillis() % 1000 != 0;
 	}
 
+	/**
+	 * Converts the given timeout to seconds.
+	 * <p>
+	 * Since a 0 timeout blocks some Redis ops indefinitely, this method will return 1 if the original value is greater
+	 * than 0 but is truncated to 0 on conversion.
+	 *
+	 * @param duration The duration to convert
+	 * @return The converted timeout
+	 * @since 2.3
+	 */
+	public static long toSeconds(Duration duration) {
+		return roundUpIfNecessary(duration.toMillis(), duration.getSeconds());
+	}
+
 	/**
 	 * Converts the given timeout to seconds.
 	 * <p>
diff --git a/src/test/java/org/springframework/data/redis/core/DefaultListOperationsTests.java b/src/test/java/org/springframework/data/redis/core/DefaultListOperationsTests.java
index 58feef079..68e75a2b9 100644
--- a/src/test/java/org/springframework/data/redis/core/DefaultListOperationsTests.java
+++ b/src/test/java/org/springframework/data/redis/core/DefaultListOperationsTests.java
@@ -18,6 +18,7 @@ package org.springframework.data.redis.core;
 import static org.assertj.core.api.Assertions.*;
 import static org.junit.Assume.*;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -34,6 +35,7 @@ import org.junit.runners.Parameterized.Parameters;
 import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.ObjectFactory;
 import org.springframework.data.redis.RedisTestProfileValueSource;
+import org.springframework.data.redis.StringObjectFactory;
 
 /**
  * Integration test of {@link DefaultListOperations}
@@ -129,10 +131,41 @@ public class DefaultListOperationsTests<K, V> {
 		assertThat(listOps.range(key, 0, -1)).contains(v3, v2, v1);
 	}
 
+	@Test
+	public void testLeftPopDuration() {
+		// 1 ms timeout gets upgraded to 1 sec timeout at the moment
+		assumeTrue(RedisTestProfileValueSource.matches("runLongTests", "true"));
+		assumeTrue(valueFactory instanceof StringObjectFactory);
+
+		K key = keyFactory.instance();
+		V v1 = valueFactory.instance();
+		V v2 = valueFactory.instance();
+
+		assertThat(listOps.leftPop(key, Duration.ofSeconds(1))).isNull();
+		listOps.rightPushAll(key, v1, v2);
+		assertThat(listOps.leftPop(key, Duration.ofSeconds(1))).isEqualTo(v1);
+	}
+
+	@Test
+	public void testRightPopDuration() {
+		// 1 ms timeout gets upgraded to 1 sec timeout at the moment
+		assumeTrue(RedisTestProfileValueSource.matches("runLongTests", "true"));
+		assumeTrue(valueFactory instanceof StringObjectFactory);
+
+		K key = keyFactory.instance();
+		V v1 = valueFactory.instance();
+		V v2 = valueFactory.instance();
+
+		assertThat(listOps.rightPop(key, Duration.ofSeconds(1))).isNull();
+		listOps.rightPushAll(key, v1, v2);
+		assertThat(listOps.rightPop(key, Duration.ofSeconds(1))).isEqualTo(v2);
+	}
+
 	@Test
 	public void testRightPopAndLeftPushTimeout() {
 		// 1 ms timeout gets upgraded to 1 sec timeout at the moment
 		assumeTrue(RedisTestProfileValueSource.matches("runLongTests", "true"));
+		assumeTrue(valueFactory instanceof StringObjectFactory);
 
 		K key = keyFactory.instance();
 		K key2 = keyFactory.instance();
@@ -143,6 +176,21 @@ public class DefaultListOperationsTests<K, V> {
 		assertThat(listOps.rightPopAndLeftPush(key, key2, 1, TimeUnit.MILLISECONDS)).isEqualTo(v1);
 	}
 
+	@Test // DATAREDIS-611
+	public void testRightPopAndLeftPushDuration() {
+		// 1 ms timeout gets upgraded to 1 sec timeout at the moment
+		assumeTrue(RedisTestProfileValueSource.matches("runLongTests", "true"));
+		assumeTrue(valueFactory instanceof StringObjectFactory);
+
+		K key = keyFactory.instance();
+		K key2 = keyFactory.instance();
+		V v1 = valueFactory.instance();
+
+		assertThat(listOps.rightPopAndLeftPush(key, key2, Duration.ofMillis(1))).isNull();
+		listOps.leftPush(key, v1);
+		assertThat(listOps.rightPopAndLeftPush(key, key2, Duration.ofMillis(1))).isEqualTo(v1);
+	}
+
 	@Test
 	public void testRightPopAndLeftPush() {
 
diff --git a/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java b/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java
index d839e3e64..43d1d774a 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java
@@ -20,6 +20,9 @@ import static org.assertj.core.api.Assumptions.assumeThat;
 import static org.junit.Assume.*;
 import static org.springframework.data.redis.SpinBarrier.*;
 
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 
@@ -526,9 +529,6 @@ public class RedisTemplateTests<K, V> {
 	@Test // DATAREDIS-526
 	public void testGetExpireMillis() {
 
-		assumeTrue(redisTemplate.getConnectionFactory() instanceof JedisConnectionFactory
-				|| redisTemplate.getConnectionFactory() instanceof LettuceConnectionFactory);
-
 		K key = keyFactory.instance();
 		redisTemplate.boundValueOps(key).set(valueFactory.instance());
 		redisTemplate.expire(key, 1, TimeUnit.DAYS);
@@ -539,6 +539,19 @@ public class RedisTemplateTests<K, V> {
 		assertThat(ttl).isLessThan(25L);
 	}
 
+	@Test // DATAREDIS-611
+	public void testGetExpireDuration() {
+
+		K key = keyFactory.instance();
+		redisTemplate.boundValueOps(key).set(valueFactory.instance());
+		redisTemplate.expire(key, Duration.ofDays(1));
+
+		Long ttl = redisTemplate.getExpire(key, TimeUnit.HOURS);
+
+		assertThat(ttl).isGreaterThanOrEqualTo(23L);
+		assertThat(ttl).isLessThan(25L);
+	}
+
 	@Test // DATAREDIS-526
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public void testGetExpireMillisUsingTransactions() {
@@ -592,24 +605,6 @@ public class RedisTemplateTests<K, V> {
 		assertThat(((Long) result.get(2))).isLessThan(25L);
 	}
 
-	@Test
-	public void testGetExpireMillisNotSupported() {
-
-		assumeTrue(redisTemplate.getConnectionFactory() instanceof JedisConnectionFactory);
-
-		K key1 = keyFactory.instance();
-		V value1 = valueFactory.instance();
-
-		assumeTrue(key1 instanceof String && value1 instanceof String);
-
-		StringRedisTemplate template2 = new StringRedisTemplate(redisTemplate.getConnectionFactory());
-		template2.boundValueOps((String) key1).set((String) value1);
-		template2.expire((String) key1, 5, TimeUnit.SECONDS);
-		long expire = template2.getExpire((String) key1, TimeUnit.MILLISECONDS);
-		// we should still get expire in milliseconds if requested
-		assertThat(expire > 1000 && expire <= 5000).isTrue();
-	}
-
 	@Test
 	public void testExpireAt() {
 		K key1 = keyFactory.instance();
@@ -619,6 +614,15 @@ public class RedisTemplateTests<K, V> {
 		waitFor(() -> (!redisTemplate.hasKey(key1)), 5L);
 	}
 
+	@Test // DATAREDIS-611
+	public void testExpireAtInstant() {
+		K key1 = keyFactory.instance();
+		V value1 = valueFactory.instance();
+		redisTemplate.boundValueOps(key1).set(value1);
+		redisTemplate.expireAt(key1, Instant.now().plus(5, ChronoUnit.MILLIS));
+		waitFor(() -> (!redisTemplate.hasKey(key1)), 5L);
+	}
+
 	@Test
 	public void testExpireAtMillisNotSupported() {