我复制了它；重新平衡时，kafka 本身（auto.comit.enabled=true）似乎存在问题，kafka 将未读分区（the offset of the <i>next record</i> that will be fetched (if a record with that offset exists)）的“位置”报告为分区结束。
其实我在使用kafka-consumer-groups工具的时候，未读分区的偏移量已经在“端”了。当我只使用一个消费者运行它时，它正在读取第一个分区，我看到...
$ kafka-consumer-groups --bootstrap-server localhost:9092 --describe -group so43405009
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
tracking 0 37 40 3 client1-8129bb3d-3a83-4c83-9128-3a2762ede758 /10.0.0.6 client1
tracking 1 40 40 0 client1-8129bb3d-3a83-4c83-9128-3a2762ede758 /10.0.0.6 client1
tracking 2 40 40 0 client1-8129bb3d-3a83-4c83-9128-3a2762ede758 /10.0.0.6 client1
tracking 3 40 40 0 client1-8129bb3d-3a83-4c83-9128-3a2762ede758 /10.0.0.6 client1
tracking 4 40 40 0 client1-8129bb3d-3a83-4c83-9128-3a2762ede758 /10.0.0.6 client1
注意 CURRENT_OFFSET 列。
在下一次运行时，我运行了两次，一次是在处理第一个分区时，然后再一次......
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
tracking 0 41 44 3 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
tracking 1 44 44 0 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
tracking 2 44 44 0 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
tracking 3 44 44 0 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
tracking 4 44 44 0 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
和
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
tracking 0 44 44 0 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
tracking 1 44 44 0 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
tracking 2 41 44 3 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
tracking 3 44 44 0 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
tracking 4 44 44 0 client1-56d78a4b-f2df-4e31-8c5d-f8a1d8f64cf8 /10.0.0.6 client1
查看分区 2 的当前偏移量如何从 44 下降到 41。
禁用自动提交为我解决了这个问题...
spring.kafka.consumer.enable-auto-commit=false
...
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
tracking 0 52 52 0 client1-59413599-81e8-49dd-bbd7-8a62152f11e5 /10.0.0.6 client1
tracking 1 49 52 3 client1-59413599-81e8-49dd-bbd7-8a62152f11e5 /10.0.0.6 client1
tracking 2 49 52 3 client2-edfe34f9-08d5-4825-80d0-4a6cf9526e42 /10.0.0.6 client2
tracking 3 48 52 4 client2-edfe34f9-08d5-4825-80d0-4a6cf9526e42 /10.0.0.6 client2
tracking 4 51 52 1 client3-20da8742-af38-403e-b125-5d0c7c771319 /10.0.0.6 client3
这是我的测试程序：
@SpringBootApplication
public class So43405009Application implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication.run(So43405009Application.class, args);
}
@Autowired
private KafkaTemplate<String, String> template;
@Value("${spring.kafka.consumer.client-id}")
private String clientId;
@Override
public void run(String... args) throws Exception {
if (this.clientId.endsWith("1")) {
for (int i = 0; i < 20; i++) {
this.template.sendDefault("foo" + i);
}
}
}
@Bean
public KafkaMessageListenerContainer<String, String> container(ConsumerFactory<String, String> cf) {
ContainerProperties containerProperties = new ContainerProperties("tracking");
containerProperties.setMessageListener((MessageListener<?, ?>) d -> {
System.out.println(d);
try {
Thread.sleep(5_000);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(cf,
containerProperties);
return container;
}
}
有属性
spring.kafka.listener.ack-mode=record
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.group-id=so43405009
spring.kafka.consumer.client-id=client1
spring.kafka.template.default-topic=tracking
我在 0.10.2.0 中也看到了相同的结果。
编辑
原来是spring-kafka bug；它可以在启用自动提交的情况下使用，但您必须明确启用它
spring.kafka.consumer.enable-auto-commit=true
否则容器会假定它是false 并导致上述奇怪的行为 - 如果启用了自动提交，看起来客户端不喜欢调用消费者的提交方法。 #288.
我通常建议设置为 false，然后选择容器的 AckModes 之一；例如RECORD 在记录后提交，BATCH 在轮询收到的每个批次后提交（默认）。