ExponentialBackoffRetryLogic implements RetryLogic { @Override public <T> CompletionStage<T> retryAsync( Supplier<CompletionStage<T>> work ) { CompletableFuture<T> resultFuture = new CompletableFuture<>(); executeWorkInEventLoop( resultFuture, work ); return resultFuture; } ExponentialBackoffRetryLogic( RetrySettings settings, EventExecutorGroup eventExecutorGroup, Clock clock,
            Logging logging );   ExponentialBackoffRetryLogic( long maxRetryTimeMs, long initialRetryDelayMs, double multiplier,
            double jitterFactor, EventExecutorGroup eventExecutorGroup, Clock clock, Logging logging ); @Override T retry( Supplier<T> work ); @Override CompletionStage<T> retryAsync( Supplier<CompletionStage<T>> work ); @Override Publisher<T> retryRx( Publisher<T> work ); @Experimental static boolean isRetryable( Throwable error );  }
@Test void nextDelayCalculatedAccordingToMultiplierAsync() { String result = "The Result"; int retries = 14; int initialDelay = 1; int multiplier = 2; int noJitter = 0; ExponentialBackoffRetryLogic retryLogic = newRetryLogic( MAX_VALUE, initialDelay, multiplier, noJitter, Clock.SYSTEM ); CompletionStage<Object> future = retryAsync( retryLogic, retries, result ); assertEquals( result, await( future ) ); assertEquals( delaysWithoutJitter( initialDelay, multiplier, retries ), eventExecutor.scheduleDelays() ); }
@Test void nextDelayCalculatedAccordingToJitterAsync() { String result = "The Result"; int retries = 24; double jitterFactor = 0.2; int initialDelay = 1; int multiplier = 2; ExponentialBackoffRetryLogic retryLogic = newRetryLogic( MAX_VALUE, initialDelay, multiplier, jitterFactor, mock( Clock.class ) ); CompletionStage<Object> future = retryAsync( retryLogic, retries, result ); assertEquals( result, await( future ) ); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); List<Long> delaysWithoutJitter = delaysWithoutJitter( initialDelay, multiplier, retries ); assertDelaysApproximatelyEqual( delaysWithoutJitter, scheduleDelays, jitterFactor ); }
@Test void doesNotRetryWhenMaxRetryTimeExceededAsync() { long retryStart = Clock.SYSTEM.millis(); int initialDelay = 100; int multiplier = 2; long maxRetryTimeMs = 45; Clock clock = mock( Clock.class ); when( clock.millis() ).thenReturn( retryStart ) .thenReturn( retryStart + maxRetryTimeMs - 5 ) .thenReturn( retryStart + maxRetryTimeMs + 7 ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( maxRetryTimeMs, initialDelay, multiplier, 0, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); SessionExpiredException error = sessionExpired(); when( workMock.get() ).thenReturn( failedFuture( error ) ); CompletionStage<Object> future = retryLogic.retryAsync( workMock ); Exception e = assertThrows( Exception.class, () -> await( future ) ); assertEquals( error, e ); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); assertEquals( 2, scheduleDelays.size() ); assertEquals( initialDelay, scheduleDelays.get( 0 ).intValue() ); assertEquals( initialDelay * multiplier, scheduleDelays.get( 1 ).intValue() ); verify( workMock, times( 3 ) ).get(); }
@Test void schedulesRetryOnServiceUnavailableExceptionAsync() { String result = "The Result"; Clock clock = mock( Clock.class ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( 1, 42, 1, 0, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); ServiceUnavailableException error = serviceUnavailable(); when( workMock.get() ).thenReturn( failedFuture( error ) ).thenReturn( completedFuture( result ) ); assertEquals( result, await( retryLogic.retryAsync( workMock ) ) ); verify( workMock, times( 2 ) ).get(); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); assertEquals( 1, scheduleDelays.size() ); assertEquals( 42, scheduleDelays.get( 0 ).intValue() ); }
@Test void schedulesRetryOnSessionExpiredExceptionAsync() { String result = "The Result"; Clock clock = mock( Clock.class ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( 1, 4242, 1, 0, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); SessionExpiredException error = sessionExpired(); when( workMock.get() ).thenReturn( failedFuture( error ) ).thenReturn( completedFuture( result ) ); assertEquals( result, await( retryLogic.retryAsync( workMock ) ) ); verify( workMock, times( 2 ) ).get(); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); assertEquals( 1, scheduleDelays.size() ); assertEquals( 4242, scheduleDelays.get( 0 ).intValue() ); }
@Test void schedulesRetryOnTransientExceptionAsync() { String result = "The Result"; Clock clock = mock( Clock.class ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( 1, 23, 1, 0, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); TransientException error = transientException(); when( workMock.get() ).thenReturn( failedFuture( error ) ).thenReturn( completedFuture( result ) ); assertEquals( result, await( retryLogic.retryAsync( workMock ) ) ); verify( workMock, times( 2 ) ).get(); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); assertEquals( 1, scheduleDelays.size() ); assertEquals( 23, scheduleDelays.get( 0 ).intValue() ); }
@Test void doesNotRetryOnUnknownErrorAsync() { Clock clock = mock( Clock.class ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( 1, 1, 1, 1, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); IllegalStateException error = new IllegalStateException(); when( workMock.get() ).thenReturn( failedFuture( error ) ); Exception e = assertThrows( Exception.class, () -> await( retryLogic.retryAsync( workMock ) ) ); assertEquals( error, e ); verify( workMock ).get(); assertEquals( 0, eventExecutor.scheduleDelays().size() ); }
@Test void doesNotRetryOnTransactionTerminatedErrorAsync() { Clock clock = mock( Clock.class ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( 1, 13, 1, 0, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); TransientException error = new TransientException( "Neo.TransientError.Transaction.Terminated", "" ); when( workMock.get() ).thenReturn( failedFuture( error ) ); Exception e = assertThrows( Exception.class, () -> await( retryLogic.retryAsync( workMock ) ) ); assertEquals( error, e ); verify( workMock ).get(); assertEquals( 0, eventExecutor.scheduleDelays().size() ); }
@Test void doesNotRetryOnTransactionLockClientStoppedErrorAsync() { Clock clock = mock( Clock.class ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( 1, 15, 1, 0, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); TransientException error = new TransientException( "Neo.TransientError.Transaction.LockClientStopped", "" ); when( workMock.get() ).thenReturn( failedFuture( error ) ); Exception e = assertThrows( Exception.class, () -> await( retryLogic.retryAsync( workMock ) ) ); assertEquals( error, e ); verify( workMock ).get(); assertEquals( 0, eventExecutor.scheduleDelays().size() ); }
@Test void collectsSuppressedErrorsAsync() { String result = "The Result"; long maxRetryTime = 20; int initialDelay = 15; int multiplier = 2; Clock clock = mock( Clock.class ); when( clock.millis() ).thenReturn( 0L ).thenReturn( 10L ).thenReturn( 15L ).thenReturn( 25L ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( maxRetryTime, initialDelay, multiplier, 0, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); SessionExpiredException error1 = sessionExpired(); SessionExpiredException error2 = sessionExpired(); ServiceUnavailableException error3 = serviceUnavailable(); TransientException error4 = transientException(); when( workMock.get() ).thenReturn( failedFuture( error1 ) ) .thenReturn( failedFuture( error2 ) ) .thenReturn( failedFuture( error3 ) ) .thenReturn( failedFuture( error4 ) ) .thenReturn( completedFuture( result ) ); Exception e = assertThrows( Exception.class, () -> await( retryLogic.retryAsync( workMock ) ) ); assertEquals( error4, e ); Throwable[] suppressed = e.getSuppressed(); assertEquals( 3, suppressed.length ); assertEquals( error1, suppressed[0] ); assertEquals( error2, suppressed[1] ); assertEquals( error3, suppressed[2] ); verify( workMock, times( 4 ) ).get(); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); assertEquals( 3, scheduleDelays.size() ); assertEquals( initialDelay, scheduleDelays.get( 0 ).intValue() ); assertEquals( initialDelay * multiplier, scheduleDelays.get( 1 ).intValue() ); assertEquals( initialDelay * multiplier * multiplier, scheduleDelays.get( 2 ).intValue() ); }
@Test void doesNotCollectSuppressedErrorsWhenSameErrorIsThrownAsync() { long maxRetryTime = 20; int initialDelay = 15; int multiplier = 2; Clock clock = mock( Clock.class ); when( clock.millis() ).thenReturn( 0L ).thenReturn( 10L ).thenReturn( 25L ); ExponentialBackoffRetryLogic retryLogic = newRetryLogic( maxRetryTime, initialDelay, multiplier, 0, clock ); Supplier<CompletionStage<Object>> workMock = newWorkMock(); SessionExpiredException error = sessionExpired(); when( workMock.get() ).thenReturn( failedFuture( error ) ); Exception e = assertThrows( Exception.class, () -> await( retryLogic.retryAsync( workMock ) ) ); assertEquals( error, e ); assertEquals( 0, e.getSuppressed().length ); verify( workMock, times( 3 ) ).get(); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); assertEquals( 2, scheduleDelays.size() ); assertEquals( initialDelay, scheduleDelays.get( 0 ).intValue() ); assertEquals( initialDelay * multiplier, scheduleDelays.get( 1 ).intValue() ); }
@Test void doesRetryOnClientExceptionWithRetryableCauseAsync() { Clock clock = mock( Clock.class ); Logging logging = mock( Logging.class ); Logger logger = mock( Logger.class ); when( logging.getLog( anyString() ) ).thenReturn( logger ); ExponentialBackoffRetryLogic logic = new ExponentialBackoffRetryLogic( RetrySettings.DEFAULT, eventExecutor, clock, logging ); AtomicBoolean exceptionThrown = new AtomicBoolean( false ); String result = await( logic.retryAsync( () -> { if ( exceptionThrown.compareAndSet( false, true ) ) { throw clientExceptionWithValidTerminationCause(); } return CompletableFuture.completedFuture( "Done" ); } ) ); assertEquals( "Done", result ); }
@Test void doesNotRetryOnRandomClientExceptionAsync() { Clock clock = mock( Clock.class ); Logging logging = mock( Logging.class ); Logger logger = mock( Logger.class ); when( logging.getLog( anyString() ) ).thenReturn( logger ); ExponentialBackoffRetryLogic logic = new ExponentialBackoffRetryLogic( RetrySettings.DEFAULT, eventExecutor, clock, logging ); AtomicBoolean exceptionThrown = new AtomicBoolean( false ); ClientException exception = Assertions.assertThrows( ClientException.class, () -> await( logic.retryAsync( () -> { if ( exceptionThrown.compareAndSet( false, true ) ) { throw randomClientException(); } return CompletableFuture.completedFuture( "Done" ); } ) ) ); assertEquals( "Meeh", exception.getMessage() ); }
@Test void eachRetryIsLoggedAsync() { String result = "The Result"; int retries = 9; Clock clock = mock( Clock.class ); Logging logging = mock( Logging.class ); Logger logger = mock( Logger.class ); when( logging.getLog( anyString() ) ).thenReturn( logger ); ExponentialBackoffRetryLogic logic = new ExponentialBackoffRetryLogic( RetrySettings.DEFAULT, eventExecutor, clock, logging ); assertEquals( result, await( retryAsync( logic, retries, result ) ) ); verify( logger, times( retries ) ).warn( startsWith( "Async transaction failed and is scheduled to retry" ), any( ServiceUnavailableException.class ) ); }
@Test void nothingIsLoggedOnFatalFailureAsync() { Logging logging = mock( Logging.class ); Logger logger = mock( Logger.class ); when( logging.getLog( anyString() ) ).thenReturn( logger ); ExponentialBackoffRetryLogic logic = new ExponentialBackoffRetryLogic( RetrySettings.DEFAULT, eventExecutor, mock( Clock.class ), logging ); RuntimeException error = assertThrows( RuntimeException.class, () -> await( logic.retryAsync( () -> failedFuture( new RuntimeException( "Fatal async" ) ) ) ) ); assertEquals( "Fatal async", error.getMessage() ); verifyZeroInteractions( logger ); }
@Test void correctNumberOfRetiesAreLoggedOnFailureAsync() { Clock clock = mock( Clock.class ); Logging logging = mock( Logging.class ); Logger logger = mock( Logger.class ); when( logging.getLog( anyString() ) ).thenReturn( logger ); RetrySettings settings = RetrySettings.DEFAULT; RetryLogic logic = new ExponentialBackoffRetryLogic( settings, eventExecutor, clock, logging ); SessionExpiredException error = assertThrows( SessionExpiredException.class, () -> await( logic.retryAsync( new Supplier<CompletionStage<Void>>() { volatile boolean invoked; @Override public CompletionStage<Void> get() { if ( invoked ) { when( clock.millis() ).thenReturn( settings.maxRetryTimeMs() + 42 ); } else { invoked = true; } return failedFuture( new SessionExpiredException( "Session no longer valid" ) ); } } ) ) ); assertEquals( "Session no longer valid", error.getMessage() ); verify( logger ).warn( startsWith( "Async transaction failed and is scheduled to retry" ), any( SessionExpiredException.class ) ); }