Merge pull request #475 from splitio/SDKS-8067

[SDKS-8067] Fix deadlock when calling sendUniqueKeys

Author: nmayorsplit

CHANGES.txt

@@ -1,3 +1,6 @@
+4.11.1 (Feb 29, 2024)
+- Fixed deadlock in UniqueKeysTracker when sending Unique Keys.
+
 4.11.0 (Jan 9, 2024)
 - Added impressionsListener method in the IntegrationConfig builder to set Sync or Async Listener execution.
 - Fixed localhost to read files with yml ending.

client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.11.0</version>
+        <version>4.11.1</version>
     </parent>
     <artifactId>java-client</artifactId>
     <packaging>jar</packaging>

client/src/main/java/io/split/client/impressions/UniqueKeysTrackerImp.java

@@ -14,10 +14,13 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class UniqueKeysTrackerImp implements UniqueKeysTracker{
     private static final Logger _log = LoggerFactory.getLogger(UniqueKeysTrackerImp.class);
@@ -31,6 +34,7 @@ public class UniqueKeysTrackerImp implements UniqueKeysTracker{
     private final ConcurrentHashMap<String,HashSet<String>> uniqueKeysTracker;
     private final int _uniqueKeysRefreshRate;
     private final int _filterRefreshRate;
+    private final AtomicBoolean sendGuard = new AtomicBoolean(false);
     private static final Logger _logger = LoggerFactory.getLogger(UniqueKeysTrackerImp.class);
 
     public UniqueKeysTrackerImp(TelemetrySynchronizer telemetrySynchronizer, int uniqueKeysRefreshRate, int filterRefreshRate,
@@ -46,19 +50,19 @@ public UniqueKeysTrackerImp(TelemetrySynchronizer telemetrySynchronizer, int uni
     }
 
     @Override
-    public synchronized boolean track(String featureFlagName, String key) {
+    public boolean track(String featureFlagName, String key) {
         if (!filterAdapter.add(featureFlagName, key)) {
             _logger.debug("The feature flag " + featureFlagName + " and key " + key + " exist in the UniqueKeysTracker");
             return false;
         }
-        HashSet<String> value = new HashSet<>();
-        if(uniqueKeysTracker.containsKey(featureFlagName)){
-            value = uniqueKeysTracker.get(featureFlagName);
-        }
-        value.add(key);
-        uniqueKeysTracker.put(featureFlagName, value);
+        uniqueKeysTracker.compute(featureFlagName,
+                (feature, current) -> {
+                    HashSet<String> keysByFeature = Optional.ofNullable(current).orElse(new HashSet<>());
+                    keysByFeature.add(key);
+                    return keysByFeature;
+                });
         _logger.debug("The feature flag " + featureFlagName + " and key " + key + " was added");
-        if (uniqueKeysTracker.size() == MAX_AMOUNT_OF_TRACKED_UNIQUE_KEYS){
+        if (uniqueKeysTracker.size() >= MAX_AMOUNT_OF_TRACKED_UNIQUE_KEYS){
             _logger.warn("The UniqueKeysTracker size reached the maximum limit");
             try {
                 sendUniqueKeys();
@@ -107,17 +111,25 @@ public HashMap<String,HashSet<String>> popAll(){
     }
 
     private void sendUniqueKeys(){
-        if (uniqueKeysTracker.size() == 0) {
-           _log.warn("The Unique Keys Tracker is empty");
-           return;
+        if (!sendGuard.compareAndSet(false, true)) {
+            _log.debug("SendUniqueKeys already running");
+            return;
         }
-        HashMap<String, HashSet<String>> uniqueKeysHashMap = popAll();
-        List<UniqueKeys.UniqueKey> uniqueKeysFromPopAll = new ArrayList<>();
-        for (String featureFlag : uniqueKeysHashMap.keySet()) {
-            UniqueKeys.UniqueKey uniqueKey = new UniqueKeys.UniqueKey(featureFlag, new ArrayList<>(uniqueKeysHashMap.get(featureFlag)));
-            uniqueKeysFromPopAll.add(uniqueKey);
+        try {
+            if (uniqueKeysTracker.size() == 0) {
+                _log.warn("The Unique Keys Tracker is empty");
+                return;
+            }
+            HashMap<String, HashSet<String>> uniqueKeysHashMap = popAll();
+            List<UniqueKeys.UniqueKey> uniqueKeysFromPopAll = new ArrayList<>();
+            for (Map.Entry<String, HashSet<String>> uniqueKeyEntry : uniqueKeysHashMap.entrySet()) {
+                UniqueKeys.UniqueKey uniqueKey = new UniqueKeys.UniqueKey(uniqueKeyEntry.getKey(), new ArrayList<>(uniqueKeyEntry.getValue()));
+                uniqueKeysFromPopAll.add(uniqueKey);
+            }
+            _telemetrySynchronizer.synchronizeUniqueKeys(new UniqueKeys(uniqueKeysFromPopAll));
+        } finally {
+            sendGuard.set(false);
         }
-        _telemetrySynchronizer.synchronizeUniqueKeys(new UniqueKeys(uniqueKeysFromPopAll));
     }
 
     private interface ExecuteUniqueKeysAction{
@@ -138,4 +150,8 @@ public void execute() {
             sendUniqueKeys();
         }
     }
+
+    public AtomicBoolean getSendGuard() {
+        return sendGuard;
+    }
 }

client/src/main/java/io/split/client/impressions/filters/BloomFilterImp.java

@@ -6,7 +6,7 @@
 
 public class BloomFilterImp implements Filter {
 
-    private BloomFilter bloomFilter;
+    private BloomFilter<String> bloomFilter;
     private final int size;
     private final double errorMargin;
 
@@ -17,17 +17,17 @@ public BloomFilterImp(int size, double errorMargin) {
     }
 
     @Override
-    public synchronized boolean add(String data) {
+    public boolean add(String data) {
         return bloomFilter.put(data);
     }
 
     @Override
-    public synchronized boolean contains(String data) {
+    public boolean contains(String data) {
         return bloomFilter.mightContain(data);
     }
 
     @Override
-    public synchronized void clear() {
+    public void clear() {
         bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_16), size, errorMargin);
 
     }

client/src/main/java/io/split/engine/common/SyncManagerImp.java

@@ -29,7 +29,7 @@
 import static io.split.client.utils.SplitExecutorFactory.buildExecutorService;
 
 public class SyncManagerImp implements SyncManager {
-    private static final Logger _log = LoggerFactory.getLogger(SyncManager.class);
+    private static final Logger _log = LoggerFactory.getLogger(SyncManagerImp.class);
 
     private final AtomicBoolean _streamingEnabledConfig;
     private final Synchronizer _synchronizer;

client/src/main/java/io/split/engine/sse/AuthApiClientImp.java

@@ -22,7 +22,7 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 public class AuthApiClientImp implements AuthApiClient {
-    private static final Logger _log = LoggerFactory.getLogger(AuthApiClient.class);
+    private static final Logger _log = LoggerFactory.getLogger(AuthApiClientImp.class);
 
     private final CloseableHttpClient _httpClient;
     private final String _target;

client/src/test/java/io/split/client/impressions/UniqueKeysTrackerImpTest.java

@@ -90,6 +90,7 @@ public void testStopSynchronization() throws Exception {
         TelemetrySynchronizer telemetrySynchronizer = Mockito.mock(TelemetryInMemorySubmitter.class);
         UniqueKeysTrackerImp uniqueKeysTrackerImp = new UniqueKeysTrackerImp(telemetrySynchronizer, 1, 2, null);
         uniqueKeysTrackerImp.start();
+        Assert.assertFalse(uniqueKeysTrackerImp.getSendGuard().get());
         Assert.assertTrue(uniqueKeysTrackerImp.track("feature1","key1"));
         Assert.assertTrue(uniqueKeysTrackerImp.track("feature1","key2"));
         Assert.assertTrue(uniqueKeysTrackerImp.track("feature2","key3"));

pluggable-storage/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <artifactId>java-client-parent</artifactId>
         <groupId>io.split.client</groupId>
-        <version>4.11.0</version>
+        <version>4.11.1</version>
     </parent>
 
     <version>2.1.0</version>

pom.xml

@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>io.split.client</groupId>
     <artifactId>java-client-parent</artifactId>
-    <version>4.11.0</version>
+    <version>4.11.1</version>
     <dependencyManagement>
         <dependencies>
             <dependency>

redis-wrapper/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <artifactId>java-client-parent</artifactId>
         <groupId>io.split.client</groupId>
-        <version>4.11.0</version>
+        <version>4.11.1</version>
     </parent>
     <artifactId>redis-wrapper</artifactId>
     <version>3.1.0</version>