ExponentialBackoffRetryLogic implements RetryLogic { @Override public <T> Publisher<T> retryRx( Publisher<T> work ) { return Flux.from( work ).retryWhen( retryRxCondition() ); } 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 nextDelayCalculatedAccordingToMultiplierRx() { 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 ); Mono<Object> single = Flux.from( retryRx( retryLogic, retries, result ) ).single(); assertEquals( result, await( single ) ); assertEquals( delaysWithoutJitter( initialDelay, multiplier, retries ), eventExecutor.scheduleDelays() ); }
@Test void nextDelayCalculatedAccordingToJitterRx() { 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 ) ); Mono<Object> single = Flux.from( retryRx( retryLogic, retries, result ) ).single(); assertEquals( result, await( single ) ); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); List<Long> delaysWithoutJitter = delaysWithoutJitter( initialDelay, multiplier, retries ); assertDelaysApproximatelyEqual( delaysWithoutJitter, scheduleDelays, jitterFactor ); }
@Test void doesNotRetryWhenMaxRetryTimeExceededRx() { 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 ); SessionExpiredException error = sessionExpired(); AtomicInteger executionCount = new AtomicInteger(); Publisher<Object> publisher = retryLogic.retryRx( Mono.error( error ).doOnTerminate( executionCount::getAndIncrement ) ); Exception e = assertThrows( Exception.class, () -> await( publisher ) ); 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() ); assertThat( executionCount.get(), equalTo( 3 ) ); }
@Test void collectsSuppressedErrorsRx() throws Exception { 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 logic = newRetryLogic( maxRetryTime, initialDelay, multiplier, 0, clock ); SessionExpiredException error1 = sessionExpired(); SessionExpiredException error2 = sessionExpired(); ServiceUnavailableException error3 = serviceUnavailable(); TransientException error4 = transientException(); Mono<String> mono = createMono( "A result", error1, error2, error3, error4 ); StepVerifier.create( logic.retryRx( mono ) ).expectErrorSatisfies( e -> { assertEquals( error4, e ); Throwable[] suppressed = e.getSuppressed(); assertEquals( 3, suppressed.length ); assertEquals( error1, suppressed[0] ); assertEquals( error2, suppressed[1] ); assertEquals( error3, suppressed[2] ); } ).verify(); 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 doesNotCollectSuppressedErrorsWhenSameErrorIsThrownRx() { 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 ); SessionExpiredException error = sessionExpired(); StepVerifier.create( retryLogic.retryRx( Mono.error( error ) ) ).expectErrorSatisfies( e -> assertEquals( error, e ) ).verify(); List<Long> scheduleDelays = eventExecutor.scheduleDelays(); assertEquals( 2, scheduleDelays.size() ); assertEquals( initialDelay, scheduleDelays.get( 0 ).intValue() ); assertEquals( initialDelay * multiplier, scheduleDelays.get( 1 ).intValue() ); }
@Test void doesRetryOnClientExceptionWithRetryableCauseRx() { 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( Mono.from( logic.retryRx( Mono.fromSupplier( () -> { if ( exceptionThrown.compareAndSet( false, true ) ) { throw clientExceptionWithValidTerminationCause(); } return "Done"; } ) ) ) ); assertEquals( "Done", result ); }
@Test void doesNotRetryOnRandomClientExceptionRx() { 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( Mono.from( logic.retryRx( Mono.fromSupplier( () -> { if ( exceptionThrown.compareAndSet( false, true ) ) { throw randomClientException(); } return "Done"; } ) ) ) ) ); assertEquals( "Meeh", exception.getMessage() ); }
@Test void eachRetryIsLoggedRx() { 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( Flux.from( retryRx( logic, retries, result ) ).single() ) ); verify( logger, times( retries ) ).warn( startsWith( "Reactive transaction failed and is scheduled to retry" ), any( ServiceUnavailableException.class ) ); }
@Test void nothingIsLoggedOnFatalFailureRx() { 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 ); Publisher<Object> retryRx = logic.retryRx( Mono.error( new RuntimeException( "Fatal rx" ) ) ); RuntimeException error = assertThrows( RuntimeException.class, () -> await( retryRx ) ); assertEquals( "Fatal rx", error.getMessage() ); verifyZeroInteractions( logger ); }
@Test void correctNumberOfRetiesAreLoggedOnFailureRx() { 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 ); AtomicBoolean invoked = new AtomicBoolean( false ); SessionExpiredException error = assertThrows( SessionExpiredException.class, () -> await( logic.retryRx( Mono.create( e -> { if ( invoked.get() ) { when( clock.millis() ).thenReturn( settings.maxRetryTimeMs() + 42 ); } else { invoked.set( true ); } e.error( new SessionExpiredException( "Session no longer valid" ) ); } ) ) ) ); assertEquals( "Session no longer valid", error.getMessage() ); verify( logger ).warn( startsWith( "Reactive transaction failed and is scheduled to retry" ), any( SessionExpiredException.class ) ); }
@Test void shouldRetryWithBackOffRx() { Exception exception = new TransientException( "Unknown", "Retry this error." ); Clock clock = mock( Clock.class ); when( clock.millis() ).thenReturn( 0L, 100L, 200L, 400L, 800L ); ExponentialBackoffRetryLogic retryLogic = new ExponentialBackoffRetryLogic( 500, 100, 2, 0, eventExecutor, clock, DEV_NULL_LOGGING ); Flux<Integer> source = Flux.concat( Flux.range( 0, 2 ), Flux.error( exception ) ); Flux<Integer> retriedSource = Flux.from( retryLogic.retryRx( source ) ); StepVerifier.create( retriedSource ) .expectNext( 0, 1 ) .expectNext( 0, 1, 0, 1, 0, 1, 0, 1 ) .verifyErrorSatisfies( e -> assertThat( e, equalTo( exception ) ) ); List<Long> delays = eventExecutor.scheduleDelays(); assertThat( delays.size(), equalTo( 4 ) ); assertThat( delays, contains( 100L, 200L, 400L, 800L ) ); }
@Test void shouldRetryWithRandomBackOffRx() { Exception exception = new TransientException( "Unknown", "Retry this error." ); Clock clock = mock( Clock.class ); when( clock.millis() ).thenReturn( 0L, 100L, 200L, 400L, 800L ); ExponentialBackoffRetryLogic retryLogic = new ExponentialBackoffRetryLogic( 500, 100, 2, 0.1, eventExecutor, clock, DEV_NULL_LOGGING ); Flux<Integer> source = Flux.concat( Flux.range( 0, 2 ), Flux.error( exception ) ); Flux<Integer> retriedSource = Flux.from( retryLogic.retryRx( source ) ); StepVerifier.create( retriedSource ) .expectNext( 0, 1 ) .expectNext( 0, 1, 0, 1, 0, 1, 0, 1 ) .verifyErrorSatisfies( e -> assertThat( e, equalTo( exception ) ) ); List<Long> delays = eventExecutor.scheduleDelays(); assertThat( delays.size(), equalTo( 4 ) ); assertThat( delays.get( 0 ), allOf( greaterThanOrEqualTo( 90L ), lessThanOrEqualTo( 110L ) ) ); assertThat( delays.get( 1 ), allOf( greaterThanOrEqualTo( 180L ), lessThanOrEqualTo( 220L ) ) ); assertThat( delays.get( 2 ), allOf( greaterThanOrEqualTo( 260L ), lessThanOrEqualTo( 440L ) ) ); assertThat( delays.get( 3 ), allOf( greaterThanOrEqualTo( 720L ), lessThanOrEqualTo( 880L ) ) ); }