java - 无法捕获 Kafka TopicExistsException

Question

我编写了一个程序，它在使用 AdminClient 启动时创建 Kafka 主题。如果主题存在，则抛出 TopicExistsException。但是我无法捕获它。相关代码:

try {
    CreateTopicsResult result = adminClient.createTopics(Collections.singleton(new NewTopic(topic, 2, (short) 1)));
    result.all().get(5, TimeUnit.SECONDS);
} catch (TopicExistsException e){
    log.info("topic exists: {}", topic);
} catch ( InterruptedException | TimeoutException | ExecutionException e){
    log.error("Topic create failed with: {}", e.getMessage());
}

预期的行为是执行第一个捕获，但根据日志，这不是发生的情况:

2020-12-01 13:18:56,462 ERROR [restartedMain] com.mypackage.KafkaService@lambda$configureTopics$0:75 - Topic create failed with: org.apache.kafka.common.errors.TopicExistsException: Topic 'MEASUREMENT_TEST' already exists.

我的进口是为了异常(exception):

import org.apache.kafka.common.errors.TopicExistsException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import java.lang.InterruptedException;

有什么建议吗？有什么我想念的吗？

Answer 1

这是因为主题存在异常被包裹在 ExecutionException 中。当 KafkaFuture 被抛出时完成。
您可以通过记录完整的堆栈跟踪来查看这一点，而不仅仅是 getMessage()部分:

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: Topic 'hello' already exists.
    at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45) ~[kafka-clients-2.0.0.jar:?]
    at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32) ~[kafka-clients-2.0.0.jar:?]
    at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89) ~[kafka-clients-2.0.0.jar:?]
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:262) ~[kafka-clients-2.0.0.jar:?]
Caused by: org.apache.kafka.common.errors.TopicExistsException: Topic 'hello' already exists.

所以如果你想 catch TopicExistsException ，您必须执行以下操作:

try {
    CreateTopicsResult result = adminClient.createTopics(Collections.singleton(new NewTopic(topic, 2, (short) 1)));
    result.all().get(5, TimeUnit.SECONDS);
} catch (ExecutionException e){
    if(e.getCause() != null && e.getCause() instanceof TopicExistsException) {
        log.info("topic exists: {}", topic);
    } else {
        log.error("Topic create failed with: {}", e);
    }
} catch ( InterruptedException | TimeoutException){
    log.error("Topic create failed with: {}", e.getMessage());
}