Camel：检测Kafka错误IP

Question

我正在将Kafka配置为RouteBuilder中的源。我的目标是处理Kafka断线问题。我的RouteBuilder如下：

new RouteBuilder() {
        public void configure() {
            onException(Exception.class).process(exchange -> {
                final Exception exception = exchange.getException();
                logger.error(exception.getMessage());
                // will do more processing here
            });
            from(String.format("kafka:%s?brokers=%s:%s", topicName, host, port)).bean(getMyService(), "myMethod")
            .process(new Processor() {
                @Override
                public void process(Exchange exchange) throws Exception {
                    // some more processing
                }
            });
        }
    };

我提供了错误的主机和端口，并期望看到一个异常。但是，在日志中没有看到异常，并且没有调用onException处理。知道我做错了什么吗？

通过在本地运行https://github.com/apache/camel/blob/master/examples/camel-example-kafka/src/main/java/org/apache/camel/example/Kafka/messageConsumerClient.java而不运行任何Kafka服务器，可以重现类似的问题。这样做会导致持续的消息流：

Connection to node -1 could not be established. Broker may not be available.

有没有方法抛出异常？如有任何帮助，我们将不胜感激。

Answer 1

当您有消息要路由时，RouteBuilder中的OnException将被触发，但由于您无法连接到Kafka集群，所以您没有。这就是您看不到异常处理的原因。