我现在陷入了一个手动管理kafka偏移量和提交的简单示例中。我有一个springcloudstreams应用程序 enable.auto.commit = false
(在打印consumervalues时在启动日志中看到),但是当我解析消息时,它仍然不提供确认头。
这是我的听众:
@StreamListener(Sink.INPUT)
public void handleSchedulerMessage(@Payload SchedulerEvent event, @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {
log.debug("[message={}]", event);
// todo: processing
log.debug("Event processed successfully [event={}]", event);
}
配置的yaml也非常简单:
spring:
application:
name: scheduler
cloud:
stream:
kafka:
binder:
brokers: *kafka-broker*:9092
zkNodes: *zookeeper*:2181
bindings:
input:
destination: scheduler
contentType: application/json
consumer:
autoCommitOffset: false
当我发送信息时,会立即弹出一个错误:
2018-05-22 11:38:32.470 ERROR 11651 --- [container-0-C-1] o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageHandlingException: Missing header 'kafka_acknowledgment' for method parameter type [interface org.springframework.kafka.support.Acknowledgment], failedMessage=GenericMessage [payload=byte[38], headers={kafka_offset=9, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@fdac355, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=scheduler, kafka_receivedTimestamp=1526981909241, contentType=application/json}]
收到的消息不包含所需的标头,这与禁用autocommit时文档所述相反:
Whether to autocommit offsets when a message has been processed. If set to false, a header with the key kafka_acknowledgment of the type org.springframework.kafka.support.Acknowledgment header will be present in the inbound message. Applications may use this header for acknowledging messages.
代码并不复杂,我也没有使用任何预先生成的项目。举例说明不了我做了什么,所以我不知道我可能会错过什么。
1条答案
按热度按时间uz75evzq1#
看起来你的yaml中减少了一级缩进。
根据文档,属性必须如下所示:
但你的样本是这样的:
多加注意
.kafka.
在中间。我不知道如何帮助正确管理yaml,但这是我们必须要让它工作。