You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/12/24 03:27:16 UTC
[GitHub] [pulsar] liangyepianzhou opened a new pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
liangyepianzhou opened a new pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481
### Motivation
Avoid concurrent processing of ack, avoid causing disorder of ack.
### Modification
HandleCacheRequest before change to ready.
If the request `isInCacheRequest == true`, there is no need to judge if ready.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API: (yes / no)
- The schema: (yes / no / don't know)
- The default values of configurations: (yes / no)
- The wire protocol: (yes / no)
- The rest endpoints: (yes / no)
- The admin cli options: (yes / no)
- Anything that affects deployment: (yes / no / don't know)
### Documentation
Check the box below and label this PR (if you have committer privilege).
Need to update docs?
- [ ] `doc-required`
(If you need help on updating docs, create a doc issue)
- [x] `no-need-doc`
(Please explain why)
- [ ] `doc`
(If this PR contains doc changes)
--
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
[GitHub] [pulsar] congbobo184 commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r776121803
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java
##########
@@ -44,7 +44,7 @@ public MLPendingAckReplyCallBack(PendingAckHandleImpl pendingAckHandle) {
@Override
public void replayComplete() {
- synchronized (pendingAckHandle) {
+ pendingAckHandle.getInternalPinnedExecutor().submit(() -> {
Review comment:
use execute method. if you use submit, it will return future, we don't need the future
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -291,46 +275,51 @@ private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
return completableFuture;
}
- private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
- List<PositionImpl> positions,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> cumulativeAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
- }
-
@Override
- public CompletableFuture<Void> cumulativeAcknowledgeMessage(TxnID txnID,
- List<PositionImpl> positions,
+ public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
+ List<MutablePair<PositionImpl, Integer>> positions,
boolean isInCacheRequest) {
- if (!checkIfReady()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- synchronized (PendingAckHandleImpl.this) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ internalPinnedExecutor.submit(() -> {
+ if (!checkIfReady()) {
switch (state) {
case Initializing:
- addCumulativeAcknowledgeMessageRequest(txnID, positions, completableFuture);
- return completableFuture;
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return;
case None:
- addCumulativeAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
initPendingAckStore();
- return completableFuture;
+ return;
case Error:
completableFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
- return completableFuture;
+ return;
case Close:
completableFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
- return completableFuture;
+ return;
default:
break;
-
}
}
- }
+ internalIndividualAcknowledgeMessage(txnID, positions, completableFuture);
+ });
+ return completableFuture;
+ }
+ private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
+ List<PositionImpl> positions,
+ CompletableFuture<Void> completableFuture) {
+ acceptQueue.add(() -> internalCumulativeAcknowledgeMessage(txnID, positions, new CompletableFuture<>())
Review comment:
same above `addIndividualAcknowledgeMessageRequest `
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -610,6 +586,37 @@ private void addAbortTxnRequest(TxnID txnId, Consumer consumer, long lowWaterMar
return abortFuture;
}
+ @Override
+ public CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer,
+ long lowWaterMark, boolean isInCacheRequest) {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ internalPinnedExecutor.submit(() -> {
Review comment:
use excute
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
##########
@@ -305,6 +307,8 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
.newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-stats-updater"));
this.authorizationService = new AuthorizationService(
pulsar.getConfiguration(), pulsar().getPulsarResources());
+ this.internalExecutorService = new ExecutorProvider(pulsar.getConfiguration().getNumIOThreads(),
Review comment:
now only transaction pending ack will use it, so when enable transaction we can init it
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -126,76 +132,54 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
completeHandleFuture();
}
});
+
+ internalPinnedExecutor = persistentSubscription
+ .getTopic()
+ .getBrokerService()
+ .getInternalExecutorService()
+ .getExecutor();
}
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
- });
- }
+ if (!checkIfClose()) {
+ this.pendingAckStoreFuture =
+ pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
+ this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
+ pendingAckStore.replayAsync(this,
+ persistentSubscription.getTopic().getBrokerService()
+ .getPulsar().getTransactionReplayExecutor());
+ }).exceptionally(e -> {
+ acceptQueue.clear();
+ changeToErrorState();
+ log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
+ return null;
+ });
}
}
}
private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
List<MutablePair<PositionImpl, Integer>> positions,
CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> individualAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
+ acceptQueue.add(() -> internalIndividualAcknowledgeMessage(txnID, positions, new CompletableFuture<>())
Review comment:
why we need to create a CompletableFuture? If `internalIndividualAcknowledgeMessage` can handle completableFuture, we don't need to create a new future
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -291,46 +275,51 @@ private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
return completableFuture;
}
- private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
- List<PositionImpl> positions,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> cumulativeAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
- }
-
@Override
- public CompletableFuture<Void> cumulativeAcknowledgeMessage(TxnID txnID,
- List<PositionImpl> positions,
+ public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
+ List<MutablePair<PositionImpl, Integer>> positions,
boolean isInCacheRequest) {
- if (!checkIfReady()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- synchronized (PendingAckHandleImpl.this) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ internalPinnedExecutor.submit(() -> {
Review comment:
use excute
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -394,56 +381,50 @@ private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
return completableFuture;
}
- private void addCommitTxnRequest(TxnID txnId, Map<String, Long> properties, long lowWaterMark,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> commitTxn(txnId, properties, lowWaterMark, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
- }
-
@Override
- public synchronized CompletableFuture<Void> commitTxn(TxnID txnID, Map<String, Long> properties,
- long lowWaterMark, boolean isInCacheRequest) {
- if (!checkIfReady()) {
- synchronized (PendingAckHandleImpl.this) {
- if (state == State.Initializing) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
- return completableFuture;
- } else if (state == State.None) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
- initPendingAckStore();
- return completableFuture;
- } else if (checkIfReady()) {
-
- } else {
- if (state == State.Error) {
- return FutureUtil.failedFuture(
+ public CompletableFuture<Void> cumulativeAcknowledgeMessage(TxnID txnID,
+ List<PositionImpl> positions,
+ boolean isInCacheRequest) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ internalPinnedExecutor.submit(() -> {
Review comment:
use excute
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -502,55 +483,50 @@ private void addCommitTxnRequest(TxnID txnId, Map<String, Long> properties, long
return commitFuture;
}
- private void addAbortTxnRequest(TxnID txnId, Consumer consumer, long lowWaterMark,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> abortTxn(txnId, consumer, lowWaterMark, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
- }
-
@Override
- public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer,
- long lowWaterMark, boolean isInCacheRequest) {
- if (!checkIfReady()) {
- synchronized (PendingAckHandleImpl.this) {
+ public CompletableFuture<Void> commitTxn(TxnID txnID, Map<String, Long> properties,
+ long lowWaterMark, boolean isInCacheRequest) {
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ internalPinnedExecutor.submit(() -> {
+ if (!checkIfReady()) {
if (state == State.Initializing) {
CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addAbortTxnRequest(txnId, consumer, lowWaterMark, completableFuture);
- return completableFuture;
+ addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
+ return;
} else if (state == State.None) {
CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addAbortTxnRequest(txnId, consumer, lowWaterMark, completableFuture);
+ addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
initPendingAckStore();
- return completableFuture;
+ return;
} else if (checkIfReady()) {
} else {
if (state == State.Error) {
- return FutureUtil.failedFuture(
+ commitFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
} else {
- return FutureUtil.failedFuture(
+ commitFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
}
+ return;
}
}
- }
-
+ internalCommitTxn(txnID, properties, lowWaterMark, commitFuture);
+ });
+ return commitFuture;
+ }
- if (!acceptQueue.isEmpty() && !isInCacheRequest) {
- synchronized (PendingAckHandleImpl.this) {
- if (!acceptQueue.isEmpty()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addAbortTxnRequest(txnId, consumer, lowWaterMark, completableFuture);
- return completableFuture;
- }
- }
- }
+ private void addAbortTxnRequest(TxnID txnId, Consumer consumer, long lowWaterMark,
+ CompletableFuture<Void> completableFuture) {
+ acceptQueue.add(() -> internalAbortTxn(txnId, consumer, lowWaterMark, new CompletableFuture<>())
Review comment:
same above `addIndividualAcknowledgeMessageRequest `
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -394,56 +381,50 @@ private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
return completableFuture;
}
- private void addCommitTxnRequest(TxnID txnId, Map<String, Long> properties, long lowWaterMark,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> commitTxn(txnId, properties, lowWaterMark, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
- }
-
@Override
- public synchronized CompletableFuture<Void> commitTxn(TxnID txnID, Map<String, Long> properties,
- long lowWaterMark, boolean isInCacheRequest) {
- if (!checkIfReady()) {
- synchronized (PendingAckHandleImpl.this) {
- if (state == State.Initializing) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
- return completableFuture;
- } else if (state == State.None) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
- initPendingAckStore();
- return completableFuture;
- } else if (checkIfReady()) {
-
- } else {
- if (state == State.Error) {
- return FutureUtil.failedFuture(
+ public CompletableFuture<Void> cumulativeAcknowledgeMessage(TxnID txnID,
+ List<PositionImpl> positions,
+ boolean isInCacheRequest) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ internalPinnedExecutor.submit(() -> {
+ if (!checkIfReady()) {
+ switch (state) {
+ case Initializing:
+ addCumulativeAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return;
+ case None:
+ addCumulativeAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ initPendingAckStore();
+ return;
+ case Error:
+ completableFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
- } else {
- return FutureUtil.failedFuture(
+ return;
+ case Close:
+ completableFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
- }
-
+ return;
+ default:
+ break;
}
}
- }
+ internalCumulativeAcknowledgeMessage(txnID, positions, completableFuture);
+ });
- if (!acceptQueue.isEmpty() && !isInCacheRequest) {
- synchronized (PendingAckHandleImpl.this) {
- if (!acceptQueue.isEmpty()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
- return completableFuture;
- }
- }
- }
+ return completableFuture;
+ }
- CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ private void addCommitTxnRequest(TxnID txnId, Map<String, Long> properties, long lowWaterMark,
Review comment:
same above `addIndividualAcknowledgeMessageRequest `
--
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
[GitHub] [pulsar] congbobo184 commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r791797179
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -500,58 +466,43 @@ private void addCommitTxnRequest(TxnID txnId, Map<String, Long> properties, long
return null;
});
}
- return commitFuture;
- }
-
- private void addAbortTxnRequest(TxnID txnId, Consumer consumer, long lowWaterMark,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> abortTxn(txnId, consumer, lowWaterMark, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
}
@Override
- public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer,
- long lowWaterMark, boolean isInCacheRequest) {
- if (!checkIfReady()) {
- synchronized (PendingAckHandleImpl.this) {
+ public CompletableFuture<Void> commitTxn(TxnID txnID, Map<String, Long> properties, long lowWaterMark) {
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ internalPinnedExecutor.execute(() -> {
+ if (!checkIfReady()) {
if (state == State.Initializing) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addAbortTxnRequest(txnId, consumer, lowWaterMark, completableFuture);
- return completableFuture;
+ addCommitTxnRequest(txnID, properties, lowWaterMark, commitFuture);
Review comment:
use switch case
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -130,73 +142,40 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- exceptionHandleFuture(e.getCause());
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
- });
- }
+ if (!checkIfClose()) {
+ this.pendingAckStoreFuture =
+ pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
+ this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
+ pendingAckStore.replayAsync(this,
+ (ScheduledExecutorService) internalPinnedExecutor);
+ }).exceptionally(e -> {
+ acceptQueue.clear();
+ changeToErrorState();
+ log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
+ exceptionHandleFuture(e.getCause());
Review comment:
when exceptionHandleFuture() we should handle request witch in cache and don't clear the queue. may can implement it on next pr.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -611,6 +562,34 @@ private void addAbortTxnRequest(TxnID txnId, Consumer consumer, long lowWaterMar
return abortFuture;
}
+ @Override
+ public CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer, long lowWaterMark) {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ internalPinnedExecutor.execute(() -> {
+ if (!checkIfReady()) {
+ if (state == State.Initializing) {
Review comment:
use switch case is better
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -309,11 +310,10 @@ public PulsarService(ServiceConfiguration config,
new DefaultThreadFactory("zk-cache-callback"));
if (config.isTransactionCoordinatorEnabled()) {
- this.transactionReplayExecutor = Executors.newScheduledThreadPool(
- config.getNumTransactionReplayThreadPoolSize(),
- new DefaultThreadFactory("transaction-replay"));
+ this.transactionExecutorProvider = new ExecutorProvider(this.getConfiguration().getNumIOThreads(),
Review comment:
use config.getNumTransactionReplayThreadPoolSize()
--
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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r776639384
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentSubscriptionTest.java
##########
@@ -318,7 +318,7 @@ public void testCanAcknowledgeAndAbortForTransaction() throws BrokerServiceExcep
assertTrue(e.getCause() instanceof TransactionConflictException);
assertEquals(e.getCause().getMessage(),"[persistent://prop/use/ns-abc/successTopic]" +
"[subscriptionName] Transaction:(1,2) try to cumulative batch ack position: " +
- "2:50 within range of current currentPosition: 1:100");
+ "2:50 within range of current currentPosition: 2:1");
Review comment:
Why the test can get passed before?
--
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
[GitHub] [pulsar] congbobo184 commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r784461740
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
##########
@@ -2324,13 +2324,6 @@
private String transactionPendingAckStoreProviderClassName =
"org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStoreProvider";
- @FieldContext(
- category = CATEGORY_TRANSACTION,
- doc = "Number of threads to use for pulsar transaction replay PendingAckStore or TransactionBuffer."
- + "Default is 5"
- )
- private int numTransactionReplayThreadPoolSize = Runtime.getRuntime().availableProcessors();
Review comment:
why delete this?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
##########
@@ -105,7 +105,7 @@ public TopicTransactionBuffer(PersistentTopic topic) {
this.takeSnapshotIntervalTime = topic.getBrokerService().getPulsar()
.getConfiguration().getTransactionBufferSnapshotMinTimeInMillis();
this.maxReadPosition = (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry();
- this.topic.getBrokerService().getPulsar().getTransactionReplayExecutor()
+ this.topic.getBrokerService().getPulsar().getTransactionExecutorProvider().getExecutor()
Review comment:
this.topic.getBrokerService().getPulsar().getTransactionExecutorProvider().getExecutor(this)
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -126,76 +133,51 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
completeHandleFuture();
}
});
+
+ internalPinnedExecutor = persistentSubscription
+ .getTopic()
+ .getBrokerService()
+ .getPulsar()
+ .getTransactionExecutorProvider()
+ .getExecutor();
Review comment:
.getExecutor(this)
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -610,6 +570,37 @@ private void addAbortTxnRequest(TxnID txnId, Consumer consumer, long lowWaterMar
return abortFuture;
}
+ @Override
+ public CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer,
+ long lowWaterMark, boolean isInCacheRequest) {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ internalPinnedExecutor.execute(() -> {
+ if (!checkIfReady()) {
+ if (state == State.Initializing) {
+ addAbortTxnRequest(txnId, consumer, lowWaterMark, abortFuture);
+ return;
+ } else if (state == State.None) {
+ addAbortTxnRequest(txnId, consumer, lowWaterMark, abortFuture);
+ initPendingAckStore();
+ return;
+ } else if (checkIfReady()) {
Review comment:
we don't need to checkIfReady, because change to ready state use the same thread with abortTxn. like commitTxn or individual ack or cumulative ack.
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/EmbeddedPulsarCluster.java
##########
@@ -130,7 +130,6 @@ private ServiceConfiguration getConf() {
conf.setManagedLedgerNumSchedulerThreads(1);
conf.setManagedLedgerNumWorkerThreads(1);
conf.setWebSocketNumIoThreads(1);
- conf.setNumTransactionReplayThreadPoolSize(1);
Review comment:
don't delete this
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -126,76 +133,51 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
completeHandleFuture();
}
});
+
+ internalPinnedExecutor = persistentSubscription
+ .getTopic()
+ .getBrokerService()
+ .getPulsar()
+ .getTransactionExecutorProvider()
+ .getExecutor();
}
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
- });
- }
+ if (!checkIfClose()) {
+ this.pendingAckStoreFuture =
+ pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
+ this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
+ pendingAckStore.replayAsync(this,
+ (ScheduledExecutorService) persistentSubscription.getTopic().getBrokerService()
+ .getPulsar().getTransactionExecutorProvider().getExecutor(this));
Review comment:
use local variable internalPinnedExecutor
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -126,76 +133,51 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
completeHandleFuture();
}
});
+
+ internalPinnedExecutor = persistentSubscription
+ .getTopic()
+ .getBrokerService()
+ .getPulsar()
+ .getTransactionExecutorProvider()
+ .getExecutor();
}
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
- });
- }
+ if (!checkIfClose()) {
+ this.pendingAckStoreFuture =
+ pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
+ this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
+ pendingAckStore.replayAsync(this,
+ (ScheduledExecutorService) persistentSubscription.getTopic().getBrokerService()
+ .getPulsar().getTransactionExecutorProvider().getExecutor(this));
+ }).exceptionally(e -> {
+ acceptQueue.clear();
+ changeToErrorState();
+ log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
+ return null;
+ });
}
}
}
private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
List<MutablePair<PositionImpl, Integer>> positions,
CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> individualAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
+ acceptQueue.add(() -> internalIndividualAcknowledgeMessage(txnID, positions, completableFuture));
}
- @Override
- public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions,
- boolean isInCacheRequest) {
-
- if (!checkIfReady()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- synchronized (PendingAckHandleImpl.this) {
- switch (state) {
- case Initializing:
- addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
- return completableFuture;
- case None:
- addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
- initPendingAckStore();
- return completableFuture;
- case Error:
- completableFuture.completeExceptionally(
- new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
- return completableFuture;
- case Close:
- completableFuture.completeExceptionally(
- new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
- return completableFuture;
- default:
- break;
- }
- }
- }
-
+ public CompletableFuture<Void> internalIndividualAcknowledgeMessage(TxnID txnID,
Review comment:
don't need return future
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -394,56 +375,46 @@ private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
return completableFuture;
}
- private void addCommitTxnRequest(TxnID txnId, Map<String, Long> properties, long lowWaterMark,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> commitTxn(txnId, properties, lowWaterMark, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
- }
-
@Override
- public synchronized CompletableFuture<Void> commitTxn(TxnID txnID, Map<String, Long> properties,
- long lowWaterMark, boolean isInCacheRequest) {
- if (!checkIfReady()) {
- synchronized (PendingAckHandleImpl.this) {
- if (state == State.Initializing) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
- return completableFuture;
- } else if (state == State.None) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
- initPendingAckStore();
- return completableFuture;
- } else if (checkIfReady()) {
-
- } else {
- if (state == State.Error) {
- return FutureUtil.failedFuture(
+ public CompletableFuture<Void> cumulativeAcknowledgeMessage(TxnID txnID,
+ List<PositionImpl> positions,
+ boolean isInCacheRequest) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ internalPinnedExecutor.execute(() -> {
+ if (!checkIfReady()) {
+ switch (state) {
+ case Initializing:
+ addCumulativeAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return;
+ case None:
+ addCumulativeAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ initPendingAckStore();
+ return;
+ case Error:
+ completableFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
- } else {
- return FutureUtil.failedFuture(
+ return;
+ case Close:
+ completableFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
- }
-
+ return;
+ default:
+ break;
}
}
- }
+ internalCumulativeAcknowledgeMessage(txnID, positions, completableFuture);
+ });
- if (!acceptQueue.isEmpty() && !isInCacheRequest) {
- synchronized (PendingAckHandleImpl.this) {
- if (!acceptQueue.isEmpty()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addCommitTxnRequest(txnID, properties, lowWaterMark, completableFuture);
- return completableFuture;
- }
- }
- }
+ return completableFuture;
+ }
- CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ private void addCommitTxnRequest(TxnID txnId, Map<String, Long> properties, long lowWaterMark,
+ CompletableFuture<Void> completableFuture) {
+ acceptQueue.add(() -> internalCommitTxn(txnId, properties, lowWaterMark, completableFuture));
+ }
+ private CompletableFuture<Void> internalCommitTxn(TxnID txnID, Map<String, Long> properties, long lowWaterMark,
Review comment:
don't need return future
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -126,76 +133,51 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
completeHandleFuture();
}
});
+
+ internalPinnedExecutor = persistentSubscription
+ .getTopic()
+ .getBrokerService()
+ .getPulsar()
+ .getTransactionExecutorProvider()
+ .getExecutor();
}
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
- });
- }
+ if (!checkIfClose()) {
+ this.pendingAckStoreFuture =
+ pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
+ this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
+ pendingAckStore.replayAsync(this,
+ (ScheduledExecutorService) persistentSubscription.getTopic().getBrokerService()
+ .getPulsar().getTransactionExecutorProvider().getExecutor(this));
+ }).exceptionally(e -> {
+ acceptQueue.clear();
+ changeToErrorState();
+ log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
+ return null;
+ });
}
}
}
private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
List<MutablePair<PositionImpl, Integer>> positions,
CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> individualAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
+ acceptQueue.add(() -> internalIndividualAcknowledgeMessage(txnID, positions, completableFuture));
}
- @Override
- public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions,
- boolean isInCacheRequest) {
-
- if (!checkIfReady()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- synchronized (PendingAckHandleImpl.this) {
- switch (state) {
- case Initializing:
- addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
- return completableFuture;
- case None:
- addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
- initPendingAckStore();
- return completableFuture;
- case Error:
- completableFuture.completeExceptionally(
- new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
- return completableFuture;
- case Close:
- completableFuture.completeExceptionally(
- new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
- return completableFuture;
- default:
- break;
- }
- }
- }
-
+ public CompletableFuture<Void> internalIndividualAcknowledgeMessage(TxnID txnID,
+ List<MutablePair<PositionImpl, Integer>>
+ positions,
Review comment:
code style
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -291,46 +273,47 @@ private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
return completableFuture;
}
- private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
- List<PositionImpl> positions,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> cumulativeAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
- }
-
@Override
- public CompletableFuture<Void> cumulativeAcknowledgeMessage(TxnID txnID,
- List<PositionImpl> positions,
+ public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
+ List<MutablePair<PositionImpl, Integer>> positions,
boolean isInCacheRequest) {
- if (!checkIfReady()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- synchronized (PendingAckHandleImpl.this) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ internalPinnedExecutor.execute(() -> {
+ if (!checkIfReady()) {
switch (state) {
case Initializing:
- addCumulativeAcknowledgeMessageRequest(txnID, positions, completableFuture);
- return completableFuture;
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return;
case None:
- addCumulativeAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
initPendingAckStore();
- return completableFuture;
+ return;
case Error:
completableFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
- return completableFuture;
+ return;
case Close:
completableFuture.completeExceptionally(
new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
- return completableFuture;
+ return;
default:
break;
-
}
}
- }
+ internalIndividualAcknowledgeMessage(txnID, positions, completableFuture);
+ });
+ return completableFuture;
+ }
+ private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
+ List<PositionImpl> positions,
+ CompletableFuture<Void> completableFuture) {
+ acceptQueue.add(() -> internalCumulativeAcknowledgeMessage(txnID, positions, completableFuture));
+ }
+
+ public CompletableFuture<Void> internalCumulativeAcknowledgeMessage(TxnID txnID,
Review comment:
don't need return future
--
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
[GitHub] [pulsar] liangyepianzhou commented on pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#issuecomment-1002646849
/pulsarbot-run-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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r776568821
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckPersistentTest.java
##########
@@ -344,4 +346,42 @@ private void testDeleteTopicThenDeletePendingAckManagedLedger() throws Exception
assertFalse(topics.contains(MLPendingAckStore.getTransactionPendingAckStoreSuffix(topic, subName2)));
assertFalse(topics.contains(topic));
}
+
+ @Test
+ public void testTxnCumulativeAckInOrder() throws Exception{
Review comment:
Without this PR, the test still can get passed, could you please help check if this is a correct test for cover the new changes?
--
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
[GitHub] [pulsar] liangyepianzhou commented on pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#issuecomment-1002648178
/pulsarbot run-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
[GitHub] [pulsar] liangyepianzhou commented on pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#issuecomment-1002895597
/pulsarbot run-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
[GitHub] [pulsar] congbobo184 commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r786562199
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -130,73 +142,39 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- exceptionHandleFuture(e.getCause());
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
- });
- }
+ if (!checkIfClose()) {
+ this.pendingAckStoreFuture =
+ pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
+ this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
+ pendingAckStore.replayAsync(this,
+ (ScheduledExecutorService) internalPinnedExecutor);
+ }).exceptionally(e -> {
+ acceptQueue.clear();
+ changeToErrorState();
+ log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
+ return null;
+ });
}
}
}
private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
List<MutablePair<PositionImpl, Integer>> positions,
CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> individualAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
+ acceptQueue.add(() -> internalIndividualAcknowledgeMessage(txnID, positions, completableFuture));
}
- @Override
- public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions,
- boolean isInCacheRequest) {
-
- if (!checkIfReady()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- synchronized (PendingAckHandleImpl.this) {
- switch (state) {
- case Initializing:
- addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
- return completableFuture;
- case None:
- addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
- initPendingAckStore();
- return completableFuture;
- case Error:
- completableFuture.completeExceptionally(
- new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
- return completableFuture;
- case Close:
- completableFuture.completeExceptionally(
- new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
- return completableFuture;
- default:
- break;
- }
- }
- }
-
+ public void internalIndividualAcknowledgeMessage(TxnID txnID, List<MutablePair<PositionImpl, Integer>> positions,
+ CompletableFuture<Void> completableFuture) {
if (txnID == null) {
- return FutureUtil.failedFuture(new NotAllowedException("TransactionID can not be null."));
+ FutureUtil.failedFuture(new NotAllowedException("TransactionID can not be null."));
Review comment:
completableFuture.completeExceptionally(new NotAllowedException("TransactionID can not be null."));
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -289,67 +267,68 @@ private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
completableFuture.completeExceptionally(e);
return null;
});
- return completableFuture;
- }
-
- private void addCumulativeAcknowledgeMessageRequest(TxnID txnID,
- List<PositionImpl> positions,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> cumulativeAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
}
@Override
- public CompletableFuture<Void> cumulativeAcknowledgeMessage(TxnID txnID,
- List<PositionImpl> positions,
+ public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
+ List<MutablePair<PositionImpl, Integer>> positions,
boolean isInCacheRequest) {
Review comment:
delete isInCacheRequest
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -130,73 +142,39 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- exceptionHandleFuture(e.getCause());
Review comment:
dont delete this
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -130,73 +142,39 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- exceptionHandleFuture(e.getCause());
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
- });
- }
+ if (!checkIfClose()) {
+ this.pendingAckStoreFuture =
+ pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
+ this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
+ pendingAckStore.replayAsync(this,
+ (ScheduledExecutorService) internalPinnedExecutor);
+ }).exceptionally(e -> {
+ acceptQueue.clear();
+ changeToErrorState();
+ log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
+ return null;
+ });
}
}
}
private void addIndividualAcknowledgeMessageRequest(TxnID txnID,
List<MutablePair<PositionImpl, Integer>> positions,
CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> individualAcknowledgeMessage(txnID, positions, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
+ acceptQueue.add(() -> internalIndividualAcknowledgeMessage(txnID, positions, completableFuture));
}
- @Override
- public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions,
- boolean isInCacheRequest) {
-
- if (!checkIfReady()) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- synchronized (PendingAckHandleImpl.this) {
- switch (state) {
- case Initializing:
- addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
- return completableFuture;
- case None:
- addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
- initPendingAckStore();
- return completableFuture;
- case Error:
- completableFuture.completeExceptionally(
- new ServiceUnitNotReadyException("PendingAckHandle not replay error!"));
- return completableFuture;
- case Close:
- completableFuture.completeExceptionally(
- new ServiceUnitNotReadyException("PendingAckHandle have been closed!"));
- return completableFuture;
- default:
- break;
- }
- }
- }
-
+ public void internalIndividualAcknowledgeMessage(TxnID txnID, List<MutablePair<PositionImpl, Integer>> positions,
+ CompletableFuture<Void> completableFuture) {
if (txnID == null) {
- return FutureUtil.failedFuture(new NotAllowedException("TransactionID can not be null."));
+ FutureUtil.failedFuture(new NotAllowedException("TransactionID can not be null."));
+ return;
+
}
if (positions == null) {
- return FutureUtil.failedFuture(new NotAllowedException("Positions can not be null."));
+ FutureUtil.failedFuture(new NotAllowedException("Positions can not be null."));
Review comment:
completableFuture.completeExceptionally(new NotAllowedException("Positions can not be null."));
--
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
[GitHub] [pulsar] liangyepianzhou commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r776570542
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
##########
@@ -105,7 +105,7 @@ public TopicTransactionBuffer(PersistentTopic topic) {
this.takeSnapshotIntervalTime = topic.getBrokerService().getPulsar()
.getConfiguration().getTransactionBufferSnapshotMinTimeInMillis();
this.maxReadPosition = (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry();
- this.topic.getBrokerService().getPulsar().getTransactionReplayExecutor()
+ this.topic.getBrokerService().getPulsar().getInternalExecutorService().getExecutor()
Review comment:
The name is changed here, and the thread pool is used uniformly.
--
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
[GitHub] [pulsar] liangyepianzhou removed a comment on pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
liangyepianzhou removed a comment on pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#issuecomment-1002646849
/pulsarbot-run-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
[GitHub] [pulsar] congbobo184 commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r791797179
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -500,58 +466,43 @@ private void addCommitTxnRequest(TxnID txnId, Map<String, Long> properties, long
return null;
});
}
- return commitFuture;
- }
-
- private void addAbortTxnRequest(TxnID txnId, Consumer consumer, long lowWaterMark,
- CompletableFuture<Void> completableFuture) {
- acceptQueue.add(() -> abortTxn(txnId, consumer, lowWaterMark, true).thenAccept(v ->
- completableFuture.complete(null)).exceptionally(e -> {
- completableFuture.completeExceptionally(e);
- return null;
- }));
}
@Override
- public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer,
- long lowWaterMark, boolean isInCacheRequest) {
- if (!checkIfReady()) {
- synchronized (PendingAckHandleImpl.this) {
+ public CompletableFuture<Void> commitTxn(TxnID txnID, Map<String, Long> properties, long lowWaterMark) {
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ internalPinnedExecutor.execute(() -> {
+ if (!checkIfReady()) {
if (state == State.Initializing) {
- CompletableFuture<Void> completableFuture = new CompletableFuture<>();
- addAbortTxnRequest(txnId, consumer, lowWaterMark, completableFuture);
- return completableFuture;
+ addCommitTxnRequest(txnID, properties, lowWaterMark, commitFuture);
Review comment:
use switch case
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -130,73 +142,40 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
private void initPendingAckStore() {
if (changeToInitializingState()) {
- synchronized (PendingAckHandleImpl.this) {
- if (!checkIfClose()) {
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- acceptQueue.clear();
- changeToErrorState();
- exceptionHandleFuture(e.getCause());
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
- });
- }
+ if (!checkIfClose()) {
+ this.pendingAckStoreFuture =
+ pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
+ this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
+ pendingAckStore.replayAsync(this,
+ (ScheduledExecutorService) internalPinnedExecutor);
+ }).exceptionally(e -> {
+ acceptQueue.clear();
+ changeToErrorState();
+ log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
+ exceptionHandleFuture(e.getCause());
Review comment:
when exceptionHandleFuture() we should handle request witch in cache and don't clear the queue. may can implement it on next pr.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -611,6 +562,34 @@ private void addAbortTxnRequest(TxnID txnId, Consumer consumer, long lowWaterMar
return abortFuture;
}
+ @Override
+ public CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer, long lowWaterMark) {
+ CompletableFuture<Void> abortFuture = new CompletableFuture<>();
+ internalPinnedExecutor.execute(() -> {
+ if (!checkIfReady()) {
+ if (state == State.Initializing) {
Review comment:
use switch case is better
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -309,11 +310,10 @@ public PulsarService(ServiceConfiguration config,
new DefaultThreadFactory("zk-cache-callback"));
if (config.isTransactionCoordinatorEnabled()) {
- this.transactionReplayExecutor = Executors.newScheduledThreadPool(
- config.getNumTransactionReplayThreadPoolSize(),
- new DefaultThreadFactory("transaction-replay"));
+ this.transactionExecutorProvider = new ExecutorProvider(this.getConfiguration().getNumIOThreads(),
Review comment:
use config.getNumTransactionReplayThreadPoolSize()
--
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
[GitHub] [pulsar] congbobo184 merged pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
congbobo184 merged pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481
--
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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r776169414
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
##########
@@ -305,6 +307,12 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
.newSingleThreadScheduledExecutor(new DefaultThreadFactory("pulsar-stats-updater"));
this.authorizationService = new AuthorizationService(
pulsar.getConfiguration(), pulsar().getPulsarResources());
+ if (pulsar.getConfiguration().isTransactionCoordinatorEnabled()) {
Review comment:
We have added a transaction thread pool `transactionReplayExecutor`, it's better to consider using one thread pool for transaction buffer and pending ack state.
For example:
Using a unified transaction thread pool.
One topic pinned one thread to handle the transaction buffer operations and pending ack operations.
--
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
[GitHub] [pulsar] codelipenghui commented on a change in pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#discussion_r776565752
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -308,11 +309,10 @@ public PulsarService(ServiceConfiguration config,
new DefaultThreadFactory("zk-cache-callback"));
if (config.isTransactionCoordinatorEnabled()) {
- this.transactionReplayExecutor = Executors.newScheduledThreadPool(
- config.getNumTransactionReplayThreadPoolSize(),
- new DefaultThreadFactory("transaction-replay"));
+ this.internalExecutorService = new ExecutorProvider(this.getConfiguration().getNumIOThreads(),
+ "pulsar-executorProvider-internal");
Review comment:
```suggestion
"pulsar-transaction-executor");
```
Currently, no other domains use this executor provider, we can just give a name that is used by transactions.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -252,7 +253,7 @@
private PulsarResources pulsarResources;
private TransactionPendingAckStoreProvider transactionPendingAckStoreProvider;
- private final ScheduledExecutorService transactionReplayExecutor;
+ private final ExecutorProvider internalExecutorService;
Review comment:
```suggestion
private final ExecutorProvider transactionExecutorProvider;
```
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
##########
@@ -105,7 +105,7 @@ public TopicTransactionBuffer(PersistentTopic topic) {
this.takeSnapshotIntervalTime = topic.getBrokerService().getPulsar()
.getConfiguration().getTransactionBufferSnapshotMinTimeInMillis();
this.maxReadPosition = (PositionImpl) topic.getManagedLedger().getLastConfirmedEntry();
- this.topic.getBrokerService().getPulsar().getTransactionReplayExecutor()
+ this.topic.getBrokerService().getPulsar().getInternalExecutorService().getExecutor()
Review comment:
We don't need a pinned executor here right?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
##########
@@ -252,7 +253,7 @@
private PulsarResources pulsarResources;
private TransactionPendingAckStoreProvider transactionPendingAckStoreProvider;
- private final ScheduledExecutorService transactionReplayExecutor;
+ private final ExecutorProvider internalExecutorService;
Review comment:
And we can remove `numTransactionReplayThreadPoolSize` in the ServiceConfiguration.
--
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
[GitHub] [pulsar] liangyepianzhou commented on pull request #13481: [Transaction] PendingAckHandleImpl handle isInCacheRequest
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on pull request #13481:
URL: https://github.com/apache/pulsar/pull/13481#issuecomment-1014998463
/pulsarbot run-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