spring – kafka在重新平衡后停止使用来自新分配的分区的消息
作者：互联网
我对kafka(也是英语……)很新,我面对这个问题,不能谷歌任何解决方案.
我使用spring-boot,spring-kafka支持,我在本地机器上安装了kafka_2.11-0.10.1.1(只有一个代理0)
s1.然后我创建主题
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 5 --topic tracking
我的消费者配置：
applitions.properties：
kafka.servers.bootstrap=localhost:9092
kafka.topic.tracking=tracking
kafka.group.id=trackingGroup
kafka.client.id=client-1
S2.然后我通过更改’kafka.client.id’启动3个消费者并运行spring-boot主类.在eclipse控制台上,我可以检查分区分配：
client-1: partitions assigned:[tracking-4, tracking-3]
client-2: partitions assigned:[tracking-2, tracking-1]
client-3: partitions assigned:[tracking-0]
S3.启动pruducer向主题发送20条消息,每条消息开始消耗特定分区的消息
S4.我关闭消耗1,kafka自动进行重新平衡,
新分区分配：
client-1: partitions assigned:[]
client-2: partitions assigned:[tracking-2,tracking-1, tracking-0]
client-3: partitions assigned:[tracking-4,tracking-3]
S5.我发现分区’tracking-3’上的消息没有消耗!!
问题可以每次都重现,在新分配的分区中丢失一些消息,你能不能提出任何建议？请帮帮我,谢谢
解决方法:
我复制了它;在重新平衡中kafka本身看起来像是一个问题(使用auto.comit.enabled = true),kafka报告未读分区的“位置”(< i>下一条记录的偏移量< / i>被提取(如果存在具有该偏移的记录))作为分区的结尾.
事实上,当我使用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.consumer.enable-auto-commit=true
否则容器认为它是假的并导致上述奇怪的行为 – 看起来客户端不喜欢在启用自动提交时调用消费者的提交方法. #288.
我通常建议设置为false,然后选择一个容器的AckModes;例如RECORD在每次轮询接收后的记录,BATCH之后提交(默认).
标签：spring,apache-kafka,spring-kafka 来源： https://codeday.me/bug/20190622/1265419.html