Fix lookup issue for MQTT-5

[Technoboy-]

Motivation

Fix the lookup issue. this pr only fix this issue for Qos1 under MQTT-5.

2023-07-25T08:27:12,754+0000 [pulsar-2-1] ERROR io.streamnative.pulsar.handlers.mqtt.support.MQTTBrokerProtocolMethodProcessor - [Publish] [bench/mqtt/4296] Write MqttPublishMessage[fixedHeader=MqttFixedHeader[messageType=PUBLISH, isDup=false, qosLevel=AT_MOST_ONCE, isRetain=false, remainingLength=274], variableHeader=MqttPublishVariableHeader[topicName=bench/mqtt/4296, packetId=-1], payload=PooledSlicedByteBuf(ridx: 0, widx: 256, cap: 256/256, unwrapped: PooledUnsafeDirectByteBuf(ridx: 277, widx: 277, cap: 277))] to Pulsar topic failed.
org.apache.pulsar.broker.service.BrokerServiceException$ServiceUnitNotReadyException: Namespace bundle for topic (persistent://public/default/bench%2Fmqtt%2F4296) not served by this instance. Please redo the lookup. Request is denied: namespace=public/default
	at org.apache.pulsar.broker.service.BrokerService.lambda$checkTopicNsOwnership$84(BrokerService.java:1925) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1106) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2235) ~[?:?]
	at org.apache.pulsar.broker.service.BrokerService.checkTopicNsOwnership(BrokerService.java:1917) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at org.apache.pulsar.broker.service.BrokerService.loadOrCreatePersistentTopic(BrokerService.java:1393) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at org.apache.pulsar.broker.service.BrokerService.lambda$getTopic$27(BrokerService.java:976) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.put(ConcurrentOpenHashMap.java:404) ~[io.streamnative-pulsar-common-2.10.3.7.jar:2.10.3.7]
	at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.computeIfAbsent(ConcurrentOpenHashMap.java:238) ~[io.streamnative-pulsar-common-2.10.3.7.jar:2.10.3.7]
	at org.apache.pulsar.broker.service.BrokerService.getTopic(BrokerService.java:975) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at org.apache.pulsar.broker.service.BrokerService.getTopic(BrokerService.java:940) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at io.streamnative.pulsar.handlers.mqtt.utils.PulsarTopicUtils.lambda$getTopicReference$4(PulsarTopicUtils.java:88) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1072) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073) ~[?:?]
	at org.apache.pulsar.broker.namespace.NamespaceService.lambda$searchForCandidateBroker$16(NamespaceService.java:605) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at java.util.concurrent.CompletableFuture.uniAcceptNow(CompletableFuture.java:753) ~[?:?]
	at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:731) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:2108) ~[?:?]
	at org.apache.pulsar.broker.namespace.NamespaceService.searchForCandidateBroker(NamespaceService.java:605) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at org.apache.pulsar.broker.namespace.NamespaceService.lambda$findBrokerServiceUrl$8(NamespaceService.java:408) ~[io.streamnative-pulsar-broker-2.10.3.7.jar:2.10.3.7]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[?:?]
	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:1128) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.netty-netty-common-4.1.87.Final.jar:4.1.87.Final]
	at java.lang.Thread.run(Thread.java:829) ~[?:?]
2023-07-25T08:27:12,754+0000 [pulsar-2-2] ERROR io.streamnative.pulsar.handlers.mqtt.support.MQTTBrokerProtocolMethodProcessor - [Publish] [bench/mqtt/2116] Write MqttPublishMessage[fixedHeader=MqttFixedHeader[messageType=PUBLISH, isDup=false, qosLevel=AT_MOST_ONCE, isRetain=false, remainingLength=274], variableHeader=MqttPublishVariableHeader[topicName=bench/mqtt/2116, packetId=-1], payload=PooledSlicedByteBuf(ridx: 0, widx: 256, cap: 256/256, unwrapped: PooledUnsafeDirectByteBuf(ridx: 277, widx: 277, cap: 277))] to Pulsar topic failed.

Modifications

Documentation

Check the box below.

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

  (Please explain why)

• doc

  (If this PR contains doc changes)