如何让KafkaProducer抛出并捕获UNKNOWN_TOPIC_OR_PARTITION错误？

我完全懂你的处境：Kafka禁用了自动创建主题，你想通过捕获「未知主题」的异常来触发外部API创建主题，但默认情况下KafkaProducer只会循环打印UNKNOWN_TOPIC_OR_PARTITION的WARN日志，根本不会抛出异常让你处理。你已经通过设置max.block.ms=0让它立刻抛出超时异常，但觉得这个方案太激进，想找更合适的替代方法，对吧？

先给你理清楚默认行为的本质：KafkaProducer在发送消息前会先尝试获取目标主题的元数据，默认情况下它会阻塞最多1分钟（由max.block.ms默认值60000ms控制），期间不断重试获取元数据，所以只会打日志而不会立刻报错。

接下来给你分析几个可选的方案，各有优劣，你可以根据业务场景选择：

1. 优化当前的max.block.ms方案（温和版） #

你现在设的max.block.ms=0确实能让Producer立刻抛出TimeoutException（会被包装在ExecutionException里），但这个值太极端——毕竟除了未知主题的场景，正常情况下Producer也可能需要短暂阻塞获取元数据（比如主题分区变化时），设为0会导致这些正常场景也直接超时。

建议改成一个合理的小值，比如1000ms（1秒）：

properties.setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, "1000");

这样既不会像默认那样等1分钟才报错，也不会影响正常的元数据获取逻辑。捕获异常后，你可以结合日志或者异常栈判断是否是因为主题不存在导致的超时，再触发创建逻辑。

2. 主动用KafkaAdminClient检查主题存在性（最可靠） #

这个方案是最精准的：在发送消息前，先用AdminClient主动查询主题是否存在，这样能明确判断原因，而不是靠超时间接推断。

示例代码：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.DescribeTopicsResult;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class TopicChecker {
    public static boolean topicExists(String bootstrapServers, String topicName) {
        Properties adminProps = new Properties();
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        
        try (AdminClient adminClient = AdminClient.create(adminProps)) {
            DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(topicName));
            // 设置5秒超时，避免无限阻塞
            return result.all().get(5, TimeUnit.SECONDS).containsKey(topicName);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            // 可进一步拆解异常链，区分是网络问题还是主题不存在
            return false;
        }
    }
}

使用时，在发送消息前先做检查：

String topic = "unknown_topic";
if (!TopicChecker.topicExists("***:9092", topic)) {
    // 调用外部API创建主题
    createTopicViaExternalAPI(topic);
}
// 再发送消息
producer.send(new ProducerRecord<>(topic, "hello world")).get();

这个方案的好处是：完全明确主题是否存在，不会把网络超时、Broker不可用等其他场景误判为主题不存在，可靠性最高。

3. 主动获取元数据并设置超时（轻量中间方案） #

如果你不想额外封装AdminClient逻辑，也可以在发送前主动调用Producer.partitionsFor()方法，这个方法会尝试获取主题的分区信息，如果主题不存在，会在max.block.ms超时后抛出异常。

示例：

String topic = "unknown_topic";
try {
    // 主动获取主题分区信息，超时由max.block.ms控制
    producer.partitionsFor(topic);
} catch (TimeoutException e) {
    // 结合日志判断是否为主题不存在导致的超时
    createTopicViaExternalAPI(topic);
}
// 再发送消息
producer.send(new ProducerRecord<>(topic, "hello world")).get();

这个方案比直接发送更主动，但本质还是依赖超时判断，不如AdminClient直接明确。

总结一下 #

• 若追求简单快速，且能接受轻微误判风险，优化max.block.ms到合理小值是最省事的选择；
• 若要最高的可靠性和精准性，优先用KafkaAdminClient提前检查主题；
• 轻量场景下，主动调用partitionsFor()也是一个可选的中间方案。

备注：内容来源于stack exchange，提问作者Makrushin Evgenii