[improve] [broker] Do not try to open ML when the topic meta does not exist and do not expect to create a new one. #21995

[poorbarcode]

Motivation

When we call pulsar-admin topics stats <topic name>, Pulsar will try to open a managed ledger even if the topic metadata does not exist.

It costs a lot of resources when there are many requests trying to call pulsar-admin topics stats <topic name> on a deleted partition.

2024-01-30T15:52:17,143+0000 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] ERROR org.apache.pulsar.broker.admin.v2.PersistentTopics - [flink-cluster@belvedere.auth.streamnative.cloud] Failed to get stats for persistent://xx/xx/xx-partition-0
java.util.concurrent.CompletionException: org.apache.pulsar.broker.web.RestException: Topic partitions were not yet created
        at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315) ~[?:?]
        at java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:687) ~[?:?]
        at java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:662) ~[?:?]
        at java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2168) ~[?:?]
        at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.topicNotFoundReasonAsync(PersistentTopicsBase.java:4519) ~[io.streamnative-pulsar-broker-3.0.2.1.jar:3.0.2.1]
        at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.lambda$getTopicReferenceAsync$438(PersistentTopicsBase.java:4499) ~[io.streamnative-pulsar-broker-3.0.2.1.jar:3.0.2.1]
        at java.util.Optional.orElseGet(Optional.java:364) ~[?:?]
        at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.lambda$getTopicReferenceAsync$439(PersistentTopicsBase.java:4499) ~[io.streamnative-pulsar-broker-3.0.2.1.jar:3.0.2.1]
        at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1150) ~[?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
        at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) ~[?:?]
        at org.apache.pulsar.broker.service.BrokerService$2.openLedgerFailed(BrokerService.java:1800) ~[io.streamnative-pulsar-broker-3.0.2.1.jar:3.0.2.1]
        at org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl.lambda$asyncOpen$8(ManagedLedgerFactoryImpl.java:421) ~[io.streamnative-managed-ledger-3.0.2.1.jar:3.0.2.1]
        at java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:990) ~[?:?]
        at java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:974) ~[?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2162) ~[?:?]
        at org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl$2.initializeFailed(ManagedLedgerFactoryImpl.java:416) ~[io.streamnative-managed-ledger-3.0.2.1.jar:3.0.2.1]
        at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl$1.operationFailed(ManagedLedgerImpl.java:458) ~[io.streamnative-managed-ledger-3.0.2.1.jar:3.0.2.1]
        at org.apache.bookkeeper.mledger.impl.MetaStoreImpl.lambda$getManagedLedgerInfo$3(MetaStoreImpl.java:150) ~[io.streamnative-managed-ledger-3.0.2.1.jar:3.0.2.1]
        at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718) ~[?:?]
        at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482) ~[?:?]
        at org.apache.bookkeeper.common.util.OrderedExecutor$TimedRunnable.run(OrderedExecutor.java:201) ~[org.apache.bookkeeper-bookkeeper-common-4.16.3.jar:4.16.3]
        at org.apache.bookkeeper.common.util.SingleThreadSafeScheduledExecutorService$SafeRunnable.run(SingleThreadSafeScheduledExecutorService.java:46) ~[org.apache.bookkeeper-bookkeeper-common-4.16.3.jar:4.16.3]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[?:?]
        at java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[?:?]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[?:?]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty-netty-common-4.1.100.Final.jar:4.1.100.Final]
        at java.lang.Thread.run(Thread.java:840) ~[?:?]     
Caused by: org.apache.pulsar.broker.web.RestException: Topic partitions were not yet created
        at org.apache.pulsar.broker.admin.impl.PersistentTopicsBase.lambda$topicNotFoundReasonAsync$442(PersistentTopicsBase.java:4521) ~[io.streamnative-pulsar-broker-3.0.2.1.jar:3.0.2.1]
        at java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:684) ~[?:?]
        ... 29 more
2024-01-30T15:52:17,143+0000 [bookkeeper-ml-scheduler-OrderedScheduler-2-0] INFO  org.eclipse.jetty.server.RequestLog - 127.0.0.6 - - [30/Jan/2024:15:52:17 +0000] "GET /admin/v2/persistent/xx/xx/xx-partition-0/stats?getPreciseBacklog=true&authoritative=false&subscriptionBacklogSize=true&getEarliestTimeInBacklog=false HTTP/1.1" 404 50 "-" "Pulsar-Java-v2.11.1" 7

Modifications

• Do not try to open ML when the topic meta does not exist and do not expect to create a new one.

Behavior changes of TopicEventsListener

• Before changes: the listener will receive a Topic Load event when calling pulsar-admin topics stats <topic name> even if the topic was not created.
• After changes: the listener will not receive a Topic Load event when calling pulsar-admin topics stats <topic name> if the topic was not created.

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[dao-jun]

lgtm

[dao-jun]

@poorbarcode there is a test keeps failing, could you please take a check

[poorbarcode]

there is a test keeps failing, could you please take a check

Sure, thanks

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 90.32258% with 3 lines in your changes are missing coverage. Please review.

Project coverage is 73.60%. Comparing base (0b6bd70) to head (cdecb66).
Report is 8 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #22004      +/-   ##
============================================
+ Coverage     73.57%   73.60%   +0.02%     
+ Complexity    32553    32102     -451     
============================================
  Files          1874     1874              
  Lines        139252   139351      +99     
  Branches      15260    15279      +19     
============================================
+ Hits         102461   102574     +113     
+ Misses        28866    28854      -12     
+ Partials       7925     7923       -2     

Flag	Coverage Δ
inttests	24.60% <51.61%> (-0.29%)	⬇️
systests	24.23% <51.61%> (-0.23%)	⬇️
unittests	72.89% <90.32%> (+0.06%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...rg/apache/pulsar/broker/service/BrokerService.java	81.03% <90.32%> (-0.20%)	⬇️

... and 74 files with indirect coverage changes