[client] Lookup sender need retry to send when leader not found in metadata cache

Author: swuferhong

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

@@ -148,11 +148,13 @@ private Map<Tuple2<Integer, LookupType>, List<AbstractLookupQuery<?>>> groupByLe
             // lookup the leader node
             TableBucket tb = lookup.tableBucket();
             try {
-                // TODO this can be a re-triable operation. We should retry here instead of
-                // throwing exception.
                 leader = metadataUpdater.leaderFor(tb);
             } catch (Exception e) {
-                lookup.future().completeExceptionally(e);
+                if (lookup.retries() < maxRetries && !lookup.future().isDone()) {
+                    reEnqueueLookup(lookup);
+                } else {
+                    lookup.future().completeExceptionally(e);
+                }
                 continue;
             }
             lookupBatchesByLeader
@@ -167,9 +169,19 @@ void sendLookups(
             int destination, LookupType lookupType, List<AbstractLookupQuery<?>> lookupBatches) {
         TabletServerGateway gateway = metadataUpdater.newTabletServerClientForNode(destination);
         if (gateway == null) {
-            // TODO handle this exception, like retry.
-            throw new LeaderNotAvailableException(
-                    "Server " + destination + " is not found in metadata cache.");
+            lookupBatches.forEach(
+                    lookup -> {
+                        if (lookup.retries() < maxRetries && !lookup.future().isDone()) {
+                            reEnqueueLookup(lookup);
+                        } else {
+                            lookup.future()
+                                    .completeExceptionally(
+                                            new LeaderNotAvailableException(
+                                                    "Server "
+                                                            + destination
+                                                            + " is not found in metadata cache."));
+                        }
+                    });
         }
 
         if (lookupType == LookupType.LOOKUP) {