You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "BewareMyPower (via GitHub)" <gi...@apache.org> on 2025/07/21 05:52:25 UTC
[PR] [fix][client] Close orphan producer or consumer when the creation is interrupted [pulsar]
BewareMyPower opened a new pull request, #24539:
URL: https://github.com/apache/pulsar/pull/24539
### Motivation
The producer or reader creation has the same issue like what https://github.com/apache/pulsar/pull/24100 fixes for consumer: when the thread is interrupted, the returned future will complete exceptionally with the wrapped `InterruptedException`, but the asynchronous creation will continue in the background.
### Modifications
- Add a common `FutureUtils.wait` method to call a callback on the completed result.
- Apply the method above to the creation of producer, consumer and reader.
- Add `ClientInterruptTest` to cover the fixes above and remove duplicated `ConsumerCloseTest#testInterruptedWhenCreateConsumer`.
It should be noted that the complicated fix approach of https://github.com/apache/pulsar/pull/24100 is replaced by the solution in this PR.
This PR also reverts the change that the thread interrupt state is set, which should be the caller's responsibility to do that. This is actually a wrong behavior for many existing synchronous APIs.
Without this change, the 2nd creation of the consumer in the code snippet will fail when the 1st creation is interrupted:
```java
try {
@Cleanup final var consumer = client.newConsumer().topic("tp").subscriptionName("sub")
.subscribe();
fail();
} catch (PulsarClientException e) {
assertTrue(e.getCause() instanceof InterruptedException);
}
try {
@Cleanup final var consumer = client.newConsumer().topic("another-tp").subscriptionName("sub")
.subscribe();
} catch (PulsarClientException e) {
fail();
}
```
If we want to make the 2nd call fail, the caller should handle the `InterruptException` like:
```java
} catch (PulsarClientException e) {
if (e.getCause() instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
}
```
In short,
- Before https://github.com/apache/pulsar/pull/24100, after the creation of producer, consumer or reader is interrupted, the thread interrupt flag won't be set.
- After https://github.com/apache/pulsar/pull/24100, after the creation of consumer is interrupted, the thread interrupt flag will be set.
- After this PR, after the creation of producer, consumer or reader is interrupted, the thread interrupt flag won't be set.
### 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 -->
<!--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#discussion_r2218308684
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -381,4 +382,19 @@ public static void safeRunAsync(Runnable runnable,
return null;
});
}
+
+ /**
+ * When calling {@link CompletableFuture#get()} on a future, it could throw a {@link InterruptedException} when
+ * the thread is interrupted. However, the asynchronous task might not be interrupted and the result could be
+ * created successfully eventually. In this case, we might need to close the result.
+ */
+ public static <T> T wait(CompletableFuture<T> future, Consumer<T> closeCallback)
+ throws InterruptedException, ExecutionException {
Review Comment:
```suggestion
/**
* Blocks to get the result of a CompletableFuture, while ensuring resources are cleaned up
* if the wait is interrupted.
* <p>
* If the current thread is interrupted while waiting, this method registers a cleanup action
* to be executed when the future eventually completes. This prevents resource leaks that
* could otherwise occur when an interruption happens but the underlying asynchronous task
* finishes successfully later. After registering the action, it re-throws the
* {@link InterruptedException}.
*
* @param future The CompletableFuture to wait for.
* @param cleanupAction A consumer that performs a cleanup action (e.g., closing a resource)
* on the result if the wait is interrupted.
* @param <T> The type of the future's result.
* @return The computed result from the future.
* @throws InterruptedException if the current thread was interrupted while waiting.
* @throws ExecutionException if the future completed exceptionally.
*/
public static <T> T getAndCleanupOnInterrupt(CompletableFuture<T> future, Consumer<T> cleanupAction)
throws InterruptedException, ExecutionException {
```
Change the method name to make it more clearer.
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3096964259
> the user's intent was to stop the entire thread.
You said, it's the user's intent. Then user can manually set the thread's interruption flag to fail the subsequent `receive` calls:
```java
// 1. First, try to create consumer
Consumer<byte[]> consumer = null;
try {
/* ... */
} catch (PulsarClientException e) {
// HERE
if (e.getCause() instanceof InterruptedException) {
Thread.currentThread().interrupt();
}
}
```
**Why does it have to be the client SDK that does that?** If the user don't want to stop the thread before, there is no way unless a very ugly solution like:
```java
// 1. First, try to create consumer
Consumer<byte[]> consumer = null;
try {
/* ... */
} catch (PulsarClientException e) {
// ...
}
// HERE
try {
Thread.sleep(1);
} catch (InterruptedException ignored) {
}
```
My [previous comment](https://github.com/apache/pulsar/pull/24539#issuecomment-3095802016) already mentions this example. This is a real world case.
The key point is, JDK provides the ability to set the flag by `Thread#interrupt` but does not provide the ability to clear the flag directly (as my example shows, users can unset the flag by manually clearing the `InterruptedException`). When the creation of a consumer fails, the `PulsarClient`'s state is still healthy. There is no reason to fail the subsequent creation.
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3096850561
hi, @BewareMyPower @lhotari. Thank you all for discussing this.
> BTW, @shibd I'm wondering that if you know the impact when you added Thread.currentThread().interrupt(); in https://github.com/apache/pulsar/pull/24100?
Initially, I didn't think through the use cases carefully and just followed the best practice, as mentioned [here](https://github.com/apache/pulsar/pull/24539#pullrequestreview-3037017249)
I've reconsidered this case, and I can think of a scenario like this:
Suppose a user tries to create a consumer `twice` within a single thread. If the creation of the first consumer fails, the code then attempts to create the second one and subsequently receive messages.
If we don't restore the interrupt flag, the creation of the second consumer will also fail. This is not the expected behavior, because the user's intent was to stop the entire thread.
``` java
@Test
public void testInterruptedWhenCreateConsumer() throws InterruptedException {
String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
String subName = "test-sub";
String mlCursorPath = BrokerService.MANAGED_LEDGER_PATH_ZNODE + "/"
+ TopicName.get(tpName).getPersistenceNamingEncoding() + "/" + subName;
// Make create cursor delay 1s
CountDownLatch topicLoadLatch = new CountDownLatch(1);
for (int i = 0; i < 5; i++) {
mockZooKeeper.delay(1000, (op, path) -> {
if (mlCursorPath.equals(path)) {
topicLoadLatch.countDown();
return true;
}
return false;
});
}
Thread startConsumer = new Thread(() -> {
// 1. First, try to create consumer
Consumer<byte[]> consumer = null;
try {
consumer = pulsarClient.newConsumer()
.topic(tpName)
.subscriptionName(subName)
.subscribe();
} catch (PulsarClientException e) {
log.error("failed to create consumer");
}
log.info("Thread stats {}", Thread.currentThread().isInterrupted());
// 2. if creating pulsar consumer failed, try to create consumer again
try {
if (consumer == null) {
consumer = pulsarClient.newConsumer()
.topic(tpName)
.subscriptionName(subName)
.subscribe();
}
// receive message and do somthing
} catch (PulsarClientException e) {
throw new RuntimeException(e);
}
while (!Thread.currentThread().isInterrupted()) {
try {
Message<byte[]> receive = consumer.receive();
log.info("Receive message {}", receive);
} catch (PulsarClientException e) {
throw new RuntimeException(e);
}
}
});
startConsumer.start();
topicLoadLatch.await();
// user try to interrupt the thread
startConsumer.interrupt();
PulsarClientImpl clientImpl = (PulsarClientImpl) pulsarClient;
Awaitility.await().ignoreExceptions().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
Assert.assertEquals(clientImpl.consumersCount(), 0);
});
log.info("Thread is alive {}", startConsumer.isAlive());
}
```
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3095316816
Actually @nodece raised the opinion about thread interruption here, but it's not addressed, maybe due to the misunderstanding. https://github.com/apache/pulsar/pull/24100#pullrequestreview-2704697654
IMO, the key point should be:
> When using the Pulsar Client/Admin, the users should not call the thread interrupt.
The client SDK should guarantee when the thread is interrupted, there will be no resource leak. But the SDK should not set the thread interruption flag, which might lead to unexpected behavior.
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3101671834
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24539?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 74.28%. 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 [(`25a4687`)](https://app.codecov.io/gh/apache/pulsar/commit/25a4687df996b5c60ce7db169f7bef5636458e9e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1210 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24539?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24539 +/- ##
============================================
+ Coverage 73.57% 74.28% +0.70%
- Complexity 32624 32905 +281
============================================
Files 1877 1869 -8
Lines 139502 146081 +6579
Branches 15299 16760 +1461
============================================
+ Hits 102638 108510 +5872
- Misses 28908 28953 +45
- Partials 7956 8618 +662
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24539/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/24539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.72% <57.14%> (+2.14%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24539/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.34% <57.14%> (-0.99%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24539/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.92%)` | :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/24539?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...apache/pulsar/client/impl/ConsumerBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24539?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FConsumerBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVyQnVpbGRlckltcGwuamF2YQ==) | `86.83% <100.00%> (-0.11%)` | :arrow_down: |
| [...apache/pulsar/client/impl/ProducerBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24539?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FProducerBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1Byb2R1Y2VyQnVpbGRlckltcGwuamF2YQ==) | `83.07% <100.00%> (+0.50%)` | :arrow_up: |
| [...g/apache/pulsar/client/impl/ReaderBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24539?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FReaderBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1JlYWRlckJ1aWxkZXJJbXBsLmphdmE=) | `87.23% <100.00%> (+2.12%)` | :arrow_up: |
| [...java/org/apache/pulsar/common/util/FutureUtil.java](https://app.codecov.io/gh/apache/pulsar/pull/24539?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Futil%2FFutureUtil.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NvbW1vbi91dGlsL0Z1dHVyZVV0aWwuamF2YQ==) | `74.21% <100.00%> (-0.33%)` | :arrow_down: |
... and [1096 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24539/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] [fix][client] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3102649521
Let me merge this PR first, since the key point of this PR is to close orphan producer or consumer.
But how to handle the interruption will be another interesting topic. This is a part that previous PRs hardly touched and discussed in details. I'm preparing a PIP at the moment. We can discuss more after it's out.
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3101433113
> > Whenever code in some method calls another method that throws InterruptedException and it is not propagated to the caller (the method doesn't throw InterruptedException in the method signature)
>
> The key point is that from my perspective, the `InterruptedException` is propagated to the caller from the underlying of `PulsarClientException`. It's an issue with the poor JavaDocs for client APIs. The synchronous API should note how should the caller handle the `PulsarClientException` except for simply logging it.
>
> The real use case is that I'm using a single thread executor to perform two types of tasks:
>
> * LOAD: create a reader and read a topic from a given position to load key-values of this topic
> * UNLOAD: remove the key-values from this topic
>
> They're synchronous tasks executed in a single thread, executed by `ExecutorService#submit`. Before scheduling each task, I will cancel the previous task (typically LOAD because it's blocking) by calling `.cancel(true)` and handling the possible `InterruptedException` after unwrapping `PulsarClientException`. If the interruption flag is set internally by the client SDK, there is no graceful way to cancel this flag because this thread could schedule new tasks in future.
>
> You can see https://gist.github.com/BewareMyPower/fbd1e67fa947967c71b4722289ba7cea for reference.
>
> Since I used `Reader` rather than `Consumer`, I didn't suffer the breaking change brought by #24100.
>
> Since this PR uses a common method to handle interruption for creation of producer, reader and consumer. I have to choose whether to set the interruption flag. I chose NOT from the perspective of the downstream application developer.
Java Concurrency in Practice is pretty clear about the correct handling of InterruptedException:
<img width="702" height="674" alt="image" src="https://github.com/user-attachments/assets/81f03141-963b-4d11-b4fe-e8d04faff762" />
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3101483689
Yes, the key point is that `InterruptedException` has already been propagated as the cause of the `PulsarClientException`. The issue is that it's not documented to be a formal behavior, but all methods, including the method that has set the interruption flag, all have the same behavior.
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3101394015
> The key point is, JDK provides the ability to set the flag by `Thread#interrupt` but does not provide the ability to clear the flag directly (as my example shows, users can unset the flag by manually clearing the `InterruptedException`).
@BewareMyPower The interrupt flag can be cleared with the `interrupted` method in `java.lang.Thread`. It's not very intuitively named method.
> When the creation of a consumer fails, the `PulsarClient`'s state is still healthy. There is no reason to fail the subsequent creation.
When the current thread is interrupted, there's some sort of shutdown in progress. I agree that it's very context specific whether subsequent calls should fail or not. It depends a lot on which thread is in question. In most cases, when a thread's interrupted flag is set, that thread should finish up work after cleanup. Since when a thread is interrupted, there's some sort of shutdown in progress. Typical cases are that a task running in a ExecutorService is cancelled, or the complete JVM is shutdown and it's part of the shutdown sequence that threads get interrupted.
If there's a need to keep on calling blocking methods to cleanup, there are ways to handle that.
For example, Guava contains `com.google.common.util.concurrent.Uninterruptibles` utilities which will run the work in the current thread after resetting the possible interrupted status and then restore the interrupted status.
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3095693805
> The client SDK should guarantee when the thread is interrupted, there will be no resource leak. But the SDK should not set the thread interruption flag, which might lead to unexpected behavior.
As mention in my previous comment, I disagree on this point.
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3095802016
> Whenever code in some method calls another method that throws InterruptedException and it is not propagated to the caller (the method doesn't throw InterruptedException in the method signature)
The key point is that from my perspective, the `InterruptedException` is propagated to the caller from the underlying of `PulsarClientException`. It's an issue with the poor JavaDocs for client APIs. The synchronous API should note how should the caller handle the `PulsarClientException` except for simply logging it.
The real use case is that I'm using a single thread executor to perform two types of tasks:
- LOAD: create a reader and read a topic from a given position to load key-values of this topic
- UNLOAD: remove the key-values from this topic
They're synchronous tasks executed in a single thread, executed by `ExecutorService#submit`. Before scheduling each task, I will cancel the previous task (typically LOAD because it's blocking) by calling `.cancel(true)` and handling the possible `InterruptedException` after unwrapping `PulsarClientException`. If the interruption flag is set internally by the client SDK, there is no graceful way to cancel this flag because this thread could schedule new tasks in future.
You can see https://gist.github.com/BewareMyPower/fbd1e67fa947967c71b4722289ba7cea for reference.
Since I used `Reader` rather than `Consumer`, I didn't suffer the breaking change brought by https://github.com/apache/pulsar/pull/24100.
Since this PR uses a common method to handle interruption for creation of producer, reader and consumer. I have to choose whether to set the interruption flag. I chose NOT from the perspective of the downstream application developer.
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24539:
URL: https://github.com/apache/pulsar/pull/24539#issuecomment-3095921674
> But the SDK should not set the thread interruption flag, which might lead to unexpected behavior.
I'm correcting the description to "in certain cases".
For example, setting the interruption flag in `Consumer#receive` makes sense. Because once it's interrupted, it means the internal blocking queue has been interrupted. There is no way from the caller side to continue calling `receive` safely, so the client SDK must set the interruption flag to fail all subsequent `receive` calls in the same thread.
However, for this case, take consumer for example, if the creation failed:
```java
@Cleanup final var consumer = client.newConsumer().topic("tp").subscriptionName("sub")
.subscribe();
```
The `consumer` variable will never be touched by the caller. Setting the interruption flag could fail all subsequent blocking calls that might throw an `InterruptedException` (e.g. `Thread.sleep`), this would be really confusing.
BTW, @shibd I'm wondering that if you know the impact when you added `Thread.currentThread().interrupt();` in https://github.com/apache/pulsar/pull/24100?
--
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] Close orphan producer or consumer when the creation is interrupted [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #24539:
URL: https://github.com/apache/pulsar/pull/24539
--
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