一般来说，我认为您已经知道问题的答案，使用您所描述的分区方案，如果您要重新处理客户机，您必须阅读所有分区，因为消息将分布在所有分区中。
在重新处理整个客户机时，我唯一能想到的限制开销的方法是实现一个分区方案，该方案为一个客户机将多个分区分组，然后在这些分区上分配用户，以避免用一个特别“大”的客户机使一个分区过载。希望这张照片能澄清我可能无法用文字解释的问题。。

实现此发行版的自定义分区器可能类似于以下代码。请恕我直言，到目前为止，这完全是理论上的，从来没有被测试过（甚至没有运行过），但它应该说明原理。

public class ClientUserPartitioner implements Partitioner {
int partitionGroupSize = 10;

@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    // For this we expect the key to be of the format "client/user"
    String[] splitValues = ((String)key).split("/");
    String client = splitValues[0];
    String user = splitValues[1];

    // Check that partitioncount is divisible by group size
    if (cluster.availablePartitionsForTopic(topic).size() % partitionGroupSize != 0) {
        throw new ConfigException("Partitioncount must be divisible by "+ partitionGroupSize +" for this partitioner but is " +
                cluster.availablePartitionsForTopic(topic).size() + " for topic " + topic);
    }

    // Calculate partition group from client and specific partition from user
    int clientPartitionOffset = Utils.murmur2(client.getBytes()) % partitionGroupSize * partitionGroupSize;
    int userPartition = Utils.murmur2(user.getBytes()) % partitionGroupSize;

    // Combine group and specific value to get final partition
    return clientPartitionOffset + userPartition;
}

@Override
public void configure(Map<String, ?> configs) {
    if (configs.containsKey("partition.group.size")) {
        this.partitionGroupSize = Integer.parseInt((String)configs.get("partition.group.size"));
    }
}

@Override
public void close() {
}

}
这样做当然会影响您的分发，您可能需要使用partitiongroupsize的不同值和具有代表性的数据样本来运行一些模拟，以估计分发的均匀性以及重新处理整个客户机时节省的开销。