Java Apache Kafka Producer 元数据更新器和重试逻辑

Question

我正在为 Apache Kafka 使用 Spring，并创建了一个服务，该服务通过 Spring 的 KafkaTemplate 使用 Kafka 生产者 (org.apache.kafka.clients.producer) 向主题发送消息。在目标 Kafka 集群上，我禁用了自动主题创建。使用此处列出的生产者配置组合https://kafka.apache.org/documentation/#producerconfigs 我成功控制了请求的重试次数、重试间隔时间等。

如果我提供的主题不存在，则请求会在我期望的时候超时（达到 ma​​x.block.ms 的值）。但是，在超时后，我继续以 retry.backoff.ms 设置的间隔获取日志条目（如下面的条目），直到 300000 ms / 5 分钟到达。

我无法确定生产者或代理上的哪个配置属性可以更改以阻止生产者检查 5 分钟以查看主题是否已创建。

有人可以指出正确的设置，让我在请求超时后减少这种情况或让它停止检查吗？

日志条目示例：

WARN  [kafka-producer-network-thread | producer-1] org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater: [Producer clientId=producer-1] Error while fetching metadata with correlation id 9 : {<specified_topic>=UNKNOWN_TOPIC_OR_PARTITION}

使用的生产者配置：

• delivery.timeout.ms = 5000
• linger.ms = 1000
• max.block.ms = 8000
• request.timeout.ms= 4000
• max.retry.count = 0
• retry.backoff.ms = 2000

Answer 1

Kafka Producer 在第一个 send 之前检索和缓存主题/分区元数据。然后，它会定期尝试刷新此元数据，每隔metadata.max.age.ms（默认=5 分钟）刷新一次“好”主题，每隔retry.backoff.ms 刷新一次“无效”主题。这些元数据刷新尝试是您在日志中观察到的。

为了防止缓存不受控制地增长，根据这些source comments，在一定时间后从缓存中删除未使用的主题。目前，此到期时间在ProducerMetadata.java 中被硬编码为 5 分钟。

  public class ProducerMetadata extends Metadata {
      private static final long TOPIC_EXPIRY_NEEDS_UPDATE = -1L;
      static final long TOPIC_EXPIRY_MS = 5 * 60 * 1000;
        ...

您实际上可以通过将生产者日志级别设置为DEBUG 来观察所有这些活动。