为什么使用Kafka绑定器哈希键的Spring Cloud Stream与标准Kafka Producer不同？

Question

我遇到了一个问题，我需要将一个现有的主题(< code>source)重新分区到一个新的主题(< code>target)，这个新主题具有更大的分区数量(是以前分区数量的倍数)。

Source主题是使用Kafka Binder使用Spring Cloud Stream写入的。目标主题正在使用KStreams应用程序写入。

< code>source主题中的记录基于标题进行分区，其中< code>key=null。我试图显式提取这个头，并为< code>target主题中的记录设置一个消息键，并注意到具有相同分区键的记录位于完全不同的分区中。

经过一番调查，我发现罪魁祸首如下：

org.springframework.cloud.stream.binder.PartitionHandler.DefaultPartitionSelector

    private static class DefaultPartitionSelector implements PartitionSelectorStrategy {

        @Override
        public int selectPartition(Object key, int partitionCount) {
            int hashCode = key.hashCode();
            if (hashCode == Integer.MIN_VALUE) {
                hashCode = 0;
            }
            return Math.abs(hashCode);
        }
    }

< code > org . spring framework . cloud . stream . binder . partition handler

    public int determinePartition(Message<?> message) {
        // ... non relevant code omitted

        partition = this.partitionSelectorStrategy.selectPartition(key,
                    this.partitionCount);
        // protection in case a user selector returns a negative.
        return Math.abs(partition % this.partitionCount);

虽然默认的 Kafka 分区策略可以：

org.apache.kafka.clients.producer.internals.DefaultPartitioner

    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster,
                         int numPartitions) {
        if (keyBytes == null) {
            return stickyPartitionCache.partition(topic, cluster);
        }
        // hash the keyBytes to choose a partition
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }

本质上，使用Spring Cloud Stream之外的东西永远不允许与非Spring Cloud Stream应用程序编写的主题进行共同分区，除非使用自定义分区器(这并不难做到)。

但是，应该注意的是，上面的 DefaultPartitionSelector 并不位于 Kafka Binder 模块中，而是位于更高级别的 spring-cloud-stream 模块中。

这种设计选择的理由是什么？我认为默认的分区器适用于所有的绑定器，而不仅仅是Kafka，但是为什么Kafka绑定器不实现它自己的分区器，默认情况下它允许与非Spring Cloud Stream应用程序进行开箱即用的联合分区？

Answer 1

正如我在评论中所说

活页夹级别的分区适用于本机不支持分区的基础设施；不要使用它，让Kafka自己进行分区。

也就是说，你的意思并不完全清楚；Spring分区器是很久以前编写的，早于KIP 480引入的粘性缓存。但是，如果应用程序重新启动时分区数量发生变化，即使是那个分区器也会更改分区——如果有密钥，它会根据分区数量进行修改；如果没有密钥，则会选择一个随机（粘性）分区。

用10个分区，然后是20个分区运行这个，你会看到。

@SpringBootApplication
public class So73207602Application {

    public static void main(String[] args) {
        SpringApplication.run(So73207602Application.class, args).close();
    }

    @Bean
    ApplicationRunner runner(KafkaTemplate<String, String> template, NewTopic topic, KafkaAdmin admin) {
        return args -> {
            System.out.println(template.send("topic1", "foo", "bar").get().getRecordMetadata());
        };
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("topic1").partitions(10).replicas(1).build();
    }

}

用一个< code>null键，你每次都会得到一个不同的(随机的)分区。