[client] Allow bootstrap endpoint with unknown server type

bindings/elixir/lib/fluss/server_node.ex

@@ -29,7 +29,7 @@ defmodule Fluss.ServerNode do
           uid: String.t(),
           host: String.t(),
           port: non_neg_integer(),
-          server_type: :tablet_server | :coordinator_server
+          server_type: :tablet_server | :coordinator_server | :unknown
         }
 
   @spec url(t()) :: String.t()

bindings/elixir/native/fluss_nif/src/admin.rs

@@ -30,6 +30,7 @@ use std::sync::Arc;
 pub enum NifServerType {
     TabletServer,
     CoordinatorServer,
+    Unknown,
 }
 
 #[derive(NifStruct)]
@@ -52,6 +53,7 @@ impl NifServerNode {
             server_type: match node.server_type() {
                 ServerType::TabletServer => NifServerType::TabletServer,
                 ServerType::CoordinatorServer => NifServerType::CoordinatorServer,
+                ServerType::Unknown => NifServerType::Unknown,
             },
         }
     }

crates/fluss/src/client/metadata.rs

@@ -88,7 +88,7 @@ impl Metadata {
             -1,
             socket_address.ip().to_string(),
             socket_address.port() as u32,
-            ServerType::CoordinatorServer,
+            ServerType::Unknown,
         );
         let con = connections.get_connection(&server_node).await?;
 

crates/fluss/src/cluster/mod.rs

@@ -36,12 +36,14 @@ pub struct ServerNode {
 
 impl ServerNode {
     pub fn new(id: i32, host: String, port: u32, server_type: ServerType) -> ServerNode {
+        let uid = match &server_type {
+            ServerType::CoordinatorServer => format!("cs-{id}"),
+            ServerType::TabletServer => format!("ts-{id}"),
+            ServerType::Unknown => format!("unknown-{host}:{port}"),
+        };
         ServerNode {
             id,
-            uid: match server_type {
-                ServerType::CoordinatorServer => format!("cs-{id}"),
-                ServerType::TabletServer => format!("ts-{id}"),
-            },
+            uid,
             host,
             port,
             server_type,
@@ -77,20 +79,23 @@ impl ServerNode {
 pub enum ServerType {
     TabletServer,
     CoordinatorServer,
+    Unknown,
 }
 
 impl ServerType {
     pub fn to_type_id(&self) -> i32 {
         match self {
             ServerType::CoordinatorServer => 1,
             ServerType::TabletServer => 2,
+            ServerType::Unknown => -1,
         }
     }
 
     pub fn from_type_id(type_id: i32) -> Option<ServerType> {
         match type_id {
             1 => Some(ServerType::CoordinatorServer),
             2 => Some(ServerType::TabletServer),
+            -1 => Some(ServerType::Unknown),
             _ => None,
         }
     }
@@ -101,6 +106,7 @@ impl fmt::Display for ServerType {
         match self {
             ServerType::TabletServer => write!(f, "TabletServer"),
             ServerType::CoordinatorServer => write!(f, "CoordinatorServer"),
+            ServerType::Unknown => write!(f, "Unknown"),
         }
     }
 }

crates/fluss/src/rpc/server_connection.rs

@@ -156,6 +156,10 @@ fn validate_server_type(
     expected: &ServerType,
     response_server_type: Option<i32>,
 ) -> Result<(), Error> {
+    if *expected == ServerType::Unknown {
+        return Ok(());
+    }
+
     // For forward-compat with servers that do not populate `server_type`, validation is skipped.
     let Some(type_id) = response_server_type else {
         return Ok(());
@@ -1268,6 +1272,21 @@ mod tests {
             )
             .is_ok()
         );
+        assert!(
+            validate_server_type(
+                &ServerType::Unknown,
+                Some(ServerType::CoordinatorServer.to_type_id()),
+            )
+            .is_ok()
+        );
+        assert!(
+            validate_server_type(
+                &ServerType::Unknown,
+                Some(ServerType::TabletServer.to_type_id()),
+            )
+            .is_ok()
+        );
+        assert!(validate_server_type(&ServerType::Unknown, Some(99),).is_ok());
 
         // Mismatch: connected to a coordinator while expecting a tablet server
         // (and vice versa).