Fix race in (JCache) cache creation

Fixes #17284 on 3.12.z branch

Backported from a58cce0

hazelcast-client/src/main/java/com/hazelcast/client/cache/impl/AbstractClientInternalCacheProxy.java

@@ -176,11 +176,16 @@ protected void onInitialize() {
     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/main/java/com/hazelcast/cache/impl/AbstractInternalCacheProxy.java

@@ -113,11 +113,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/test/java/com/hazelcast/cache/CacheCreationTest.java

@@ -51,6 +51,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.config.EvictionConfig.MaxSizePolicy.ENTRY_COUNT;
 import static java.util.Collections.singletonList;
@@ -85,6 +87,48 @@ public void createSingleCache() {
         cache.get(1);
     }
 
+    @Test
+    public void concurrentCacheCreation() throws InterruptedException {
+        // see https://github.com/hazelcast/hazelcast/issues/17284
+        final String cacheName = "myCache";
+        int threadCount = Runtime.getRuntime().availableProcessors() * 20;
+        Config config = new Config().addCacheConfig(new CacheSimpleConfig().setName(cacheName));
+        final CacheManager cacheManager = createCachingProvider(config).getCacheManager();
+
+        final CountDownLatch startLatch = new CountDownLatch(1);
+        final AtomicInteger errorCounter = new AtomicInteger();
+        Runnable getCache = new Runnable() {
+            @Override
+            public void run() {
+                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);