Skip to content

Streaming retry #199

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Apr 6, 2021
Merged

Streaming retry #199

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ecc9ed1
Fixing streaming retry
ldecheverz-split Mar 29, 2021
f93ae28
Adding tests
ldecheverz-split Apr 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions client/src/main/java/io/split/engine/common/SynchronizerImp.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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++;
}
}

Expand All @@ -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);
Expand All @@ -93,6 +97,7 @@ public void refreshSegment(String segmentName, long changeNumber) {
catch (NullPointerException np){
throw new NullPointerException();
}
retries++;
}
}
}
23 changes: 22 additions & 1 deletion client/src/test/java/io/split/engine/common/SynchronizerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);
}
Expand Down Expand Up @@ -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());
}

}