failureCauseContains("Could not GET '${backingServer.uri}/redirected/group/projectA/1.0/ivy-1.0.xml'")
        failureCauseContains("Read timed out")
    }

    if (cancel) {
      cancel = false
      eventSource.cancel()
    }
  }

  private fun nextEvent(): Any {
    return events.poll(10, TimeUnit.SECONDS)
      ?: throw AssertionError("Timed out waiting for event.")
  }

  fun assertExhausted() {
    assertThat(events).isEmpty()
  }

  fun assertEvent(
    id: String?,
    type: String?,
    data: String,
  ) {

//
//	func semasleep(ns int64) int32
//		If ns < 0, acquire m's semaphore and return 0.
//		If ns >= 0, try to acquire m's semaphore for at most ns nanoseconds.
//		Return 0 if the semaphore was acquired, -1 if interrupted or timed out.
//
//	func semawakeup(mp *m)
//		Wake up mp, which is or will soon be sleeping on its semaphore.
const (
	locked uintptr = 1

	active_spin     = 4
	active_spin_cnt = 30
	passive_spin    = 1
)

			tmp, err := os.CreateTemp("", "minio-goroutines-*.txt")
			if err == nil {
				_ = pprof.Lookup("goroutine").WriteTo(tmp, 1)
				tmp.Close()
				return errors.New("timed out. some connections are still active. goroutines written to " + tmp.Name())
			}
			return errors.New("timed out. some connections are still active")
		case <-timer.C:
			if atomic.LoadInt32(&srv.requestCount) <= 0 {
				return nil
			}

		}
	}

	timeoutRefCheck := func(t *testing.T, key string, count int64) {
		t.Helper()
		c := time.After(4 * time.Second)
		for {
			select {
			case <-c:
				t.Fatal("timed out waiting for expected ref count")
			default:
				e, ok := cc.Load(key)
				if !ok && count != 0 {
					t.Fatal("cache does not contain expected key")
				} else if count == 0 && !ok {
					return
				}

		time.Duration(500*time.Millisecond),
		func() (bool, error) {
			if dsw.PluginExists(socketPath) {
				return true, nil
			}
			return false, nil
		},
	)
	if err != nil {
		t.Fatalf("Timed out waiting for plugin to be added to desired state of world cache:\n%s.", socketPath)
	}
}

func waitForUnregistration(
	t *testing.T,
	socketPath string,
	dsw cache.DesiredStateOfWorld) {

	select {
	case rev := <-revisionChan:
		if rev != expected {
			t.Fatalf("expected revision %q to be produced on chan, got %q", expected, rev)
		}
	case <-time.After(time.Second * 5):
		t.Fatalf("timed out waiting for value on default revision chan")
	}
}

func TestNoDefaultRevision(t *testing.T) {
	stop := make(chan struct{})
	client := kube.NewFakeClient()
	w := NewDefaultWatcher(client, "default")

// Start begins the core controller loops that must exist for bootstrapping
// a cluster.
func (c *Controller) Start(stopCh <-chan struct{}) {
	if !cache.WaitForCacheSync(stopCh, c.serviceSynced) {
		runtime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}
	// Reconcile during first run removing itself until server is ready.
	endpointPorts := createEndpointPortSpec(c.PublicServicePort, "https")

	m.IngressClass = "foo"
	writeMessage(t, path, m)

	select {
	case <-doneCh:
		assert.Equal(t, newM, m)
		assert.Equal(t, w.Mesh(), newM)
		break
	case <-time.After(time.Second * 5):
		t.Fatal("timed out waiting for update")
	}
}

func newWatcher(t testing.TB, filename string, multi bool) mesh.Watcher {
	t.Helper()
	w, err := mesh.NewFileWatcher(filewatcher.NewWatcher(), filename, multi)
	if err != nil {

	select {
	case <-g.stopped:
		t.Stop()
		if g.result.TotalRequests == 0 {
			g.t.Fatal("no requests completed before stopping the traffic generator")
		}
		return g.result
	case <-t.C:
		g.t.Fatal("timed out waiting for result")
	}
	// Can never happen, but the compiler doesn't know that Fatal terminates
	return Result{}
}

func fillInDefaults(cfg *Config) {
	if cfg.Interval == 0 {