@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() )
                {
                    // move clock forward to stop retries
                    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 )
    );
}