Kafka Streams恢复消费者无法更新IP（DNS问题）

问题背景 #

• Kafka Streams应用通过Aiven Kafka提供的公网域名public-my-kafka.aivencloud.com:25624连接bootstrap地址
• Broker升级更换全新VM及IP后，添加新KStream实例触发重平衡，发现恢复消费者（restore-consumer）持续尝试连接旧Broker IP
• 无致命错误日志，但存在循环连接超时/失败的重复日志：

  INFO | kstream-sample-67928ec5-cdc1-416e-a680-a6686c020023-StreamThread-1 | org.apache.kafka.clients.NetworkClient
        | [Consumer clientId=kstream-sample-67928ec5-cdc1-416e-a680-a6686c020023-StreamThread-1-restore-consumer, groupId=null] Disconnecting from node 1 due to socket connection setup timeout. The timeout value is 25409 ms.
  

  WARN | kstream-sample-67928ec5-cdc1-416e-a680-a6686c020023-StreamThread-1 | org.apache.kafka.clients.NetworkClient
        | [Consumer clientId=kstream-sample-67928ec5-cdc1-416e-a680-a6686c020023-StreamThread-1-restore-consumer, groupId=null] Connection to node 5 (20.56.29.123/20.56.29.123:25624) could not be established. Broker may not be available.
  

• 已尝试的配置/操作：
  • 为恢复消费者设置CLIENT_DNS_LOOKUP_CONFIG=RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY和METADATA_MAX_AGE_CONFIG=500
  • 通过代码设置JVM DNS缓存TTL：java.security.Security.setProperty("networkaddress.cache.ttl" , "1");
  • 使用Kafka版本：3.2.3

进一步调试建议 #

1. 验证恢复消费者的配置是否生效 #

Kafka Streams的恢复消费者配置需通过StreamsConfig.restoreConsumerPrefix()前缀注入，需确认配置未被覆盖：

• 开启客户端DEBUG日志，查看org.apache.kafka.clients.consumer.ConsumerConfig的初始化日志，确认恢复消费者的client.dns.lookup和metadata.max.age.ms参数为预期值
• 确保配置在创建KafkaStreams实例前设置，且未被其他配置源（如外部配置文件）覆盖

2. 强制恢复消费者刷新元数据 #

恢复消费者启动时拉取Broker元数据，若元数据缓存未过期，可能不会重新查询DNS：

• 手动触发元数据刷新：在StreamThread的恢复逻辑中，调用restoreConsumer.partitionsFor(topic)或restoreConsumer.listTopics()强制刷新元数据（需注意线程安全）
• 缩短重连退避时间，让恢复消费者更快重试并触发元数据刷新：

  props.put(StreamsConfig.restoreConsumerPrefix(ConsumerConfig.RECONNECT_BACKOFF_MS_CONFIG), 100);
  props.put(StreamsConfig.restoreConsumerPrefix(ConsumerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG), 500);
  

3. 检查Aiven Kafka的Broker公告地址 #

Aiven Kafka默认可能将内部IP公告给客户端，需确认公网环境下的公告配置：

• 登录Aiven控制台，查看Kafka服务的「Advanced configuration」，确认advertised.listeners包含公网域名而非旧IP
• 若advertised.listeners仍为旧IP，联系Aiven支持更新配置

4. 确认JVM DNS缓存实际生效 #

JVM DNS缓存可能受系统级缓存或安全策略文件影响：

• 直接读取当前JVM缓存配置：

  System.out.println("networkaddress.cache.ttl: " + java.security.Security.getProperty("networkaddress.cache.ttl"));
  System.out.println("networkaddress.cache.negative.ttl: " + java.security.Security.getProperty("networkaddress.cache.negative.ttl"));
  

• 若代码设置无效，尝试在JVM启动参数中添加：-Dnetworkaddress.cache.ttl=1 -Dnetworkaddress.cache.negative.ttl=1（优先级高于代码设置）

5. 自定义SocketSupplier强制DNS解析（临时方案） #

若上述方法无效，可通过自定义SocketSupplier绕过JVM缓存，手动解析域名：

props.put(StreamsConfig.restoreConsumerPrefix(CommonClientConfigs.SOCKET_SUPPLIER_CLASS_CONFIG), CustomDnsSocketSupplier.class.getName());

public class CustomDnsSocketSupplier implements SocketSupplier {
    @Override
    public Socket connect(String host, int port, int connectTimeoutMs) throws IOException {
        // 手动解析域名，绕过JVM缓存
        InetAddress[] addresses = InetAddress.getAllByName(host);
        if (addresses.length == 0) {
            throw new UnknownHostException(host);
        }
        // 选择第一个有效地址（或实现轮询逻辑）
        Socket socket = new Socket();
        socket.connect(new InetSocketAddress(addresses[0], port), connectTimeoutMs);
        return socket;
    }
}

Kafka-Client中恢复消费者的DNS查询代码位置 #

在Kafka 3.2.3版本中，恢复消费者的DNS查询逻辑集中在以下核心类：

1. org.apache.kafka.clients.NetworkClient #

• 建立新连接时，调用InetAddressResolver.resolve()解析Broker地址
• 核心代码路径：NetworkClient.initiateConnect() -> InetAddressResolver.resolve()

2. org.apache.kafka.common.network.InetAddressResolver #

• 实现DNS解析逻辑，根据client.dns.lookup配置决定解析策略：
  • RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY：仅对bootstrap服务器解析规范名称，Broker地址从元数据获取
  • USE_ALL_DNS_IPS：解析所有IP并尝试连接
• 核心代码路径：InetAddressResolver.resolve() -> InetAddress.getAllByName()（最终调用JVM DNS解析）

3. org.apache.kafka.streams.processor.internals.StandbyTaskCreator #

• 恢复消费者的创建逻辑在此类中，配置通过restoreConsumerPrefix注入
• 核心代码路径：StandbyTaskCreator.createRestoreConsumer() -> 初始化KafkaConsumer并应用恢复消费者专属配置

内容的提问来源于stack exchange，提问作者DarVar