[fix][broker] Fix forced topic/namespace deletion hanging or failing when compaction is in progress

[lhotari]

Fixes #24148
Related: #22736

Motivation

Symptom (who is affected)

On a topic that uses compaction, a forced deletion can hang forever or fail on every retry:

• admin.topics().delete(topic, true) never completes (or keeps failing), and
• admin.namespaces().deleteNamespace(ns, true) never completes — it cascades into deleting the namespace's topics, and the always-compacted __change_events system topic (plus any user topic with a compaction threshold) hits the same problem.

This is the long-standing issue reported in #24148 and the flakiness in #22736 (e.g. deleteNamespaceWithRetry timing out). In production it shows up as a force-delete that hangs, or that fails on every attempt until the topic is unloaded/reloaded or the broker restarts.

The trigger is a compaction that is running when the forced deletion fences the topic. Forcing deletion is expected to abort the in-flight compaction and then delete the topic; instead, the deletion gets stuck on the compaction.

Root causes

Two independent bugs can each wedge a forced deletion. They sit at different stages of the same flow, so depending on timing you hit one or the other.

1. A failed/aborted compaction permanently poisons the compaction-cursor deletion. When the deletion fences the topic, the compaction aborts and currentCompaction completes exceptionally (the reader-close behavior from [fix][broker]Fix deadlock when compaction and topic deletion execute concurrently #24366). But asyncDeleteCursorWithCleanCompactionLedger() chained the cursor deletion with currentCompaction.thenCompose(...), so an exceptional completion skipped the cursor deletion and failed the unsubscribe instead. That failed currentCompaction future is never reset (a new compaction can't start while the topic is deleting), so every subsequent delete attempt fails immediately until the topic instance is reloaded.

2. The compaction's reader can hang, so currentCompaction never completes at all. The compactor reads the topic through an internal RawReader. When the forced deletion makes that reader hit an unrecoverable error and close, a readNextAsync() issued by the compactor at just the wrong instant was enqueued after the consumer's close had already drained its pending-receive queue — leaving that read future uncompleted forever. currentCompaction then never completes, and the cursor deletion that waits on it blocks the entire forced deletion indefinitely.

What about unloads?

Unloading a topic does not itself hang — unload does not wait on currentCompaction. However, an unload/fence of a compacting topic is one of the events that can leave a compaction's reader in the hung state of cause 2 (the compaction reader is configured to treat "service not ready" as unrecoverable). The user-visible deadlock is on forced deletion, but the cause-2 fix applies no matter what drove the reader to close — a delete, a namespace delete, or an unload.

Modifications

1. PersistentTopic.asyncDeleteCursorWithCleanCompactionLedger() — proceed with the compaction-cursor deletion once currentCompaction has finished, whether it completed normally or exceptionally (only an in-flight compaction is awaited). Fixes cause 1.
2. RawReaderImpl.RawConsumerImpl.receiveRawAsync() — after enqueuing a pending receive, re-check the consumer state and fail the pending raw receives if the consumer is in a terminal state (Closing/Closed/Failed). This completes the read future deterministically once the consumer is terminal, regardless of why it closed. Fixes cause 2. Re-checking after the enqueue is what closes the race: every close path moves the consumer to a terminal state before it drains the pending-receive queue, so the recheck either observes the terminal state (and self-drains) or is ordered before the close's drain (which then covers it).
3. Strengthened testConcurrentCompactionAndTopicDelete to assert the deletion actually succeeded (deleteTopicFuture.get(...)), instead of only asserting it completed (which is also true when it fails).
4. Added regression tests:
   • CompactionTest.testForcedDeleteSucceedsAfterFailedCompaction — forced topic deletion after a deterministically failed compaction (cause 1).
   • CompactionTest.testForcedNamespaceDeleteWithInflightCompaction — end-to-end forced namespace deletion with an in-flight compaction ([Bug] Compaction of __change_events topic is blocking forceful namespace/topic deletion #24148 reproduction).
   • RawReaderTest.testReadNextAsyncCompletesAfterConsumerClosed — a read issued once the consumer is terminal must complete rather than hang (cause 2).

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Each of RawReaderTest.testReadNextAsyncCompletesAfterConsumerClosed, CompactionTest.testForcedDeleteSucceedsAfterFailedCompaction, and the strengthened testConcurrentCompactionAndTopicDelete was verified red without its corresponding fix and green with it.
• CompactionTest.testForcedNamespaceDeleteWithInflightCompaction and the full CompactionTest / RawReaderTest suites pass locally.

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

If the box was checked, please highlight the changes

• 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

[void-ptr974]

Reviewed the patch and did not find a blocking issue. The failed-compaction handling in PersistentTopic and the terminal-state recheck in RawReaderImpl both look consistent with the deletion race being fixed.

LGTM.