java - Kafka Spring Batch Consumer - 提交单个偏移量
问题描述
SpringBoot 中的 Kafka 批处理侦听器有问题。
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, this.maxPollRecords);
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, this.autoOffsetReset);
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, this.maxFetchBytesMaxPartition);
props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, this.maxFetchBytesMax);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
props.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, ByteArrayDeserializer.class);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, receiveBuffer);
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, heartbeatInterval);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, pollInterval);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, minFetch);
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, maxWaitFetch);
return props;
}
@Bean
public DefaultKafkaConsumerFactory<String, byte[]> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, byte[]> factory = new ConcurrentKafkaListenerContainerFactory<>();
try {
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckOnError(false);
factory.setBatchListener(true);
factory.getContainerProperties().setSyncCommits(false);
factory.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());
factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
} catch(Exception e) {
logger.error("Error KafkaListenerContainerFactory: {}", e.getMessage());
}
return factory;
}
所以,这是@KafkaListener
@KafkaListener(autoStartup = "${kafka-startup}", groupId = "${kafka-group}", topics = "${queue}",
containerFactory = "kafkaListenerContainerFactory", concurrency = "${concurrency}")
public void listen(@Payload List<byte[]> messages,
@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<String> keys,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
@Header(KafkaHeaders.RECEIVED_TIMESTAMP) List<Long> timestamps,
@Header(KafkaHeaders.OFFSET) List<Long> offsets,
Acknowledgment ack) throws Exception {
int indexQueue = new Random().nextInt(queues.size());
for (int i = 0; i < messages.size(); i++) {
//Do somethings
ack.acknowledge();
}
}
这个解决方案对于我的问题不起作用,因为ack.acknowledge()
批量提交。对于我的解决方案,我需要提交单个消息的偏移量。
我曾尝试使用KafkaConsumer<String, byte[]> consumer
with consumer.commitAsync()
,但情况是一样的。为了测试它,脚本读取了一个批次(由 3 mex 组成):在第 thirth 消息时,脚本启动了一个异常。
例如:消息 1 -> 偏移量 10;消息 2 -> 偏移量 11,消息 3 -> 偏移量 12
脚本正在阅读:
- 消息 1(偏移量 10)-> 确定
- 消息 2(偏移量 11)-> 好的
- 消息 3(偏移量 12)-> 异常
在下一个循环中,脚本重新读取偏移量为 10 的消息 1,但我预计消息 3 的偏移量为 12。
你有什么想法吗?你能帮我吗 ?
谢谢
解决方案
批处理侦听器Acknowledgment
应该只调用一次。
您现在可以(从 2.3 开始)调用acknowledgment.nack(thisOneFailed, sleep);
见https://docs.spring.io/spring-kafka/docs/current/reference/html/#committing-offsets
从 2.3 版开始,该
Acknowledgment
接口有两个附加方法nack(long sleep)
和nack(int index, long sleep)
. 第一个用于记录侦听器,第二个用于批处理侦听器。为您的侦听器类型调用错误的方法将引发 IllegalStateException。
推荐阅读
- javascript - 是否有等效于 Python 的 getattr() 函数的 Javascript?
- apache-kafka - Kafka Connect:如何将自定义连接器 (SMT) 添加到现有的 Kafka Connect docker 安装中?
- firebase - Cloud Firestore:获取集合中的文档数
- uwp - 如何恢复旧版本的 UWP 应用程序?
- excel - 如何在数据透视表 VB 脚本中仅删除(空白)?
- kdb - KDB如何通过引用更新表而不会出现分配错误?
- git - Git:使用 VS Code 比较来自 2 个 Git 提交的文件
- html - 当我使用异步管道时 *ngFor 和 *ngForOf 不起作用
- git - 将代码推送到 aws codeCommit 时出现此错误 --> repo : RPC failed; curl 56 LibreSSL SSL_read: SSL_ERROR_SYSCALL, errno 54
- authentication - 二维码登录如何在 Discord 中工作,我该如何重新创建它?