[historyserver] Fix flaky TestShutdown_DrainsRetriedTasks

[1fanwang]

TestShutdown_DrainsRetriedTasks fails intermittently in CI on PRs that change nothing under historyserver/:

eventcollector_test.go:618:
    Error:    Should be true
    Test:     TestShutdown_DrainsRetriedTasks
    Messages: retried task never uploaded; keys: []
--- FAIL: TestShutdown_DrainsRetriedTasks (0.00s)

The test waits for the mock writer to clear failNext and treats that as "the retry goroutine now exists":

require.Eventually(t, func() bool { return !writer.failPending() }, ...)
ec.shutdown()

But failNext is cleared at the start of the failing write, before it even returns an error — so the signal fires well before processRotatedFile handles that error and calls retryProcess. If shutdown() closes stopProducers inside that window, retryProcess declines the retry, which is deliberate:

select {
case <-ec.stopProducers:
    logrus.Errorf("Giving up on retrying %s during shutdown; file left on disk", task.path)
    return
default:
}

The product behaviour is correct — the declined file stays on disk and is picked up by resumePendingFiles on the next start. Only the test is wrong: it wants the "retry already registered, shutdown cuts its backoff short" path, but it never actually guarantees registration first.

Driving the failing attempt inline makes that ordering a fact rather than a race. processRotatedFile returns only after retryProcess has registered the goroutine with consumerWG, so shutdown() cannot get there first. The assertions and what they cover are unchanged, and failPending goes with it since nothing calls it any more.

Testing Done

The flake is load-sensitive, so it will not reproduce on an idle machine — 200 plain runs and 200 -race runs pass both before and after. To reproduce deterministically I widened the window the race needs, simulating the worker being descheduled exactly where CI descheduled it:

if m.failNext {
    m.failNext = false
    m.mu.Unlock()
    time.Sleep(50 * time.Millisecond) // REPRO only, not committed
    return assertErr("mock write failure")
}

With that delay, the old test fails with the exact CI error and the new one passes:

Raw logs

=== OLD form + repro delay (5 runs) ===
    eventcollector_test.go:616:
        	Error Trace:	.../eventcollector_test.go:616
        	Error:      	Should be true
        	Test:       	TestShutdown_DrainsRetriedTasks
        	Messages:   	retried task never uploaded; keys: []
FAIL
FAIL	github.com/ray-project/kuberay/historyserver/pkg/collector/eventcollector	0.619s

=== FIXED form + repro delay (5 runs) ===
ok  	github.com/ray-project/kuberay/historyserver/pkg/collector/eventcollector	0.604s

=== FIXED form, 300 runs with -race, delay removed ===
ok  	github.com/ray-project/kuberay/historyserver/pkg/collector/eventcollector	2.331s

=== full package ===
ok  	github.com/ray-project/kuberay/historyserver/pkg/collector/eventcollector	0.542s

golangci-lint run ./pkg/collector/eventcollector/... reports 52 findings on this package both before and after the change — none new. gofmt clean.

[win5923]

Thanks! Did you also run the test locally around 20 more times to make sure it passes consistently?

[win5923]

This makes sense. Now synchronously driving processRotatedFile, which guarantees that the retry is registered before shutdown() starts, instead of relying on failNext as an indirect signal. This make the test deterministic.