[PR] [fix][client] Recover listener delivery after executor rejection [pulsar]

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

Denovo1998 opened a new pull request, #26640:
URL: https://github.com/apache/pulsar/pull/26640

   ### Motivation
   
   When a `MessageListenerExecutor` rejects a task, the message has already been removed from the receiver queue but never reaches the listener callback. Its permit is not returned, and repeated rejections can exhaust the receive window, leaving consumption stalled even after the executor recovers. Pooled messages can also remain unreleased.
   
   This follows up on #26531 by recovering rejected messages in addition to keeping listener drain scheduling alive.
   
   ### Modifications
   
   - Retain the rejected message and retry submission with exponential backoff from 100 ms to 1 second.
   - Pause subsequent listener submissions until the retained message is accepted, preserving submission order.
   - Cancel pending retries on close and seek, handling concurrent submissions without releasing messages still in use.
   - Include the rejected message when calculating the restart position for non-durable subscriptions.
   - Document the rejection contract for `MessageListenerExecutor`.
   - Add real broker tests using a saturated bounded executor to cover recovery, permit replenishment, pooled-message cleanup, close races, seek, and reconnect. Preserve the existing rejection scheduling tests with updated recovery assertions, and add separate coverage for other submission failures.
   
   ### Verifying this change
   
   - [x] Make sure that the change passes the CI checks.
   
   *(Please pick either of the following options)*
   
   This change is a trivial rework / code cleanup without any test coverage.
   
   *(or)*
   
   This change is already covered by existing tests, such as *(please describe tests)*.
   
   *(or)*
   
   This change added tests and can be verified as follows:
   
   *(example:)*
     - *Added integration tests for end-to-end deployment with large payloads (10MB)*
     - *Extended integration test for recovery after broker failure*
   
   ### Does this pull request potentially affect one of the following parts:
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   *If the box was checked, please highlight the changes*
   
   - [ ] Dependencies (add or upgrade a dependency)
   - [ ] The public API
   - [ ] The schema
   - [ ] The default values of configurations
   - [ ] The threading model
   - [ ] The binary protocol
   - [ ] The REST endpoints
   - [ ] The admin CLI options
   - [ ] The metrics
   - [ ] Anything that affects deployment


-- 
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: commits-unsubscribe@pulsar.apache.org

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