You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "sandeep-mst (via GitHub)" <gi...@apache.org> on 2026/02/03 18:29:55 UTC
[PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
sandeep-mst opened a new pull request, #25207:
URL: https://github.com/apache/pulsar/pull/25207
<!-- Either this PR fixes an issue, -->
Fixes #25201
### Motivation
<!-- Explain here the context, and why you're making that change. What is the problem you're trying to solve. -->
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
<!-- Describe the modifications you've done. -->
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
- [x] 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:
<!-- 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
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
https://github.com/cognitree/pulsar/pull/29
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2762409297
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2304,16 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
Review Comment:
We do not access `pendingMessages` without the monitor lock.
Even though, it has monitor lock this method is executed by a timer thread and when [op.sendComplete(ex)](https://github.com/cognitree/pulsar/blob/cab92f0164947e3cbb0da1a014951561f1599add/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2315) is called within that method it switches to client logic where on the same thread they can send a retry message synchronously (`producer.sendAsync`) which will get enqueued into the pendingMessages [here](https://github.com/apache/pulsar/blob/cab92f0164947e3cbb0da1a014951561f1599add/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2468).
So, monitor lock is not preventing the issue from occurring because of reentrancy as it's the same thread holding the lock.
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2762975381
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2304,16 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
if (cnx == null) {
final AtomicInteger releaseCount = new AtomicInteger();
final boolean batchMessagingEnabled = isBatchMessagingEnabled();
- pendingMessages.forEach(op -> {
+ final ArrayList<OpSendMsg> failingMessages = new ArrayList<>(pendingMessages.size());
Review Comment:
Went for a different approach altogether. Not creating a new ArrayList.
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "heesung-sohn (via GitHub)" <gi...@apache.org>.
heesung-sohn commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2761466486
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2304,16 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
Review Comment:
do we access pendingMessages without monitor lock? can you share the reentrance code without the monitor lock?
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#issuecomment-3846230925
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)
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#issuecomment-3846373987
> > 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.
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2769198237
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2299,17 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
Review Comment:
The comment perhaps can be updated a bit.
"fails the messages that were pending at the start of the call, without dropping newly enqueued retry messages"
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2763700209
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java:
##########
@@ -67,4 +70,79 @@ public void testPopulateMessageSchema() {
assertTrue(producer.populateMessageSchema(msg, null));
verify(msg).setSchemaState(MessageImpl.SchemaState.Ready);
}
+
+ @Test
+ public void testFailPendingMessagesSyncRetry()
+ throws Exception {
+ ProducerImpl<byte[]> producer =
+ Mockito.mock(ProducerImpl.class, Mockito.CALLS_REAL_METHODS);
+ Mockito.doNothing()
+ .when(producer)
+ .semaphoreRelease(Mockito.anyInt());
+ Mockito.doReturn(false)
+ .when(producer)
+ .isBatchMessagingEnabled();
+
+ // Real pending queue
+ ProducerImpl.OpSendMsgQueue pendingQueue = new ProducerImpl.OpSendMsgQueue();
+ Field pendingField = ProducerImpl.class.getDeclaredField("pendingMessages");
+ pendingField.setAccessible(true);
+ pendingField.set(producer, pendingQueue);
Review Comment:
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.
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#issuecomment-3854613907
@lhotari Added the new integration test in `pulsar-broker` module. Please review and let me know if anything else is required.
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#issuecomment-3855734244
/pulsarbot rerun-failure-checks
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2762792887
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2304,16 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
Review Comment:
yes, it would. Doing the same.
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2762972839
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2304,16 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
Review Comment:
done
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2769331962
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2299,17 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
Review Comment:
yes makes sense. Done
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "heesung-sohn (via GitHub)" <gi...@apache.org>.
heesung-sohn commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2762781509
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2304,16 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
Review Comment:
would this work, instead of the snapshot array?
int pendingMessageCountToFail = pendingMessages.size();
for(int i =0; i< pendingMessageCountToFail; i++){
var op = pendingMessages.remove()
// fail op;
}
Then, we don't need to pendingMessages.clear(); at the line 2327.
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "dlg99 (via GitHub)" <gi...@apache.org>.
dlg99 commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2760532935
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2304,16 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
if (cnx == null) {
final AtomicInteger releaseCount = new AtomicInteger();
final boolean batchMessagingEnabled = isBatchMessagingEnabled();
- pendingMessages.forEach(op -> {
+ final ArrayList<OpSendMsg> failingMessages = new ArrayList<>(pendingMessages.size());
Review Comment:
how many pending messages there can be? is this a good idea to allocate new ArrayList all the time?
method is synchronized, so it should be safe to create a static ArrayList of size e.g. 100 and drain min(100, pendingMessages.size()) and process and repeat until pendingMessages.size() == 0
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2760602728
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2298,11 +2304,16 @@ public void run(Timeout timeout) throws Exception {
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ @VisibleForTesting
+ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
if (cnx == null) {
final AtomicInteger releaseCount = new AtomicInteger();
final boolean batchMessagingEnabled = isBatchMessagingEnabled();
- pendingMessages.forEach(op -> {
+ final ArrayList<OpSendMsg> failingMessages = new ArrayList<>(pendingMessages.size());
Review Comment:
At this point, the pending messages should consist only the timed out messages from the producer timeout config. So, it relatively will be low but it makes sense to size it down. Will update the code.
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#issuecomment-3846260765
> LGTM, great work @sandeep-mst . Just a wish to add a test mentioned in the description.
Will add it. That test will go to `pulsar-broker` module. Should I remove the currently added test afterwards?
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "sandeep-mst (via GitHub)" <gi...@apache.org>.
sandeep-mst commented on code in PR #25207:
URL: https://github.com/apache/pulsar/pull/25207#discussion_r2769358090
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java:
##########
@@ -67,4 +70,79 @@ public void testPopulateMessageSchema() {
assertTrue(producer.populateMessageSchema(msg, null));
verify(msg).setSchemaState(MessageImpl.SchemaState.Ready);
}
+
+ @Test
+ public void testFailPendingMessagesSyncRetry()
+ throws Exception {
+ ProducerImpl<byte[]> producer =
+ Mockito.mock(ProducerImpl.class, Mockito.CALLS_REAL_METHODS);
+ Mockito.doNothing()
+ .when(producer)
+ .semaphoreRelease(Mockito.anyInt());
+ Mockito.doReturn(false)
+ .when(producer)
+ .isBatchMessagingEnabled();
+
+ // Real pending queue
+ ProducerImpl.OpSendMsgQueue pendingQueue = new ProducerImpl.OpSendMsgQueue();
+ Field pendingField = ProducerImpl.class.getDeclaredField("pendingMessages");
+ pendingField.setAccessible(true);
+ pendingField.set(producer, pendingQueue);
Review Comment:
done. Used `FieldUtils` as the `@Getter` approach is not possible for this producer mock as the `pendingMessages` field is `final` .
--
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
Re: [PR] [fix][client] Fix producer synchronous retry handling in failPendingMessages method [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #25207:
URL: https://github.com/apache/pulsar/pull/25207
--
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