[BUG] Topic not served by this instance Exception when consume topic

[hangc0276]

Describe the bug
When consume topic message using kafka client, the kop throws topic not served by this instance exception. The Exception as follow:

14:55:10.789 [ForkJoinPool.commonPool-worker-53] WARN  org.apache.pulsar.broker.service.BrokerService - Namespace bundle for topic (persistent://public/default/chenhang_kop_perf_consume_v2-partition-0) not served by this instance. Please redo the lookup. Request is denied: namespace=public/default
14:55:10.789 [ForkJoinPool.commonPool-worker-53] ERROR io.streamnative.pulsar.handlers.kop.coordinator.group.OffsetAcker - [persistent://public/default/chenhang_kop_perf_consume_v2-partition-0] get topic failed when ack for kop_perf_group_v1.
org.apache.pulsar.broker.service.BrokerServiceException$ServiceUnitNotReadyException: Namespace bundle for topic (persistent://public/default/chenhang_kop_perf_consume_v2-partition-0) not served by this instance. Please redo the lookup. Request is denied: namespace=public/default
        at org.apache.pulsar.broker.service.BrokerService.lambda$checkTopicNsOwnershipAsync$53(BrokerService.java:1515) ~[org.apache.pulsar-pulsar-broker-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.uniWhenCompleteStage(CompletableFuture.java:883) ~[?:?]
        at java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:2251) ~[?:?]
        at org.apache.pulsar.broker.service.BrokerService.checkTopicNsOwnershipAsync(BrokerService.java:1507) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.broker.service.BrokerService.loadOrCreatePersistentTopic(BrokerService.java:1050) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.broker.service.BrokerService.lambda$getTopic$16(BrokerService.java:779) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.put(ConcurrentOpenHashMap.java:276) ~[org.apache.pulsar-pulsar-common-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.computeIfAbsent(ConcurrentOpenHashMap.java:129) ~[org.apache.pulsar-pulsar-common-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at org.apache.pulsar.broker.service.BrokerService.getTopic(BrokerService.java:778) ~[org.apache.pulsar-pulsar-broker-2.8.0-rc-202101252233.jar:2.8.0-rc-202101252233]
        at io.streamnative.pulsar.handlers.kop.coordinator.group.OffsetAcker.lambda$null$3(OffsetAcker.java:93) ~[?:?]
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859) ~[?:?]
        at java.util.concurrent.CompletableFuture.uniWhenCompleteStage(CompletableFuture.java:883) ~[?:?]
        at java.util.concurrent.CompletableFuture.whenComplete(CompletableFuture.java:2251) ~[?:?]
        at io.streamnative.pulsar.handlers.kop.coordinator.group.OffsetAcker.lambda$ackOffsets$4(OffsetAcker.java:86) ~[?:?]
        at java.util.HashMap.forEach(HashMap.java:1336) ~[?:?]
        at io.streamnative.pulsar.handlers.kop.coordinator.group.OffsetAcker.ackOffsets(OffsetAcker.java:82) ~[?:?]
        at io.streamnative.pulsar.handlers.kop.coordinator.group.GroupCoordinator.lambda$handleCommitOffsets$39(GroupCoordinator.java:815) ~[?:?]
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859) [?:?]
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837) [?:?]
        at java.util.concurrent.CompletableFuture$Completion.exec(CompletableFuture.java:479) [?:?]
        at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290) [?:?]
        at java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020) [?:?]
        at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656) [?:?]
        at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594) [?:?]
        at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183) [?:?]

To Reproduce
Steps to reproduce the behavior:

1. Start Kop with pulsar broker
2. Start kafka consumer to consume message from topic
3. The broker log throw those exception

[BewareMyPower]

Is it resolved by #404?