Address comments

fresh-borzoni · fresh-borzoni · commit 36e8cdc4edbe · 2026-02-26T01:10:14.000Z
diff --git a/bindings/cpp/test/test_sasl_auth.cpp b/bindings/cpp/test/test_sasl_auth.cpp
@@ -26,6 +26,9 @@ class SaslAuthTest : public ::testing::Test {
     const std::string& sasl_servers() {
         return fluss_test::FlussTestEnvironment::Instance()->GetSaslBootstrapServers();
     }
+    const std::string& plaintext_servers() {
+        return fluss_test::FlussTestEnvironment::Instance()->GetBootstrapServers();
+    }
 };
 
 TEST_F(SaslAuthTest, SaslConnectWithValidCredentials) {
@@ -55,6 +58,26 @@ TEST_F(SaslAuthTest, SaslConnectWithValidCredentials) {
     ASSERT_OK(admin.DropDatabase(db_name, true, true));
 }
 
+TEST_F(SaslAuthTest, SaslConnectWithSecondUser) {
+    fluss::Configuration config;
+    config.bootstrap_servers = sasl_servers();
+    config.security_protocol = "sasl";
+    config.security_sasl_mechanism = "PLAIN";
+    config.security_sasl_username = "alice";
+    config.security_sasl_password = "alice-secret";
+
+    fluss::Connection conn;
+    ASSERT_OK(fluss::Connection::Create(config, conn));
+
+    fluss::Admin admin;
+    ASSERT_OK(conn.GetAdmin(admin));
+
+    // Basic operation to confirm functional connection
+    bool exists = false;
+    ASSERT_OK(admin.DatabaseExists("some_nonexistent_db_alice", exists));
+    ASSERT_FALSE(exists);
+}
+
 TEST_F(SaslAuthTest, SaslConnectWithWrongPassword) {
     fluss::Configuration config;
     config.bootstrap_servers = sasl_servers();
@@ -87,3 +110,16 @@ TEST_F(SaslAuthTest, SaslConnectWithUnknownUser) {
     EXPECT_NE(result.error_message.find("Authentication failed"), std::string::npos)
         << "Expected 'Authentication failed' in: " << result.error_message;
 }
+
+TEST_F(SaslAuthTest, SaslClientToPlaintextServer) {
+    fluss::Configuration config;
+    config.bootstrap_servers = plaintext_servers();
+    config.security_protocol = "sasl";
+    config.security_sasl_mechanism = "PLAIN";
+    config.security_sasl_username = "admin";
+    config.security_sasl_password = "admin-secret";
+
+    fluss::Connection conn;
+    auto result = fluss::Connection::Create(config, conn);
+    ASSERT_FALSE(result.Ok()) << "SASL client connecting to plaintext server should fail";
+}
diff --git a/bindings/cpp/test/test_utils.h b/bindings/cpp/test/test_utils.h
@@ -63,6 +63,29 @@ static constexpr int kPlainClientTabletPort = 9224;
 /// Execute a shell command and return its exit code.
 inline int RunCommand(const std::string& cmd) { return system(cmd.c_str()); }
 
+/// Join property lines with the escaped newline separator used by `printf` in docker commands.
+inline std::string JoinProps(const std::vector<std::string>& lines) {
+    std::string result;
+    for (size_t i = 0; i < lines.size(); ++i) {
+        if (i > 0) result += "\\n";
+        result += lines[i];
+    }
+    return result;
+}
+
+/// Build a `docker run` command with FLUSS_PROPERTIES.
+inline std::string DockerRunCmd(const std::string& name, const std::string& props,
+                                const std::vector<std::string>& port_mappings,
+                                const std::string& server_type) {
+    std::string cmd = "docker run -d --rm --name " + name + " --network " + kNetworkName;
+    for (const auto& pm : port_mappings) {
+        cmd += " -p " + pm;
+    }
+    cmd += " -e FLUSS_PROPERTIES=\"$(printf '" + props + "')\"";
+    cmd += " " + std::string(kFlussImage) + ":" + kFlussVersion + " " + server_type;
+    return cmd;
+}
+
 /// Wait until a TCP port is accepting connections, or timeout.
 inline bool WaitForPort(const std::string& host, int port, int timeout_seconds = 60) {
     auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(timeout_seconds);
@@ -129,28 +152,24 @@ class FlussTestCluster {
         std::string sasl_jaas =
             "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required"
             " user_admin=\"admin-secret\" user_alice=\"alice-secret\";";
-        std::string coord_props =
-            "zookeeper.address: " + std::string(kZookeeperName) +
-            ":2181\\n"
-            "bind.listeners: INTERNAL://" +
-            std::string(kCoordinatorName) + ":0, CLIENT://" + std::string(kCoordinatorName) +
-            ":9123, PLAIN_CLIENT://" + std::string(kCoordinatorName) +
-            ":9223\\n"
-            "advertised.listeners: CLIENT://localhost:9123, PLAIN_CLIENT://localhost:9223\\n"
-            "internal.listener.name: INTERNAL\\n"
-            "security.protocol.map: CLIENT:sasl\\n"
-            "security.sasl.enabled.mechanisms: plain\\n"
-            "security.sasl.plain.jaas.config: " +
-            sasl_jaas +
-            "\\n"
-            "netty.server.num-network-threads: 1\\n"
-            "netty.server.num-worker-threads: 3";
-
-        std::string coord_cmd = std::string("docker run -d --rm") + " --name " + kCoordinatorName +
-                                " --network " + kNetworkName + " -p 9123:9123" + " -p 9223:9223" +
-                                " -e FLUSS_PROPERTIES=\"$(printf '" + coord_props + "')\"" + " " +
-                                std::string(kFlussImage) + ":" + kFlussVersion +
-                                " coordinatorServer";
+
+        std::string coord = std::string(kCoordinatorName);
+        std::string zk = std::string(kZookeeperName);
+        std::string coord_props = JoinProps({
+            "zookeeper.address: " + zk + ":2181",
+            "bind.listeners: INTERNAL://" + coord + ":0, CLIENT://" + coord +
+                ":9123, PLAIN_CLIENT://" + coord + ":9223",
+            "advertised.listeners: CLIENT://localhost:9123, PLAIN_CLIENT://localhost:9223",
+            "internal.listener.name: INTERNAL",
+            "security.protocol.map: CLIENT:sasl",
+            "security.sasl.enabled.mechanisms: plain",
+            "security.sasl.plain.jaas.config: " + sasl_jaas,
+            "netty.server.num-network-threads: 1",
+            "netty.server.num-worker-threads: 3",
+        });
+
+        std::string coord_cmd = DockerRunCmd(kCoordinatorName, coord_props,
+                                             {"9123:9123", "9223:9223"}, "coordinatorServer");
         if (RunCommand(coord_cmd) != 0) {
             std::cerr << "Failed to start Coordinator Server" << std::endl;
             Stop();
@@ -165,33 +184,26 @@ class FlussTestCluster {
         }
 
         // Start Tablet Server (dual listeners: CLIENT=SASL on 9123, PLAIN_CLIENT=plaintext on 9223)
-        std::string ts_props =
-            "zookeeper.address: " + std::string(kZookeeperName) +
-            ":2181\\n"
-            "bind.listeners: INTERNAL://" +
-            std::string(kTabletServerName) + ":0, CLIENT://" + std::string(kTabletServerName) +
-            ":9123, PLAIN_CLIENT://" + std::string(kTabletServerName) +
-            ":9223\\n"
-            "advertised.listeners: CLIENT://localhost:" +
-            std::to_string(kTabletServerPort) +
-            ", PLAIN_CLIENT://localhost:" + std::to_string(kPlainClientTabletPort) +
-            "\\n"
-            "internal.listener.name: INTERNAL\\n"
-            "security.protocol.map: CLIENT:sasl\\n"
-            "security.sasl.enabled.mechanisms: plain\\n"
-            "security.sasl.plain.jaas.config: " +
-            sasl_jaas +
-            "\\n"
-            "tablet-server.id: 0\\n"
-            "netty.server.num-network-threads: 1\\n"
-            "netty.server.num-worker-threads: 3";
-
-        std::string ts_cmd = std::string("docker run -d --rm") + " --name " + kTabletServerName +
-                             " --network " + kNetworkName + " -p " +
-                             std::to_string(kTabletServerPort) + ":9123" + " -p " +
-                             std::to_string(kPlainClientTabletPort) + ":9223" +
-                             " -e FLUSS_PROPERTIES=\"$(printf '" + ts_props + "')\"" + " " +
-                             std::string(kFlussImage) + ":" + kFlussVersion + " tabletServer";
+        std::string ts = std::string(kTabletServerName);
+        std::string ts_props = JoinProps({
+            "zookeeper.address: " + zk + ":2181",
+            "bind.listeners: INTERNAL://" + ts + ":0, CLIENT://" + ts + ":9123, PLAIN_CLIENT://" +
+                ts + ":9223",
+            "advertised.listeners: CLIENT://localhost:" + std::to_string(kTabletServerPort) +
+                ", PLAIN_CLIENT://localhost:" + std::to_string(kPlainClientTabletPort),
+            "internal.listener.name: INTERNAL",
+            "security.protocol.map: CLIENT:sasl",
+            "security.sasl.enabled.mechanisms: plain",
+            "security.sasl.plain.jaas.config: " + sasl_jaas,
+            "tablet-server.id: 0",
+            "netty.server.num-network-threads: 1",
+            "netty.server.num-worker-threads: 3",
+        });
+
+        std::string ts_cmd = DockerRunCmd(kTabletServerName, ts_props,
+                                          {std::to_string(kTabletServerPort) + ":9123",
+                                           std::to_string(kPlainClientTabletPort) + ":9223"},
+                                          "tabletServer");
         if (RunCommand(ts_cmd) != 0) {
             std::cerr << "Failed to start Tablet Server" << std::endl;
             Stop();
diff --git a/bindings/python/src/config.rs b/bindings/python/src/config.rs
@@ -95,21 +95,21 @@ impl Config {
                             }
                         };
                     }
-                    "client.connect-timeout" => {
+                    "connect-timeout" => {
                         config.connect_timeout_ms = value.parse::<u64>().map_err(|e| {
                             FlussError::new_err(format!("Invalid value '{value}' for '{key}': {e}"))
                         })?;
                     }
-                    "client.security.protocol" => {
+                    "security.protocol" => {
                         config.security_protocol = value;
                     }
-                    "client.security.sasl.mechanism" => {
+                    "security.sasl.mechanism" => {
                         config.security_sasl_mechanism = value;
                     }
-                    "client.security.sasl.username" => {
+                    "security.sasl.username" => {
                         config.security_sasl_username = value;
                     }
-                    "client.security.sasl.password" => {
+                    "security.sasl.password" => {
                         config.security_sasl_password = value;
                     }
                     _ => {
diff --git a/bindings/python/test/conftest.py b/bindings/python/test/conftest.py
@@ -159,6 +159,13 @@ def sasl_bootstrap_servers(fluss_cluster):
     return sasl_addr
 
 
+@pytest.fixture(scope="session")
+def plaintext_bootstrap_servers(fluss_cluster):
+    """Bootstrap servers for the plaintext (non-SASL) listener."""
+    plaintext_addr, _sasl_addr = fluss_cluster
+    return plaintext_addr
+
+
 @pytest_asyncio.fixture(scope="session")
 async def admin(connection):
     """Session-scoped admin client."""
diff --git a/bindings/python/test/test_sasl_auth.py b/bindings/python/test/test_sasl_auth.py
@@ -29,10 +29,10 @@ async def test_sasl_connect_with_valid_credentials(sasl_bootstrap_servers):
     """Verify that a client with correct SASL credentials can connect and perform operations."""
     config = fluss.Config({
         "bootstrap.servers": sasl_bootstrap_servers,
-        "client.security.protocol": "sasl",
-        "client.security.sasl.mechanism": "PLAIN",
-        "client.security.sasl.username": "admin",
-        "client.security.sasl.password": "admin-secret",
+        "security.protocol": "sasl",
+        "security.sasl.mechanism": "PLAIN",
+        "security.sasl.username": "admin",
+        "security.sasl.password": "admin-secret",
     })
     conn = await fluss.FlussConnection.create(config)
     admin = await conn.get_admin()
@@ -48,14 +48,31 @@ async def test_sasl_connect_with_valid_credentials(sasl_bootstrap_servers):
     conn.close()
 
 
+async def test_sasl_connect_with_second_user(sasl_bootstrap_servers):
+    """Verify that a second user can also authenticate successfully."""
+    config = fluss.Config({
+        "bootstrap.servers": sasl_bootstrap_servers,
+        "security.protocol": "sasl",
+        "security.sasl.mechanism": "PLAIN",
+        "security.sasl.username": "alice",
+        "security.sasl.password": "alice-secret",
+    })
+    conn = await fluss.FlussConnection.create(config)
+    admin = await conn.get_admin()
+
+    # Basic operation to confirm functional connection
+    assert not await admin.database_exists("some_nonexistent_db_alice")
+    conn.close()
+
+
 async def test_sasl_connect_with_wrong_password(sasl_bootstrap_servers):
     """Verify that wrong credentials are rejected with AUTHENTICATE_EXCEPTION."""
     config = fluss.Config({
         "bootstrap.servers": sasl_bootstrap_servers,
-        "client.security.protocol": "sasl",
-        "client.security.sasl.mechanism": "PLAIN",
-        "client.security.sasl.username": "admin",
-        "client.security.sasl.password": "wrong-password",
+        "security.protocol": "sasl",
+        "security.sasl.mechanism": "PLAIN",
+        "security.sasl.username": "admin",
+        "security.sasl.password": "wrong-password",
     })
     with pytest.raises(fluss.FlussError) as exc_info:
         await fluss.FlussConnection.create(config)
@@ -67,12 +84,25 @@ async def test_sasl_connect_with_unknown_user(sasl_bootstrap_servers):
     """Verify that a nonexistent user is rejected with AUTHENTICATE_EXCEPTION."""
     config = fluss.Config({
         "bootstrap.servers": sasl_bootstrap_servers,
-        "client.security.protocol": "sasl",
-        "client.security.sasl.mechanism": "PLAIN",
-        "client.security.sasl.username": "nonexistent_user",
-        "client.security.sasl.password": "some-password",
+        "security.protocol": "sasl",
+        "security.sasl.mechanism": "PLAIN",
+        "security.sasl.username": "nonexistent_user",
+        "security.sasl.password": "some-password",
     })
     with pytest.raises(fluss.FlussError) as exc_info:
         await fluss.FlussConnection.create(config)
 
     assert exc_info.value.error_code == fluss.ErrorCode.AUTHENTICATE_EXCEPTION
+
+
+async def test_sasl_client_to_plaintext_server(plaintext_bootstrap_servers):
+    """Verify that a SASL-configured client fails when connecting to a plaintext server."""
+    config = fluss.Config({
+        "bootstrap.servers": plaintext_bootstrap_servers,
+        "security.protocol": "sasl",
+        "security.sasl.mechanism": "PLAIN",
+        "security.sasl.username": "admin",
+        "security.sasl.password": "admin-secret",
+    })
+    with pytest.raises(fluss.FlussError):
+        await fluss.FlussConnection.create(config)
diff --git a/crates/fluss/tests/integration/fluss_cluster.rs b/crates/fluss/tests/integration/fluss_cluster.rs
@@ -378,6 +378,11 @@ impl FlussTestingCluster {
         &self.sasl_users
     }
 
+    /// Returns the plaintext (non-SASL) bootstrap servers address.
+    pub fn plaintext_bootstrap_servers(&self) -> &str {
+        &self.bootstrap_servers
+    }
+
     pub async fn get_fluss_connection(&self) -> FlussConnection {
         let config = Config {
             writer_acks: "all".to_string(),
diff --git a/crates/fluss/tests/integration/sasl_auth.rs b/crates/fluss/tests/integration/sasl_auth.rs
@@ -18,6 +18,8 @@
 #[cfg(test)]
 mod sasl_auth_test {
     use crate::integration::utils::get_shared_cluster;
+    use fluss::client::FlussConnection;
+    use fluss::config::Config;
     use fluss::error::FlussError;
     use fluss::metadata::DatabaseDescriptorBuilder;
 
@@ -102,6 +104,29 @@ mod sasl_auth_test {
         );
     }
 
+    /// Verify that a SASL-configured client fails when connecting to a plaintext server.
+    #[tokio::test]
+    async fn test_sasl_client_to_plaintext_server() {
+        let cluster = get_shared_cluster();
+        let plaintext_addr = cluster.plaintext_bootstrap_servers().to_string();
+
+        let config = Config {
+            writer_acks: "all".to_string(),
+            bootstrap_servers: plaintext_addr,
+            security_protocol: "sasl".to_string(),
+            security_sasl_mechanism: "PLAIN".to_string(),
+            security_sasl_username: SASL_USERNAME.to_string(),
+            security_sasl_password: SASL_PASSWORD.to_string(),
+            ..Default::default()
+        };
+
+        let result = FlussConnection::new(config).await;
+        assert!(
+            result.is_err(),
+            "SASL client connecting to plaintext server should fail"
+        );
+    }
+
     /// Verify that a nonexistent user is rejected with a typed error.
     #[tokio::test]
     async fn test_sasl_connect_with_unknown_user() {
diff --git a/website/docs/user-guide/python/example/configuration.md b/website/docs/user-guide/python/example/configuration.md
@@ -32,11 +32,11 @@ with await fluss.FlussConnection.create(config) as conn:
 | `scanner.remote-log.prefetch-num` | Number of remote log segments to prefetch                                             | `4`                |
 | `remote-file.download-thread-num` | Number of threads for remote log downloads                                            | `3`                |
 | `scanner.log.max-poll-records`    | Max records returned in a single poll()                                               | `500`              |
-| `client.connect-timeout`          | TCP connect timeout in milliseconds                                                   | `120000`           |
-| `client.security.protocol`        | `PLAINTEXT` (default) or `sasl` for SASL auth                                        | `PLAINTEXT`        |
-| `client.security.sasl.mechanism`  | SASL mechanism (only `PLAIN` is supported)                                            | `PLAIN`            |
-| `client.security.sasl.username`   | SASL username (required when protocol is `sasl`)                                      | (empty)            |
-| `client.security.sasl.password`   | SASL password (required when protocol is `sasl`)                                      | (empty)            |
+| `connect-timeout`                 | TCP connect timeout in milliseconds                                                   | `120000`           |
+| `security.protocol`               | `PLAINTEXT` (default) or `sasl` for SASL auth                                        | `PLAINTEXT`        |
+| `security.sasl.mechanism`         | SASL mechanism (only `PLAIN` is supported)                                            | `PLAIN`            |
+| `security.sasl.username`          | SASL username (required when protocol is `sasl`)                                      | (empty)            |
+| `security.sasl.password`          | SASL password (required when protocol is `sasl`)                                      | (empty)            |
 
 ## SASL Authentication
 
@@ -45,10 +45,10 @@ To connect to a Fluss cluster with SASL/PLAIN authentication enabled:
 ```python
 config = fluss.Config({
     "bootstrap.servers": "127.0.0.1:9123",
-    "client.security.protocol": "sasl",
-    "client.security.sasl.mechanism": "PLAIN",
-    "client.security.sasl.username": "admin",
-    "client.security.sasl.password": "admin-secret",
+    "security.protocol": "sasl",
+    "security.sasl.mechanism": "PLAIN",
+    "security.sasl.username": "admin",
+    "security.sasl.password": "admin-secret",
 })
 conn = await fluss.FlussConnection.create(config)
 ```

Original file line number	Diff line number	Diff line change
`@@ -95,21 +95,21 @@ impl Config {`
`95`	`95`	`}`
`96`	`96`	`};`
`97`	`97`	`}`
`98`		`- "client.connect-timeout" => {`
	`98`	`+ "connect-timeout" => {`
`99`	`99`	`config.connect_timeout_ms = value.parse::<u64>().map_err(\|e\| {`
`100`	`100`	`FlussError::new_err(format!("Invalid value '{value}' for '{key}': {e}"))`
`101`	`101`	`})?;`
`102`	`102`	`}`
`103`		`- "client.security.protocol" => {`
	`103`	`+ "security.protocol" => {`
`104`	`104`	`config.security_protocol = value;`
`105`	`105`	`}`
`106`		`- "client.security.sasl.mechanism" => {`
	`106`	`+ "security.sasl.mechanism" => {`
`107`	`107`	`config.security_sasl_mechanism = value;`
`108`	`108`	`}`
`109`		`- "client.security.sasl.username" => {`
	`109`	`+ "security.sasl.username" => {`
`110`	`110`	`config.security_sasl_username = value;`
`111`	`111`	`}`
`112`		`- "client.security.sasl.password" => {`
	`112`	`+ "security.sasl.password" => {`
`113`	`113`	`config.security_sasl_password = value;`
`114`	`114`	`}`
`115`	`115`	`_ => {`