Merge branch 'main' into lakesense

Author: xx789633

.github/ISSUE_TEMPLATE/bug.yml

@@ -39,9 +39,8 @@ body:
       description: What Fluss version are you using?
       multiple: false
       options:
-        - "0.7.0 (latest release)"
-        - "0.6.0"
-        - "0.5.0"
+        - "0.8.0 (latest release)"
+        - "0.7.0"
         - "main (development)"
     validations:
       required: true

.github/workflows/docs-check.yaml

@@ -44,7 +44,7 @@ jobs:
         run: ./build_versioned_docs.sh
       - uses: actions/setup-node@v4
         with:
-          node-version: 18
+          node-version: 20
       - name: Install dependencies
         run: npm install
       - name: Test build website

README.md

@@ -61,15 +61,12 @@ Prerequisites for building Apache Fluss:
 - Unix-like environment (we use Linux, Mac OS X, Cygwin, WSL)
 - Git
 - Maven (we require version >= 3.8.6)
-- Java 8 or 11
+- Java 11
 
 ```bash
 git clone https://github.com/apache/fluss.git
 cd fluss
-# in case of java 11
 ./mvnw clean package -DskipTests
-# or in case of java 8
-./mvnw clean package -DskipTests -Pjava8
 ```
 
 Apache Fluss is now installed in `build-target`. The build command uses Maven Wrapper (`mvnw`) which ensures the correct Maven version is used.

docker/quickstart-flink/prepare_build.sh

