apache / apache/pulsar

[Transaction] internalPinnedExecutor will stuck in acquire semaphore

Open
#14,866 2 comments 0 reactions 0 assignees View on GitHub
lifecycle/stale Stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

**Describe the bug**
1. TransactionMetaStoreHandler use a internalPinnedExecutor to execute` new transaction``addPublishPartitionToTxnAsync` and so on.
2. We add a semaphore in TransactionMetaStoreHandler before execute ops.
3. If we new transaction and produce message in callback of new transaction, internalPinnedExecutor will stuck in acquire semaphore, and we also need this internalPinnedExecutor to release semaphore.

**To Reproduce**
Run this test.

>```
> public void testExternalThreadToAcquireSemaphore() throws Exception {
> long st = System.currentTimeMillis();
> String topic = NAMESPACE1 + "/test";
> int numOfTransaction = 1000;
> int transactionTimeoutInSec = 30;
> int numOfMessagesPerTransaction = 20;
>
> Producer producer = pulsarClient.newProducer(Schema.INT64)
> .producerName("transaction-send-test")
> .sendTimeout(0, TimeUnit.SECONDS)
> .topic(topic)
> .enableBatching(false)
> .create();
> CountDownLatch waitSendCompletely = new CountDownLatch(numOfTransaction);
> AtomicBoolean exitByException = new AtomicBoolean(false);
>
> for (int i = 0; i < numOfTransaction; i++) {
> pulsarClient.newTransaction()
> .withTransactionTimeout(transactionTimeoutInSec, TimeUnit.SECONDS)
> .build().whenComplete((transaction, throwable) -> {
> if (throwable != null) {
> log.error("Failed to open transaction", throwable);
> exitByException.set(true);
> } else {
> for (long j = 0; j < numOfMessagesPerTransaction; j++) {
> producer.newMessage(transaction)
> .value(j)
> .sendAsync()
> .whenComplete((messageId, exception) -> {
> if (exception != null) {
> log.error("Failed to send messages", exception);
> exitByException.set(true);
> }
> });
> }
> transaction.commit().thenRun(() -> {
> waitSendCompletely.countDown();
> }).exceptionally(throwable1 -> {
> log.error("Failed to commit the transaction [{}]", transaction, throwable1);
> waitSendCompletely.countDown();
> exitByException.set(true);
> return null;
> });
> }
> });
> }
> new Thread(() -> {
> try {
> Thread.sleep(30000);
> exitByException.set(true);
> waitSendCompletely.countDown();
> } catch (InterruptedException e) {
> e.printStackTrace();
> }
> });
> waitSendCompletely.await();
> assertFalse(exitByException.get());
> log.info("test completely in [{}] secs", (System.currentTimeMillis() - st) / 1000);
> }
>

**Expected behavior**
A clear and concise description of what you expected to happen.

**Screenshots**
If applicable, add screenshots to help explain your problem.

**Desktop (please complete the following information):**
- OS: [e.g. iOS]

**Additional context**
Add any other context about the problem here.

Contributor guide

Open the contributing guide

Research direction

Start with TransactionMetaStoreHandler, especially internalPinnedExecutor and the semaphore around transaction operations such as new transaction and addPublishPartitionToTxnAsync. Run testExternalThreadToAcquireSemaphore with the provided transaction and producer workload. Done means the test completes without the executor stalling while acquiring the semaphore, and the final assertion passes.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.