[BUG] Kop startup depends on other Pulsar broker

[hangc0276]

Describe the bug
When rolling restart pulsar broker equiped with KOP, the pulsar broker startup depends on other pulsar broker.
The exception log list as follow.

java.util.concurrent.ExecutionException: org.apache.pulsar.client.admin.PulsarAdminException: java.util.concurrent.CompletionException: org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector$RetryException: Could not complete the operation. Number of retries has been exhausted. Failed reason: Connection refused: /xx.xx.xx.xx:8210
        at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395) ~[?:?]
        at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999) ~[?:?]
        at org.apache.pulsar.broker.admin.v2.NonPersistentTopics.lambda$getList$1(NonPersistentTopics.java:283) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:930) ~[?:?]
        at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:907) ~[?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]
        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088) ~[?:?]
        at org.apache.pulsar.client.admin.internal.TopicsImpl$5.failed(TopicsImpl.java:233) ~[org.apache.pulsar-pulsar-client-admin-original-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.glassfish.jersey.client.JerseyInvocation$1.failed(JerseyInvocation.java:839) ~[org.glassfish.jersey.core-jersey-client-2.31.jar:?]
        at org.glassfish.jersey.client.ClientRuntime.processFailure(ClientRuntime.java:247) ~[org.glassfish.jersey.core-jersey-client-2.31.jar:?]
        at org.glassfish.jersey.client.ClientRuntime.processFailure(ClientRuntime.java:242) ~[org.glassfish.jersey.core-jersey-client-2.31.jar:?]
        at org.glassfish.jersey.client.ClientRuntime.access$100(ClientRuntime.java:62) ~[org.glassfish.jersey.core-jersey-client-2.31.jar:?]
        at org.glassfish.jersey.client.ClientRuntime$2.lambda$failure$1(ClientRuntime.java:178) ~[org.glassfish.jersey.core-jersey-client-2.31.jar:?]
        at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248) ~[org.glassfish.jersey.core-jersey-common-2.31.jar:?]
        at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244) ~[org.glassfish.jersey.core-jersey-common-2.31.jar:?]
        at org.glassfish.jersey.internal.Errors.process(Errors.java:292) ~[org.glassfish.jersey.core-jersey-common-2.31.jar:?]
        at org.glassfish.jersey.internal.Errors.process(Errors.java:274) ~[org.glassfish.jersey.core-jersey-common-2.31.jar:?]
        at org.glassfish.jersey.internal.Errors.process(Errors.java:244) ~[org.glassfish.jersey.core-jersey-common-2.31.jar:?]
        at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:288) ~[org.glassfish.jersey.core-jersey-common-2.31.jar:?]
        at org.glassfish.jersey.client.ClientRuntime$2.failure(ClientRuntime.java:178) ~[org.glassfish.jersey.core-jersey-client-2.31.jar:?]
        at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$apply$1(AsyncHttpConnector.java:202) ~[org.apache.pulsar-pulsar-client-admin-original-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859) ~[?:?]
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837) ~[?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]
        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088) ~[?:?]
        at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$retryOperation$3(AsyncHttpConnector.java:244) ~[org.apache.pulsar-pulsar-client-admin-original-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859) [?:?]
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837) [?:?]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) [?:?]
        at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088) [?:?]
        at org.asynchttpclient.netty.NettyResponseFuture.abort(NettyResponseFuture.java:273) [org.asynchttpclient-async-http-client-2.12.1.jar:?]
        at org.asynchttpclient.netty.channel.NettyConnectListener.onFailure(NettyConnectListener.java:181) [org.asynchttpclient-async-http-client-2.12.1.jar:?]
        at org.asynchttpclient.netty.channel.NettyChannelConnector$1.onFailure(NettyChannelConnector.java:108) [org.asynchttpclient-async-http-client-2.12.1.jar:?]
        at org.asynchttpclient.netty.SimpleChannelFutureListener.operationComplete(SimpleChannelFutureListener.java:28) [org.asynchttpclient-async-http-client-2.12.1.jar:?]
        at org.asynchttpclient.netty.SimpleChannelFutureListener.operationComplete(SimpleChannelFutureListener.java:20) [org.asynchttpclient-async-http-client-2.12.1.jar:?]
        at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:577) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:570) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:549) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:490) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:615) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:608) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:117) [io.netty-netty-common-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:321) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:337) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:702) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) [io.netty-netty-transport-4.1.51.Final.jar:4.1.51.Final]

...

16:27:55.281 [main] ERROR org.apache.pulsar.broker.PulsarService - null
java.lang.IllegalStateException: null
        at com.google.common.base.Preconditions.checkState(Preconditions.java:492) ~[com.google.guava-guava-30.1-jre.jar:?]
        at io.streamnative.pulsar.handlers.kop.KafkaProtocolHandler.newChannelInitializers(KafkaProtocolHandler.java:286) ~[?:?]
        at org.apache.pulsar.broker.protocol.ProtocolHandlerWithClassLoader.newChannelInitializers(ProtocolHandlerWithClassLoader.java:71) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.broker.protocol.ProtocolHandlers.newChannelInitializers(ProtocolHandlers.java:124) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.broker.PulsarService.start(PulsarService.java:649) [org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.PulsarBrokerStarter$BrokerStarter.start(PulsarBrokerStarter.java:255) [org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.PulsarBrokerStarter.main(PulsarBrokerStarter.java:326) [org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
16:27:55.282 [main] ERROR org.apache.pulsar.PulsarBrokerStarter - Failed to start pulsar service.
org.apache.pulsar.broker.PulsarServerException: java.lang.IllegalStateException
        at org.apache.pulsar.broker.PulsarService.start(PulsarService.java:675) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.PulsarBrokerStarter$BrokerStarter.start(PulsarBrokerStarter.java:255) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.PulsarBrokerStarter.main(PulsarBrokerStarter.java:326) [org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
Caused by: java.lang.IllegalStateException
        at com.google.common.base.Preconditions.checkState(Preconditions.java:492) ~[com.google.guava-guava-30.1-jre.jar:?]
        at io.streamnative.pulsar.handlers.kop.KafkaProtocolHandler.newChannelInitializers(KafkaProtocolHandler.java:286) ~[?:?]
        at org.apache.pulsar.broker.protocol.ProtocolHandlerWithClassLoader.newChannelInitializers(ProtocolHandlerWithClassLoader.java:71) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.broker.protocol.ProtocolHandlers.newChannelInitializers(ProtocolHandlers.java:124) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.broker.PulsarService.start(PulsarService.java:649) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        ... 2 more

The case doesn't occurs in each rolling restart.

[BewareMyPower]

This issue might be fixed by #522 because the previous behavior is

                // Check to see if the partitions all exist
                partitionSet.removeAll(
                        topics.getList(kafkaMetadataNamespace).stream()
                                .filter((topic) -> topic.startsWith(kopTopic.getFullName()))
                                .collect(Collectors.toList())
                );

The exception was thrown by topics.getList from MetadataUtils.java line 194 before the PR.

It looks like to be a bug with Pulsar's Topics#getList call. But now it won't affect KoP anymore because we don't call this method now. So close this PR first.