return
		}

		// async timers can send one stale value (then the timer is disabled).
		drain1()
		if isTicker {
			// async tickers can send two stale values: there may be one
			// sitting in the channel buffer, and there may also be one
			// send racing with the Reset/Stop+drain that arrives after
			// the first drain1 has pulled the value out.
			// This is rare, but it does happen on overloaded builder machines.

	}

	// Let all the established get request loops soak
	time.Sleep(5 * time.Second)

	// Tear down
	close(stopChan)

	// Let loops drain
	drained = make(chan struct{})
	go func() {
		defer close(drained)
		wg.Wait()
	}()

	select {
	case <-drained:
	case <-time.After(wait.ForeverTestTimeout):
		t.Fatal("timed out waiting for clients to complete")
	}

	if network.AllIPv6(e.NodeIPs) {
		proxyLocalAddressType = "v6"
	}
	startupArgs := []string{
		"-c", fname,
		"--drain-time-s", fmt.Sprint(int(e.DrainDuration.AsDuration().Seconds())),
		"--drain-strategy", "immediate", // Clients are notified as soon as the drain process starts.
		"--local-address-ip-version", proxyLocalAddressType,

			select {
			case currentGlobalBuffer <- v:
			default:
			}
		}

		if mainWorker {
		drain:
			for {
				select {
				case v, ok := <-h.logCh:
					if !ok {
						break drain
					}

					currentGlobalBuffer <- v
				default:
					break drain
				}
			}
		}
	}()

	var entry interface{}
	var ok bool
	var err error

	// ShutdownWatchTerminationGracePeriod, if set to a positive value,
	// is the maximum duration the apiserver will wait for all active
	// watch request(s) to drain.
	// Once this grace period elapses, the apiserver will no longer
	// wait for any active watch request(s) in flight to drain, it will
	// proceed to the next step in the graceful server shutdown process.
	// If set to a positive value, the apiserver will keep track of the

        "//tensorflow/core/tfrt/saved_model/...",
        "//tensorflow/core/tfrt/tfrt_session/...",
    ] + if_google([
        "//learning/brain/experimental/mlir/tflite/tfmrt/...",
        "//learning/brain/experimental/mlir/tfrt_compiler/...",
        "//learning/brain/experimental/tfrt/...",
        "//learning/brain/tfrt/...",
        "//learning/infra/mira/...",
        "//learning/serving/contrib/tfrt/mlir/...",

	go w.processInterval(context.Background(), intervalFromEvents(initEvents), 1)
	if !w.add(makeWatchCacheEvent(7), time.NewTimer(1*time.Second)) {
		t.Fatal("failed adding an even to the watcher")
	}
	forget(true) // drain the watcher

	eventCount := 0
	for range w.ResultChan() {
		eventCount++
	}
	if eventCount != 3 {
		t.Errorf("Unexpected number of objects received: %d, expected: 3", eventCount)
	}

		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
		return
	}
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write([]byte("OK"))
	log.Infof("handling %s, starting drain", drainPath)
	s.drain()
}

func (s *Server) handleAppProbe(w http.ResponseWriter, req *http.Request) {
	// Validate the request first.
	path := req.URL.Path
	if !strings.HasPrefix(path, "/") {

		newContainerAllocateResponse(
			withDevices(map[string]string{"/dev/r1dev1": "/dev/r1dev1", "/dev/r1dev2": "/dev/r1dev2"}),
			withMounts(map[string]string{"/home/r1lib1": "/usr/r1lib1"}),
			withCDIDevices("domain1.com/resource1=dev1", "domain1.com/resource1=dev2"),
		),
	)
	testManager.podDevices.insert("pod1", "con1", resourceName2,
		constructDevices([]string{"dev1", "dev2"}),
		newContainerAllocateResponse(

func (s *Stream) Results(next func(b []byte) error) (err error) {
	done := false
	defer func() {
		if s.cancel != nil {
			s.cancel(err)
		}

		if !done {
			// Drain channel.
			for range s.responses {
			}
		}
	}()
	doneCh := s.ctx.Done()
	for {
		select {
		case <-doneCh:
			if err := context.Cause(s.ctx); !errors.Is(err, errStreamEOF) {
				return err
			}