Kafka 对于分区返回 endOffset 15,但可以使用的最后一条消息的偏移量为 13,而不是我期望的 14。我想知道为什么。
The 卡夫卡文档 https://kafka.apache.org/28/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#endOffsets(java.util.Collection) read
在默认的read_uncommissed隔离级别下,结束偏移量是高水位线(即最后成功复制的消息的偏移量加一)。对于read_commited消费者来说,结束偏移量是最后一个稳定偏移量(LSO),它是高水位线的最小值和任何打开事务的最小偏移量。
这是 kafkacat 的输出。我使用 kafkacat,因为它可以打印消息偏移量:
$ kafkacat -Ce -p0 -tTK -f'offset: %o key: %k\n'
offset: 0 key: 0108
offset: 1 key: 0253
offset: 4 key: 0278
offset: 5 key: 0198
offset: 8 key: 0278
offset: 9 key: 0210
offset: 10 key: 0253
offset: 11 key: 1058
offset: 12 key: 0141
offset: 13 key: 1141
% Reached end of topic TK [0] at offset 15: exiting
同样令人困惑的是——而且很可能是相关的——偏移量不是连续的,尽管我还没有设置压缩等。
更多细节:
$ kafka-topics.sh --bootstrap-server localhost:9092 --topic TK --describe
Topic: TK PartitionCount: 2 ReplicationFactor: 1 Configs: segment.bytes=1073741824
Topic: TK Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: TK Partition: 1 Leader: 0 Replicas: 0 Isr: 0
通过 kafka-console-consumer.sh 打印密钥:
$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic TK \
--offset earliest --partition 0 --timeout-ms 5000 \
--property print.key=true --property print.value=false
0108
0253
0278
0198
0278
0210
0253
1058
0141
1141
[2021-09-15 10:54:06,556] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.TimeoutException
Processed a total of 10 messages
N.B. This topic has been produced to without involvement of transactions, and *) consumption is being done in read_uncommitted
mode.
*) Actually, processing.guarantee is set to exactly_once_beta
, so that would amount to using transactions.
更多信息事实证明,我可以使用我的 Streams 应用程序可靠地重现这种情况(1. 擦除 kafka/zookeeper 数据,2. 重新创建主题,3. 运行应用程序),其输出是显示此问题的主题。
同时,我已将 Streams 应用程序精简为这种无操作拓扑,并且仍然可以重现它:
Topologies:
Sub-topology: 0
Source: KSTREAM-SOURCE-0000000000 (topics: [TK1])
--> KSTREAM-SINK-0000000001
Sink: KSTREAM-SINK-0000000001 (topic: TK)
<-- KSTREAM-SOURCE-0000000000
News同时,我已将本地运行的 Kafka 代理 (2.5.0) 替换为在 Docker 容器中运行的代理 (wurstmeister/kafka:2.13-2.6.0)。问题仍然存在。
该应用程序使用版本为 6.0.1-ccs 的 kafka 库,对应于 2.6.0。