You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "3pacccccc (via GitHub)" <gi...@apache.org> on 2025/07/23 15:10:34 UTC
[PR] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
3pacccccc opened a new pull request, #24550:
URL: https://github.com/apache/pulsar/pull/24550
### Motivation
Currently, when `consumer.receive()` is called, it blocks indefinitely even when the consumer is being closed. This can lead to threads hanging indefinitely in applications that need to gracefully shutdown consumers. The issue can be reproduced with:
``` java
Thread thread = new Thread(() -> {
try {
// sleep 0.1 second to close consumer to ensure consumer.receive() is triggerd
Thread.sleep(1000);
consumer.close();
} catch (Exception e) {
throw new RuntimeException(e);
}
});
thread.start();
// hangs forever
consumer.receive();
```
in fact, this could happen in many scenario as long as ```consumer.receive()``` called before ```consumer.close()```
**so, this PR aims to ensure that `consumer.receive()` will immediately throw an `IllegalStateException` when the consumer is being closed, providing clear feedback and preventing thread hangs**
### Modifications
1.Added termination handling in GrowableArrayBlockingQueue to:
- Check termination flag in take() and poll() methods
- Throw IllegalStateException with "Queue is terminated" message when terminated
2.Modified ```ConsumerImpl.closeConsumerTasks()``` to terminate the incoming message queue
3.Added tests covering:
- Regular consumer close during receive()
- Zero queue size consumer receives a batchMessage and another thread close consuer
<!-- Describe the modifications you've done. -->
### Verifying this change
- [x] Make sure that the change passes the CI checks.
*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: https://github.com/3pacccccc/pulsar/pull/15
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2226469075
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -194,6 +194,9 @@ public T take() throws InterruptedException {
try {
while (SIZE_UPDATER.get(this) == 0) {
+ if (terminated) {
+ throw new IllegalStateException("Queue is terminated");
+ }
isNotEmpty.await();
}
Review Comment:
there would need to be another check for "terminated" here since a thread might be waiting when the queue gets terminated.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1246,6 +1246,7 @@ private void closeConsumerTasks() {
if (poolMessages) {
releasePooledMessagesAndStopAcceptNew();
}
+ incomingMessages.terminate();
Review Comment:
releasePooledMessagesAndStopAcceptNew already calls the other terminate method. Refactor this code since there should be a single terminate method in GroableArrayBlockingQueue that handles termination. I'd suggest renaming releasePooledMessagesAndStopAcceptNew and call it in all cases with the consumer to release new messages only in the case of pooled messages.
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -438,6 +444,20 @@ public void terminate(@Nullable Consumer<T> itemAfterTerminatedHandler) {
}
}
+
+ public void terminate() {
+ // 1. Set terminated flag to prevent new items
+ terminate(itemAfterTerminatedHandler);
+
+ // 2. Wake up waiting threads with proper handling
+ headLock.lock();
+ try {
+ isNotEmpty.signal();
+ } finally {
+ headLock.unlock();
+ }
+ }
Review Comment:
instead of adding a new terminate method, It's recommend adding the logic to signal the wake up directly in the existing terminate method.
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -194,6 +194,9 @@ public T take() throws InterruptedException {
try {
while (SIZE_UPDATER.get(this) == 0) {
+ if (terminated) {
+ throw new IllegalStateException("Queue is terminated");
+ }
Review Comment:
instead of throwing an exception, I'd recommend returning null. GrowableArrayBlockingQueue implements a BlockingQueue and using IllegalStateException would be violating the contract.
One possible solution would be to return null and the caller could check the consumer state when a null is returned and throw AlreadyClosedException in that case to indicate that the consumer has been closed.
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#issuecomment-3119921843
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24550?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 74.26%. Comparing base ([`bbc6224`](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`b4a02da`](https://app.codecov.io/gh/apache/pulsar/commit/b4a02dad09b9c710f0d46fb43a7046c8a5e9b903?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1229 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24550?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24550 +/- ##
============================================
+ Coverage 73.57% 74.26% +0.69%
+ Complexity 32624 32595 -29
============================================
Files 1877 1876 -1
Lines 139502 146326 +6824
Branches 15299 16781 +1482
============================================
+ Hits 102638 108673 +6035
- Misses 28908 29021 +113
- Partials 7956 8632 +676
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24550/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24550/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.70% <50.00%> (+2.11%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24550/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.28% <50.00%> (-1.04%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24550/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.77% <100.00%> (+0.93%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24550?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...va/org/apache/pulsar/client/impl/ConsumerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24550?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FConsumerImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVySW1wbC5qYXZh) | `79.24% <100.00%> (+1.67%)` | :arrow_up: |
| [...n/util/collections/GrowableArrayBlockingQueue.java](https://app.codecov.io/gh/apache/pulsar/pull/24550?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Futil%2Fcollections%2FGrowableArrayBlockingQueue.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi91dGlsL2NvbGxlY3Rpb25zL0dyb3dhYmxlQXJyYXlCbG9ja2luZ1F1ZXVlLmphdmE=) | `97.58% <100.00%> (+2.15%)` | :arrow_up: |
... and [1111 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24550/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#issuecomment-3117450611
> IMO, because of the `headLock` ensures that there is at most one thread can be waiting on the condition. so, I think the scenario `handle the case where more than one thread is waiting.` might not be existed.
> @lhotari
`isNotEmpty` has been created with `headLock.newCondition()`. When `.await()` is called, the lock is released. When `.signal` is called, it will acquire the lock again. The difference between `.signal` and `.signalAll` is that the first one will only wake up one thread. `.signalAll` will wake up one thread at a time.
There is a need to handle the multi-thread case.
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2227577606
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -194,6 +194,9 @@ public T take() throws InterruptedException {
try {
while (SIZE_UPDATER.get(this) == 0) {
+ if (terminated) {
+ throw new IllegalStateException("Queue is terminated");
+ }
Review Comment:
> instead of throwing an exception, I'd recommend returning null. GrowableArrayBlockingQueue implements a BlockingQueue and using IllegalStateException would be violating the contract. One possible solution would be to return null and the caller could check the consumer state when a null is returned and throw AlreadyClosedException in that case to indicate that the consumer has been closed.
Actually return null also violating the contract, because the method ```BlockingQueue#take``` declared as ```@NotNull```, so I throw a InterruptedException, I think it might be more properly
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2228906555
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1243,18 +1243,22 @@ private void closeConsumerTasks() {
}
negativeAcksTracker.close();
stats.getStatTimeout().ifPresent(Timeout::cancel);
- if (poolMessages) {
- releasePooledMessagesAndStopAcceptNew();
- }
+ processIncomingMessagesWhenClose();
}
/**
- * If enabled pooled messages, we should release the messages after closing consumer and stop accept the new
+ * when consumer is closed.we should
+ * 1. terminate incomingMessages queue
+ * 2. If enabled pooled messages, we should release the messages after closing consumer and stop accept the new
* messages.
*/
- private void releasePooledMessagesAndStopAcceptNew() {
- incomingMessages.terminate(message -> message.release());
- clearIncomingMessages();
+ private void processIncomingMessagesWhenClose() {
+ if (poolMessages) {
+ incomingMessages.terminate(message -> message.release());
+ clearIncomingMessages();
+ } else {
+ incomingMessages.terminate(null);
+ }
Review Comment:
> btw. it's odd that "clearIncomingMessages" is only called in the case of poolMessage. Since message.release() is a no-op when the message isn't a pooled message, it would be fine to remove the condition for poolMessages. The implication of not calling "clearIncomingMessages" in all cases is that it won't release memory back to the memory controller for messages that are in the queue. That might also be a problem with the terminate callback.
@lhotari good point, I've pushed a new commit, LTAL
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2227577606
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -194,6 +194,9 @@ public T take() throws InterruptedException {
try {
while (SIZE_UPDATER.get(this) == 0) {
+ if (terminated) {
+ throw new IllegalStateException("Queue is terminated");
+ }
Review Comment:
> instead of throwing an exception, I'd recommend returning null. GrowableArrayBlockingQueue implements a BlockingQueue and using IllegalStateException would be violating the contract. One possible solution would be to return null and the caller could check the consumer state when a null is returned and throw AlreadyClosedException in that case to indicate that the consumer has been closed.
Actually return null also violating the contract, because the method ```BlockingQueue#take``` declared as ```@NotNull```, so I throw a InterruptedException, I think it might be more propertly
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2228545652
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -194,6 +194,9 @@ public T take() throws InterruptedException {
try {
while (SIZE_UPDATER.get(this) == 0) {
+ if (terminated) {
+ throw new IllegalStateException("Queue is terminated");
+ }
isNotEmpty.await();
}
Review Comment:
oh yes, I guess it's fine since in terminate, putting to the queue will be blocked first and then the signaling will happen.
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#issuecomment-3119155892
> > IMO, because of the `headLock` ensures that there is at most one thread can be waiting on the condition. so, I think the scenario `handle the case where more than one thread is waiting.` might not be existed.
> > @lhotari
>
> `isNotEmpty` has been created with `headLock.newCondition()`. When `.await()` is called, the lock is released. When `.signal` is called, the thread that was awaiting will acquire the lock again. The difference between `.signal` and `.signalAll` is that the first one will only wake up one thread. `.signalAll` will wake up one thread at a time. There is a need to handle the multi-thread case.
@lhotari yes! you're right. Thank you so much for pointing this out, I've pushed a new commit, PTAL
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2227601123
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -194,6 +194,9 @@ public T take() throws InterruptedException {
try {
while (SIZE_UPDATER.get(this) == 0) {
+ if (terminated) {
+ throw new IllegalStateException("Queue is terminated");
+ }
isNotEmpty.await();
}
Review Comment:
> there would need to be another check for "terminated" here since a thread might be waiting when the queue gets terminated.
I think no need to, because if a thread can access here, it means there is data exist in queue, so just return data no matter ```terminated``` or not. or maybe I don't get your point, we can dive deeper into this discussion
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2228557266
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1243,18 +1243,22 @@ private void closeConsumerTasks() {
}
negativeAcksTracker.close();
stats.getStatTimeout().ifPresent(Timeout::cancel);
- if (poolMessages) {
- releasePooledMessagesAndStopAcceptNew();
- }
+ processIncomingMessagesWhenClose();
}
/**
- * If enabled pooled messages, we should release the messages after closing consumer and stop accept the new
+ * when consumer is closed.we should
+ * 1. terminate incomingMessages queue
+ * 2. If enabled pooled messages, we should release the messages after closing consumer and stop accept the new
* messages.
*/
- private void releasePooledMessagesAndStopAcceptNew() {
- incomingMessages.terminate(message -> message.release());
- clearIncomingMessages();
+ private void processIncomingMessagesWhenClose() {
+ if (poolMessages) {
+ incomingMessages.terminate(message -> message.release());
+ clearIncomingMessages();
+ } else {
+ incomingMessages.terminate(null);
+ }
Review Comment:
btw. it's odd that "clearIncomingMessages" is only called in the case of poolMessage. Since message.release() is a no-op when the message isn't a pooled message, it would be fine to remove the condition for poolMessages.
The implication of not calling "clearIncomingMessages" in all cases is that it won't release memory back to the memory controller for messages that are in the queue. That might also be a problem with the terminate callback.
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2228560035
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1243,18 +1243,22 @@ private void closeConsumerTasks() {
}
negativeAcksTracker.close();
stats.getStatTimeout().ifPresent(Timeout::cancel);
- if (poolMessages) {
- releasePooledMessagesAndStopAcceptNew();
- }
+ processIncomingMessagesWhenClose();
}
/**
- * If enabled pooled messages, we should release the messages after closing consumer and stop accept the new
+ * when consumer is closed.we should
+ * 1. terminate incomingMessages queue
+ * 2. If enabled pooled messages, we should release the messages after closing consumer and stop accept the new
* messages.
*/
- private void releasePooledMessagesAndStopAcceptNew() {
- incomingMessages.terminate(message -> message.release());
- clearIncomingMessages();
+ private void processIncomingMessagesWhenClose() {
+ if (poolMessages) {
+ incomingMessages.terminate(message -> message.release());
+ clearIncomingMessages();
+ } else {
+ incomingMessages.terminate(null);
+ }
Review Comment:
This is a different matter, but it would be great to check it while we are on this 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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #24550:
URL: https://github.com/apache/pulsar/pull/24550
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2230487301
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -436,6 +443,14 @@ public void terminate(@Nullable Consumer<T> itemAfterTerminatedHandler) {
} finally {
tailLock.unlockWrite(stamp);
}
+
+ // Signal waiting consumer threads to prevent indefinite blocking after termination
+ headLock.lock();
+ try {
+ isNotEmpty.signal();
Review Comment:
We should wake up all waiting threads.
```suggestion
isNotEmpty.signalAll();
```
Please add a test case for more than one thread so that this case is also covered.
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#discussion_r2230491006
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:
##########
@@ -436,6 +443,14 @@ public void terminate(@Nullable Consumer<T> itemAfterTerminatedHandler) {
} finally {
tailLock.unlockWrite(stamp);
}
+
+ // Signal waiting consumer threads to prevent indefinite blocking after termination
+ headLock.lock();
+ try {
+ isNotEmpty.signal();
Review Comment:
I guess it would be extraordinary to use the Consumer API this way, but that's possible and the API doesn't prevent multiple threads to call `.receive()` at the same time.
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#issuecomment-3112229517
> Good catch! left some review comments
thank you so such for pointing these out, I've pushed a new commit, please take a look again.
--
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] [improve][client] Terminate consumer.receive() when consumer is closed [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on PR #24550:
URL: https://github.com/apache/pulsar/pull/24550#issuecomment-3116931997
> There's a need to handle the case where more than one thread is waiting.
IMO, because of the ```headLock``` ensures that there is at most one thread can be waiting on the condition. so, I think the scenario ```handle the case where more than one thread is waiting.``` might not be existed.
@lhotari
--
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