[fix][broker][branch-3.0] Backport missed CompactedTopicImpl deadlock fixes #20697 and #20984 to branch 3.0

[lhotari]

Fixes #24034

Motivation & Modifications

This backports missed CompactedTopicImpl deadlock fixes #20697 and #20984 to branch 3.0

Open issue

While testing locally, this test fails with the changes:

[ERROR] Tests run: 43, Failures: 1, Errors: 0, Skipped: 38, Time elapsed: 192.271 s <<< FAILURE! - in org.apache.pulsar.compaction.CompactedTopicTest
[ERROR] org.apache.pulsar.compaction.CompactedTopicTest.testCompactionWithTopicUnloading  Time elapsed: 40.658 s  <<< FAILURE!
org.awaitility.core.ConditionTimeoutException: Assertion condition defined as a org.apache.pulsar.compaction.CompactedTopicTest expected [1000] but found [501] within 30 seconds.
        at org.awaitility.core.ConditionAwaiter.await(ConditionAwaiter.java:167)
        at org.awaitility.core.AssertionCondition.await(AssertionCondition.java:119)
        at org.awaitility.core.AssertionCondition.await(AssertionCondition.java:31)
        at org.awaitility.core.ConditionFactory.until(ConditionFactory.java:985)
        at org.awaitility.core.ConditionFactory.untilAsserted(ConditionFactory.java:769)
        at org.apache.pulsar.compaction.CompactedTopicTest.testCompactionWithTopicUnloading(CompactedTopicTest.java:628)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:569)
        at org.testng.internal.invokers.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:139)
        at org.testng.internal.invokers.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:47)
        at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:76)
        at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:11)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: java.lang.AssertionError: expected [1000] but found [501]
        at org.testng.Assert.fail(Assert.java:110)
        at org.testng.Assert.failNotEquals(Assert.java:1577)
        at org.testng.Assert.assertEqualsImpl(Assert.java:149)
        at org.testng.Assert.assertEquals(Assert.java:131)
        at org.testng.Assert.assertEquals(Assert.java:979)
        at org.testng.Assert.assertEquals(Assert.java:955)
        at org.testng.Assert.assertEquals(Assert.java:989)
        at org.apache.pulsar.compaction.CompactedTopicTest.lambda$testCompactionWithTopicUnloading$16(CompactedTopicTest.java:631)
        at org.awaitility.core.AssertionCondition.lambda$new$0(AssertionCondition.java:53)
        at org.awaitility.core.ConditionAwaiter$ConditionPoller.call(ConditionAwaiter.java:248)
        at org.awaitility.core.ConditionAwaiter$ConditionPoller.call(ConditionAwaiter.java:235)
        ... 4 more

Documentation

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

[lhotari]

branch-3.0 is broken, #24038

[lhotari]

Last required fix is #24131

[lhotari]

Closing this PR since #20697 and #20984 have been directly backported and cherry-picked to branch 3.0