Fix flaky test: KafkaTopicConsumerManagerTest.testTopicManagerClose

[Demogorgon314]

Because the KafkaRequestHandler.close() is called by channelInActive, when channelInActive called, the tcp connection is already closed. We need ensure topicManager.close() is called.

channelInActive call root

We can easily reproduce by add some sleep in KafkaRequestHandler.close() like this:

@Override
protected void close() {
    if (isActive.getAndSet(false)) {
        super.close();
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        topicManager.close();
        String clientHost = ctx.channel().remoteAddress().toString();
        if (currentConnectedGroup.containsKey(clientHost)){
            log.info("currentConnectedGroup remove {}", clientHost);
            currentConnectedGroup.remove(clientHost);
        }
        producePurgatory.shutdown();
        fetchPurgatory.shutdown();

        // update alive channel count stat
        RequestStats.ALIVE_CHANNEL_COUNT_INSTANCE.decrementAndGet();
    }
}