Kafka adminClient 抛出 TimeoutException答案

【问题标题】：Kafka adminClient throws TimeoutExceptionKafka adminClient 抛出 TimeoutException
【发布时间】：2020-05-25 19:24:37
【问题描述】：

我有一个运行状况线程，它每 5 秒从我的工作应用程序检查一次 Kafka 集群的状态。但是，我时不时会收到TimeoutException：

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Aborted due to timeout.
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)

我也有工具可以从外部监视我的集群（Cruise Control、Grafana），但它们都没有指出集群中的任何问题。此外，我的工作应用程序不断消耗消息，似乎没有一个失败。

为什么我偶尔会收到此超时？如果经纪人没有关闭，那么我认为我的配置中的某些内容已关闭。我将超时设置为 5 秒，这似乎绰绰有余。

我的 AdminClient 配置：

 @Bean
public AdminClient adminClient() {
    return KafkaAdminClient.create(adminClientConfigs());
}

 public Map<String, Object> adminClientConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, serverAddress);
    props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 5000);
    return props;
}

我如何检查集群（我比在代理列表上运行逻辑）：

@Autowired
private AdminClient adminClient;

 private void addCluster() throws ExecutionException, InterruptedException {
        adminClient.describeCluster().nodes().get().forEach(node -> brokers.add(node.host()));
    }

【问题讨论】：

标签： java apache-kafka spring-kafka

【解决方案1】：

两件事：

默认请求超时为 30 秒。通过将其设置为较小的值，您会增加慢请求超时的风险。如果 1000 个请求中的一个 (0.1%) 耗时超过 5 秒，因为您每隔几秒查询一次，您每天都会看到几次失败。
要调查为什么某些调用需要更长的时间，您可以做几件事：
- 检查 Kafka 客户端日志。 describeCluster() 可能需要启动与集群的新连接。在这种情况下，客户端还必须发送ApiVersionsRequest，并且根据您的配置，可能会建立 TLS 连接和/或执行 SASL 身份验证。如果发生任何这些情况，应在客户端日志中明确。（您可能需要稍微提高日志级别才能看到所有这些）。
- 检查代理请求指标。 describeCluster() 转换为 MetadataRequest 发送给经纪人。您可以跟踪处理请求所需的时间。请参阅docs 中描述的指标，在您的情况下，尤其是：kafka.network:type=RequestMetrics,name=*,request=Metadata

【讨论】：