Polishing.

Inline valueFromLoader method. Refine tests to not rely on the number of runtime CPU cores.

See #2079
Original pull request: #2082.

Author: mp911de

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

@@ -132,13 +132,19 @@ public <T> T get(Object key, Callable<T> valueLoader) {
 
 	@SuppressWarnings("unchecked")
 	private synchronized <T> T getSynchronized(Object key, Callable<T> valueLoader) {
+
 		ValueWrapper result = get(key);
 
 		if (result != null) {
 			return (T) result.get();
 		}
 
-		T value = valueFromLoader(key, valueLoader);
+		T value;
+		try {
+			value = valueLoader.call();
+		} catch (Exception e) {
+			throw new ValueRetrievalException(key, valueLoader, e);
+		}
 		put(key, value);
 		return value;
 	}
@@ -391,12 +397,4 @@ private String prefixCacheKey(String key) {
 		return cacheConfig.getKeyPrefixFor(name) + key;
 	}
 
-	private static <T> T valueFromLoader(Object key, Callable<T> valueLoader) {
-
-		try {
-			return valueLoader.call();
-		} catch (Exception e) {
-			throw new ValueRetrievalException(key, valueLoader, e);
-		}
-	}
 }

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

@@ -18,22 +18,26 @@
 import static org.assertj.core.api.Assertions.*;
 import static org.assertj.core.api.Assumptions.*;
 
+import io.netty.util.concurrent.DefaultThreadFactory;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.RequiredArgsConstructor;
 
 import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
-import java.util.stream.Stream;
+import java.util.stream.IntStream;
 
 import org.junit.jupiter.api.BeforeEach;
 
@@ -48,13 +52,15 @@
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.test.extension.parametrized.MethodSource;
 import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest;
+import org.springframework.lang.Nullable;
 
 /**
  * Tests for {@link RedisCache} with {@link DefaultRedisCacheWriter} using different {@link RedisSerializer} and
  * {@link RedisConnectionFactory} pairs.
  *
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Piotr Mionskowski
  */
 @MethodSource("testParams")
 public class RedisCacheTests {
@@ -415,35 +421,85 @@ void cacheShouldFailOnNonConvertibleCacheKey() {
 	}
 
 	@ParameterizedRedisTest // GH-2079
-	void multipleThreadsLoadValueOnce() {
+	void multipleThreadsLoadValueOnce() throws InterruptedException {
 
-		int threadCount = 5;
+		int threadCount = 2;
 
-		ConcurrentMap<Integer, Integer> valuesByThreadId = new ConcurrentHashMap<>(threadCount);
+		CountDownLatch prepare = new CountDownLatch(threadCount);
+		CountDownLatch prepareForReturn = new CountDownLatch(1);
+		CountDownLatch finished = new CountDownLatch(threadCount);
+		AtomicInteger retrievals = new AtomicInteger();
+		AtomicReference<byte[]> storage = new AtomicReference<>();
 
-		CountDownLatch waiter = new CountDownLatch(threadCount);
+		cache = new RedisCache("foo", new RedisCacheWriter() {
+			@Override
+			public void put(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
+				storage.set(value);
+			}
 
-		AtomicInteger threadIds = new AtomicInteger(0);
+			@Override
+			public byte[] get(String name, byte[] key) {
 
-		AtomicInteger currentValueForKey = new AtomicInteger(0);
+				prepare.countDown();
+				try {
+					prepareForReturn.await(1, TimeUnit.MINUTES);
+				} catch (InterruptedException e) {
+					throw new RuntimeException(e);
+				}
 
-		Stream.generate(threadIds::getAndIncrement)
-				.limit(threadCount)
-				.parallel()
-				.forEach((threadId) -> {
-					waiter.countDown();
-					try {
-						waiter.await();
-					} catch (InterruptedException e) {
-						e.printStackTrace();
-					}
-					Integer valueForThread = cache.get("key", currentValueForKey::incrementAndGet);
-					valuesByThreadId.put(threadId, valueForThread);
-				});
+				return storage.get();
+			}
 
-		valuesByThreadId.forEach((thread, valueForThread) -> {
-			assertThat(valueForThread).isEqualTo(currentValueForKey.get());
-		});
+			@Override
+			public byte[] putIfAbsent(String name, byte[] key, byte[] value, @Nullable Duration ttl) {
+				return new byte[0];
+			}
+
+			@Override
+			public void remove(String name, byte[] key) {
+
+			}
+
+			@Override
+			public void clean(String name, byte[] pattern) {
+
+			}
+
+			@Override
+			public void clearStatistics(String name) {
+
+			}
+
+			@Override
+			public RedisCacheWriter withStatisticsCollector(CacheStatisticsCollector cacheStatisticsCollector) {
+				return null;
+			}
+
+			@Override
+			public CacheStatistics getCacheStatistics(String cacheName) {
+				return null;
+			}
+		}, RedisCacheConfiguration.defaultCacheConfig());
+
+		ThreadPoolExecutor tpe = new ThreadPoolExecutor(threadCount, threadCount, 1, TimeUnit.MINUTES,
+				new LinkedBlockingDeque<>(), new DefaultThreadFactory("RedisCacheTests"));
+
+		IntStream.range(0, threadCount).forEach(it -> tpe.submit(() -> {
+			cache.get("foo", retrievals::incrementAndGet);
+			finished.countDown();
+		}));
+
+		// wait until all Threads have arrived in RedisCacheWriter.get(…)
+		prepare.await();
+
+		// let all threads continue
+		prepareForReturn.countDown();
+
+		// wait until ThreadPoolExecutor has completed.
+		finished.await();
+		tpe.shutdown();
+
+		assertThat(retrievals).hasValue(1);
 	}
 
 	void doWithConnection(Consumer<RedisConnection> callback) {