[client][rpc][server] Add getClusterVersion API to expose the server version

fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java

@@ -799,6 +799,25 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
      */
     CompletableFuture<ClusterHealth> getClusterHealth();
 
+    /**
+     * Get the human-readable version of the cluster asynchronously: the server's Maven project
+     * version, e.g. {@code "0.10.0"} for a release build or {@code "1.0-SNAPSHOT"} for a snapshot
+     * build.
+     *
+     * <p>Like {@link #getClusterHealth()}, this is answered by the Coordinator, so the result does
+     * not depend on which server the client happens to be connected to. During a rolling upgrade,
+     * this reflects the Coordinator's own version, not necessarily every Tablet Server's.
+     *
+     * <p>Returns {@code "unknown"} if the Coordinator could not determine its own version.
+     *
+     * <p>Servers that do not yet implement this RPC complete the returned future exceptionally with
+     * {@link org.apache.fluss.exception.UnsupportedVersionException}.
+     *
+     * @return a {@link CompletableFuture} that completes with the cluster version string.
+     * @since 1.0
+     */
+    CompletableFuture<String> getClusterVersion();
+
     /**
      * List per-bucket remote log manifest entries for a table or partition scope.
      *

fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java

@@ -71,6 +71,8 @@
 import org.apache.fluss.rpc.messages.DropDatabaseRequest;
 import org.apache.fluss.rpc.messages.DropTableRequest;
 import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
+import org.apache.fluss.rpc.messages.GetClusterVersionRequest;
+import org.apache.fluss.rpc.messages.GetClusterVersionResponse;
 import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
 import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataRequest;
 import org.apache.fluss.rpc.messages.GetLakeSnapshotRequest;
@@ -942,6 +944,12 @@ public CompletableFuture<ClusterHealth> getClusterHealth() {
                 .thenApply(ClientRpcMessageUtils::toClusterHealth);
     }
 
+    @Override
+    public CompletableFuture<String> getClusterVersion() {
+        return gateway.getClusterVersion(new GetClusterVersionRequest())
+                .thenApply(GetClusterVersionResponse::getVersion);
+    }
+
     @VisibleForTesting
     public AdminGateway getAdminGateway() {
         return gateway;

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -89,6 +89,7 @@
 import org.apache.fluss.server.zk.data.ServerTags;
 import org.apache.fluss.types.DataTypeChecks;
 import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.VersionInfo;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -2872,4 +2873,13 @@ void testClusterHealthDuringRollingUpgrade() throws Exception {
         assertThat(afterRecovery.getNumLeaderReplicas())
                 .isEqualTo(afterRecovery.getActiveLeaderReplicas());
     }
+
+    @Test
+    void testGetClusterVersion() throws Exception {
+        String version = admin.getClusterVersion().get();
+        assertThat(version)
+                .isEqualTo(VersionInfo.getVersion())
+                .isNotEqualTo("unknown")
+                .matches("^\\d+\\.\\d+.*");
+    }
 }

fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java

@@ -73,6 +73,7 @@
 import org.apache.fluss.server.zk.data.TableRegistration;
 import org.apache.fluss.shaded.guava32.com.google.common.collect.Lists;
 import org.apache.fluss.utils.CloseableIterator;
+import org.apache.fluss.utils.VersionInfo;
 
 import org.assertj.core.api.ThrowableAssert;
 import org.junit.jupiter.api.AfterEach;
@@ -852,6 +853,32 @@ void testDynamicConfigs() throws ExecutionException, InterruptedException {
                                 ConfigEntry.ConfigSource.INITIAL_SERVER_CONFIG));
     }
 
+    @Test
+    void testGetClusterVersion() throws ExecutionException, InterruptedException {
+        assertThatThrownBy(() -> guestAdmin.getClusterVersion().get())
+                .rootCause()
+                .hasMessageContaining(
+                        String.format(
+                                "Principal %s have no authorization to operate DESCRIBE on resource Resource{type=CLUSTER, name='fluss-cluster'}",
+                                guestPrincipal));
+        rootAdmin
+                .createAcls(
+                        Collections.singletonList(
+                                new AclBinding(
+                                        Resource.cluster(),
+                                        new AccessControlEntry(
+                                                guestPrincipal,
+                                                "*",
+                                                OperationType.DESCRIBE,
+                                                PermissionType.ALLOW))))
+                .all()
+                .get();
+        assertThat(guestAdmin.getClusterVersion().get())
+                .isEqualTo(VersionInfo.getVersion())
+                .isNotEqualTo("unknown")
+                .matches("^\\d+\\.\\d+.*");
+    }
+
     @Test
     void testControlledShutdown() throws Exception {
         ControlledShutdownRequest request =

fluss-common/pom.xml

@@ -166,6 +166,25 @@
     </dependencies>
 
     <build>
+        <resources>
+            <!-- Filter only fluss-version.properties, so VersionInfo can read the resolved
+                 project version; every other resource here (e.g. META-INF/services) stays
+                 unfiltered. -->
+            <resource>
+                <directory>src/main/resources</directory>
+                <filtering>true</filtering>
+                <includes>
+                    <include>org/apache/fluss/utils/fluss-version.properties</include>
+                </includes>
+            </resource>
+            <resource>
+                <directory>src/main/resources</directory>
+                <filtering>false</filtering>
+                <excludes>
+                    <exclude>org/apache/fluss/utils/fluss-version.properties</exclude>
+                </excludes>
+            </resource>
+        </resources>
         <plugins>
             <plugin>
                 <artifactId>maven-assembly-plugin</artifactId>

fluss-common/src/main/java/org/apache/fluss/utils/VersionInfo.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.utils;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.annotation.VisibleForTesting;
+
+import javax.annotation.Nullable;
+
+import java.io.InputStream;
+import java.util.Properties;
+
+/** Utility for looking up the human-readable Fluss version of the running build. */
+@Internal
+public class VersionInfo {
+
+    // Resolved relative to this class so that a downstream shade relocating org.apache.fluss moves
+    // the resource along with it.
+    private static final String VERSION_RESOURCE = "fluss-version.properties";
+    private static final String VERSION_KEY = "version";
+    private static final String UNKNOWN_VERSION = "unknown";
+
+    private static final String VERSION = readVersion();
+
+    private VersionInfo() {}
+
+    /**
+     * Returns the Fluss version (e.g. {@code "0.10.0"} for a release, {@code "1.0-SNAPSHOT"} for a
+     * snapshot build), read once at class initialization from the build-time-filtered {@code
+     * fluss-version.properties} resource on the classpath.
+     *
+     * <p>Returns {@code "unknown"} when the resource is missing, unreadable, has no {@code version}
+     * key, or was copied without Maven resource filtering and so still carries the literal {@code
+     * ${project.version}} token.
+     */
+    public static String getVersion() {
+        return VERSION;
+    }
+
+    private static String readVersion() {
+        try (InputStream stream = VersionInfo.class.getResourceAsStream(VERSION_RESOURCE)) {
+            return parseVersion(stream);
+        } catch (Exception e) {
+            return UNKNOWN_VERSION;
+        }
+    }
+
+    @VisibleForTesting
+    static String parseVersion(@Nullable InputStream stream) {
+        if (stream == null) {
+            return UNKNOWN_VERSION;
+        }
+        try {
+            Properties properties = new Properties();
+            properties.load(stream);
+            String version = properties.getProperty(VERSION_KEY, UNKNOWN_VERSION);
+            // An unfiltered copy of the resource still carries the raw Maven token.
+            return version.startsWith("${") ? UNKNOWN_VERSION : version;
+        } catch (Exception e) {
+            return UNKNOWN_VERSION;
+        }
+    }
+}

