[fix][broker] Race condition causes perpetual backlog on internal topics

[darinspivey]

Motivation

Topics with compaction enabled — most visibly __change_events system topics — accumulate an unbounded __compaction subscription backlog that is never drained, polluting backlog metrics across the cluster.

Modifications

The broker disconnects the compaction consumer as part of PersistentSubscription.resetCursorInternal, then resets the cursor and sends the success response. This races with the client: channelInactive fires on the consumer's ClientCnx and fails the in-flight seek future with ConnectException before the broker's success response arrives.

AbstractTwoPhaseCompactor.phaseTwoSeekThenLoop has no retry for this transient failure, so every compaction attempt aborts and the __compaction subscription backlog grows without bound. This is most visible on __change_events system topics, where compaction is triggered on any non-zero backlog.

The seek is idempotent and the cursor is already repositioned server-side, so retry the seek on ConnectException with a bounded backoff (100ms -> 1s, 10s budget). Non-transient failures propagate immediately. A unit test reproduces the race via an injection hook and asserts the compactor recovers.

Fixes #25279

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added an integration test following existing styles for testing the race condition that happens and the implemented fix for retrying

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

NONE of these are affected with this change

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[darinspivey]

@merlimat Thanks for your attention and approval on this PR. I've added a commit that hopefully fixes the test injection design for my new test.

[darinspivey]

The remaining CI failures appear to be out of my control

[lhotari]

The remaining CI failures appear to be out of my control

when CI fails due to flaky tests, you can add a comment "/pulsarbot rerun" to trigger a retry.
For early stage PRs, it's worth setting up "Personal CI" where you have full control.

[lhotari]

/pulsarbot rerun

[darinspivey]

Thanks for that tip on personal CI, @lhotari . I'll do that in the future. In the meantime, I meant to sync my fork's master branch, but accidentally sync'd the feature branch. CI will have to be kicked off again, sorry. I won't touch it from here on out. Good to know about "/pulsarbot rerun" too!