[BUG] testMutiBrokerUnloadReload cannot pass

[BewareMyPower]

Describe the bug
We disabled the DistributedClusterTest in #207 because these tests cannot pass in CI environment though they can pass in local environment. The track issue is #184 .

However, I found the testMutiBrokerUnloadReload cannot pass even in local environment. The other two tests of DistributedClusterTest work well.

To Reproduce
Steps to reproduce the behavior:

$ mvn clean install -DskipTests
$ mvn test '-Dtest=DistributedClusterTest#testMutiBrokerUnloadReload' -pl tests

Then the test timed out:

[ERROR] testMutiBrokerUnloadReload(io.streamnative.pulsar.handlers.kop.DistributedClusterTest)  Time elapsed: 30.013 s  <<< FAILURE!
org.testng.internal.thread.ThreadTimeoutException: Method io.streamnative.pulsar.handlers.kop.DistributedClusterTest.testMutiBrokerUnloadReload() didn't finish within the time-out 30000
	at java.util.concurrent.locks.ReentrantLock.unlock(ReentrantLock.java:457)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:299)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:242)
	at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1247)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1187)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1154)
	at io.streamnative.pulsar.handlers.kop.DistributedClusterTest.kafkaConsumeCommitMessage(DistributedClusterTest.java:237)
	at io.streamnative.pulsar.handlers.kop.DistributedClusterTest.testMutiBrokerUnloadReload(DistributedClusterTest.java:447)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:124)
	at org.testng.internal.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:54)
	at org.testng.internal.InvokeMethodRunnable.run(InvokeMethodRunnable.java:44)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Expected behavior
testMutiBrokerUnloadReload should pass.

Additional context
The issue seems to be introduced from #193 , see following two commits:

d684f4bd295ed686aa6c6f371864f29f91c02e00 (HEAD -> upstream-master) remove unnecessary register producer (#193)
57a3cac44e55b0cb2c5d881cfe3db250a001c721 Add Kafka version coverage tests (#196)

When I reset to 57a3cac and run test again, it succeed.

[BewareMyPower]

When I revert the changes of #193 , the tests passed. I'll push a PR to revert the changes later.

And @aloyszhang could you help take a look at testMutiBrokerUnloadReload issue?

[aloyszhang]

@BewareMyPower I'll check this.

[aloyszhang]

This problem is not caused not by PR #193.
The reason is that after unload the namespace, when consumer continue consume, broker still use the origin cursor which has been closed.

The logic of handleFetchRequest is

public CompletableFuture<AbstractResponse> handleFetch(
            CompletableFuture<AbstractResponse> fetchResponse,
            KafkaHeaderAndRequest fetchRequest) {
        LinkedHashMap<TopicPartition, PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();

        // Map of partition and related tcm.
        Map<TopicPartition, CompletableFuture<KafkaTopicConsumerManager>> topicsAndCursor =
            ((FetchRequest) fetchRequest.getRequest())
                .fetchData().entrySet().stream()
                .map(entry -> {
                    CompletableFuture<KafkaTopicConsumerManager> consumerManager =
                        requestHandler.getTopicManager().getTopicConsumerManager(KopTopic.toString(entry.getKey()));

                    return Pair.of(
                        entry.getKey(),
                        consumerManager);
                })
                .collect(Collectors.toMap(Pair::getKey, Pair::getValue));

First, we get the topicsAndCursor by invoke KafkaTopicManager#getTopicConsumerManager which return KafkaTopicConsumerManager using consumerTopicManagers.computeIfAbsen(. So, if already have TopicConsumerManager for a topic in consumerTopicManagers, consumer will re-use the old TopicConsumerManager which has been closed when namespace being unload.

public CompletableFuture<KafkaTopicConsumerManager> getTopicConsumerManager(String topicName) {
        return consumerTopicManagers.computeIfAbsent(
            topicName,
            t -> {
                CompletableFuture<PersistentTopic> topic = getTopic(t);
                checkState(topic != null);

                return topic.thenApply(t2 -> {
                    if (log.isDebugEnabled()) {
                        log.debug("[{}] Call getTopicConsumerManager for {}, and create TCM for {}.",
                            requestHandler.ctx.channel(), topicName, t2);
                    }

                    if (t2 == null) {
                        return null;
                    }
                    // return consumer manager
                    return new KafkaTopicConsumerManager(requestHandler, t2);
                });
            }
        );

[BewareMyPower]

What I'm still confused is that why the outdated TopicConsumerManager wouldn't be reused after applying

diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java
index 48ba56a..557aed6 100644
--- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java
+++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaTopicManager.java
@@ -296,6 +296,12 @@ public class KafkaTopicManager {
                     }
                     if (t2.isPresent()) {
                         PersistentTopic persistentTopic = (PersistentTopic) t2.get();
+                        try {
+                            references.putIfAbsent(t, registerInPersistentTopic(persistentTopic));
+                        } catch (Exception e) {
+                            log.info("[{}] Failed to get client in registerInPersistentTopic {}. exception:",
+                                    requestHandler.ctx.channel(), t, e);
+                        }
                         topicCompletableFuture.complete(persistentTopic);
                     } else {
                         log.error("[{}]Get empty topic for name {}",

Since it only affects the producer side but just solve the problem.

[aloyszhang]

Before PR #193, when invoke KafkaTopicManager#getTopic from sendMessage request or Fetch request, broker will register a internal producer in topic. So, when unload the namespace, connection will be closed for kafka producer or consumer. But it's incorrect since we should not register a producer for kafka consuemr.

And RP #193 remove the un-necessary register for kafka consumer, connnect for kafka consuemr will not be closed when unload namespace, so it will re-use the old TopicConsumerManager and cursors are already closed.

[BewareMyPower]

@aloyszhang Could you take a look at this issue again?