[docs] Add security documents including authentication and authorization. (#1060)

Author: loserwang1024

fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java

@@ -177,6 +177,53 @@ public class ConfigOptions {
                             "The maximum number of buckets that can be created for a table."
                                     + "The default value is 128000");
 
+    /**
+     * The network address and port the server binds to for accepting connections.
+     *
+     * <p>This specifies the interface and port where the server will listen for incoming requests.
+     * The format is {@code listener_name://host:port}, supporting multiple addresses separated by
+     * commas.
+     *
+     * <p>The default value {@code "CLIENT://localhost:9123"} is suitable for local development.
+     */
+    public static final ConfigOption<String> BIND_LISTENERS =
+            key("bind.listeners")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The network address and port to which the server binds for accepting connections. "
+                                    + "This defines the interface and port where the server will listen for incoming requests. "
+                                    + "The format is `listener_name://host:port`, and multiple addresses can be specified, separated by commas. "
+                                    + "Use `0.0.0.0` for the `host` to bind to all available interfaces which is dangerous on production and not suggested for production usage. "
+                                    + "The `listener_name` serves as an identifier for the address in the configuration. For example, "
+                                    + "`internal.listener.name` specifies the address used for internal server communication. "
+                                    + "If multiple addresses are configured, ensure that the `listener_name` values are unique.");
+
+    /**
+     * The externally advertised address and port for client connections.
+     *
+     * <p>This specifies the address other nodes/clients should use to connect to this server. It is
+     * required when the bind address ({@link #BIND_LISTENERS}) is not publicly reachable (e.g.,
+     * when using {@code localhost} in {@code bind.listeners}). <b>Must be configured in distributed
+     * environments</b> to ensure proper cluster discovery. If not explicitly set, the value of
+     * {@code bind.listeners} will be used as fallback.
+     */
+    public static final ConfigOption<String> ADVERTISED_LISTENERS =
+            key("advertised.listeners")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The externally advertised address and port for client connections. "
+                                    + "Required in distributed environments when the bind address is not publicly reachable. "
+                                    + "Format matches `bind.listeners` (listener_name://host:port). "
+                                    + "Defaults to the value of `bind.listeners` if not explicitly configured.");
+
+    public static final ConfigOption<String> INTERNAL_LISTENER_NAME =
+            key("internal.listener.name")
+                    .stringType()
+                    .defaultValue(DEFAULT_LISTENER_NAME)
+                    .withDescription("The listener for server internal communication.");
+
     public static final ConfigOption<List<String>> SERVER_SASL_ENABLED_MECHANISMS_CONFIG =
             key("security.sasl.enabled.mechanisms").stringType().asList().noDefaultValue();
 
@@ -237,53 +284,6 @@ public class ConfigOptions {
                                     + " (“50100,50101”), ranges (“50100-50200”) or a combination of both."
                                     + "This option is deprecated. Please use bind.listeners instead, which provides a more flexible configuration for multiple ports");
 
-    /**
-     * The network address and port the server binds to for accepting connections.
-     *
-     * <p>This specifies the interface and port where the server will listen for incoming requests.
-     * The format is {@code listener_name://host:port}, supporting multiple addresses separated by
-     * commas.
-     *
-     * <p>The default value {@code "CLIENT://localhost:9123"} is suitable for local development.
-     */
-    public static final ConfigOption<String> BIND_LISTENERS =
-            key("bind.listeners")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription(
-                            "The network address and port to which the server binds for accepting connections. "
-                                    + "This defines the interface and port where the server will listen for incoming requests. "
-                                    + "The format is `listener_name://host:port`, and multiple addresses can be specified, separated by commas. "
-                                    + "Use `0.0.0.0` for the `host` to bind to all available interfaces which is dangerous on production and not suggested for production usage. "
-                                    + "The `listener_name` serves as an identifier for the address in the configuration. For example, "
-                                    + "`internal.listener.name` specifies the address used for internal server communication. "
-                                    + "If multiple addresses are configured, ensure that the `listener_name` values are unique.");
-
-    /**
-     * The externally advertised address and port for client connections.
-     *
-     * <p>This specifies the address other nodes/clients should use to connect to this server. It is
-     * required when the bind address ({@link #BIND_LISTENERS}) is not publicly reachable (e.g.,
-     * when using {@code localhost} in {@code bind.listeners}). <b>Must be configured in distributed
-     * environments</b> to ensure proper cluster discovery. If not explicitly set, the value of
-     * {@code bind.listeners} will be used as fallback.
-     */
-    public static final ConfigOption<String> ADVERTISED_LISTENERS =
-            key("advertised.listeners")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription(
-                            "The externally advertised address and port for client connections. "
-                                    + "Required in distributed environments when the bind address is not publicly reachable. "
-                                    + "Format matches `bind.listeners` (listener_name://host:port). "
-                                    + "Defaults to the value of `bind.listeners` if not explicitly configured.");
-
-    public static final ConfigOption<String> INTERNAL_LISTENER_NAME =
-            key("internal.listener.name")
-                    .stringType()
-                    .defaultValue(DEFAULT_LISTENER_NAME)
-                    .withDescription("The listener for server internal communication.");
-
     public static final ConfigOption<Integer> COORDINATOR_IO_POOL_SIZE =
             key("coordinator.io-pool.size")
                     .intType()

