FMWK-48 Spring Data Aerospike Cacheable sync option

src/main/java/org/springframework/data/aerospike/cache/AerospikeCache.java

@@ -112,20 +112,48 @@ public Object getNativeCache() {
      * @return The value (bins) to which this cache maps the specified key.
      */
     @Override
-    @SuppressWarnings({"unchecked", "NullableProblems"})
+    @SuppressWarnings({"NullableProblems"})
     public <T> T get(Object key, Callable<T> valueLoader) {
-        T value = (T) client.get(null, getKey(key)).getValue(VALUE);
-        if (Objects.isNull(value)) {
-            try {
-                value = valueLoader.call();
-                if (Objects.nonNull(value)) {
-                    put(key, value);
+        Key dbKey = getKey(key);
+        Record record = client.get(null, dbKey);
+        if (record == null && valueLoader != null) {
+            synchronized (this) {
+                record = client.get(null, dbKey);
+                if (record == null) {
+                    T value = callValueLoader(valueLoader, key);
+                    if (Objects.nonNull(value)) {
+                        put(key, value);
+                    }
+                    return value;
                 }
-            } catch (Exception e) {
-                throw new Cache.ValueRetrievalException(key, valueLoader, e);
             }
         }
-        return value;
+        if (record.getValue(VALUE) != null) {
+            AerospikeReadData data = AerospikeReadData.forRead(dbKey, record);
+            Class<T> type = getValueType(valueLoader); // determine the class of T
+            return aerospikeConverter.read(type, data);
+        } else {
+            return null;
+        }
+    }
+
+    private <T> T callValueLoader(Callable<T> valueLoader, Object key) {
+        try {
+            return valueLoader.call();
+        } catch (Exception e) {
+            throw new Cache.ValueRetrievalException(key, valueLoader, e);
+        }
+    }
+
+    // Helper method to determine the class of T
+    @SuppressWarnings("unchecked")
+    private static <T> Class<T> getValueType(Callable<T> valueLoader) {
+        try {
+            // Use reflection to get the return type of the Callable
+            return (Class<T>) valueLoader.getClass().getMethod("call").getReturnType();
+        } catch (NoSuchMethodException e) {
+            throw new IllegalArgumentException("Cannot determine value type", e);
+        }
     }
 
     /**

src/test/java/org/springframework/data/aerospike/cache/AerospikeCacheManagerIntegrationTests.java

@@ -66,6 +66,35 @@ public void shouldCache() {
         assertThat(cachingComponent.getNoOfCalls()).isEqualTo(1);
     }
 
+    @Test
+    public void testCacheableMethodSync() throws InterruptedException {
+        assertThat(cachingComponent.getNoOfCalls() == 0).isTrue();
+
+        // Creating two threads that will call cacheableMethod concurrently
+        Thread thread1 = new Thread(() -> {
+            CachedObject response = cachingComponent.cacheableMethodSynchronized(KEY);
+            assertThat(response).isNotNull();
+            assertThat(response.getValue()).isEqualTo(VALUE);
+        });
+
+        Thread thread2 = new Thread(() -> {
+            CachedObject response = cachingComponent.cacheableMethodSynchronized(KEY);
+            assertThat(response).isNotNull();
+            assertThat(response.getValue()).isEqualTo(VALUE);
+        });
+
+        // Starting both threads
+        thread1.start();
+        thread2.start();
+
+        // Waiting for both threads to complete
+        thread1.join();
+        thread2.join();
+
+        // Expecting method to be called only once due to synchronization
+        assertThat(cachingComponent.getNoOfCalls() == 1).isTrue();
+    }
+
     @Test
     public void shouldEvictCache() {
         CachedObject response1 = cachingComponent.cacheableMethod(KEY);
@@ -185,6 +214,12 @@ public CachedObject cacheableMethod(String param) {
             return new CachedObject("id", VALUE);
         }
 
+        @Cacheable(value = "TEST", sync = true)
+        public CachedObject cacheableMethodSynchronized(String param) {
+            noOfCalls++;
+            return new CachedObject("id", VALUE);
+        }
+
         @Cacheable(value = "CACHE-WITH-TTL")
         public CachedObject cacheableMethodWithTTL(String param) {
             noOfCalls++;