Merge pull request #208 from splitio/stream-retry-delay

Adding delay on streaming retry

Author: ldecheverz-split

client/src/main/java/io/split/client/SplitClientConfig.java

@@ -46,6 +46,7 @@ public class SplitClientConfig {
     private final int _streamingReconnectBackoffBase;
     private final String _authServiceURL;
     private final String _streamingServiceURL;
+    private final int _onDemandFetchRetryDelayMs;
 
     // Proxy configs
     private final HttpHost _proxy;
@@ -89,7 +90,8 @@ private SplitClientConfig(String endpoint,
                               int authRetryBackoffBase,
                               int streamingReconnectBackoffBase,
                               String authServiceURL,
-                              String streamingServiceURL) {
+                              String streamingServiceURL,
+                              int onDemandFetchRetryDelayMs) {
         _endpoint = endpoint;
         _eventsEndpoint = eventsEndpoint;
         _featuresRefreshRate = pollForFeatureChangesEveryNSeconds;
@@ -120,6 +122,7 @@ private SplitClientConfig(String endpoint,
         _streamingReconnectBackoffBase = streamingReconnectBackoffBase;
         _authServiceURL = authServiceURL;
         _streamingServiceURL = streamingServiceURL;
+        _onDemandFetchRetryDelayMs = onDemandFetchRetryDelayMs;
 
         Properties props = new Properties();
         try {
@@ -248,6 +251,8 @@ public String streamingServiceURL() {
         return _streamingServiceURL;
     }
 
+    public int streamingRetryDelay() {return _onDemandFetchRetryDelayMs;}
+
     public static final class Builder {
 
         private String _endpoint = "https://sdk.split.io";
@@ -283,6 +288,7 @@ public static final class Builder {
         private int _streamingReconnectBackoffBase = 1;
         private String _authServiceURL = "https://auth.split.io/api/auth";
         private String _streamingServiceURL = "https://streaming.split.io/sse";
+        private int _onDemandFetchRetryDelayMs = 50;
 
         public Builder() {
         }
@@ -674,6 +680,16 @@ public Builder streamingServiceURL(String streamingServiceURL) {
             return this;
         }
 
+        /**
+         * Set Streaming retry delay.
+         * @param onDemandFetchRetryDelayMs
+         * @return
+         */
+        public Builder streamingRetryDelay(int onDemandFetchRetryDelayMs) {
+            _onDemandFetchRetryDelayMs = onDemandFetchRetryDelayMs;
+            return this;
+        }
+
         public SplitClientConfig build() {
             if (_featuresRefreshRate < 5 ) {
                 throw new IllegalArgumentException("featuresRefreshRate must be >= 5: " + _featuresRefreshRate);
@@ -744,6 +760,10 @@ public SplitClientConfig build() {
                 throw new IllegalArgumentException("streamingServiceURL must not be null");
             }
 
+            if(_onDemandFetchRetryDelayMs <= 0) {
+                throw new IllegalStateException("streamingRetryDelay must be > 0");
+            }
+
             return new SplitClientConfig(
                     _endpoint,
                     _eventsEndpoint,
@@ -774,7 +794,8 @@ public SplitClientConfig build() {
                     _authRetryBackoffBase,
                     _streamingReconnectBackoffBase,
                     _authServiceURL,
-                    _streamingServiceURL);
+                    _streamingServiceURL,
+                    _onDemandFetchRetryDelayMs);
         }
     }
 }

client/src/main/java/io/split/client/SplitFactoryImpl.java

@@ -141,7 +141,7 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         _eventClient = EventClientImpl.create(_httpclient, _eventsRootTarget, config.eventsQueueSize(), config.eventFlushIntervalInMillis(), config.waitBeforeShutdown());
 
         // SyncManager
-        _syncManager = SyncManagerImp.build(config.streamingEnabled(), _splitSynchronizationTask, _splitFetcher, _segmentSynchronizationTaskImp, _splitCache, config.authServiceURL(), _httpclient, config.streamingServiceURL(), config.authRetryBackoffBase(), buildSSEdHttpClient(config), _segmentCache);
+        _syncManager = SyncManagerImp.build(config.streamingEnabled(), _splitSynchronizationTask, _splitFetcher, _segmentSynchronizationTaskImp, _splitCache, config.authServiceURL(), _httpclient, config.streamingServiceURL(), config.authRetryBackoffBase(), buildSSEdHttpClient(config), _segmentCache, config.streamingRetryDelay());
         _syncManager.start();
 
         // Evaluator

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

@@ -59,9 +59,10 @@ public static SyncManagerImp build(boolean streamingEnabledConfig,
                                        String streamingServiceUrl,
                                        int authRetryBackOffBase,
                                        CloseableHttpClient sseHttpClient,
-                                       SegmentCache segmentCache) {
+                                       SegmentCache segmentCache,
+                                       int streamingRetryDelay) {
         LinkedBlockingQueue<PushManager.Status> pushMessages = new LinkedBlockingQueue<>();
-        Synchronizer synchronizer = new SynchronizerImp(splitSynchronizationTask, splitFetcher, segmentSynchronizationTaskImp, splitCache, segmentCache);
+        Synchronizer synchronizer = new SynchronizerImp(splitSynchronizationTask, splitFetcher, segmentSynchronizationTaskImp, splitCache, segmentCache, streamingRetryDelay);
         PushManager pushManager = PushManagerImp.build(synchronizer, streamingServiceUrl, authUrl, httpClient, pushMessages, sseHttpClient);
         return new SyncManagerImp(streamingEnabledConfig, synchronizer, pushManager, pushMessages, authRetryBackOffBase);
     }

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

@@ -27,17 +27,20 @@ public class SynchronizerImp implements Synchronizer {
     private final ScheduledExecutorService _syncAllScheduledExecutorService;
     private final SplitCache _splitCache;
     private final SegmentCache _segmentCache;
+    private final int _onDemandFetchRetryDelayMs;
 
     public SynchronizerImp(SplitSynchronizationTask splitSynchronizationTask,
                            SplitFetcher splitFetcher,
                            SegmentSynchronizationTask segmentSynchronizationTaskImp,
                            SplitCache splitCache,
-                           SegmentCache segmentCache) {
+                           SegmentCache segmentCache,
+                           int onDemandFetchRetryDelayMs) {
         _splitSynchronizationTask = checkNotNull(splitSynchronizationTask);
         _splitFetcher = checkNotNull(splitFetcher);
         _segmentSynchronizationTaskImp = checkNotNull(segmentSynchronizationTaskImp);
         _splitCache = checkNotNull(splitCache);
         _segmentCache = checkNotNull(segmentCache);
+        _onDemandFetchRetryDelayMs = checkNotNull(onDemandFetchRetryDelayMs);
 
         ThreadFactory splitsThreadFactory = new ThreadFactoryBuilder()
                 .setDaemon(true)
@@ -70,10 +73,23 @@ public void stopPeriodicFetching() {
 
     @Override
     public void refreshSplits(long targetChangeNumber) {
-        int retries = 1;
-        while(targetChangeNumber > _splitCache.getChangeNumber() && retries <= RETRIES_NUMBER) {
+        int retries = RETRIES_NUMBER;
+        while(targetChangeNumber > _splitCache.getChangeNumber()) {
+            retries--;
             _splitFetcher.forceRefresh(true);
-            retries++;
+            if (targetChangeNumber <= _splitCache.getChangeNumber()) {
+                _log.debug("Refresh completed in %s attempts.", RETRIES_NUMBER - retries);
+                return;
+            } else if (retries <= 0) {
+                _log.warn("No changes fetched after %s attempts.", RETRIES_NUMBER);
+                return;
+            }
+            try {
+                Thread.sleep(_onDemandFetchRetryDelayMs);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                _log.debug("Error trying to sleep current Thread.");
+            }
         }
     }
 

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

@@ -26,7 +26,7 @@ public void beforeMethod() {
         _splitCache = Mockito.mock(SplitCache.class);
         _segmentCache = Mockito.mock(SegmentCache.class);
 
-        _synchronizer = new SynchronizerImp(_refreshableSplitFetcherTask, _splitFetcher, _segmentFetcher, _splitCache, _segmentCache);
+        _synchronizer = new SynchronizerImp(_refreshableSplitFetcherTask, _splitFetcher, _segmentFetcher, _splitCache, _segmentCache, 50);
     }
 
     @Test