@@ -170,12 +170,12 @@ main() {
     # Iceberg Support
     log_info "Downloading Iceberg connector JARs..."
 
-    # Download iceberg-flink-runtime for Flink 1.20 (version 1.9.1)
+    # Download iceberg-flink-runtime for Flink 1.20 (version 1.10.0)
     download_jar \
-        "https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-flink-runtime-1.20/1.9.1/iceberg-flink-runtime-1.20-1.9.1.jar" \
-        "./lib/iceberg-flink-runtime-1.20-1.9.1.jar" \
+        "https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-flink-runtime-1.20/1.10.0/iceberg-flink-runtime-1.20-1.10.0.jar" \
+        "./lib/iceberg-flink-runtime-1.20-1.10.0.jar" \
         "" \
-        "iceberg-flink-runtime-1.20-1.9.1"
+        "iceberg-flink-runtime-1.20-1.10.0"
 
 
     # Prepare lake tiering JAR
@@ -201,7 +201,7 @@ verify_jars() {
         "flink-faker-0.5.3.jar"
         "hadoop-apache-3.3.5-2.jar"
         "paimon-flink-1.20-1.2.0.jar"
-        "iceberg-flink-runtime-1.20-1.9.1.jar"
+        "iceberg-flink-runtime-1.20-1.10.0.jar"
     )
 
     local opt_jars=(
@@ -250,7 +250,7 @@ show_summary() {
     echo "  ✓ Fluss Flink 1.20 connector"
     echo "  ✓ Fluss Lake Paimon connector"
     echo "  ✓ Fluss Lake Iceberg connector"
-    echo "  ✓ Iceberg Flink runtime 1.20 (v1.9.1)"
+    echo "  ✓ Iceberg Flink runtime 1.20 (v1.10.0)"
     echo "  ✓ Paimon Flink 1.20 (v1.2.0)"
     echo "  ✓ Hadoop Apache (v3.3.5-2)"
     echo "  ✓ Flink Faker (v0.5.3)"

fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java

@@ -29,11 +29,11 @@
 import org.apache.fluss.client.token.SecurityTokenManager;
 import org.apache.fluss.client.token.SecurityTokenProvider;
 import org.apache.fluss.client.write.WriterClient;
+import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.fs.FileSystem;
-import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.metrics.registry.MetricRegistry;
 import org.apache.fluss.rpc.GatewayClientProxy;
@@ -43,6 +43,7 @@
 
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 
 import static org.apache.fluss.client.utils.MetadataUtils.getOneAvailableTabletServerNode;
@@ -61,6 +62,7 @@ public final class FlussConnection implements Connection {
     private volatile LookupClient lookupClient;
     private volatile RemoteFileDownloader remoteFileDownloader;
     private volatile SecurityTokenManager securityTokenManager;
+    private volatile Admin admin;
 
     FlussConnection(Configuration conf) {
         this(conf, MetricRegistry.create(conf, null));
@@ -93,19 +95,15 @@ public Configuration getConfiguration() {
 
     @Override
     public Admin getAdmin() {
-        return new FlussAdmin(rpcClient, metadataUpdater);
+        return getOrCreateAdmin();
     }
 
     @Override
     public Table getTable(TablePath tablePath) {
-        // force to update the table info from server to avoid stale data in cache
+        // force to update the table info from server to avoid stale data in cache.
         metadataUpdater.updateTableOrPartitionMetadata(tablePath, null);
-        TableInfo tableInfo = metadataUpdater.getTableInfoOrElseThrow(tablePath);
-        return new FlussTable(this, tablePath, tableInfo);
-    }
-
-    public RpcClient getRpcClient() {
-        return rpcClient;
+        Admin admin = getOrCreateAdmin();
+        return new FlussTable(this, tablePath, admin.getTableInfo(tablePath).join());
     }
 
     public MetadataUpdater getMetadataUpdater() {
@@ -140,6 +138,17 @@ public LookupClient getOrCreateLookupClient() {
         return lookupClient;
     }
 
+    public Admin getOrCreateAdmin() {
+        if (admin == null) {
+            synchronized (this) {
+                if (admin == null) {
+                    admin = new FlussAdmin(rpcClient, metadataUpdater);
+                }
+            }
+        }
+        return admin;
+    }
+
     public RemoteFileDownloader getOrCreateRemoteFileDownloader() {
         if (remoteFileDownloader == null) {
             synchronized (this) {
@@ -155,9 +164,17 @@ public RemoteFileDownloader getOrCreateRemoteFileDownloader() {
                     // todo: may add retry logic when no any available tablet server?
                     AdminReadOnlyGateway gateway =
                             GatewayClientProxy.createGatewayProxy(
-                                    () ->
-                                            getOneAvailableTabletServerNode(
-                                                    metadataUpdater.getCluster()),
+                                    () -> {
+                                        ServerNode serverNode =
+                                                getOneAvailableTabletServerNode(
+                                                        metadataUpdater.getCluster(),
+                                                        new HashSet<>());
+                                        if (serverNode == null) {
+                                            throw new FlussRuntimeException(
+                                                    "no available tablet server");
+                                        }
+                                        return serverNode;
+                                    },
                                     rpcClient,
                                     AdminReadOnlyGateway.class);
                     SecurityTokenProvider securityTokenProvider =

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

@@ -88,8 +88,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
 
+import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeAlterTableRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeCreatePartitionRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeDropPartitionRequest;
 import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeListOffsetsRequest;
@@ -248,18 +248,8 @@ public CompletableFuture<Void> createTable(
     public CompletableFuture<Void> alterTable(
             TablePath tablePath, List<TableChange> tableChanges, boolean ignoreIfNotExists) {
         tablePath.validate();
-        AlterTableRequest request = new AlterTableRequest();
-
-        List<PbAlterConfig> pbFlussTableChanges =
-                tableChanges.stream()
-                        .map(ClientRpcMessageUtils::toPbAlterConfigs)
-                        .collect(Collectors.toList());
-
-        request.addAllConfigChanges(pbFlussTableChanges)
-                .setIgnoreIfNotExists(ignoreIfNotExists)
-                .setTablePath()
-                .setDatabaseName(tablePath.getDatabaseName())
-                .setTableName(tablePath.getTableName());
+        AlterTableRequest request =
+                makeAlterTableRequest(tablePath, tableChanges, ignoreIfNotExists);
         return gateway.alterTable(request).thenApply(r -> null);
     }
 
@@ -420,15 +410,21 @@ private ListOffsetsResult listOffsets(
             OffsetSpec offsetSpec) {
         Long partitionId = null;
         metadataUpdater.updateTableOrPartitionMetadata(physicalTablePath.getTablePath(), null);
-        long tableId = metadataUpdater.getTableId(physicalTablePath.getTablePath());
+        TableInfo tableInfo = getTableInfo(physicalTablePath.getTablePath()).join();
+
         // if partition name is not null, we need to check and update partition metadata
         if (physicalTablePath.getPartitionName() != null) {
             metadataUpdater.updatePhysicalTableMetadata(Collections.singleton(physicalTablePath));
             partitionId = metadataUpdater.getPartitionIdOrElseThrow(physicalTablePath);
         }
         Map<Integer, ListOffsetsRequest> requestMap =
                 prepareListOffsetsRequests(
-                        metadataUpdater, tableId, partitionId, buckets, offsetSpec);
+                        metadataUpdater,
+                        tableInfo.getTableId(),
+                        partitionId,
+                        buckets,
+                        offsetSpec,
+                        tableInfo.getTablePath());
         Map<Integer, CompletableFuture<Long>> bucketToOffsetMap = MapUtils.newConcurrentHashMap();
         for (int bucket : buckets) {
             bucketToOffsetMap.put(bucket, new CompletableFuture<>());
@@ -545,10 +541,13 @@ private static Map<Integer, ListOffsetsRequest> prepareListOffsetsRequests(
             long tableId,
             @Nullable Long partitionId,
             Collection<Integer> buckets,
-            OffsetSpec offsetSpec) {
+            OffsetSpec offsetSpec,
+            TablePath tablePath) {
         Map<Integer, List<Integer>> nodeForBucketList = new HashMap<>();
         for (Integer bucketId : buckets) {
-            int leader = metadataUpdater.leaderFor(new TableBucket(tableId, partitionId, bucketId));
+            int leader =
+                    metadataUpdater.leaderFor(
+                            tablePath, new TableBucket(tableId, partitionId, bucketId));
             nodeForBucketList.computeIfAbsent(leader, k -> new ArrayList<>()).add(bucketId);
         }
 

fluss-client/src/main/java/org/apache/fluss/client/lookup/AbstractLookupQuery.java

@@ -19,29 +19,46 @@
 
 import org.apache.fluss.annotation.Internal;
 import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
 
 import java.util.concurrent.CompletableFuture;
 
 /** Abstract Class to represent a lookup operation. */
 @Internal
 public abstract class AbstractLookupQuery<T> {
 
+    private final TablePath tablePath;
     private final TableBucket tableBucket;
     private final byte[] key;
+    private int retries;
 
-    public AbstractLookupQuery(TableBucket tableBucket, byte[] key) {
+    public AbstractLookupQuery(TablePath tablePath, TableBucket tableBucket, byte[] key) {
+        this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.key = key;
+        this.retries = 0;
     }
 
     public byte[] key() {
         return key;
     }
 
+    public TablePath tablePath() {
+        return tablePath;
+    }
+
     public TableBucket tableBucket() {
         return tableBucket;
     }
 
+    public int retries() {
+        return retries;
+    }
+
+    public void incrementRetries() {
+        retries++;
+    }
+
     public abstract LookupType lookupType();
 
     public abstract CompletableFuture<T> future();