Improve atomicity in DefaultRedisCacheWriter.doLock().

Original pull request spring-projects#518
Closes spring-projects#1686

src/main/java/org/springframework/data/redis/cache/DefaultRedisCacheWriter.java

@@ -319,12 +319,15 @@ void lock(String name) {
 	}
 
 	@Nullable
-	private Boolean doLock(String name, Object contextualKey, @Nullable Object contextualValue,
+	protected Boolean doLock(String name, Object contextualKey, @Nullable Object contextualValue,
 			RedisConnection connection) {
 
 		Expiration expiration = Expiration.from(this.lockTtl.getTimeToLive(contextualKey, contextualValue));
 
-		return connection.stringCommands().set(createCacheLockKey(name), new byte[0], expiration, SetOption.SET_IF_ABSENT);
+		while (!ObjectUtils.nullSafeEquals(connection.stringCommands().set(createCacheLockKey(name), new byte[0], expiration, SetOption.SET_IF_ABSENT),true)) {
+			checkAndPotentiallyWaitUntilUnlocked(name, connection);
+		}
+		return true;
 	}
 
 	/**

src/test/java/org/springframework/data/redis/cache/DefaultRedisCacheWriterTests.java

@@ -22,13 +22,12 @@
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Collection;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
@@ -38,6 +37,7 @@
 import org.springframework.data.redis.test.condition.RedisDriver;
 import org.springframework.data.redis.test.extension.parametrized.MethodSource;
 import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest;
+import org.springframework.lang.Nullable;
 
 /**
  * Integration tests for {@link DefaultRedisCacheWriter}.
@@ -419,6 +419,46 @@ void noOpStatisticsCollectorReturnsEmptyStatsInstance() {
 		assertThat(stats.getPuts()).isZero();
 	}
 
+	@ParameterizedRedisTest
+	@Disabled
+	void doLockShouldGetLock() throws InterruptedException {
+
+		int threadCount = 3;
+
+		DefaultRedisCacheWriter cw = new DefaultRedisCacheWriter(connectionFactory, Duration.ofMillis(50),
+				BatchStrategies.keys()){
+			@Nullable
+			protected Boolean doLock(String name, Object contextualKey, @Nullable Object contextualValue,
+								   RedisConnection connection) {
+				Boolean doLock = super.doLock(name, contextualKey, contextualValue, connection);
+				assertThat(doLock).isTrue();
+				return doLock;
+			}
+		};
+
+		CountDownLatch beforeWrite = new CountDownLatch(threadCount);
+		CountDownLatch afterWrite = new CountDownLatch(threadCount);
+
+		cw.lock(CACHE_NAME);
+
+		for (int i = 0; i < threadCount; i++) {
+			Thread th = new Thread(() -> {
+				beforeWrite.countDown();
+				cw.putIfAbsent(CACHE_NAME, binaryCacheKey, binaryCacheValue, Duration.ZERO);
+				afterWrite.countDown();
+			});
+
+			th.start();
+		}
+
+		beforeWrite.await();
+
+		Thread.sleep(200);
+
+		cw.unlock(CACHE_NAME);
+		afterWrite.await();
+
+	}
 	private void doWithConnection(Consumer<RedisConnection> callback) {
 
 		try (RedisConnection connection = connectionFactory.getConnection()) {