public List<MetadataResponseData.MetadataResponseTopic> handleMetadataResponse(ClientIdMetadata clientIdMetadata,
        List<MetadataResponseData.MetadataResponseTopic> topics) {
        String clientRack = clientIdMetadata.rack();
        if (clientRack == null) {
            return withSnapshotReadFollowers(topics);
        }
        Map<Integer, BrokerRegistration> clientRackMain2proxy = main2proxyByRack.get(clientRack);
        if (clientRackMain2proxy == null) {
            // If the cluster doesn't cover the client rack, the producer should directly send records to the main node.
            return withSnapshotReadFollowers(topics);
        }
        // If the client config rack in clientId, we need to replace the leader id with the proxy leader id.
        topics.forEach(metadataResponseTopic -> {
            metadataResponseTopic.partitions().forEach(metadataResponsePartition -> {
                int leaderMainNodeId = metadataResponsePartition.leaderId();
                if (leaderMainNodeId != -1) {
                    BrokerRegistration proxy = clientRackMain2proxy.get(leaderMainNodeId);
                    if (proxy != null) {
                        int proxyLeaderId = proxy.id();
                        if (proxyLeaderId != leaderMainNodeId) {
                            metadataResponsePartition.setLeaderId(proxyLeaderId);
                            metadataResponsePartition.setIsrNodes(List.of(proxyLeaderId));
                            metadataResponsePartition.setReplicaNodes(List.of(proxyLeaderId));
                        }
                    }
                }
            });
        });
        return topics;
    }
Get the proxy leader node when NOT_LEADER_OR_FOLLOWER happens.
core/src/main/java/kafka/automq/zerozone/ProxyNodeMapping.java