servicetalk-loadbalancer/src/test/java/io/servicetalk/loadbalancer/RoundRobinLoadBalancerTest.java
// Concurrency test, run multiple times (at least 1000).
Concurrency test, run multiple times (at least 1000).
Concurrency test run multiple times ( at least 1000 ) .
@Test
    void hostUnhealthyDoesntRaceToRunHealthCheck() throws Exception {
        serviceDiscoveryPublisher.onComplete();
        final Single<TestLoadBalancedConnection> properConnection = newRealizedConnectionSingle("address-1");
        final int timeAdvancementsTillHealthy = 3;
        final UnhealthyHostConnectionFactory unhealthyHostConnectionFactory = new UnhealthyHostConnectionFactory(
                "address-1", timeAdvancementsTillHealthy, properConnection);
        final DelegatingConnectionFactory connectionFactory = unhealthyHostConnectionFactory.createFactory();
        lb = defaultLb(connectionFactory);
        sendServiceDiscoveryEvents(upEvent("address-1"));
        ExecutorService executor = Executors.newFixedThreadPool(3);
        try {
            final Runnable runnable = () ->
                    assertThrows(ExecutionException.class, () -> lb.selectConnection(any()).toFuture().get());
            for (int i = 0; i < 1000; i++) {
                executor.submit(runnable);
            }
            final Executor executorForRetries = io.servicetalk.concurrent.api.Executors.newFixedSizeExecutor(1);
            try {
                awaitIndefinitely(lb.selectConnection(any()).retryWhen(retryWithConstantBackoffFullJitter((t) ->
                                t instanceof DeliberateException || testExecutor.scheduledTasksPending() == 0,
                        Duration.ofMillis(30), executorForRetries)));
            } catch (Exception e) {
                assertThat(e.getCause(), instanceOf(NoAvailableHostException.class));
            } finally {
                executorForRetries.closeAsync().toFuture().get();
            }
            for (int i = 0; i < timeAdvancementsTillHealthy - 1; ++i) {
                unhealthyHostConnectionFactory.advanceTime(testExecutor);
                Exception e = assertThrows(ExecutionException.class, () -> lb.selectConnection(any()).toFuture().get());
                assertThat(e.getCause(), instanceOf(NoAvailableHostException.class));
            }
        } finally {
            executor.shutdownNow();
            executor.awaitTermination(10, SECONDS);
        }
        unhealthyHostConnectionFactory.advanceTime(testExecutor);
        final TestLoadBalancedConnection selectedConnection = lb.selectConnection(any()).toFuture().get();
        assertThat(selectedConnection, equalTo(properConnection.toFuture().get()));
    }
@ Test void hostUnhealthyDoesntRaceToRunHealthCheck ( ) throws Exception { serviceDiscoveryPublisher . onComplete ( ) ; final Single < TestLoadBalancedConnection > properConnection = newRealizedConnectionSingle ( "address-1" ) ; final int timeAdvancementsTillHealthy = 3 ; final UnhealthyHostConnectionFactory unhealthyHostConnectionFactory = new UnhealthyHostConnectionFactory ( "address-1" , timeAdvancementsTillHealthy , properConnection ) ; final DelegatingConnectionFactory connectionFactory = unhealthyHostConnectionFactory . createFactory ( ) ; lb = defaultLb ( connectionFactory ) ; sendServiceDiscoveryEvents ( upEvent ( "address-1" ) ) ; ExecutorService executor = Executors . newFixedThreadPool ( 3 ) ; try { final Runnable runnable = ( ) -> assertThrows ( ExecutionException . class , ( ) -> lb . selectConnection ( any ( ) ) . toFuture ( ) . get ( ) ) ; for ( int i = 0 ; i < 1000 ; i ++ ) { executor . submit ( runnable ) ; } final Executor executorForRetries = io . servicetalk . concurrent . api . Executors . newFixedSizeExecutor ( 1 ) ; try { awaitIndefinitely ( lb . selectConnection ( any ( ) ) . retryWhen ( retryWithConstantBackoffFullJitter ( ( t ) -> t instanceof DeliberateException || testExecutor . scheduledTasksPending ( ) == 0 , Duration . ofMillis ( 30 ) , executorForRetries ) ) ) ; } catch ( Exception e ) { assertThat ( e . getCause ( ) , instanceOf ( NoAvailableHostException . class ) ) ; } finally { executorForRetries . closeAsync ( ) . toFuture ( ) . get ( ) ; } for ( int i = 0 ; i < timeAdvancementsTillHealthy - 1 ; ++ i ) { unhealthyHostConnectionFactory . advanceTime ( testExecutor ) ; Exception e = assertThrows ( ExecutionException . class , ( ) -> lb . selectConnection ( any ( ) ) . toFuture ( ) . get ( ) ) ; assertThat ( e . getCause ( ) , instanceOf ( NoAvailableHostException . class ) ) ; } } finally { executor . shutdownNow ( ) ; executor . awaitTermination ( 10 , SECONDS ) ; } unhealthyHostConnectionFactory . advanceTime ( testExecutor ) ; final TestLoadBalancedConnection selectedConnection = lb . selectConnection ( any ( ) ) . toFuture ( ) . get ( ) ; assertThat ( selectedConnection , equalTo ( properConnection . toFuture ( ) . get ( ) ) ) ; }
Concurrency test, run multiple times (at least 1000).
Concurrency test run multiple times ( at least 1000 ) .
// Imitate concurrency by running multiple threads attempting to establish connections. // From test main thread, wait until the host becomes UNHEALTHY, which is apparent from // NoHostAvailableException being thrown from selection AFTER a health check was scheduled by any thread. // DeliberateException comes from connection opening, check for that first // Next, NoAvailableHostException is thrown when the host is unhealthy, // but we still wait until the health check is scheduled and only then stop retrying. // try to prevent stack overflow // At this point, either the above selection caused the host to be marked as UNHEALTHY, // or any background thread. We also know that a health check is pending to be executed. // Now we can validate if there is just one health check happening and confirm that by asserting the host // is not selected. If our assumption doesn't hold, it means more than one health check was scheduled. // Assert still unhealthy // Shutdown the concurrent validation of unhealthiness.