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 5bd90eb9c..57b86e8af 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
+++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
@@ -69,6 +69,7 @@ import org.springframework.util.ObjectUtils;
  * @author Ninad Divadkar
  * @author Tugdual Grall
  * @author Andrey Shlykov
+ * @author dengliming
  */
 public class DefaultStringRedisConnection implements StringRedisConnection, DecoratedRedisConnection {
 
@@ -992,6 +993,15 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertAndReturn(delegate.set(key, value, expiration, option), identityConverter);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#set(byte[], byte[], org.springframework.data.redis.core.types.Expiration, org.springframework.data.redis.connection.RedisStringCommands.SetOptions, boolean)
+	 */
+	@Override
+	public Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option, boolean keepTtl) {
+		return convertAndReturn(delegate.set(key, value, expiration, option, keepTtl), identityConverter);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#setBit(byte[], long, boolean)
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 4b66f1fdf..0cf9bbbf0 100644
--- a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java
+++ b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java
@@ -294,6 +294,13 @@ public interface DefaultedRedisConnection extends RedisConnection {
 		return stringCommands().set(key, value, expiration, option);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
+	@Override
+	@Deprecated
+	default Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option, boolean keepTtl) {
+		return stringCommands().set(key, value, expiration, option, keepTtl);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
 	@Override
 	@Deprecated
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 fab800afd..8e888c343 100644
--- a/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java
@@ -92,6 +92,22 @@ public interface RedisStringCommands {
 	@Nullable
 	Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option);
 
+	/**
+	 * Set {@code value} for {@code key} applying timeouts from {@code expiration} if set and inserting/updating values
+	 * depending on {@code option}.
+	 *
+	 * @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.
+	 * @param option must not be {@literal null}. Use {@link SetOption#upsert()} to add non existing.
+	 * @param keepTtl set the value and retain the existing TTL.
+	 * @return {@literal null} when used in pipeline / transaction.
+	 * @since 2.4
+	 * @see <a href="https://redis.io/commands/set">Redis Documentation: SET</a>
+	 */
+	@Nullable
+	Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option, boolean keepTtl);
+
 	/**
 	 * 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 cc39ec59f..99d67bc48 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
@@ -41,6 +41,7 @@ import org.springframework.util.Assert;
  * @author Christoph Strobl
  * @author Mark Paluch
  * @author Xiaohu Zhang
+ * @author dengliming
  * @since 2.0
  */
 class JedisClusterStringCommands implements RedisStringCommands {
@@ -126,11 +127,20 @@ class JedisClusterStringCommands implements RedisStringCommands {
 	 */
 	@Override
 	public Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+		return set(key, value, expiration, option, false);
+	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#set(byte[], byte[], org.springframework.data.redis.core.types.Expiration, org.springframework.data.redis.connection.RedisStringCommands.SetOptions, boolean)
+	 */
+	@Override
+	public Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option, boolean keepTtl) {
 		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!");
+		Assert.isTrue(keepTtl, "KEEPTTL is currently not supported in jedis!");
 
 		SetParams setParams = JedisConverters.toSetCommandExPxArgument(expiration,
 				JedisConverters.toSetCommandNxXxArgument(option));
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 2b47ada13..44c654ec6 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
@@ -34,6 +34,7 @@ import org.springframework.util.Assert;
 /**
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author dengliming
  * @since 2.0
  */
 class JedisStringCommands implements RedisStringCommands {
@@ -152,11 +153,20 @@ class JedisStringCommands implements RedisStringCommands {
 	 */
 	@Override
 	public Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+		return set(key, value, expiration, option, false);
+	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#set(byte[], byte[], org.springframework.data.redis.core.types.Expiration, org.springframework.data.redis.connection.RedisStringCommands.SetOption, boolean)
+	 */
+	@Override
+	public Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option, boolean keepTtl) {
 		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!");
+		Assert.isTrue(keepTtl, "KEEPTTL is currently not supported in jedis!");
 
 		SetParams params = JedisConverters.toSetCommandExPxArgument(expiration,
 				JedisConverters.toSetCommandNxXxArgument(option));
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 acd8442bf..aa22c4284 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
@@ -82,6 +82,7 @@ import org.springframework.util.StringUtils;
  * @author Thomas Darimont
  * @author Mark Paluch
  * @author Ninad Divadkar
+ * @author dengliming
  */
 abstract public class LettuceConverters extends Converters {
 
@@ -767,6 +768,18 @@ abstract public class LettuceConverters extends Converters {
 	 * @since 1.7
 	 */
 	public static SetArgs toSetArgs(@Nullable Expiration expiration, @Nullable SetOption option) {
+		return toSetArgs(expiration, option, false);
+	}
+
+	/**
+	 * Converts a given {@link Expiration} and {@link SetOption} to the according {@link SetArgs}.<br />
+	 *
+	 * @param expiration can be {@literal null}.
+	 * @param option can be {@literal null}.
+	 * @param keepTtl set the value and retain the existing TTL.
+	 * @since 2.4
+	 */
+	public static SetArgs toSetArgs(@Nullable Expiration expiration, @Nullable SetOption option, boolean keepTtl) {
 
 		SetArgs args = new SetArgs();
 		if (expiration != null && !expiration.isPersistent()) {
@@ -781,6 +794,10 @@ abstract public class LettuceConverters extends Converters {
 			}
 		}
 
+		if (keepTtl) {
+			args.keepttl();
+		}
+
 		if (option != null) {
 
 			switch (option) {
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 0de36a677..f60374898 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
@@ -36,6 +36,7 @@ import org.springframework.util.Assert;
 /**
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author dengliming
  * @since 2.0
  */
 class LettuceStringCommands implements RedisStringCommands {
@@ -156,7 +157,15 @@ class LettuceStringCommands implements RedisStringCommands {
 	 */
 	@Override
 	public Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option) {
+		return set(key, value, expiration, option, false);
+	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#set(byte[], byte[], org.springframework.data.redis.core.types.Expiration, org.springframework.data.redis.connection.RedisStringCommands.SetOption, boolean)
+	 */
+	@Override
+	public Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option, boolean keepTtl) {
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(value, "Value must not be null!");
 		Assert.notNull(expiration, "Expiration must not be null!");
@@ -165,18 +174,18 @@ class LettuceStringCommands implements RedisStringCommands {
 		try {
 			if (isPipelined()) {
 				pipeline(connection.newLettuceResult(
-						getAsyncConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)),
+						getAsyncConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option, keepTtl)),
 						Converters.stringToBooleanConverter(), () -> false));
 				return null;
 			}
 			if (isQueueing()) {
 				transaction(connection.newLettuceResult(
-						getAsyncConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)),
+						getAsyncConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option, keepTtl)),
 						Converters.stringToBooleanConverter(), () -> false));
 				return null;
 			}
 			return Converters
-					.stringToBoolean(getConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)));
+					.stringToBoolean(getConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option, keepTtl)));
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
 		}
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 46d278569..2d525ed71 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
@@ -16,6 +16,7 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.data.Offset.offset;
 import static org.springframework.data.redis.connection.BitFieldSubCommands.*;
 import static org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldIncrBy.Overflow.*;
@@ -2466,4 +2467,19 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
 										.valueAt(BitFieldSubCommands.Offset.offset(1L).multipliedByTypeLength()))).containsExactly(100L,
 												-56L);
 	}
+
+	@Test // DATAREDIS-1103
+	public void setKeepTTL() {
+
+		long expireSeconds = 10;
+		assertThat(clusterConnection.setEx(KEY_1_BYTES, expireSeconds, VALUE_2_BYTES));
+
+		assertThat(clusterConnection.get(KEY_1_BYTES)).isEqualTo(VALUE_2_BYTES);
+
+		assertThat(clusterConnection.set(KEY_1_BYTES, VALUE_2_BYTES, Expiration.persistent(), SetOption.upsert(), true));
+
+		long ttl = clusterConnection.ttl(KEY_1_BYTES);
+
+		assertThat(0 < ttl && ttl <= expireSeconds);
+	}
 }