fluss-common/src/main/resources/org/apache/fluss/utils/fluss-version.properties

@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+version=${project.version}

fluss-common/src/test/java/org/apache/fluss/utils/VersionInfoTest.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the {@link org.apache.fluss.utils.VersionInfo}. */
+class VersionInfoTest {
+
+    @Test
+    void testGetVersionReadsTheFilteredProjectVersion() {
+        // fluss-common's pom.xml filters fluss-version.properties with ${project.version}, and
+        // that filtering already ran by the time surefire loads classes from target/classes, so
+        // this asserts the real project version rather than the "unknown" fallback.
+        assertThat(VersionInfo.getVersion()).isNotEqualTo("unknown").matches("^\\d+\\.\\d+.*");
+    }
+
+    @Test
+    void testGetVersionIsReadOnceAndCached() {
+        assertThat(VersionInfo.getVersion()).isSameAs(VersionInfo.getVersion());
+    }
+
+    @Test
+    void testParseVersionReadsTheVersionKey() {
+        assertThat(parse("version=0.10.0")).isEqualTo("0.10.0");
+    }
+
+    @Test
+    void testParseVersionFallsBackToUnknown() {
+        assertThat(VersionInfo.parseVersion(null)).isEqualTo("unknown");
+        // A resource copied without Maven filtering still carries the raw token.
+        assertThat(parse("version=${project.version}")).isEqualTo("unknown");
+        assertThat(parse("other=0.10.0")).isEqualTo("unknown");
+        // Properties.load rejects a malformed unicode escape with IllegalArgumentException.
+        assertThat(parse("version=\\uZZZZ")).isEqualTo("unknown");
+    }
+
+    private static String parse(String content) {
+        return VersionInfo.parseVersion(
+                new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8)));
+    }
+}

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java

