Kafka StreamsExcture TimeoutExcture：正在物化KTable的N条过期记录

Question

我的流应用程序只是从一个记录主题中物化一个KTable。与100K记录在主题，没有问题。但是，由于主题中15M记录，一旦我们得到几百万条记录，实例将崩溃，出现以下异常：

Thread“公司-de1f21f9-b445-449e-a59b-5e0cecfa54d1-StreamThread-1”组织中的例外情况。阿帕奇。Kafka。溪流。错误。StreamsException:任务[0_0]中止发送，因为以前的记录（时间戳1601327726515）捕获到错误，因此发送到主题公司。由于组织原因，读取变更日志。阿帕奇。Kafka。常见的错误。TimeoutException:公司的40条记录过期。read-changelog-0:120001 ms自批创建以来已通过]

下面是我们正在运行的服务的详细示例的要点。

令我不解的是，使我的streams应用程序崩溃的错误（如下）是引用了一个过载的生产者，然而，这个服务只是实现了一个KTable。

java prettyprint-override">streamsBuilder
  .stream(egressTopic, Consumed.with(Serdes.String(), companySerde))
  .toTable(Materialized.<String, Company, KeyValueStore<Bytes, byte[]>>as(companyKTableName)
    .withKeySerde(Serdes.String())
    .withValueSerde(companySerde));

我已经调整过的属性试图让它名义上运行：

• batch.size10000
• linger.ms1000
• request.timeout.ms300000
• max.block.ms300000
• retry.backoff.ms1000
• replication.factor3

Answer 1

每个表都有一个用于容错的changelog主题作为支持。因此，每次写入KTable也是写入相应的变更日志主题。

如果您输入的主题配置了日志压缩，您可以重写您的程序

streamsBuilder.table(
    egressTopic,
    Materialized.<String, Company, KeyValueStore<Bytes, byte[]>>as(companyKTableName)
        .withKeySerde(Serdes.String())
        .withValueSerde(companySerde)
);

此外，还可以启用拓扑。optimization=“all”：在这种情况下，输入主题将被重新用作变更日志以恢复状态，并且不会创建其他变更日志主题。