Kafka external stream keeps reporting warning after fails to be created

[zliang-min]

Describe what's wrong

When I tried to create a kafka external stream which connects to a kafka cluster which does not exist, on the client side, it reports the following error as expected:

Code: 2003. DB::Exception: Received from 127.0.0.1:8463. DB::Exception: foo topic does't exist. (RESOURCE_NOT_FOUND)

However, on the server side, it keeps printing warning logs:

%3|1698688956.289|FAIL|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT)
2023.10.30 11:02:36.289582 [ 89967498 ] {} <Warning> KafkaWALSimpleConsumer: Error occurred, error=Local: Broker transport failure, reason=localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection refused (after 0ms in state CONNECT)

%3|1698688957.294|FAIL|rdkafka#consumer-1| [thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT)
2023.10.30 11:02:37.294835 [ 89967498 ] {} <Warning> KafkaWALSimpleConsumer: Error occurred, error=Local: Broker transport failure, reason=localhost:9092/bootstrap: Connect to ipv6#[::1]:9092 failed: Connection refused (after 0ms in state CONNECT)

That means we have resource leaked.

How to reproduce

Error message and/or stacktrace

Additional context