@@ -326,6 +326,11 @@ public CompletableFuture<ClusterHealth> getClusterHealth() {
         throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
     }
 
+    @Override
+    public CompletableFuture<String> getClusterVersion() {
+        throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
+    }
+
     @Override
     public CompletableFuture<List<RemoteLogManifestInfo>> listRemoteLogManifests(
             long tableId, @Nullable Long partitionId) {

fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java

@@ -24,6 +24,8 @@
 import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
 import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
 import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
+import org.apache.fluss.rpc.messages.GetClusterVersionRequest;
+import org.apache.fluss.rpc.messages.GetClusterVersionResponse;
 import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
 import org.apache.fluss.rpc.messages.GetDatabaseInfoResponse;
 import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenRequest;
@@ -202,4 +204,13 @@ CompletableFuture<DescribeClusterConfigsResponse> describeClusterConfigs(
      */
     @RPC(api = ApiKeys.GET_CLUSTER_HEALTH)
     CompletableFuture<GetClusterHealthResponse> getClusterHealth(GetClusterHealthRequest request);
+
+    /**
+     * Get the human-readable version of the cluster.
+     *
+     * @return the cluster version response.
+     */
+    @RPC(api = ApiKeys.GET_CLUSTER_VERSION)
+    CompletableFuture<GetClusterVersionResponse> getClusterVersion(
+            GetClusterVersionRequest request);
 }

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/ServerConnection.java

@@ -44,6 +44,7 @@
 import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelFuture;
 import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelFutureListener;
 import org.apache.fluss.utils.ExponentialBackoff;
+import org.apache.fluss.utils.VersionInfo;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -263,11 +264,11 @@ private void establishConnection(ChannelFuture future) {
                         .addLast("handler", new NettyClientHandler(new ResponseCallback()));
                 // start checking api versions
                 switchState(ConnectionState.CHECKING_API_VERSIONS);
-                // TODO: set correct client software name and version, used for metrics in server
+                // TODO: client_software_name is hardcoded; no server-side metrics consumer yet.
                 ApiVersionsRequest request =
                         new ApiVersionsRequest()
                                 .setClientSoftwareName("fluss")
-                                .setClientSoftwareVersion("0.1.0");
+                                .setClientSoftwareVersion(VersionInfo.getVersion());
                 doSend(ApiKeys.API_VERSIONS, request, new CompletableFuture<>(), true)
                         .whenComplete(this::handleApiVersionsResponse);
             } else {

fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java

@@ -109,7 +109,8 @@ public enum ApiKeys {
     SCAN_KV(1061, 0, 0, PUBLIC),
     GET_CLUSTER_HEALTH(1062, 0, 0, PUBLIC),
     LIST_REMOTE_LOG_MANIFESTS(1063, 0, 0, PUBLIC),
-    LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC);
+    LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC),
+    GET_CLUSTER_VERSION(1065, 0, 0, PUBLIC);
 
     private static final Map<Integer, ApiKeys> ID_TO_TYPE =
             Arrays.stream(ApiKeys.values())

fluss-rpc/src/main/proto/FlussApi.proto

@@ -810,6 +810,12 @@ message GetClusterHealthResponse {
   required int32 status = 5; // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3
 }
 
+message GetClusterVersionRequest { }
+
+message GetClusterVersionResponse {
+  required string version = 1;
+}
+
 
 // --------------- Inner classes ----------------
 message PbApiVersion {

fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java

@@ -27,6 +27,8 @@
 import org.apache.fluss.rpc.messages.FetchLogResponse;
 import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
 import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
+import org.apache.fluss.rpc.messages.GetClusterVersionRequest;
+import org.apache.fluss.rpc.messages.GetClusterVersionResponse;
 import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
 import org.apache.fluss.rpc.messages.GetDatabaseInfoResponse;
 import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenRequest;
@@ -272,4 +274,10 @@ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
             GetClusterHealthRequest request) {
         return null;
     }
+
+    @Override
+    public CompletableFuture<GetClusterVersionResponse> getClusterVersion(
+            GetClusterVersionRequest request) {
+        return null;
+    }
 }