fix thread-safety problem of PrimaryKeyLoookuper and PrefixKeyLookuper

platinumhamburg · platinumhamburg · commit 6ec6e33c6c9f · 2025-10-31T10:11:18.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
@@ -146,7 +146,7 @@ private void validatePrefixLookup(TableInfo tableInfo, List<String> lookupColumn
     }
 
     @Override
-    public CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {
+    public synchronized CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {
         byte[] bucketKeyBytes = bucketKeyEncoder.encodeKey(prefixKey);
         int bucketId = bucketingFunction.bucketing(bucketKeyBytes, numBuckets);
 
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
@@ -101,7 +101,7 @@ public PrimaryKeyLookuper(
     }
 
     @Override
-    public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {
+    public synchronized CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {
         // encoding the key row using a compacted way consisted with how the key is encoded when put
         // a row
         byte[] pkBytes = primaryKeyEncoder.encodeKey(lookupKey);

Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,7 @@ private void validatePrefixLookup(TableInfo tableInfo, List<String> lookupColumn`
`146`	`146`	`}`
`147`	`147`
`148`	`148`	`@Override`
`149`		`- public CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {`
	`149`	`+ public synchronized CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {`
`150`	`150`	`byte[] bucketKeyBytes = bucketKeyEncoder.encodeKey(prefixKey);`
`151`	`151`	`int bucketId = bucketingFunction.bucketing(bucketKeyBytes, numBuckets);`
`152`	`152`
Original file line number	Diff line number	Diff line change
`@@ -101,7 +101,7 @@ public PrimaryKeyLookuper(`
`101`	`101`	`}`
`102`	`102`
`103`	`103`	`@Override`
`104`		`- public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {`
	`104`	`+ public synchronized CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {`
`105`	`105`	`// encoding the key row using a compacted way consisted with how the key is encoded when put`
`106`	`106`	`// a row`
`107`	`107`	`byte[] pkBytes = primaryKeyEncoder.encodeKey(lookupKey);`