[fix][broker] Consumer stuck when delete subscription __compaction failed

[poorbarcode]

Motivation

Background

• The steps of unsubscribing __compaction
  • Delete compacted ledger
  • Delete cursor
  • Remove subscription

Issue 1: consumer will stuck if deleting cursor(the step 2 above) failed

• Delete compacted ledger
• Delete cursor failed
• Reload the topic
• Compactor subscription relates to a deleted ledger
• Consumers are stuck because the compacted ledger has been deleted, and the messages were lost. The broker will keep printing the error log below

2025-02-13T17:23:51,448 - ERROR - [broker-topic-workers-OrderedExecutor-4-0:PersistentDispatcherSingleActiveConsumer] - [persistent://public/default/tp-74ea581c-d7ca-4558-999b-804b6e33faee / reader-801f712e61-Consumer{subscription=PersistentSubscription{topic=persistent://public/default/tp-74ea581c-d7ca-4558-999b-804b6e33faee, name=reader-801f712e61}, consumerId=1, consumerName=Ti6va, address=[id: 0xce1b338f, L:/127.0.0.1:52193 - R:/127.0.0.1:52199] [SR:127.0.0.1, state:Connected]}] Error reading entries at 11:0 : org.apache.bookkeeper.client.BKException$BKNoSuchLedgerExistsException: No such ledger exists on Bookies - Retrying to read in 15.0 seconds

You can reproduce the issue by the test testReadMsgsAfterDisableCompaction(true)

---

issue 2
The compaction task can concurrently execute by deleting the cursor __compaction

Modifications

• Fix issue 1
• Fix issue 2
• Print an error log if a compactor subscription is initialized twice, which may cause a ledger lost/

---

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: x

[lhotari]

@poorbarcode I fixed a checkstyle issue and a problem in using MockZooKeeper's delay method. After #23988, MockZooKeeper is correctly single threaded and a delay will block the execution of this code in ZKSessionWatcher:

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java

Lines 108 to 115 in 6e8c349

	try {
	zkClientState = future.get(tickTimeMillis, TimeUnit.MILLISECONDS);
	} catch (TimeoutException e) {
	// Consider zk disconnection if zk operation takes more than TICK_TIME
	zkClientState = Watcher.Event.KeeperState.Disconnected;
	}
	checkState(zkClientState);

This will trigger a ConnectionLost session event. That's why a delay should be kept under 2000 ms.
Currently MockZooKeeper's session timeout is hard coded to 30000ms and the check interval is 1/15 of this, therefore 2000ms.

[lhotari]

I created a separate PR to fix the invalid test that fails now: #24166

[lhotari]

LGTM. Good work @poorbarcode