feat: Support automatic DNS lookup for kafka bootstrap servers

src/common/meta/src/error.rs

@@ -340,6 +340,16 @@ pub enum Error {
         error: rskafka::client::error::Error,
     },
 
+    #[snafu(display("Failed to resolve Kafka broker endpoint {:?}", broker_endpoint))]
+    ResolveKafkaEndpoint {
+        broker_endpoint: String,
+        #[snafu(source)]
+        error: std::io::Error,
+    },
+
+    #[snafu(display("Endpoint ip not found for broker endpoint: {:?}", broker_endpoint))]
+    EndpointIpNotFound { broker_endpoint: String },
+
     #[snafu(display("Failed to build a Kafka controller client"))]
     BuildKafkaCtrlClient {
         location: Location,
@@ -425,6 +435,8 @@ impl ErrorExt for Error {
             | BuildKafkaClient { .. }
             | BuildKafkaCtrlClient { .. }
             | BuildKafkaPartitionClient { .. }
+            | ResolveKafkaEndpoint { .. }
+            | EndpointIpNotFound { .. }
             | ProduceRecord { .. }
             | CreateKafkaWalTopic { .. }
             | EmptyTopicPool { .. }

src/common/meta/src/wal_options_allocator/kafka/topic_manager.rs

@@ -26,11 +26,12 @@ use rskafka::client::{Client, ClientBuilder};
 use rskafka::record::Record;
 use rskafka::BackoffConfig;
 use snafu::{ensure, ResultExt};
+use tokio::net;
 
 use crate::error::{
     BuildKafkaClientSnafu, BuildKafkaCtrlClientSnafu, BuildKafkaPartitionClientSnafu,
-    CreateKafkaWalTopicSnafu, DecodeJsonSnafu, EncodeJsonSnafu, InvalidNumTopicsSnafu,
-    ProduceRecordSnafu, Result,
+    CreateKafkaWalTopicSnafu, DecodeJsonSnafu, EncodeJsonSnafu, EndpointIpNotFoundSnafu,
+    InvalidNumTopicsSnafu, ProduceRecordSnafu, ResolveKafkaEndpointSnafu, Result,
 };
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::store::PutRequest;
@@ -108,6 +109,26 @@ impl TopicManager {
         Ok(())
     }
 
+    async fn resolve_broker_endpoint(broker_endpoint: &str) -> Result<String> {
+        let ips: Vec<_> = net::lookup_host(broker_endpoint)
+            .await
+            .with_context(|_| ResolveKafkaEndpointSnafu {
+                broker_endpoint: broker_endpoint.to_string(),
+            })?
+            .into_iter()
+            // Not sure if we should filter out ipv6 addresses
+            .filter(|addr| addr.is_ipv4())
+            .collect();
+        if ips.is_empty() {
+            return (|| {
+                EndpointIpNotFoundSnafu {
+                    broker_endpoint: broker_endpoint.to_string(),
+                }
+                .fail()
+            })();
+        }
+        Ok(ips[0].to_string())
+    }
     /// Tries to create topics specified by indexes in `to_be_created`.
     async fn try_create_topics(&self, topics: &[String], to_be_created: &[usize]) -> Result<()> {
         // Builds an kafka controller client for creating topics.
@@ -117,7 +138,11 @@ impl TopicManager {
             base: self.config.backoff.base as f64,
             deadline: self.config.backoff.deadline,
         };
-        let client = ClientBuilder::new(self.config.broker_endpoints.clone())
+        let mut broker_endpoints = Vec::with_capacity(self.config.broker_endpoints.len());
+        for endpoint in &self.config.broker_endpoints {
+            broker_endpoints.push(Self::resolve_broker_endpoint(endpoint).await?);
+        }
+        let client = ClientBuilder::new(broker_endpoints)
             .backoff_config(backoff_config)
             .build()
             .await

src/log-store/src/error.rs

@@ -108,6 +108,16 @@ pub enum Error {
         error: rskafka::client::error::Error,
     },
 
+    #[snafu(display("Failed to resolve Kafka broker endpoint {:?}", broker_endpoint))]
+    ResolveKafkaEndpoint {
+        broker_endpoint: String,
+        #[snafu(source)]
+        error: std::io::Error,
+    },
+
+    #[snafu(display("Failed to find ip for broker endpoint: {:?}", broker_endpoint))]
+    EndpointIpNotFound { broker_endpoint: String },
+
     #[snafu(display(
         "Failed to build a Kafka partition client, topic: {}, partition: {}",
         topic,

src/log-store/src/kafka/client_manager.rs

@@ -22,9 +22,13 @@ use rskafka::client::producer::{BatchProducer, BatchProducerBuilder};
 use rskafka::client::{Client as RsKafkaClient, ClientBuilder};
 use rskafka::BackoffConfig;
 use snafu::ResultExt;
+use tokio::net;
 use tokio::sync::RwLock;
 
-use crate::error::{BuildClientSnafu, BuildPartitionClientSnafu, Result};
+use crate::error::{
+    BuildClientSnafu, BuildPartitionClientSnafu, EndpointIpNotFoundSnafu,
+    ResolveKafkaEndpointSnafu, Result,
+};
 
 // Each topic only has one partition for now.
 // The `DEFAULT_PARTITION` refers to the index of the partition.
@@ -80,7 +84,11 @@ impl ClientManager {
             base: config.backoff.base as f64,
             deadline: config.backoff.deadline,
         };
-        let client = ClientBuilder::new(config.broker_endpoints.clone())
+        let mut broker_endpoints = Vec::with_capacity(config.broker_endpoints.len());
+        for endpoint in &config.broker_endpoints {
+            broker_endpoints.push(Self::resolve_broker_endpoint(endpoint).await?);
+        }
+        let client = ClientBuilder::new(broker_endpoints)
             .backoff_config(backoff_config)
             .build()
             .await
@@ -94,6 +102,26 @@ impl ClientManager {
             client_pool: RwLock::new(HashMap::new()),
         })
     }
+    async fn resolve_broker_endpoint(broker_endpoint: &str) -> Result<String> {
+        let ips: Vec<_> = net::lookup_host(broker_endpoint)
+            .await
+            .with_context(|_| ResolveKafkaEndpointSnafu {
+                broker_endpoint: broker_endpoint.to_string(),
+            })?
+            .into_iter()
+            // Not sure if we should filter out ipv6 addresses
+            .filter(|addr| addr.is_ipv4())
+            .collect();
+        if ips.is_empty() {
+            return (|| {
+                EndpointIpNotFoundSnafu {
+                    broker_endpoint: broker_endpoint.to_string(),
+                }
+                .fail()
+            })();
+        }
+        Ok(ips[0].to_string())
+    }
 
     /// Gets the client associated with the topic. If the client does not exist, a new one will
     /// be created and returned.