[BACKPORT] Fix race in (JCache) cache creation

hazelcast/src/main/java/com/hazelcast/cache/impl/CacheProxySupport.java

@@ -220,11 +220,16 @@ public CacheManager getCacheManager() {
     public void setCacheManager(HazelcastCacheManager cacheManager) {
         assert cacheManager instanceof HazelcastServerCacheManager;
 
+        // optimistically assume the CacheManager is already set
         if (cacheManagerRef.get() == cacheManager) {
             return;
         }
 
         if (!this.cacheManagerRef.compareAndSet(null, (HazelcastServerCacheManager) cacheManager)) {
+            if (cacheManagerRef.get() == cacheManager) {
+                // some other thread managed to set the same CacheManager, we are good
+                return;
+            }
             throw new IllegalStateException("Cannot overwrite a Cache's CacheManager.");
         }
     }

hazelcast/src/main/java/com/hazelcast/client/cache/impl/ClientCacheProxySupport.java

@@ -231,10 +231,16 @@ public CacheManager getCacheManager() {
     public void setCacheManager(HazelcastCacheManager cacheManager) {
         assert cacheManager instanceof HazelcastClientCacheManager;
 
+        // optimistically assume the CacheManager is already set
         if (cacheManagerRef.get() == cacheManager) {
             return;
         }
+
         if (!cacheManagerRef.compareAndSet(null, (HazelcastClientCacheManager) cacheManager)) {
+            if (cacheManagerRef.get() == cacheManager) {
+                // some other thread managed to set the same CacheManager, we are good
+                return;
+            }
             throw new IllegalStateException("Cannot overwrite a Cache's CacheManager.");
         }
     }

hazelcast/src/test/java/com/hazelcast/cache/CacheCreationTest.java

@@ -47,6 +47,8 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static com.hazelcast.cache.CacheTestSupport.createServerCachingProvider;
 import static com.hazelcast.config.MaxSizePolicy.ENTRY_COUNT;
@@ -80,6 +82,45 @@ public void createSingleCache() {
         cache.get(1);
     }
 
+    @Test
+    public void concurrentCacheCreation() throws InterruptedException {
+        // see https://github.com/hazelcast/hazelcast/issues/17284
+        String cacheName = "myCache";
+        int threadCount = Runtime.getRuntime().availableProcessors() * 20;
+        Config config = new Config().addCacheConfig(new CacheSimpleConfig().setName(cacheName));
+        CacheManager cacheManager = createCachingProvider(config).getCacheManager();
+
+        CountDownLatch startLatch = new CountDownLatch(1);
+        AtomicInteger errorCounter = new AtomicInteger();
+        Runnable getCache = () -> {
+            try {
+                startLatch.await();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            try {
+                Cache<?, ?> cache = cacheManager.getCache(cacheName);
+                if (cache == null) {
+                    System.out.println("getCache() returned null!");
+                    errorCounter.incrementAndGet();
+                }
+            } catch (Throwable t) {
+                t.printStackTrace();
+                errorCounter.incrementAndGet();
+            }
+        };
+        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
+        for (int i = 0; i < threadCount; i++) {
+            executorService.submit(getCache);
+        }
+        // start all threads at once
+        startLatch.countDown();
+
+        executorService.shutdown();
+        executorService.awaitTermination(1, TimeUnit.MINUTES);
+        assertEquals(0, errorCounter.get());
+    }
+
     @Test
     public void createOrGetConcurrentlySingleCache_fromMultiProviders() {
         ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);