如何为Flink Kafka Consumer增加超时异常重试次数？

我们有一个基于Flink的Kafka消费作业，使用Flink-connector-Kafka库的KafkaSource进行数据消费。偶尔会出现**"Failed to get metadata for all topics"**超时异常（正在排查Kafka及网络问题），但从异常日志中可见请求仅尝试了1次。请问是否可以将重试次数增加至1次以上？如果可以，该如何操作？

异常日志 #

at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:600)
at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$QuiesceableContext.failJob(RecreateOnResetOperatorCoordinator.java:237)
at org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext.failJob(SourceCoordinatorContext.java:374)
at org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext.handleUncaughtExceptionFromAsyncCall(SourceCoordinatorContext.java:387)
at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:42)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to list subscribed topic partitions due to 
at org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.checkPartitionChanges(KafkaSourceEnumerator.java:234)
at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$null$4(ExecutorNotifier.java:133)
at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40)
... 6 more
Caused by: java.lang.RuntimeException: Failed to get metadata for all topics.
at org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriberUtils.getAllTopicMetadata(KafkaSubscriberUtils.java:37)
at org.apache.flink.connector.kafka.source.enumerator.subscriber.TopicPatternSubscriber.getSubscribedTopicPartitions(TopicPatternSubscriber.java:48)
at org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.getSubscribedTopicPartitions(KafkaSourceEnumerator.java:219)
at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$notifyReadyAsync$5(ExecutorNotifier.java:130)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
at java.base/java.util.concurrent.FutureTask.runAndReset(Unknown Source)
... 4 more
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1703430983336, tries=1, nextAllowedTryMs=1703430983497) timed out at 1703430983397 after 1 attempt(s)
at java.base/java.util.concurrent.CompletableFuture.reportGet(Unknown Source)
at java.base/java.util.concurrent.CompletableFuture.get(Unknown Source)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
at org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriberUtils.getAllTopicMetadata(KafkaSubscriberUtils.java:34)
... 9 more
Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1703430983336, tries=1, nextAllowedTryMs=1703430983497) timed out at 1703430983397 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.DisconnectException: Cancelled listTopics request with correlation id 66857 due to node 1001 being disconnected

可以通过配置Kafka客户端的重试相关参数实现，Flink KafkaSource底层依赖Kafka客户端，所有Kafka原生客户端配置都可以通过setProperty方法传递。

核心配置参数 #

• retries：设置请求的总重试次数（例如设为3，则加上初始尝试共4次请求）
• retry.backoff.ms：两次重试之间的间隔时间，避免频繁重试给集群造成压力
• request.timeout.ms：单个请求的超时时间，适当调大可以减少因网络延迟导致的超时
• metadata.max.age.ms：控制元数据的刷新间隔，缩短间隔可以更快感知集群拓扑变化

代码配置示例 #

KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
        .setBootstrapServers("your-kafka-brokers:9092")
        .setTopics("your-target-topics") // 或用setTopicPattern订阅正则匹配的主题
        .setGroupId("flink-consumer-group-id")
        .setValueOnlyDeserializer(new SimpleStringSchema())
        // 配置重试相关参数
        .setProperty("retries", "3")
        .setProperty("retry.backoff.ms", "1000")
        .setProperty("request.timeout.ms", "5000")
        .setProperty("metadata.max.age.ms", "300000") // 5分钟刷新一次元数据
        .build();

注意事项 #

• 重试次数不宜设置过高，避免因长时间等待重试导致Flink作业协调器超时
• 重试只是临时容错方案，仍需优先排查Kafka集群稳定性、网络连通性等根本问题
• 如果使用正则表达式订阅主题，元数据请求频率更高，更需合理调整上述参数

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