[fix][broker] fix MetadataStoreException$NotFoundException while doing topic lookup

[codelipenghui]

Motivation

#14839 fixes the HTTP lookup with multiple advertised listeners, and the added test
is trying to verify the advertised listeners with 2 brokers, but the config of the
the additional broker will not update, so the test gets passed.

The root cause is we are using the (advertised address + WebService port) to register
the broker under the loadbalance path, but while creating the lookup result, we will
use the WebService URL to get the data, so we will get MetadataStoreException$NotFoundException.

2022-05-17T14:18:55,658+0800 [pulsar-load-manager-38-1] INFO  org.apache.pulsar.metadata.coordination.impl.ResourceLockImpl - Acquired resource lock on /loadbalance/brokers/BROKER-0:15004
2022-05-17T14:19:01,126+0800 [pulsar-4-1] WARN  org.apache.pulsar.broker.lookup.TopicLookupBase - Failed to lookup broker for topic persistent://public/default/my-topic: org.apache.pulsar.metadata.api.MetadataStoreException$NotFoundException: /loadbalance/brokers/xxxx:15004
java.util.concurrent.CompletionException: org.apache.pulsar.metadata.api.MetadataStoreException$NotFoundException: /loadbalance/brokers/xxxx:15004
	at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:332) ~[?:?]
	at java.util.concurrent.CompletableFuture.uniAcceptNow(CompletableFuture.java:747) ~[?:?]
	at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:735) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:2182) ~[?:?]
	at org.apache.pulsar.broker.namespace.NamespaceService.searchForCandidateBroker(NamespaceService.java:599) ~[classes/:?]
	at org.apache.pulsar.broker.namespace.NamespaceService.lambda$findBrokerServiceUrl$8(NamespaceService.java:405) ~[classes/:?]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) [?:?]
	at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) [?:?]
	at java.util.concurrent.FutureTask.run(FutureTask.java) [?:?]
	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) [netty-common-4.1.76.Final.jar:4.1.76.Final]
	at java.lang.Thread.run(Thread.java:833) [?:?]
Caused by: org.apache.pulsar.metadata.api.MetadataStoreException$NotFoundException: /loadbalance/brokers/xxxx:15004
	at org.apache.pulsar.broker.namespace.NamespaceService.lambda$createLookupResult$18(NamespaceService.java:642) ~[classes/:?]
	at java.util.concurrent.CompletableFuture.uniAcceptNow(CompletableFuture.java:757) ~[?:?]
	at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:735) ~[?:?]
	at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:2182) ~[?:?]
	at org.apache.pulsar.broker.namespace.NamespaceService.createLookupResult(NamespaceService.java:620) ~[classes/:?]
	at org.apache.pulsar.broker.namespace.NamespaceService$MockitoMock$1465504970.createLookupResult$accessor$rPJehdnB(Unknown Source) ~[classes/:?]
	at org.apache.pulsar.broker.namespace.NamespaceService$MockitoMock$1465504970$auxiliary$608QTNMe.call(Unknown Source) ~[classes/:?]
	at org.mockito.internal.invocation.RealMethod$FromCallable$1.call(RealMethod.java:40) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.invocation.RealMethod$FromBehavior.invoke(RealMethod.java:62) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.invocation.InterceptedInvocation.callRealMethod(InterceptedInvocation.java:142) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.stubbing.answers.CallsRealMethods.answer(CallsRealMethods.java:45) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.Answers.answer(Answers.java:99) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.handler.MockHandlerImpl.handle(MockHandlerImpl.java:110) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.handler.NullResultGuardian.handle(NullResultGuardian.java:29) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.handler.InvocationNotifierHandler.handle(InvocationNotifierHandler.java:34) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.creation.bytebuddy.MockMethodInterceptor.doIntercept(MockMethodInterceptor.java:82) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.creation.bytebuddy.MockMethodInterceptor.doIntercept(MockMethodInterceptor.java:56) ~[mockito-core-3.12.4.jar:?]
	at org.mockito.internal.creation.bytebuddy.MockMethodInterceptor$DispatcherDefaultingToRealMethod.interceptSuperCallable(MockMethodInterceptor.java:145) ~[mockito-core-3.12.4.jar:?]
	at org.apache.pulsar.broker.namespace.NamespaceService$MockitoMock$1465504970.createLookupResult(Unknown Source) ~[classes/:?]
	at org.apache.pulsar.broker.namespace.NamespaceService.searchForCandidateBroker(NamespaceService.java:598) ~[classes/:?]
	... 9 more

Modification

• Added properties support for ResourceUnit, so that we can carry more information while getting the least loaded broker
• For creating the lookup result, use the broker znode name to get the broker data if it presents in the ResourceUnit

Verification

No new tests are needed, just fix the test that was added in #14839

Documentation

Check the box below or label this PR directly.

Need to update docs?

• no-need-doc
  (bug fix)

[github-actions]

@codelipenghui:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[github-actions]

@codelipenghui:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[github-actions]

@codelipenghui:Thanks for providing doc info!

[codelipenghui]

/pulsarbot run-failure-checks