private void waitUntilDeploymentIsScaled(final int count) {
    final BlockingQueue<Object> queue = new ArrayBlockingQueue<>(1);
    final AtomicReference<Integer> replicasRef = new AtomicReference<>(0);

    final String name = checkName(getItem());
    final String namespace = checkNamespace(getItem());

    final Runnable deploymentPoller = () -> {
      try {
        Deployment deployment = get();
        //If the deployment is gone, we shouldn't wait.
        if (deployment == null) {
          if (count == 0) {
            queue.put(true);
            return;
          } else {
            queue.put(new IllegalStateException("Can't wait for Deployment: " + checkName(getItem()) + " in namespace: " + checkName(getItem()) + " to scale. Resource is no longer available."));
            return;
          }
        }

        replicasRef.set(deployment.getStatus().getReplicas());
        int currentReplicas = deployment.getStatus().getReplicas() != null ? deployment.getStatus().getReplicas() : 0;
        long generation = deployment.getMetadata().getGeneration() != null ? deployment.getMetadata().getGeneration() : 0;
        long observedGeneration = deployment.getStatus() != null && deployment.getStatus().getObservedGeneration() != null ? deployment.getStatus().getObservedGeneration() : -1;
        if (observedGeneration >= generation && Objects.equals(deployment.getSpec().getReplicas(), currentReplicas)) {
          queue.put(true);
        } else {
          LOG.debug("Only {}/{} pods scheduled for Deployment: {} in namespace: {} seconds so waiting...",
            deployment.getStatus().getReplicas(), deployment.getSpec().getReplicas(), deployment.getMetadata().getName(), namespace);
        }
      } catch (Throwable t) {
        LOG.error("Error while waiting for Deployment to be scaled.", t);
      }
    };

    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
    ScheduledFuture poller = executor.scheduleWithFixedDelay(deploymentPoller, 0, POLL_INTERVAL_MS, TimeUnit.MILLISECONDS);
    try {
      if (Utils.waitUntilReady(queue, getConfig().getScaleTimeout(), TimeUnit.MILLISECONDS)) {
        LOG.debug("{}/{} pod(s) ready for Deployment: {} in namespace: {}.",
          replicasRef.get(), count, name, namespace);
      } else {
        LOG.error("{}/{} pod(s) ready for Deployment: {} in namespace: {}  after waiting for {} seconds so giving up",
          replicasRef.get(), count, name, namespace, TimeUnit.MILLISECONDS.toSeconds(getConfig().getScaleTimeout()));
      }
    } finally {
      poller.cancel(true);
      executor.shutdown();
    }
  }
private void waitUntilDeploymentIsScaled ( final int count ) { final BlockingQueue < Object > queue = new ArrayBlockingQueue < > ( 1 ) ; final AtomicReference < Integer > replicasRef = new AtomicReference < > ( 0 ) ; final String name = checkName ( getItem ( ) ) ; final String namespace = checkNamespace ( getItem ( ) ) ; final Runnable deploymentPoller = ( ) -> { try { Deployment deployment = get ( ) ; if ( deployment == null ) { if ( count == 0 ) { queue . put ( true ) ; return ; } else { queue . put ( new IllegalStateException ( "Can't wait for Deployment: " + checkName ( getItem ( ) ) + " in namespace: " + checkName ( getItem ( ) ) + " to scale. Resource is no longer available." ) ) ; return ; } } replicasRef . set ( deployment . getStatus ( ) . getReplicas ( ) ) ; int currentReplicas = deployment . getStatus ( ) . getReplicas ( ) != null ? deployment . getStatus ( ) . getReplicas ( ) : 0 ; long generation = deployment . getMetadata ( ) . getGeneration ( ) != null ? deployment . getMetadata ( ) . getGeneration ( ) : 0 ; long observedGeneration = deployment . getStatus ( ) != null && deployment . getStatus ( ) . getObservedGeneration ( ) != null ? deployment . getStatus ( ) . getObservedGeneration ( ) : - 1 ; if ( observedGeneration >= generation && Objects . equals ( deployment . getSpec ( ) . getReplicas ( ) , currentReplicas ) ) { queue . put ( true ) ; } else { LOG . debug ( "Only {}/{} pods scheduled for Deployment: {} in namespace: {} seconds so waiting..." , deployment . getStatus ( ) . getReplicas ( ) , deployment . getSpec ( ) . getReplicas ( ) , deployment . getMetadata ( ) . getName ( ) , namespace ) ; } } catch ( Throwable t ) { LOG . error ( "Error while waiting for Deployment to be scaled." , t ) ; } } ; ScheduledExecutorService executor = Executors . newSingleThreadScheduledExecutor ( ) ; ScheduledFuture poller = executor . scheduleWithFixedDelay ( deploymentPoller , 0 , POLL_INTERVAL_MS , TimeUnit . MILLISECONDS ) ; try { if ( Utils . waitUntilReady ( queue , getConfig ( ) . getScaleTimeout ( ) , TimeUnit . MILLISECONDS ) ) { LOG . debug ( "{}/{} pod(s) ready for Deployment: {} in namespace: {}." , replicasRef . get ( ) , count , name , namespace ) ; } else { LOG . error ( "{}/{} pod(s) ready for Deployment: {} in namespace: {}  after waiting for {} seconds so giving up" , replicasRef . get ( ) , count , name , namespace , TimeUnit . MILLISECONDS . toSeconds ( getConfig ( ) . getScaleTimeout ( ) ) ) ; } } finally { poller . cancel ( true ) ; executor . shutdown ( ) ; } }
Lets wait until there are enough Ready pods of the given Deployment