address comment for the lookuper

Author: polyzos

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

@@ -74,6 +74,6 @@ public Lookuper createLookuper() {
 
     @Override
     public <T> TypedLookuper<T> createTypedLookuper(Class<T> pojoClass) {
-        return new TypedLookuperImpl<>(createLookuper(), tableInfo, lookupColumnNames);
+        return new TypedLookuperImpl<>(createLookuper(), tableInfo, lookupColumnNames, pojoClass);
     }
 }

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

@@ -38,12 +38,17 @@ final class TypedLookuperImpl<K> implements TypedLookuper<K> {
     private final Lookuper delegate;
     private final TableInfo tableInfo;
     @Nullable private final List<String> lookupColumnNames;
+    private final PojoToRowConverter<K> keyConv;
 
     TypedLookuperImpl(
-            Lookuper delegate, TableInfo tableInfo, @Nullable List<String> lookupColumnNames) {
+            Lookuper delegate,
+            TableInfo tableInfo,
+            @Nullable List<String> lookupColumnNames,
+            Class<K> keyClass) {
         this.delegate = delegate;
         this.tableInfo = tableInfo;
         this.lookupColumnNames = lookupColumnNames;
+        this.keyConv = createPojoToRowConverter(keyClass);
     }
 
     @Override
@@ -55,17 +60,19 @@ public CompletableFuture<LookupResult> lookup(K key) {
         if (key instanceof InternalRow) {
             return delegate.lookup((InternalRow) key);
         }
+
+        InternalRow keyRow = keyConv.toRow(key);
+        return delegate.lookup(keyRow);
+    }
+
+    private PojoToRowConverter<K> createPojoToRowConverter(Class<K> keyClass) {
         RowType tableSchema = tableInfo.getRowType();
         RowType keyProjection;
         if (lookupColumnNames == null) {
             keyProjection = tableSchema.project(tableInfo.getPrimaryKeys());
         } else {
             keyProjection = tableSchema.project(lookupColumnNames);
         }
-        @SuppressWarnings("unchecked")
-        Class<K> keyClass = (Class<K>) key.getClass();
-        PojoToRowConverter<K> keyConv = PojoToRowConverter.of(keyClass, tableSchema, keyProjection);
-        InternalRow keyRow = keyConv.toRow(key);
-        return delegate.lookup(keyRow);
+        return PojoToRowConverter.of(keyClass, tableSchema, keyProjection);
     }
 }

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

@@ -26,6 +26,7 @@
 import org.apache.fluss.client.table.scanner.ScanRecord;
 import org.apache.fluss.client.table.scanner.log.LogScanner;
 import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
 import org.apache.fluss.client.table.writer.UpsertWriter;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
@@ -144,6 +145,13 @@ protected static void subscribeFromBeginning(LogScanner logScanner, Table table)
         }
     }
 
+    protected static void subscribeFromBeginning(TypedLogScanner<?> logScanner, Table table) {
+        int bucketCount = table.getTableInfo().getNumBuckets();
+        for (int i = 0; i < bucketCount; i++) {
+            logScanner.subscribeFromBeginning(i);
+        }
+    }
+
     protected static void subscribeFromTimestamp(
             TablePath tablePath,
             @Nullable String partitionName,