[I] [Client] Correct lookup batching and isolate stalled buckets [fluss]

["loserwang1024 (via GitHub)" <gi...@apache.org>]

loserwang1024 opened a new issue, #4043:
URL: https://github.com/apache/fluss/issues/4043

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
   
   
   ### Fluss version
   
   0.9.0 (latest release)
   
   ### Please describe the bug 🐞
   
   The lookup sender currently has two related problems.
   ### 1. client.lookup.max-batch-size is applied at the wrong stage
   In description of `client.lookup.max-batch-size` is:
   > The maximum batch size of merging lookup operations to one lookup request
   
   However, it's wrong:
   
   Currently, client.lookup.max-batch-size limits the number of logical lookup operations drained from the shared lookup queue in one sender iteration.
   
   After draining, the lookups are further grouped by:
   * TabletServer leader
   * Table
   * Lookup type
   * Normal or historical lookup mode
   
   Therefore, `client.lookup.max-batch-size` currently limits a global drain operation rather than the size of each final RPC.
   
   This causes two inconsistent behaviors:
   
   * During normal sending, several resulting RPCs share the same drain budget and may be unnecessarily small.
   
   * During shutdown, drainAll() can drain the whole queue, potentially producing an RPC containing more than client.lookup.max-batch-size logical lookups.
   
   For example, if client.lookup.max-batch-size is 128 and the queue contains lookups for four different leaders, the sender drains only 128 lookups in total. Each leader may receive a much smaller request even though each request could independently contain up to 128 lookups.
   
   
    ### 2. client.lookup.max-batch-size is applied at the wrong stage
   
   client.lookup.max-inflight-requests currently provides only a client-wide RPC concurrency limit.
   
   During leader failover, such as a rolling upgrade, requests sent to an old or temporarily unavailable leader may remain incomplete until the connection fails or the request times out.
   
   If many lookups target the same affected bucket, their RPCs may consume most or all global in-flight permits:
   
   `client.lookup.max-inflight-requests`
   
   Once the global permits are exhausted, the lookup sender blocks before sending additional RPCs. Requests for healthy buckets and healthy TabletServers can no longer be dispatched, even though they are unrelated to the failed bucket.
   
   
   To isolate this failure, the lookup client needs an additional per-bucket limit:
   
   `client.lookup.max-inflight-requests-per-bucket`
   
   Once one bucket reaches this limit, further requests for that bucket should remain queued, while requests for other buckets remain eligible for dispatch.
   
   This is specifically per-bucket isolation. Multiple affected buckets on the same failed TabletServer can still collectively consume global permits; the option is not a per-node circuit breaker.
   
   ### Solution
   
   | stage | configuration | Required semantics |
   | -- | -- | -- |
   | Logical admission | client.lookup.queue-size | Maximum number of admitted logical lookup operations. | 
   Batch accumulation | client.lookup.batch-timeout | Maximum wait time for an underfilled compatible batch.
   RPC packing | client.lookup.max-batch-size | Maximum number of logical lookups in each final RPC.
   Global concurrency | client.lookup.max-inflight-requests | Maximum number of unacknowledged lookup RPCs across the client.
   Bucket isolation | client.lookup.max-inflight-requests-per-bucket | Maximum number of unacknowledged RPCs containing a given bucket.
   
   ### Are you willing to submit a PR?
   
   - [x] I'm willing to submit a PR!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org