Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixed deadlock on txn semaphore permit exhaustion #14131

Merged
merged 3 commits into from
Feb 7, 2022
Merged

Fixed deadlock on txn semaphore permit exhaustion #14131

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b3f7c04
Fixed deadlock on txn semaphore permit exhaustion
merlimat Feb 5, 2022
87a329b
Removed semaphore field reflection access in test
merlimat Feb 5, 2022
4ca712f
Removed unused import
merlimat Feb 6, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 0 additions & 32 deletions ...n/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,14 +28,12 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.transaction.TransactionBufferClientException;
import org.apache.pulsar.client.api.transaction.TransactionBufferClientException.ReachMaxPendingOpsException;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.client.impl.ClientCnx;
import org.apache.pulsar.client.impl.PulsarClientImpl;
Expand All @@ -52,8 +50,6 @@ public class TransactionBufferHandlerImpl implements TransactionBufferHandler {
private final AtomicLong requestIdGenerator = new AtomicLong();
private final long operationTimeoutInMills;
private final HashedWheelTimer timer;
private final Semaphore semaphore;
private final boolean blockIfReachMaxPendingOps;
private final PulsarClient pulsarClient;

private final LoadingCache<String, CompletableFuture<ClientCnx>> cache = CacheBuilder.newBuilder()
Expand All @@ -77,8 +73,6 @@ public TransactionBufferHandlerImpl(PulsarClient pulsarClient,
this.pulsarClient = pulsarClient;
this.pendingRequests = new ConcurrentSkipListMap<>();
this.operationTimeoutInMills = 3000L;
this.semaphore = new Semaphore(10000);
this.blockIfReachMaxPendingOps = true;
this.timer = timer;
}

Expand All @@ -90,9 +84,6 @@ public CompletableFuture<TxnID> endTxnOnTopic(String topic, long txnIdMostBits,
topic, new TxnID(txnIdMostBits, txnIdLeastBits), action.getValue());
}
CompletableFuture<TxnID> cb = new CompletableFuture<>();
if (!canSendRequest(cb)) {
return cb;
}
long requestId = requestIdGenerator.getAndIncrement();
ByteBuf cmd = Commands.newEndTxnOnPartition(requestId, txnIdLeastBits, txnIdMostBits,
topic, action, lowWaterMark);
Expand All @@ -108,9 +99,6 @@ public CompletableFuture<TxnID> endTxnOnSubscription(String topic, String subscr
topic, new TxnID(txnIdMostBits, txnIdLeastBits), action.getValue());
}
CompletableFuture<TxnID> cb = new CompletableFuture<>();
if (!canSendRequest(cb)) {
return cb;
}
long requestId = requestIdGenerator.getAndIncrement();
ByteBuf cmd = Commands.newEndTxnOnSubscription(requestId, txnIdLeastBits, txnIdMostBits,
topic, subscription, action, lowWaterMark);
Expand Down Expand Up @@ -210,29 +198,9 @@ public void handleEndTxnOnSubscriptionResponse(long requestId,
onResponse(op);
}

private boolean canSendRequest(CompletableFuture<?> callback) {
try {
if (blockIfReachMaxPendingOps) {
semaphore.acquire();
} else {
if (!semaphore.tryAcquire()) {
callback.completeExceptionally(new ReachMaxPendingOpsException("Reach max pending ops."));
return false;
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
callback.completeExceptionally(TransactionBufferClientException.unwrap(e));
return false;
}
return true;
}


void onResponse(OpRequestSend op) {
ReferenceCountUtil.safeRelease(op.byteBuf);
op.recycle();
semaphore.release();
}

private static final class OpRequestSend {
Expand Down
10 changes: 0 additions & 10 deletions ...rc/test/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,16 +248,6 @@ public void testTransactionBufferLookUp() throws Exception {

@Test
public void testTransactionBufferHandlerSemaphore() throws Exception {

Field field = TransactionBufferClientImpl.class.getDeclaredField("tbHandler");
field.setAccessible(true);
TransactionBufferHandlerImpl transactionBufferHandler = (TransactionBufferHandlerImpl) field.get(tbClient);

field = TransactionBufferHandlerImpl.class.getDeclaredField("semaphore");
field.setAccessible(true);
field.set(transactionBufferHandler, new Semaphore(2));


String topic = "persistent://" + namespace + "/testTransactionBufferHandlerSemaphore";
String subName = "test";

Expand Down