Development

client/CHANGES.txt

@@ -1,5 +1,8 @@
 CHANGES
 
+4.1.5 (Apr 6, 2021)
+-Updated: Streaming retry fix.
+
 4.1.4 (Mar 19, 2021)
 - Updated: Internal cache structure refactor.
 - Updated: Streaming revamp with several bugfixes and improved log messages.

client/pom.xml

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

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

@@ -19,6 +19,7 @@
 
 public class SynchronizerImp implements Synchronizer {
     private static final Logger _log = LoggerFactory.getLogger(Synchronizer.class);
+    private static final int RETRIES_NUMBER = 10;
 
     private final SplitSynchronizationTask _splitSynchronizationTask;
     private final SplitFetcher _splitFetcher;
@@ -69,8 +70,10 @@ public void stopPeriodicFetching() {
 
     @Override
     public void refreshSplits(long targetChangeNumber) {
-        if (targetChangeNumber > _splitCache.getChangeNumber()) {
+        int retries = 1;
+        while(targetChangeNumber > _splitCache.getChangeNumber() && retries <= RETRIES_NUMBER) {
             _splitFetcher.forceRefresh(true);
+            retries++;
         }
     }
 
@@ -84,7 +87,8 @@ public void localKillSplit(String splitName, String defaultTreatment, long newCh
 
     @Override
     public void refreshSegment(String segmentName, long changeNumber) {
-        if (changeNumber > _segmentCache.getChangeNumber(segmentName)) {
+        int retries = 1;
+        while(changeNumber > _segmentCache.getChangeNumber(segmentName) && retries <= RETRIES_NUMBER) {
             SegmentFetcher fetcher = _segmentSynchronizationTaskImp.getFetcher(segmentName);
             try{
                 fetcher.fetch(true);
@@ -93,6 +97,7 @@ public void refreshSegment(String segmentName, long changeNumber) {
             catch (NullPointerException np){
                 throw new NullPointerException();
             }
+            retries++;
         }
     }
 }

client/src/test/java/io/split/engine/common/SynchronizerTest.java

@@ -4,6 +4,7 @@
 import io.split.cache.SplitCache;
 import io.split.engine.experiments.SplitFetcherImp;
 import io.split.engine.experiments.SplitSynchronizationTask;
+import io.split.engine.segments.SegmentFetcher;
 import io.split.engine.segments.SegmentSynchronizationTask;
 import org.junit.Before;
 import org.junit.Test;
@@ -15,14 +16,15 @@ public class SynchronizerTest {
     private SplitFetcherImp _splitFetcher;
     private SplitCache _splitCache;
     private Synchronizer _synchronizer;
+    private SegmentCache _segmentCache;
 
     @Before
     public void beforeMethod() {
         _refreshableSplitFetcherTask = Mockito.mock(SplitSynchronizationTask.class);
         _segmentFetcher = Mockito.mock(SegmentSynchronizationTask.class);
         _splitFetcher = Mockito.mock(SplitFetcherImp.class);
         _splitCache = Mockito.mock(SplitCache.class);
-        SegmentCache _segmentCache = Mockito.mock(SegmentCache.class);
+        _segmentCache = Mockito.mock(SegmentCache.class);
 
         _synchronizer = new SynchronizerImp(_refreshableSplitFetcherTask, _splitFetcher, _segmentFetcher, _splitCache, _segmentCache);
     }
@@ -51,4 +53,23 @@ public void stopPeriodicFetching() {
         Mockito.verify(_refreshableSplitFetcherTask, Mockito.times(1)).stop();
         Mockito.verify(_segmentFetcher, Mockito.times(1)).stop();
     }
+
+    @Test
+    public void streamingRetryOnSplit() {
+        Mockito.when(_splitCache.getChangeNumber()).thenReturn(0l).thenReturn(0l).thenReturn(1l);
+        _synchronizer.refreshSplits(1l);
+
+        Mockito.verify(_splitCache, Mockito.times(3)).getChangeNumber();
+    }
+
+    @Test
+    public void streamingRetryOnSegment() {
+        SegmentFetcher fetcher = Mockito.mock(SegmentFetcher.class);
+        Mockito.when(_segmentFetcher.getFetcher(Mockito.anyString())).thenReturn(fetcher);
+        Mockito.when(_segmentCache.getChangeNumber(Mockito.anyString())).thenReturn(0l).thenReturn(0l).thenReturn(1l);
+        _synchronizer.refreshSegment("Segment",1l);
+
+        Mockito.verify(_segmentCache, Mockito.times(3)).getChangeNumber(Mockito.anyString());
+    }
+
 }

pom.xml

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

testing/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.1.4</version>
+        <version>4.1.5</version>
     </parent>
 
     <artifactId>java-client-testing</artifactId>