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 8fe2f2c9f..3520b682e 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
+++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
@@ -770,6 +770,11 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertAndReturn(delegate.set(key, value, expiration, option), Converters.identityConverter());
 	}
 
+	@Override
+	public byte[] setGet(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+		return convertAndReturn(delegate.setGet(key, value, expiration, option), Converters.identityConverter());
+	}
+
 	@Override
 	public Boolean setBit(byte[] key, long offset, boolean value) {
 		return convertAndReturn(delegate.setBit(key, offset, value), Converters.identityConverter());
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 aa5f6de77..c56ec9830 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java
+++ b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java
@@ -326,6 +326,13 @@ public interface DefaultedRedisConnection extends RedisCommands, RedisCommandsPr
 		return stringCommands().set(key, value, expiration, option);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
+	@Override
+	@Deprecated
+	default byte[] setGet(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+		return stringCommands().setGet(key, value, expiration, option);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
 	@Override
 	@Deprecated
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java
index ab324fb46..1bce0b224 100644
--- a/src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java
@@ -47,6 +47,7 @@ import org.springframework.util.Assert;
  *
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Marcin Grzejszczak
  * @since 2.0
  */
 public interface ReactiveStringCommands {
@@ -193,6 +194,41 @@ public interface ReactiveStringCommands {
 	 */
 	Flux<BooleanResponse<SetCommand>> set(Publisher<SetCommand> commands);
 
+	/**
+	 * Set {@literal value} for {@literal key} with {@literal expiration} and {@literal options}. Return the old
+	 * string stored at key, or nil if key did not exist. An error is returned and SET aborted if the value
+	 * stored at key is not a string.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param value must not be {@literal null}.
+	 * @param expiration must not be {@literal null}. Use {@link Expiration#persistent()} for no expiration time or
+	 *          {@link Expiration#keepTtl()} to keep the existing.
+	 * @param option must not be {@literal null}.
+	 * @return
+	 * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
+	 * @since 3.4
+	 */
+	@Nullable
+	default Mono<ByteBuffer> setGet(ByteBuffer key, ByteBuffer value, Expiration expiration, SetOption option) {
+
+		Assert.notNull(key, "Key must not be null");
+		Assert.notNull(value, "Value must not be null");
+
+		return setGet(Mono.just(SetCommand.set(key).value(value).withSetOption(option).expiring(expiration))).next()
+				.map(CommandResponse::getOutput);
+	}
+
+	/**
+	 * Set each and every item separately by invoking {@link SetCommand}. Return the old
+	 * string stored at key, or nil if key did not exist. An error is returned and SET aborted if the value
+	 * stored at key is not a string.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return {@link Flux} of {@link ByteBufferResponse} holding the {@link SetCommand} along with the command result.
+	 * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
+	 */
+	Flux<ByteBufferResponse<SetCommand>> setGet(Publisher<SetCommand> commands);
+
 	/**
 	 * Get single element stored at {@literal key}.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java
index 007bbc774..39ce9f2fd 100644
--- a/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java
@@ -122,6 +122,22 @@ public interface RedisStringCommands {
 	@Nullable
 	Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option);
 
+	/**
+	 * Set {@code value} for {@code key}. Return the old string stored at key, or nil if key did not exist.
+	 * An error is returned and SET aborted if the value stored at key is not a string.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param value must not be {@literal null}.
+	 * @param expiration must not be {@literal null}. Use {@link Expiration#persistent()} to not set any ttl or
+	 *          {@link Expiration#keepTtl()} to keep the existing expiration.
+	 * @param option must not be {@literal null}. Use {@link SetOption#upsert()} to add non existing.
+	 * @return {@literal null} when used in pipeline / transaction.
+	 * @since 3.4
+	 * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
+	 */
+	@Nullable
+	byte[] setGet(byte[] key, byte[] value, Expiration expiration, SetOption option);
+
 	/**
 	 * Set {@code value} for {@code key}, only if {@code key} does not exist.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStringCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStringCommands.java
index ba02de01a..175203971 100644
--- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStringCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStringCommands.java
@@ -150,6 +150,24 @@ class JedisClusterStringCommands implements RedisStringCommands {
 		}
 	}
 
+	@Override
+	public byte[] setGet(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+
+		Assert.notNull(key, "Key must not be null");
+		Assert.notNull(value, "Value must not be null");
+		Assert.notNull(expiration, "Expiration must not be null");
+		Assert.notNull(option, "Option must not be null");
+
+		SetParams setParams = JedisConverters.toSetCommandExPxArgument(expiration,
+				JedisConverters.toSetCommandNxXxArgument(option));
+
+		try {
+			return connection.getCluster().setGet(key, value, setParams);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
 	@Override
 	public Boolean setNX(byte[] key, byte[] value) {
 
diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisStringCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisStringCommands.java
index 9b21e8e71..088c71c36 100644
--- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisStringCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisStringCommands.java
@@ -116,6 +116,20 @@ class JedisStringCommands implements RedisStringCommands {
 				.getOrElse(Converters.stringToBooleanConverter(), () -> false);
 	}
 
+	@Override
+	@Nullable
+	public byte[] setGet(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+		Assert.notNull(key, "Key must not be null");
+		Assert.notNull(value, "Value must not be null");
+		Assert.notNull(expiration, "Expiration must not be null");
+		Assert.notNull(option, "Option must not be null");
+
+		SetParams params = JedisConverters.toSetCommandExPxArgument(expiration,
+				JedisConverters.toSetCommandNxXxArgument(option));
+
+		return connection.invoke().just(Jedis::setGet, PipelineBinaryCommands::setGet, key, value, params);
+	}
+
 	@Override
 	public Boolean setNX(byte[] key, byte[] value) {
 
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java
index eaec33cb2..08d2972b5 100644
--- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java
@@ -105,6 +105,19 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 		}));
 	}
 
+	@Override
+	public Flux<ByteBufferResponse<SetCommand>> setGet(Publisher<SetCommand> commands) {
+		return this.connection.execute(reactiveCommands -> Flux.from(commands).concatMap((command) -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null");
+			Assert.notNull(command.getValue(), "Value must not be null");
+
+			return reactiveCommands.setGet(command.getKey(), command.getValue())
+					.map(v -> new ByteBufferResponse<>(command, v))
+					.defaultIfEmpty(new AbsentByteBufferResponse<>(command));
+		}));
+	}
+
 	@Override
 	public Flux<ByteBufferResponse<SetCommand>> getSet(Publisher<SetCommand> commands) {
 
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStringCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStringCommands.java
index ebaeb7475..c4c38a41e 100644
--- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStringCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStringCommands.java
@@ -115,6 +115,18 @@ class LettuceStringCommands implements RedisStringCommands {
 				.orElse(LettuceConverters.stringToBooleanConverter(), false);
 	}
 
+	@Override
+	@Nullable
+	public byte[] setGet(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+		Assert.notNull(key, "Key must not be null");
+		Assert.notNull(value, "Value must not be null");
+		Assert.notNull(expiration, "Expiration must not be null");
+		Assert.notNull(option, "Option must not be null");
+
+		return connection.invoke()
+				.just(RedisStringAsyncCommands::setGet, key, value, LettuceConverters.toSetArgs(expiration, option));
+	}
+
 	@Override
 	public Boolean setNX(byte[] key, byte[] value) {
 
diff --git a/src/main/java/org/springframework/data/redis/core/BoundValueOperations.java b/src/main/java/org/springframework/data/redis/core/BoundValueOperations.java
index 40be5358f..53b873334 100644
--- a/src/main/java/org/springframework/data/redis/core/BoundValueOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/BoundValueOperations.java
@@ -49,6 +49,19 @@ public interface BoundValueOperations<K, V> extends BoundKeyOperations<K> {
 	 */
 	void set(V value, long timeout, TimeUnit unit);
 
+	/**
+	 * Set the {@code value} and expiration {@code timeout} for the bound key. Return the old
+	 * string stored at key, or nil if key did not exist. An error is returned and SET aborted if the value
+	 * stored at key is not a string.
+	 *
+	 * @param value must not be {@literal null}.
+	 * @param timeout
+	 * @param unit must not be {@literal null}.
+	 * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
+	 * @since 3.4
+	 */
+	V setGet(V value, long timeout, TimeUnit unit);
+
 	/**
 	 * Set the {@code value} and expiration {@code timeout} for the bound key.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java
index 4ed048333..8cf8d5498 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java
@@ -77,6 +77,18 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 				stringCommands.set(rawKey(key), rawValue(value), Expiration.from(timeout), SetOption.UPSERT));
 	}
 
+	@Override
+	public Mono<V> setGet(K key, V value, Duration timeout) {
+
+		Assert.notNull(key, "Key must not be null");
+		Assert.notNull(value, "Value must not be null");
+		Assert.notNull(timeout, "Duration must not be null");
+
+		return createMono(stringCommands ->
+				stringCommands.setGet(rawKey(key), rawValue(value), Expiration.from(timeout), SetOption.UPSERT))
+				.map(this::readRequiredValue);
+	}
+
 	@Override
 	public Mono<Boolean> setIfAbsent(K key, V value) {
 
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java
index cbc44ca3c..b20e570f0 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultValueOperations.java
@@ -208,6 +208,27 @@ class DefaultValueOperations<K, V> extends AbstractOperations<K, V> implements V
 		execute(connection -> connection.set(rawKey, rawValue, Expiration.from(timeout, unit), SetOption.upsert()));
 	}
 
+	@Override
+	public V setGet(K key, V value, long timeout, TimeUnit unit) {
+		return doSetGet(key, value, Expiration.from(timeout, unit));
+	}
+
+	@Override
+	public V setGet(K key, V value, Duration duration) {
+		return doSetGet(key, value, Expiration.from(duration));
+	}
+
+	private V doSetGet(K key, V value, Expiration duration) {
+		byte[] rawValue = rawValue(value);
+		return execute( new ValueDeserializingRedisCallback(key) {
+
+			@Override
+			protected byte[] inRedis(byte[] rawKey, RedisConnection connection) {
+				return connection.stringCommands().setGet(rawKey, rawValue, duration, SetOption.UPSERT);
+			}
+		});
+	}
+
 	@Override
 	public Boolean setIfAbsent(K key, V value) {
 
diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveValueOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveValueOperations.java
index 123376c1d..3eab4929d 100644
--- a/src/main/java/org/springframework/data/redis/core/ReactiveValueOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/ReactiveValueOperations.java
@@ -58,6 +58,18 @@ public interface ReactiveValueOperations<K, V> {
 	 */
 	Mono<Boolean> set(K key, V value, Duration timeout);
 
+	/**
+	 * Set the {@code value} and expiration {@code timeout} for {@code key}. Return the old
+	 * string stored at key, or nil if key did not exist. An error is returned and SET aborted if the value
+	 * stored at key is not a string.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param value
+	 * @param timeout must not be {@literal null}.
+	 * @see <a href="https://redis.io/commands/setex">Redis Documentation: SETEX</a>
+	 */
+	Mono<V> setGet(K key, V value, Duration timeout);
+
 	/**
 	 * Set {@code key} to hold the string {@code value} if {@code key} is absent.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/core/ValueOperations.java b/src/main/java/org/springframework/data/redis/core/ValueOperations.java
index e3abd0918..ac4ac84ad 100644
--- a/src/main/java/org/springframework/data/redis/core/ValueOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/ValueOperations.java
@@ -32,6 +32,7 @@ import org.springframework.util.Assert;
  * @author Christoph Strobl
  * @author Mark Paluch
  * @author Jiahe Cai
+ * @author Marcin Grzejszczak
  */
 public interface ValueOperations<K, V> {
 
@@ -44,6 +45,33 @@ public interface ValueOperations<K, V> {
 	 */
 	void set(K key, V value);
 
+	/**
+	 * Set the {@code value} and expiration {@code timeout} for {@code key}. Return the old
+	 * string stored at key, or nil if key did not exist. An error is returned and SET aborted if the value
+	 * stored at key is not a string.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param value must not be {@literal null}.
+	 * @param timeout the key expiration timeout.
+	 * @param unit must not be {@literal null}.
+	 * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
+	 * @since 3.4
+	 */
+	V setGet(K key, V value, long timeout, TimeUnit unit);
+
+	/**
+	 * Set the {@code value} and expiration {@code timeout} for {@code key}. Return the old
+	 * string stored at key, or nil if key did not exist. An error is returned and SET aborted if the value
+	 * stored at key is not a string.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param value must not be {@literal null}.
+	 * @param duration expiration duration
+	 * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
+	 * @since 3.4
+	 */
+	V setGet(K key, V value, Duration duration);
+
 	/**
 	 * Set the {@code value} and expiration {@code timeout} for {@code key}.
 	 *
diff --git a/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java
index d66c7057c..068e87444 100644
--- a/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/connection/RedisConnectionUnitTests.java
@@ -1093,6 +1093,11 @@ class RedisConnectionUnitTests {
 			return delegate.set(key, value, expiration, options);
 		}
 
+		@Override
+		public byte[] setGet(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+			return delegate.setGet(key, value, expiration, option);
+		}
+
 		@Override
 		public List<Long> bitField(byte[] key, BitFieldSubCommands subCommands) {
 			return delegate.bitField(key, subCommands);
diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsIntegrationTests.java
index af3e43e13..1bb099161 100644
--- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsIntegrationTests.java
+++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsIntegrationTests.java
@@ -556,4 +556,29 @@ public class LettuceReactiveStringCommandsIntegrationTests extends LettuceReacti
 		assertThat(nativeBinaryCommands.ttl(KEY_1_BBUFFER)).isCloseTo(expireSeconds, Offset.offset(5L));
 		assertThat(nativeCommands.get(KEY_1)).isEqualTo(VALUE_2);
 	}
+
+	@ParameterizedRedisTest // GH-2853
+	void setGetMono() {
+		nativeCommands.set(KEY_1, VALUE_1);
+
+		connection.stringCommands().setGet(KEY_1_BBUFFER, VALUE_2_BBUFFER, Expiration.keepTtl(), SetOption.upsert())
+				.as(StepVerifier::create) //
+				.expectNext(VALUE_1_BBUFFER) //
+				.verifyComplete();
+
+		assertThat(nativeCommands.get(KEY_1)).isEqualTo(VALUE_2);
+	}
+
+	@ParameterizedRedisTest // GH-2853
+	void setGetFlux() {
+		nativeCommands.set(KEY_1, VALUE_1);
+
+		connection.stringCommands().setGet(Mono.just(SetCommand.set(KEY_1_BBUFFER).value(VALUE_2_BBUFFER).expiring(Expiration.keepTtl()).withSetOption( SetOption.upsert())))
+				.map(CommandResponse::getOutput)
+				.as(StepVerifier::create) //
+				.expectNext(VALUE_1_BBUFFER) //
+				.verifyComplete();
+
+		assertThat(nativeCommands.get(KEY_1)).isEqualTo(VALUE_2);
+	}
 }
diff --git a/src/test/java/org/springframework/data/redis/core/DefaultValueOperationsIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/DefaultValueOperationsIntegrationTests.java
index 1d73298e8..abc601bb8 100644
--- a/src/test/java/org/springframework/data/redis/core/DefaultValueOperationsIntegrationTests.java
+++ b/src/test/java/org/springframework/data/redis/core/DefaultValueOperationsIntegrationTests.java
@@ -316,6 +316,32 @@ public class DefaultValueOperationsIntegrationTests<K, V> {
 		await().atMost(Duration.ofMillis(500L)).until(() -> !redisTemplate.hasKey(key));
 	}
 
+	@ParameterizedRedisTest
+	void testSetGetWithExpiration() {
+
+		K key = keyFactory.instance();
+		V value1 = valueFactory.instance();
+		V value2 = valueFactory.instance();
+
+		valueOps.set(key, value1);
+
+		assertThat(valueOps.setGet(key, value2, 1, TimeUnit.SECONDS)).isEqualTo(value1);
+		assertThat(valueOps.get(key)).isEqualTo(value2);
+	}
+
+	@ParameterizedRedisTest
+	void testSetGetWithExpirationDuration() {
+
+		K key = keyFactory.instance();
+		V value1 = valueFactory.instance();
+		V value2 = valueFactory.instance();
+
+		valueOps.set(key, value1);
+
+		assertThat(valueOps.setGet(key, value2, Duration.ofMillis(1000))).isEqualTo(value1);
+		assertThat(valueOps.get(key)).isEqualTo(value2);
+	}
+
 	@ParameterizedRedisTest
 	void testAppend() {
 
diff --git a/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java b/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java
index de22a409c..2bb8046f8 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisTemplateIntegrationTests.java
@@ -571,6 +571,23 @@ public class RedisTemplateIntegrationTests<K, V> {
 		assertThat(ttl).isLessThan(25L);
 	}
 
+	@ParameterizedRedisTest // GH-3017
+	void testSetGetExpireMillis() {
+
+		K key = keyFactory.instance();
+		V value1 = valueFactory.instance();
+		V value2 = valueFactory.instance();
+		redisTemplate.boundValueOps(key).set(value1);
+
+		V oldValue = redisTemplate.boundValueOps(key).setGet(value2, 1, TimeUnit.DAYS);
+		redisTemplate.expire(key, 1, TimeUnit.DAYS);
+		Long ttl = redisTemplate.getExpire(key, TimeUnit.HOURS);
+
+		assertThat(oldValue).isEqualTo(value1);
+		assertThat(ttl).isGreaterThanOrEqualTo(23L);
+		assertThat(ttl).isLessThan(25L);
+	}
+
 	@ParameterizedRedisTest // DATAREDIS-611
 	void testGetExpireDuration() {