fix: avoid infinite retry in blue-green migration test

[nodece]

Motivation

The extensible-load-manager CI job timed out in TestBlueGreenMigrationTestSuite/TestTopicMigration/proxyConnection after 5 minutes.

From the stack and logs, the test was stuck waiting on WaitGroup while producer/consumer goroutines were still looping in retry paths. During migration, producer can enter terminal states (for example TopicTerminated or ProducerClosed), but the test retry loops had no terminal-exit logic, causing effectively unbounded retries and suite timeout.

Modifications

• Add terminal error handling in producer send retry loop:
  • if error is ErrTopicTerminated or ErrProducerClosed, fail fast instead of retrying forever.
• Add bounded retry windows for both producer and consumer loops (30 seconds per message stage).
• Add an error channel and stage-level wait helper around WaitGroup waits to fail early on goroutine errors.
• Add stage timeout protection while waiting for:
  • pre-unload send/receive synchronization
  • producer/consumer goroutine completion
• Keep per-iteration context cancellation immediate (cancel after each send/receive attempt).

These changes make the test deterministic under migration failures and prevent hanging until global test timeout.

[Copilot]

Pull request overview

This PR updates the blue/green migration integration test to avoid hanging indefinitely during broker migration/unload scenarios by adding bounded retry behavior, terminal-error exits, and stage-level timeouts/error propagation.

Changes:

• Add terminal error handling for producer send retries (stop retrying on ErrTopicTerminated / ErrProducerClosed).
• Bound producer and consumer retry loops to a 30s window per message.
• Add an error channel and a waitWithError helper to fail stages early and avoid waiting forever on WaitGroups.

Comments suppressed due to low confidence (2)

pulsar/blue_green_migration_test.go:193

• The pre-unload synchronization looks off by one: with i starting at 0, if i == messageCountBeforeUnload { wgSendAndReceiveMessages.Done() } signals after processing messageCountBeforeUnload+1 messages. If the intent is to wait until exactly messageCountBeforeUnload messages have been sent, this should trigger at i == messageCountBeforeUnload-1 (or increment a counter and compare to the target).

			if i == messageCountBeforeUnload {
				wgSendAndReceiveMessages.Done()
			}
		}

pulsar/blue_green_migration_test.go:228

• Same off-by-one issue on the consumer side: i is 0-based, so if i == messageCountBeforeUnload { ... } fires after messageCountBeforeUnload+1 messages have been received/acked. If you want the unload to start after exactly messageCountBeforeUnload messages, adjust the condition accordingly (e.g., messageCountBeforeUnload-1).

			if i == messageCountBeforeUnload {
				wgSendAndReceiveMessages.Done()
			}

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.