[fix][admin]Backlog quota's policy is null which causes a NPE

[poorbarcode]

Motivation

• pulsar-admin namespaces set-backlog-quota <namespace> does not allow a null attribute policy, but HTTP API POST /admin/v2/namespaces/{namespace}/backlogQuota allows.
• pulsar-admin topicPolicies set-backlog-quota <namespace> does not allow a null attribute policy, but HTTP API POST /admin/v2/namespaces/{topic}/backlogQuota allows.

It leads a NPE

2025-04-21T03:04:02,254+0000 [pulsar-backlog-quota-checker-OrderedScheduler-0-0] INFO  org.apache.pulsar.broker.service.BacklogQuotaManager - Backlog quota type destination_storage exceeded for topic [persistent://public/default/internal-partition-1]. Applying [null] policy
2025-04-21T03:04:02,254+0000 [pulsar-backlog-quota-checker-OrderedScheduler-0-0] ERROR org.apache.pulsar.broker.service.BrokerService - Error when checkBacklogQuota(persistent://public/default/internal-partition-1) in monitorBacklogQuota
java.util.concurrent.CompletionException: java.lang.NullPointerException: Cannot invoke "org.apache.pulsar.common.policies.data.BacklogQuota$RetentionPolicy.ordinal()" because the return value of "org.apache.pulsar.common.policies.data.BacklogQuota.getPolicy()" is null
        at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source) ~[?:?]
        at java.base/java.util.concurrent.CompletableFuture.uniAcceptNow(Unknown Source) ~[?:?]
        at java.base/java.util.concurrent.CompletableFuture.uniAcceptStage(Unknown Source) ~[?:?]
        at java.base/java.util.concurrent.CompletableFuture.thenAccept(Unknown Source) ~[?:?]
        at org.apache.pulsar.broker.service.BrokerService.lambda$monitorBacklogQuota$104(BrokerService.java:2221) ~[io.streamnative-pulsar-broker-3.3.5.jar:3.3.5]
        at org.apache.pulsar.broker.service.BrokerService.lambda$forEachPersistentTopic$99(BrokerService.java:2211) ~[io.streamnative-pulsar-broker-3.3.5.jar:3.3.5]
        at java.base/java.util.Optional.ifPresent(Unknown Source) ~[?:?]
        at org.apache.pulsar.broker.service.BrokerService.lambda$forEachPersistentTopic$100(BrokerService.java:2211) ~[io.streamnative-pulsar-broker-3.3.5.jar:3.3.5]
        at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(Unknown Source) ~[?:?]
        at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown Source) ~[?:?]
        at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown Source) ~[?:?]
        at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(Unknown Source) ~[?:?]
        at java.base/java.util.stream.AbstractPipeline.copyInto(Unknown Source) ~[?:?]
        at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source) ~[?:?]
        at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(Unknown Source) ~[?:?]
        at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(Unknown Source) ~[?:?]
        at java.base/java.util.stream.AbstractPipeline.evaluate(Unknown Source) ~[?:?]
        at java.base/java.util.stream.ReferencePipeline.forEach(Unknown Source) ~[?:?]
        at org.apache.pulsar.broker.service.BrokerService.forEachPersistentTopic(BrokerService.java:2211) ~[io.streamnative-pulsar-broker-3.3.5.jar:3.3.5]
        at org.apache.pulsar.broker.service.BrokerService.monitorBacklogQuota(BrokerService.java:2220) ~[io.streamnative-pulsar-broker-3.3.5.jar:3.3.5]
        at com.google.common.util.concurrent.MoreExecutors$ScheduledListeningDecorator$NeverSuccessfulListenableFutureTask.run(MoreExecutors.java:741) ~[com.google.guava-guava-32.1.2-jre.jar:?]
        at org.apache.bookkeeper.common.util.SingleThreadSafeScheduledExecutorService$SafeRunnable.run(SingleThreadSafeScheduledExecutorService.java:45) ~[io.streamnative-bookkeeper-common-4.17.1.2.jar:4.17.1.2]
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) ~[?:?]
        at java.base/java.util.concurrent.FutureTask.runAndReset(Unknown Source) ~[?:?]
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source) ~[?:?]
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[?:?]
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[?:?]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty-netty-common-4.1.119.Final.jar:4.1.119.Final]
        at java.base/java.lang.Thread.run(Unknown Source) [?:?]
Caused by: java.lang.NullPointerException: Cannot invoke "org.apache.pulsar.common.policies.data.BacklogQuota$RetentionPolicy.ordinal()" because the return value of "org.apache.pulsar.common.policies.data.BacklogQuota.getPolicy()" is null
        at org.apache.pulsar.broker.service.BacklogQuotaManager.handleExceededBacklogQuota(BacklogQuotaManager.java:93) ~[io.streamnative-pulsar-broker-3.3.5.jar:3.3.5]
        at org.apache.pulsar.broker.service.BrokerService.lambda$monitorBacklogQuota$102(BrokerService.java:2223) ~[io.streamnative-pulsar-broker-3.3.5.jar:3.3.5]
        ... 28 more
2025-04-21T03:04:02,256+0000 [pulsar-backlog-quota-checker-OrderedScheduler-0-0] INFO  org.apache.pulsar.broker.service.BacklogQuotaManager - Backlog quota type destination_storage exceeded for topic [persistent://public/default/output-partition-1]. Applying [null] policy

Modifications

Let HTTP API does not allow the null policy as well

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[poorbarcode]

/pulsarbot rerun-failure-checks

[Technoboy-]

• Pulsar CI / Build Pulsar on MacOS (pull_request)

checkstyle failed

[lhotari]

The test org.apache.pulsar.broker.admin.AdminApiTest.testRetentionAndBacklogQuotaCheck fails.

[lhotari]

This leaks resources in the test. The .close() method of the JerseyClient instance should be called. Adding a Lombok @Cleanup annotation would handle it.