java – 为什么Kafka KTable缺少条目？

2023年2月11日 298次阅读

我有一个使用Kafka Streams的KTable的单实例
java应用程序.直到最近,当突然有些消息似乎消失时,我可以使用KTable检索所有数据.那里应该有~33k个带有唯一键的消息.

当我想按键检索消息时,我没有得到一些消息.我使用ReadOnlyKeyValueStore来检索消息：

final ReadOnlyKeyValueStore<GenericRecord, GenericRecord> store = ((KafkaStreams)streams).store(storeName, QueryableStoreTypes.keyValueStore());
store.get(key);

这些是我为KafkaStreams设置的配置设置.

final Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_SERVER_CONFIG, serverId);
config.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
config.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
config.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
config.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);

卡夫卡：0.10.2.0-cp1
汇合：3.2.0

调查给我带来了一些非常令人担忧的见解.使用REST代理我手动读取分区,发现一些偏移返回错误.

请求：
/主题/ {主题} /分区/ {分区} /消息？偏移= {}偏移

{
    "error_code": 50002,
    "message": "Kafka error: Fetch response contains an error code: 1"
}

没有客户端,java和命令行都没有返回任何错误.他们只是跳过错误的丢失消息,导致KTables中缺少数据.一切都很好,没有通知似乎某些消息已经腐败.

我有两个代理,所有主题的复制因子都是2,并且完全复制.两个经纪人分别返回相同的.重启经纪人并没有什么区别.

>可能是什么原因？
>如何在客户端检测此案例？

最佳答案 通过
default Kafka Broker配置密钥cleanup.policy设置为删除.将其设置为紧凑以保留每个密钥的最新消息.
See compaction.

删除旧消息不会更改最小偏移量,因此尝试检索其下方的消息会导致错误.错误非常模糊. Kafka Streams客户端将开始从最小偏移量读取消息,因此没有错误.唯一可见的效果是缺少KTables中的数据.

由于caches应用程序正在运行,即使从Kafka本身删除消息后,所有数据仍可用.他们将在清理后消失.

点赞