private void waitUntilScaled ( final int count ) { final ArrayBlockingQueue < Object > queue = new ArrayBlockingQueue <> ( 1 ) ; final AtomicReference < Integer > replicasRef = new AtomicReference <> ( 0 ) ; final String name = checkName ( getItem ( ) ) ; final String namespace = checkNamespace ( getItem ( ) ) ; final Runnable tPoller = ( ) -> { try { T t = get ( ) ; //If the resource is gone, we shouldn't wait. if ( t == null ) { if ( count == 0 ) { queue . put ( true ) ; } else { queue . put ( new IllegalStateException ( "Can't wait for " + getType ( ) . getSimpleName ( ) + ": " + name + " in namespace: " + namespace + " to scale. Resource is no longer available." ) ) ; } return ; } int currentReplicas = getCurrentReplicas ( t ) ; int desiredReplicas = getDesiredReplicas ( t ) ; replicasRef . set ( currentReplicas ) ; long generation = t . getMetadata ( ) . getGeneration ( ) != null ? t . getMetadata ( ) . getGeneration ( ) : - 1 ; long observedGeneration = getObservedGeneration ( t ) ; if ( observedGeneration >= generation && Objects . equals ( desiredReplicas , currentReplicas ) ) { queue . put ( true ) ; } Log . debug ( "Only {}/{} replicas scheduled for {}: {} in namespace: {} seconds so waiting..." , currentReplicas , desiredReplicas , t . getKind ( ) , t . getMetadata ( ) . getName ( ) , namespace ) ; } catch ( Throwable t ) { Log . error ( "Error while waiting for resource to be scaled." , t ) ; } } ; ScheduledExecutorService executor = Executors . newSingleThreadScheduledExecutor ( ) ; ScheduledFuture poller = executor . scheduleWithFixedDelay ( tPoller , 0 , POLL_INTERVAL_MS , TimeUnit . MILLISECONDS ) ; try { if ( Utils . waitUntilReady ( queue , rollingTimeout , rollingTimeUnit ) ) { Log . debug ( "{}/{} pod(s) ready for {}: {} in namespace: {}." , replicasRef . get ( ) , count , getType ( ) . getSimpleName ( ) , name , namespace ) ; } else { Log . error ( "{}/{} pod(s) ready for {}: {} in namespace: {} after waiting for {} seconds so giving up" , replicasRef . get ( ) , count , getType ( ) . getSimpleName ( ) , name , namespace , rollingTimeUnit . toSeconds ( rollingTimeout ) ) ; } } finally { poller . cancel ( true ) ; executor . shutdown ( ) ; } }
Let s wait until there are enough Ready pods .