}
			}
			if test.wg == nil {
				return
			}
			if test.wg.inflight != 0 {
				t.Errorf("expected wait group inflight to be zero, but got: %d", test.wg.inflight)
			}
			if test.wgInvokedExpected != test.wg.invoked {
				t.Errorf("expected wait group Add to be invoked: %d times, but got: %d", test.wgInvokedExpected, test.wg.invoked)

				return // quietly
			}
			if err == io.EOF {
				c.mu.Lock()
				inFlight := len(c.inFlight)
				c.mu.Unlock()
				base.Fatalf("GOCACHEPROG exited pre-Close with %v pending requests", inFlight)
			}
			base.Fatalf("error reading JSON from GOCACHEPROG: %v", err)
		}
		c.mu.Lock()
		ch, ok := c.inFlight[res.ID]
		delete(c.inFlight, res.ID)
		c.mu.Unlock()
		if ok {
			ch <- res
		} else {

		},
		[]string{"method", "path", "server_type", "long_running"},
	)
	// HTTPInflightRequests tracks the number of the inflight http requests.
	HTTPInflightRequests = metrics.NewGaugeVec(
		&metrics.GaugeOpts{
			Subsystem:      kubeletSubsystem,
			Name:           "http_inflight_requests",
			Help:           "Number of the inflight http requests",
			StabilityLevel: metrics.ALPHA,
		},
		[]string{"method", "path", "server_type", "long_running"},

	}

	if s.MaxRequestsInFlight < 0 {
		errors = append(errors, fmt.Errorf("--max-requests-inflight can not be negative value"))
	}
	if s.MaxMutatingRequestsInFlight < 0 {
		errors = append(errors, fmt.Errorf("--max-mutating-requests-inflight can not be negative value"))
	}

	if s.RequestTimeout.Nanoseconds() < 0 {

	if idx < 0 {
		return 0, fmt.Errorf("tried to end event %v, but not in-flight", go122.Specs()[st].Name)
	}
	// Swap remove.
	desc := s.inFlight[idx].desc
	s.inFlight[idx], s.inFlight[len(s.inFlight)-1] = s.inFlight[len(s.inFlight)-1], s.inFlight[idx]
	s.inFlight = s.inFlight[:len(s.inFlight)-1]
	return desc, nil
}

// seqCounter represents a global sequence counter for a resource.
type seqCounter struct {

 *
 * <pre>{@code
 * final String name = ...;
 * inFlight.add(name);
 * ListenableFuture<Result> future = service.query(name);
 * future.addListener(new Runnable() {
 *   public void run() {
 *     processedCount.incrementAndGet();
 *     inFlight.remove(name);
 *     lastProcessed.set(name);
 *     logger.info("Done with {0}", name);
 *   }
 * }, executor);

    if [[ "${NUM_NODES}" -gt 3000 ]]; then
      params=$(append-param-if-not-present "${params}" "max-requests-inflight" 3000)
      params=$(append-param-if-not-present "${params}" "max-mutating-requests-inflight" 1000)
    elif [[ "${NUM_NODES}" -gt 500 ]]; then
      params=$(append-param-if-not-present "${params}" "max-requests-inflight" 1500)

		<-inflight.blockedCh
		w.WriteHeader(http.StatusOK)
	})
	s.Handler.NonGoRestfulMux.Handle(inflight.url, handler)
	return inflight
}

func setupInFlightWatchRequestHandler(s *GenericAPIServer) *inFlightRequest {
	inflight := &inFlightRequest{
		blockedCh: make(chan struct{}),
		startedCh: make(chan struct{}),
		resultCh:  make(chan result),
		url:       "/apis/watches.group/v1/namespaces/foo/bar?watch=true",
	}

	fcmetrics "k8s.io/apiserver/pkg/util/flowcontrol/metrics"

	"k8s.io/klog/v2"
)

const (
	// Constant for the retry-after interval on rate limiting.
	retryAfter = "1"

	// How often inflight usage metric should be updated. Because
	// the metrics tracks maximal value over period making this
	// longer will increase the metric value.
	inflightUsageMetricUpdatePeriod = time.Second
)

var (

package envoy

import (
	"bytes"
	"fmt"
	"io"
	"net/http"
	"strings"

	"istio.io/istio/pkg/log"
)

// DrainListeners drains inbound listeners of Envoy so that inflight requests
// can gracefully finish and even continue making outbound calls as needed.
func DrainListeners(adminPort uint32, inboundonly bool, skipExit bool) error {
	var drainURL string
	if inboundonly {