为什么KafkaKTable缺少条目？

Question

我有一个使用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);

Kafka：0.10.2.0-cp1
合流：3.2.0

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

请求：/topics/{topic}/partitions/{partition}/消息？偏移量={offset}

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

但是，没有客户端、java和命令行返回任何错误。他们只是跳过 错误的 丢失的消息，导致KTables中的数据丢失。一切都很好，似乎有些信息不知怎么地被破坏了。

我有两个经纪人，所有的主题都有2的复制因子，并且完全复制。两个经纪人分别返回相同的。重启经纪人没有区别。

• 原因可能是什么

Answer 1

默认情况下，Kafka Broker配置键cleanup.policy设置为删除。将其设置为紧凑以保留每个键的最新消息。参见压实。

删除旧邮件不会更改最小偏移量，因此尝试在其下方检索邮件会导致错误。错误很模糊。Kafka Streams客户端将从最小偏移量开始读取消息，因此不会出现错误。唯一可见的影响是KTables中缺少数据。

由于缓存，应用程序运行时，即使从Kafka本身删除消息，所有数据仍可能可用。它们会在清理后消失。