fix issue #252 bundle unload bug

[hangc0276]

Fix #252

Motivation

When bundle unload triggered, the consumerTopicManagers cache won't be evicted and it will return the old KafkaTopicConsumerManager instance in the next fetch request handle. However, after bundle unload, the producer/consumer/managedLedger of topics in related bundle will be closed. If we use old KafkaTopicConsumerManager instance to read messages, it will return managedLedger has been closed exception.

Changes

1. Change consumerTopicManagers, topics, references map to static attribute ConcurrentHashMap.
2. Evict related cache information for topics whose bundle trigged unload.
3. Turn on DistributedClusterTest .

[BewareMyPower]

@hangc0276 Could you take a look at the test failure? It looks like the test is still unstable after the change. I think it may not be caused by the uncleaned static cache.

[hangc0276]

@hangc0276 Could you take a look at the test failure? It looks like the test is still unstable after the change. I think it may not be caused by the uncleaned static cache.

@BewareMyPower OK, maybe there are other bugs to cause this problem, it looks like caused by loadbalance, i will address it these days.

[hangc0276]

When run DistributedClusterTest#testMutiBrokerAndCoordinator, it will block in publishMessages, and in the retry round, also blocked.

The code of ManagedLedgerImpl.java as follow:

public void asyncAddEntry(ByteBuf buffer, int numberOfMessages, AddEntryCallback callback, Object ctx) {
        log.info("[hangc] start asyncAddEntry v1...");
        if (log.isDebugEnabled()) {
            log.debug("[{}] asyncAddEntry size={} state={}", name, buffer.readableBytes(), state);
        }

        OpAddEntry addOperation = OpAddEntry.create(this, buffer, numberOfMessages, callback, ctx);

        log.info("[hangc] start execute name: {}", name);
        // Jump to specific thread to avoid contention from writers writing from different threads
        executor.executeOrdered(name, safeRun(() -> log.info("pre exec...")));
        executor.executeOrdered(name, safeRun(() -> {
            log.info("[hangc] enter thread exec internalAsyncAddEntry...");
            internalAsyncAddEntry(addOperation);
        }));

[BewareMyPower]

@hangc0276 Could you take a look at the failed testOneBrokerShutdown?

Also you can rebase to latest master because we ignored the unstable tests, see #444