Fix lookup and get schema deadlock.

loserwang1024 · loserwang1024 · commit d490cb100767 · 2025-11-30T14:14:07.000+08:00
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixKeyLookuper.java b/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;
     }
 }
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java b/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;
     }
 }
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java
@@ -56,7 +56,6 @@
 import org.apache.fluss.types.StringType;
 
 import org.apache.commons.lang3.StringUtils;
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -93,12 +92,6 @@
 /** IT case for {@link FlussTable}. */
 class FlussTableITCase extends ClientToServerITCaseBase {
 
-    @AfterEach
-    protected void teardown() throws Exception {
-        admin.dropDatabase(DATA1_TABLE_PATH_PK.getDatabaseName(), true, true).get();
-        super.teardown();
-    }
-
     @Test
     void testGetDescriptor() throws Exception {
         createTable(DATA1_TABLE_PATH_PK, DATA1_TABLE_DESCRIPTOR_PK, false);
diff --git a/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java b/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java
@@ -67,7 +67,6 @@ void beforeEach() {
         testingSchemaGetter.updateLatestSchemaInfo(new SchemaInfo(TestData.DATA2_SCHEMA, 2));
     }
 
-    // TODO: add tests for nested types
     @Test
     void testSetCurrentProjection() throws Exception {
         short schemaId = (short) 2;
@@ -283,6 +282,7 @@ void testIllegalByteOrder(byte recordBatchMagic) throws Exception {
                                         Integer.MAX_VALUE))
                 .isInstanceOf(EOFException.class)
                 .hasMessageContaining("Failed to read `arrow header` from file channel");
+        fileLogRecords2.close();
     }
 
     @ParameterizedTest
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunctionTest.java b/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,

Original file line number	Diff line number	Diff line change
`@@ -67,7 +67,6 @@ void beforeEach() {`
`67`	`67`	`testingSchemaGetter.updateLatestSchemaInfo(new SchemaInfo(TestData.DATA2_SCHEMA, 2));`
`68`	`68`	`}`
`69`	`69`
`70`		`- // TODO: add tests for nested types`
`71`	`70`	`@Test`
`72`	`71`	`void testSetCurrentProjection() throws Exception {`
`73`	`72`	`short schemaId = (short) 2;`
`@@ -283,6 +282,7 @@ void testIllegalByteOrder(byte recordBatchMagic) throws Exception {`
`283`	`282`	`Integer.MAX_VALUE))`
`284`	`283`	`.isInstanceOf(EOFException.class)`
`285`	`284`	.hasMessageContaining("Failed to read `arrow header` from file channel");
	`285`	`+ fileLogRecords2.close();`
`286`	`286`	`}`
`287`	`287`
`288`	`288`	`@ParameterizedTest`