Fixing PR comments

Author: ldecheverz-split

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

@@ -104,6 +104,7 @@ public void start() {
                 try {
                     Thread.currentThread().sleep(1000);
                 } catch (InterruptedException e) {
+                    _log.warn("Sdk Initializer thread interrupted");
                     e.printStackTrace();
                     Thread.currentThread().interrupt();
                 }

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

@@ -53,12 +53,7 @@ public SynchronizerImp(SplitSynchronizationTask splitSynchronizationTask,
 
     @Override
     public boolean syncAll() {
-        AtomicBoolean syncStatus = new AtomicBoolean(false);
-        if(_splitFetcher.fetchAll(true) &&
-        _segmentSynchronizationTaskImp.fetchAllSynchronous()) {
-            syncStatus.set(true);
-        }
-        return syncStatus.get();
+        return _splitFetcher.fetchAll(true) && _segmentSynchronizationTaskImp.fetchAllSynchronous();
     }
 
     @Override

client/src/main/java/io/split/engine/experiments/SplitFetcherImp.java

@@ -149,6 +149,7 @@ public boolean fetchAll(boolean addCacheHeader) {
         long start = _splitCache.getChangeNumber();
         try {
             runWithoutExceptionHandling(addCacheHeader);
+            return true;
         } catch (InterruptedException e) {
             _log.warn("Interrupting split fetcher task");
             Thread.currentThread().interrupt();
@@ -164,6 +165,5 @@ public boolean fetchAll(boolean addCacheHeader) {
                 _log.debug("split fetch before: " + start + ", after: " + _splitCache.getChangeNumber());
             }
         }
-        return true;
     }
 }

client/src/main/java/io/split/engine/segments/SegmentFetcher.java

@@ -9,7 +9,7 @@ public interface SegmentFetcher {
      */
     void fetch(boolean addCacheHeader);
 
-    void runWhitCacheHeader();
+    boolean runWhitCacheHeader();
 
     void fetchAll();
 }

client/src/main/java/io/split/engine/segments/SegmentFetcherImp.java

@@ -1,5 +1,6 @@
 package io.split.engine.segments;
 
+import com.google.common.annotations.VisibleForTesting;
 import io.split.cache.SegmentCache;
 import io.split.client.dtos.SegmentChange;
 import io.split.engine.SDKReadinessGates;
@@ -116,7 +117,8 @@ private String summarize(List<String> changes) {
         return bldr.toString();
     }
 
-    private void callLoopRun(boolean isFetch, boolean addCacheHeader){
+    @VisibleForTesting
+    void callLoopRun(boolean isFetch, boolean addCacheHeader){
         while (true) {
             long start = _segmentCache.getChangeNumber(_segmentName);
             runWithoutExceptionHandling(addCacheHeader);
@@ -131,23 +133,26 @@ private void callLoopRun(boolean isFetch, boolean addCacheHeader){
     }
 
     @Override
-    public void runWhitCacheHeader(){
-        this.fetchAndUpdate(true);
+    public boolean runWhitCacheHeader(){
+        return this.fetchAndUpdate(true);
     }
 
     /**
      * Calls callLoopRun and after fetchs segment.
      * @param addCacheHeader indicates if CacheHeader is required
      */
-    private void fetchAndUpdate(boolean addCacheHeader) {
+    @VisibleForTesting
+    boolean fetchAndUpdate(boolean addCacheHeader) {
         try {
             // Do this again in case the previous call errored out.
             callLoopRun(true, addCacheHeader);
+            return true;
         } catch (Throwable t) {
             _log.error("RefreshableSegmentFetcher failed: " + t.getMessage());
             if (_log.isDebugEnabled()) {
                 _log.debug("Reason:", t);
             }
+            return false;
         }
     }
 

client/src/main/java/io/split/engine/segments/SegmentSynchronizationTaskImp.java

@@ -167,7 +167,9 @@ public boolean fetchAllSynchronous() {
                 .collect(Collectors.toList())
                 .stream().forEach(future -> {
                     try {
-                        future.get();
+                        if(!future.get()) {
+                            fetchAllStatus.set(false);
+                        };
                     } catch (Exception ex) {
                         fetchAllStatus.set(false);
                         _log.error(ex.getMessage());

client/src/test/java/io/split/engine/segments/SegmentSynchronizationTaskImpTest.java

@@ -1,16 +1,21 @@
 package io.split.engine.segments;
 
+import com.google.common.collect.Maps;
 import io.split.engine.SDKReadinessGates;
 import io.split.cache.SegmentCache;
 import io.split.telemetry.storage.InMemoryTelemetryStorage;
 import io.split.telemetry.storage.TelemetryStorage;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
 import java.util.List;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -79,5 +84,53 @@ public void run() {
         assertThat(fetcher1.get(), is(sameInstance(fetcher2.get())));
     }
 
+    @Test
+    public void testFetchAllAsynchronousAndGetFalse() throws NoSuchFieldException, IllegalAccessException {
+        SDKReadinessGates gates = new SDKReadinessGates();
+        SegmentCache segmentCache = Mockito.mock(SegmentCache.class);
+        ConcurrentMap<String, SegmentFetcher> _segmentFetchers = Maps.newConcurrentMap();
+
+        SegmentChangeFetcher segmentChangeFetcher = Mockito.mock(SegmentChangeFetcher.class);
+        SegmentFetcherImp segmentFetcher = Mockito.mock(SegmentFetcherImp.class);
+        _segmentFetchers.put("SF", segmentFetcher);
+        final SegmentSynchronizationTaskImp fetchers = new SegmentSynchronizationTaskImp(segmentChangeFetcher, 1L, 1, gates, segmentCache, TELEMETRY_STORAGE);
+        Mockito.doNothing().when(segmentFetcher).callLoopRun(Mockito.anyBoolean(),Mockito.anyBoolean());
+        Mockito.when(segmentFetcher.runWhitCacheHeader()).thenReturn(false);
+        Mockito.when(segmentFetcher.fetchAndUpdate(Mockito.anyBoolean())).thenReturn(false);
+        Mockito.doNothing().when(segmentFetcher).callLoopRun(Mockito.anyBoolean(),Mockito.anyBoolean());
+
+        // Before executing, we'll update the map of segmentFecthers via reflection.
+        Field backoffBase = SegmentSynchronizationTaskImp.class.getDeclaredField("_segmentFetchers");
+        backoffBase.setAccessible(true);
+        Field modifiersField = Field.class.getDeclaredField("modifiers");
+        modifiersField.setAccessible(true);
+        modifiersField.setInt(backoffBase, backoffBase.getModifiers() & ~Modifier.FINAL);
+
+        backoffBase.set(fetchers, _segmentFetchers); // 1ms
+        fetcher1.set(segmentFetcher);
+        boolean fetch = fetchers.fetchAllSynchronous();
+        Assert.assertEquals(false, fetch);
+    }
 
+    @Test
+    public void testFetchAllAsynchronousAndGetTrue() throws NoSuchFieldException, IllegalAccessException {
+        SDKReadinessGates gates = new SDKReadinessGates();
+        SegmentCache segmentCache = Mockito.mock(SegmentCache.class);
+
+        SegmentChangeFetcher segmentChangeFetcher = Mockito.mock(SegmentChangeFetcher.class);
+        SegmentFetcherImp segmentFetcher = Mockito.mock(SegmentFetcherImp.class);
+        final SegmentSynchronizationTaskImp fetchers = new SegmentSynchronizationTaskImp(segmentChangeFetcher, 1L, 1, gates, segmentCache, TELEMETRY_STORAGE);
+
+        // Before executing, we'll update the map of segmentFecthers via reflection.
+        Field backoffBase = SegmentSynchronizationTaskImp.class.getDeclaredField("_segmentFetchers");
+        backoffBase.setAccessible(true);
+        Field modifiersField = Field.class.getDeclaredField("modifiers");
+        modifiersField.setAccessible(true);
+        modifiersField.setInt(backoffBase, backoffBase.getModifiers() & ~Modifier.FINAL);
+        Mockito.doNothing().when(segmentFetcher).callLoopRun(Mockito.anyBoolean(),Mockito.anyBoolean());
+        Mockito.when(segmentFetcher.runWhitCacheHeader()).thenReturn(true);
+        Mockito.when(segmentFetcher.fetchAndUpdate(Mockito.anyBoolean())).thenReturn(true);
+        boolean fetch = fetchers.fetchAllSynchronous();
+        Assert.assertEquals(true, fetch);
+    }
 }