Optimize the way to get the controllerId

[wenbingshen]

Related to this #1092

1. The current brokers metadata is maintained in the AdminManager, and obtaining the controllerId from here can ensure data consistency.
2. Avoid extra complex parsing listeners and handling exceptions.
3. The node ID is calculated from the hostname and port hash. If kop is not configured with ip or hostname, the program will automatically obtain the hostname. One of the problems is: when kop is running, if a new hostname address is added to the local domain name self-resolution, and as for the first position, it may cause the program to return a different host alias, the controller node id will inconsistent with the Broker cache in metadataStore. You can see the PR Support pulsar to obtain the correct unconfigured hostname listeners of KafkaProtocolHandler. #648 I submitted a long time ago.
4. Reuse testMetadataRequestForMultiListeners test validation controller.
5. A random node from the broker cache is used as the controller. There is no controller concept in kop. It is the same whether it returns to the node itself or randomly selects a node.

[BewareMyPower]

I think we should use a separate lock for controllerId. Here is my design:

    private void setControllerId(Map<String, Set<Node>> newBrokers) {
        final Map<String, Integer> listenerToControllerId = new HashMap<>();
        newBrokers.forEach((listenerName, brokers) -> {
            if (brokers.size() == 0) {
                listenerToControllerId.put(listenerName, MetadataResponse.NO_CONTROLLER_ID);
            } else {
                List<Node> nodes = Lists.newArrayList(brokers);
                listenerToControllerId.put(listenerName,
                        nodes.size() > 1 ? nodes.get(random.nextInt(brokers.size())).id() : nodes.get(0).id());
            }
        });
        synchronized (this.listenerToControllerId) {
            this.listenerToControllerId.clear();
            this.listenerToControllerId.putAll(listenerToControllerId);
        }
    }

    public int getControllerId(String listenerName) {
        synchronized (this.listenerToControllerId) {
            return listenerToControllerId.getOrDefault(listenerName, MetadataResponse.NO_CONTROLLER_ID);
        }
    }

    private final Map<String, Integer> listenerToControllerId = Maps.newHashMap();

[wenbingshen]

@BewareMyPower Thanks for your review. Sorry, there is a problem with my understanding, here reading does not require any read locks, we only need to ensure the synchronization of writing; synchronized will cause unnecessary competition between the request handle thread and KopEventThread , as well as competition between different request handle thread, This will bring serious performance problems, so we don't need any read-related lock operations here, because for the kafka client, the short-term expired metadata cache will not cause any problems for the client, the client only Just retry the metadata request.

[BewareMyPower]

Hi, could you try to migrate this PR to branch-2.8.2? Because the code of branch-2.8.2 is a little different from master branch.

[wenbingshen]

Hi, could you try to migrate this PR to branch-2.8.2? Because the code of branch-2.8.2 is a little different from master branch.

Sure, please wait a moment.