website/docs/apis/_category_.json

@@ -1,4 +1,4 @@
 {
   "label": "Fluss APIs",
-  "position": 9
+  "position": 10
 }

website/docs/apis/java-client.md

@@ -52,13 +52,14 @@ single `Connection` instance per application and use it to create multiple `Admi
 `Table` and `Admin` instances, on the other hand, are not thread-safe and should be created for each thread that needs to access them.
  Caching or pooling of `Table` and `Admin` is not recommended.
 
-Create a new `Admin` instance:
+Create a new `Admin` instance :
 ```java
 // creating Connection object to connect with Fluss cluster
 Configuration conf = new Configuration(); 
 conf.setString("bootstrap.servers", "localhost:9123");
 Connection connection = ConnectionFactory.createConnection(conf);
 
+
 // obtain Admin instance from the Connection
 Admin admin = connection.getAdmin();
 admin.listDatabases().get().forEach(System.out::println);
@@ -68,6 +69,29 @@ Table table = connection.getTable(TablePath.of("my_db", "my_table");
 System.out.println(table.getTableInfo());
 ```
 
+if you are using SASL authentication, you need to set the following properties:
+```java
+// creating Connection object to connect with Fluss cluster
+Configuration conf = new Configuration(); 
+conf.setString("bootstrap.servers", "localhost:9123");
+conf.setString("client.security.protocol", "sasl");
+conf.setString("client.security.sasl.mechanism", "PLAIN");
+conf.setString("client.security.sasl.username", "alice");
+conf.setString("client.security.sasl.password", "alice-secret");
+Connection connection = ConnectionFactory.createConnection(conf);
+
+
+// obtain Admin instance from the Connection
+Admin admin = connection.getAdmin();
+admin.listDatabases().get().forEach(System.out::println);
+
+// obtain Table instance from the Connection
+Table table = connection.getTable(TablePath.of("my_db", "my_table");
+System.out.println(table.getTableInfo());
+```
+
+
+
 ## Working Operations
 All methods in `FlussAdmin` return `CompletableFuture` objects. You can handle these in two ways:
 

website/docs/engine-flink/ddl.md

@@ -37,11 +37,13 @@ USE CATALOG fluss_catalog;
 
 The following properties can be set if using the Fluss catalog:
 
-| Option            | Required | Default | Description                                                 | 
-|-------------------|----------|---------|-------------------------------------------------------------|
-| type              | required | (none)  | Catalog type, must to be 'fluss' here.                      |
-| bootstrap.servers | required | (none)  | Comma separated list of Fluss servers.                      |
-| default-database  | optional | fluss   | The default database to use when switching to this catalog. |
+| Option                         | Required | Default   | Description                                                                                                                                                                          | 
+|--------------------------------|----------|-----------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| type                           | required | (none)    | Catalog type, must to be 'fluss' here.                                                                                                                                               |
+| bootstrap.servers              | required | (none)    | Comma separated list of Fluss servers.                                                                                                                                               |
+| default-database               | optional | fluss     | The default database to use when switching to this catalog.                                                                                                                          |
+| client.security.protocol       | optional | PLAINTEXT | The security protocol used to communicate with brokers. Currently, only `PLAINTEXT` and `SASL` are supported, the configuration value is case insensitive.                           |
+| `client.security.{protocol}.*` | optional | (none)    | Client-side configuration properties for a specific authentication protocol. E.g., client.security.sasl.jaas.config. More Details in [authentication](../security/authentication.md) | (none)        |
 
 The following introduced statements assuming the current catalog is switched to the Fluss catalog using `USE CATALOG <catalog_name>` statement.
 

website/docs/engine-flink/options.md

@@ -165,4 +165,5 @@ ALTER TABLE log_table SET ('table.log.ttl' = '7d');
 | client.filesystem.security.token.renewal.backoff    | Duration | 1h        | The time period how long to wait before retrying to obtain new security tokens for filesystem after a failure.                                                                                                                                                                                                                                                                                                    |
 | client.filesystem.security.token.renewal.time-ratio | Double   | 0.75      | Ratio of the token's expiration time when new credentials for access filesystem should be re-obtained.                                                                                                                                                                                                                                                                                                            |
 | client.metrics.enabled                              | Boolean  | false     | Enable metrics for client. When metrics is enabled, the client will collect metrics and report by the JMX metrics reporter.                                                                                                                                                                                                                                                                                       |
-| client.security.protocol                            | String   | PLAINTEXT | The authentication protocol used to authenticate the client.                                                                                                                                                                                                                                                                                                                                                      |
+| client.security.protocol                            | String   | PLAINTEXT | The security protocol used to communicate with brokers. Currently, only `PLAINTEXT` and `SASL` are supported, the configuration value is case insensitive.                                                                                                                                                                                                                                                        |
+| `client.security.{protocol}.*`                      | optional | (none)    | Client-side configuration properties for a specific authentication protocol. E.g., client.security.sasl.jaas.config. More Details in [authentication](../security/authentication.md)                                                                                                                                                                                                                              |