public void onChange(MetadataDelta delta, MetadataImage image) {
        if (!inited) {
            // When the main2proxyByRack is un-inited, we should force update.
            inited = true;
        } else {
            if (delta.clusterDelta() == null || delta.clusterDelta().changedBrokers().isEmpty()) {
                return;
            }
        }
        // categorize the brokers by rack
        Map<String, List<BrokerRegistration>> rack2brokers = new HashMap<>();
        image.cluster().brokers().forEach((nodeId, node) -> {
            if (node.fenced() || node.inControlledShutdown()) {
                return;
            }
            rack2brokers.compute(node.rack().orElse(NOOP_RACK), (rack, list) -> {
                if (list == null) {
                    list = new ArrayList<>();
                }
                list.add(node);
                return list;
            });
        });
        this.main2proxyByRack = calMain2proxyByRack(rack2brokers);
        notifyListeners(this.main2proxyByRack);
    }
Get the proxy leader node when NOT_LEADER_OR_FOLLOWER happens.
core/src/main/java/kafka/automq/zerozone/ProxyNodeMapping.java