Fix lookup and get schema deadlock.

Author: loserwang1024

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

@@ -169,28 +169,35 @@ public CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {
         }
 
         TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), partitionId, bucketId);
-        return lookupClient
-                .prefixLookup(tableBucket, bucketKeyBytes)
-                .thenApply(
-                        result -> {
-                            List<InternalRow> rowList = new ArrayList<>(result.size());
-                            for (byte[] valueBytes : result) {
-                                if (valueBytes == null) {
-                                    continue;
-                                }
-                                ValueDecoder.Value value = kvValueDecoder.decodeValue(valueBytes);
-                                InternalRow row;
-                                if (value.schemaId == tableInfo.getSchemaId()) {
-                                    row = value.row;
-                                } else {
-                                    Schema schema = schemaGetter.getSchema(value.schemaId);
-                                    row =
-                                            ProjectedRow.from(schema, tableInfo.getSchema())
-                                                    .replaceRow(value.row);
-                                }
-                                rowList.add(row);
+        CompletableFuture<LookupResult> future = new CompletableFuture<>();
+
+        CompletableFuture.runAsync(
+                () -> {
+                    try {
+                        List<byte[]> result =
+                                lookupClient.prefixLookup(tableBucket, bucketKeyBytes).get();
+                        List<InternalRow> rowList = new ArrayList<>(result.size());
+                        for (byte[] valueBytes : result) {
+                            if (valueBytes == null) {
+                                continue;
                             }
-                            return new LookupResult(rowList);
-                        });
+                            ValueDecoder.Value value = kvValueDecoder.decodeValue(valueBytes);
+                            InternalRow row;
+                            if (value.schemaId == tableInfo.getSchemaId()) {
+                                row = value.row;
+                            } else {
+                                Schema schema = schemaGetter.getSchema(value.schemaId);
+                                row =
+                                        ProjectedRow.from(schema, tableInfo.getSchema())
+                                                .replaceRow(value.row);
+                            }
+                            rowList.add(row);
+                        }
+                        future.complete(new LookupResult(rowList));
+                    } catch (Exception e) {
+                        future.complete(new LookupResult(Collections.emptyList()));
+                    }
+                });
+        return future;
     }
 }

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

@@ -132,25 +132,30 @@ public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {
 
         int bucketId = bucketingFunction.bucketing(bkBytes, numBuckets);
         TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), partitionId, bucketId);
-         lookupClient
-                .lookup(tableBucket, pkBytes)
-                .thenApply(
-                        valueBytes -> {
-                            InternalRow row = null;
-                            if (valueBytes != null) {
-                                ValueDecoder.Value value = kvValueDecoder.decodeValue(valueBytes);
-                                if (value.schemaId == tableInfo.getSchemaId()) {
-                                    row = value.row;
-                                } else {
-                                    Schema schema = schemaGetter.getSchema(value.schemaId);
-                                    checkNotNull(schema, "schema is null");
-                                    row =
-                                            ProjectedRow.from(schema, tableInfo.getSchema())
-                                                    .replaceRow(value.row);
-                                }
+        CompletableFuture<LookupResult> future = new CompletableFuture<>();
+
+        CompletableFuture.runAsync(
+                () -> {
+                    try {
+                        byte[] valueBytes = lookupClient.lookup(tableBucket, pkBytes).get();
+                        InternalRow row = null;
+                        if (valueBytes != null) {
+                            ValueDecoder.Value value = kvValueDecoder.decodeValue(valueBytes);
+                            if (value.schemaId == tableInfo.getSchemaId()) {
+                                row = value.row;
+                            } else {
+                                Schema schema = schemaGetter.getSchema(value.schemaId);
+                                checkNotNull(schema, "schema is null");
+                                row =
+                                        ProjectedRow.from(schema, tableInfo.getSchema())
+                                                .replaceRow(value.row);
                             }
-
-                            return new LookupResult(row);
-                        });
+                        }
+                        future.complete(new LookupResult(row));
+                    } catch (Exception e) {
+                        future.completeExceptionally(e);
+                    }
+                });
+        return future;
     }
 }

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunctionTest.java

@@ -195,7 +195,6 @@ void testSchemaChange() throws Exception {
         lookupFunction.close();
 
         // start lookup job after schema change.
-        admin.getTableSchema(tablePath, 1).get();
         lookupFunction =
                 new FlinkLookupFunction(
                         clientConf,