[fix][client] Fix producer synchronous retry handling in failPendingMessages method

[sandeep-mst]

Fixes #25201

Motivation

Fix a re-entrancy bug in ProducerImpl.failPendingMessages. While executing on the timer, failPendingMessages invokes sendComplete(ex) on pending messages, which can synchronously trigger a retry from client code. The subsequent pendingMessages.clear() then removes the newly enqueued retry operation, leaving the retry’s CompletableFuture unresolved and the client in a limbo state.

Modifications

Updated failPendingMessages to first drain the pendingMessages queue into a local list and clear the queue before iterating. This prevents re-entrant retries triggered during sendComplete from being inadvertently cleared.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• org.apache.pulsar.client.impl.ProducerImplTest#testFailPendingMessagesSyncRetry

Does this pull request potentially affect one of the following parts:

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

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:
cognitree#29

[lhotari]

great work @sandeep-mst .

Regarding the tests, would it be possible to add a reproducer similar as described in #25201? (test without mocking the client, just using MockedPulsarServiceBaseTest base class)

[lhotari]

LGTM, great work @sandeep-mst . Just a wish to add a test mentioned in the description.

[sandeep-mst]

LGTM, great work @sandeep-mst . Just a wish to add a test mentioned in the description.

Will add it if possible. That test will go to pulsar-broker module. Should I remove the currently added test afterwards?

[lhotari]

LGTM, great work @sandeep-mst . Just a wish to add a test mentioned in the description.

Will add it if possible. That test will go to pulsar-broker module. Should I remove the currently added test afterwards?

I think the current unit tests are fine. The additional test would be a bit like an integration test. In Pulsar, the tests which use the MockedPulsarServiceBaseTest base class are in most cases integration tests although we call them "unit tests" in CI.

[lhotari]

In many cases we use org.apache.commons.lang3.reflect.FieldUtils or add a getter to the field (could use @Getter(value = AccessLevel.PACKAGE, onMethod_ = @__({@VisibleForTesting})) to add package protected getter in Lombok.