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/06/25 07:49:44 UTC
[GitHub] [pulsar] congbobo184 opened a new pull request #11091: [Transaction] Transaction pending ack lazy init.
congbobo184 opened a new pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091
## Motivation
now, in `broker.conf` `transactionCoordinatorEnabled=true` MLPendingAck will init manageLedger, some ack will not use transaction, so don't need to init manageLedger. When this sub use transaction, we can lazy init `PendingAckHandle`.
## implement
When this sub use transaction, we can lazy init `PendingAckHandle`.
### Verifying this change
Add the tests for it
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): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701340295
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java
##########
@@ -44,16 +44,19 @@ public MLPendingAckReplyCallBack(PendingAckHandleImpl pendingAckHandle) {
@Override
public void replayComplete() {
- log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
-
- if (pendingAckHandle.changeToReadyState()) {
- pendingAckHandle.completeHandleFuture();
+ synchronized (pendingAckHandle) {
log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
- } else {
- log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+
+ if (pendingAckHandle.changeToReadyState()) {
+ pendingAckHandle.completeHandleFuture();
+ pendingAckHandle.handleCacheRequest();
+ log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
+ pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+ } else {
+ log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",
Review comment:
If the pending ack handle relay failed, how to reply again?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ private void initPendingAckStore() {
+ if (changeToInitializingState()) {
+ synchronized (PendingAckHandleImpl.this) {
Review comment:
I'm not sure why need this lock, it seems that only one thread will perform the replay operation because only one thread could change the state successfully.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -300,11 +402,36 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
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) {
+ long lowWaterMark, boolean isInCacheRequest) {
Review comment:
It seems that the param `isInCacheRequest` isn't used, do we need to add this param?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java
##########
@@ -44,16 +44,19 @@ public MLPendingAckReplyCallBack(PendingAckHandleImpl pendingAckHandle) {
@Override
public void replayComplete() {
- log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
-
- if (pendingAckHandle.changeToReadyState()) {
- pendingAckHandle.completeHandleFuture();
+ synchronized (pendingAckHandle) {
log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
- } else {
- log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+
+ if (pendingAckHandle.changeToReadyState()) {
+ pendingAckHandle.completeHandleFuture();
+ pendingAckHandle.handleCacheRequest();
+ log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
Review comment:
It seems that this log is the same as the above.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -732,7 +888,14 @@ public TransactionInPendingAckStats getTransactionInPendingAckStats(TxnID txnID)
@Override
public CompletableFuture<Void> close() {
- return this.pendingAckStoreFuture.thenAccept(PendingAckStore::closeAsync);
+ changeToCloseState();
+ synchronized (PendingAckHandleImpl.this) {
Review comment:
Do we need this lock?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+ if (!isInCacheRequest) {
+ if (!checkIfReady()) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
Review comment:
It seems that this check is repeated many times, maybe we could add a method to make this check.
such as
```
public boolean needCacheRequest() {
if (state == State.Initializing ||) {
return true;
} else if (state == State.None) {
initPendingAckStore();
return true;
} else if (checkIfReady()) {
// do nothing
} else {
return FutureUtil.failedFuture(
new ServiceUnitNotReadyException("PendingAckHandle replay failed!"));
}
}
```
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+ if (!isInCacheRequest) {
+ if (!checkIfReady()) {
+ synchronized (PendingAckHandleImpl.this) {
Review comment:
Does this lock is used to make sure the state doesn't be changed?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -367,11 +494,38 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
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) {
+ public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer,
+ long lowWaterMark, boolean isInCacheRequest) {
Review comment:
It seems that the param isInCacheRequest isn't used, do we need to add this param?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -300,11 +402,36 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
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) {
+ long lowWaterMark, boolean isInCacheRequest) {
if (!checkIfReady()) {
- return FutureUtil.failedFuture(new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));
+ 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 {
+ return FutureUtil.failedFuture(
+ new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));
Review comment:
Does this error log is right? It seems that the state is `Error` or `Close`.
--
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] hangc0276 commented on pull request #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#issuecomment-893229316
@merlimat @codelipenghui @eolivelli Would you please help review this PR ?
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701173839
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+
+ if (!checkIfReady() && !isInCacheRequest) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return completableFuture;
+ } else if (state == State.None) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ initPendingAckStore();
+ return completableFuture;
+ } else if (checkIfReady()) {
+
+ } else {
+ return FutureUtil.failedFuture(
+ new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));
+ }
+ }
+ }
+
+ if (!acceptQueue.isEmpty() && !isInCacheRequest) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (!acceptQueue.isEmpty()) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
Review comment:
We need to make sure that Ack is in order
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java
##########
@@ -44,16 +44,19 @@ public MLPendingAckReplyCallBack(PendingAckHandleImpl pendingAckHandle) {
@Override
public void replayComplete() {
- log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
-
- if (pendingAckHandle.changeToReadyState()) {
- pendingAckHandle.completeHandleFuture();
+ synchronized (pendingAckHandle) {
log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
- } else {
- log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+
+ if (pendingAckHandle.changeToReadyState()) {
+ pendingAckHandle.completeHandleFuture();
+ pendingAckHandle.handleCacheRequest();
Review comment:
In order to ensure the order of ack, I think we should continue to acquire this lock
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+
+ if (!checkIfReady() && !isInCacheRequest) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return completableFuture;
+ } else if (state == State.None) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ initPendingAckStore();
+ return completableFuture;
+ } else if (checkIfReady()) {
+
Review comment:
When ready, we should let the code continue to execute downward
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 merged pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701544865
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -732,7 +888,14 @@ public TransactionInPendingAckStats getTransactionInPendingAckStats(TxnID txnID)
@Override
public CompletableFuture<Void> close() {
- return this.pendingAckStoreFuture.thenAccept(PendingAckStore::closeAsync);
+ changeToCloseState();
+ synchronized (PendingAckHandleImpl.this) {
Review comment:
same with init
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701540231
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ private void initPendingAckStore() {
+ if (changeToInitializingState()) {
+ synchronized (PendingAckHandleImpl.this) {
Review comment:
if change Initializing state success, in this time, we can change the state to closed, when change state to closed, we don't need to do replay op.
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701545525
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+ if (!isInCacheRequest) {
+ if (!checkIfReady()) {
+ synchronized (PendingAckHandleImpl.this) {
Review comment:
yes
--
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] hangc0276 commented on pull request #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#issuecomment-893987287
move to 2.8.2.
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701526020
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java
##########
@@ -44,16 +44,19 @@ public MLPendingAckReplyCallBack(PendingAckHandleImpl pendingAckHandle) {
@Override
public void replayComplete() {
- log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
-
- if (pendingAckHandle.changeToReadyState()) {
- pendingAckHandle.completeHandleFuture();
+ synchronized (pendingAckHandle) {
log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
- } else {
- log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+
+ if (pendingAckHandle.changeToReadyState()) {
+ pendingAckHandle.completeHandleFuture();
+ pendingAckHandle.handleCacheRequest();
+ log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
+ pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+ } else {
+ log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",
Review comment:
if change ready state fail, represent that the pending ack has already close
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701540231
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ private void initPendingAckStore() {
+ if (changeToInitializingState()) {
+ synchronized (PendingAckHandleImpl.this) {
Review comment:
if change Initializing state success, in this time, we can change the state to close, when change test to close, we don't need to do replay op.
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701544338
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -367,11 +494,38 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
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) {
+ public synchronized CompletableFuture<Void> abortTxn(TxnID txnId, Consumer consumer,
+ long lowWaterMark, boolean isInCacheRequest) {
Review comment:
good catch, we should add the abort request `!isInCacheRequest` in to a queue.
--
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] eolivelli commented on a change in pull request #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r661538840
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,95 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Accept> acceptQueue = new LinkedBlockingDeque<>();
+
+ interface Accept {
Review comment:
why not "java.lang.Runnable" ?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleState.java
##########
@@ -32,14 +32,15 @@
None,
Initializing,
Ready,
+ Error,
Review comment:
how can we recover from "Error" ?
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701545393
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -300,11 +402,36 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
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) {
+ long lowWaterMark, boolean isInCacheRequest) {
if (!checkIfReady()) {
- return FutureUtil.failedFuture(new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));
+ 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 {
+ return FutureUtil.failedFuture(
+ new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));
Review comment:
good catch
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701544619
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java
##########
@@ -44,16 +44,19 @@ public MLPendingAckReplyCallBack(PendingAckHandleImpl pendingAckHandle) {
@Override
public void replayComplete() {
- log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
-
- if (pendingAckHandle.changeToReadyState()) {
- pendingAckHandle.completeHandleFuture();
+ synchronized (pendingAckHandle) {
log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
- } else {
- log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+
+ if (pendingAckHandle.changeToReadyState()) {
+ pendingAckHandle.completeHandleFuture();
+ pendingAckHandle.handleCacheRequest();
+ log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
Review comment:
good catch
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701546506
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+ if (!isInCacheRequest) {
+ if (!checkIfReady()) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
Review comment:
in different state, commit, ack and abort will do different op, so this method return what?
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701189463
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+
+ if (!checkIfReady() && !isInCacheRequest) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return completableFuture;
+ } else if (state == State.None) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ initPendingAckStore();
+ return completableFuture;
+ } else if (checkIfReady()) {
+
+ } else {
+ return FutureUtil.failedFuture(
+ new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));
+ }
+ }
+ }
+
+ if (!acceptQueue.isEmpty() && !isInCacheRequest) {
Review comment:
optimized, please review again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701558104
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+ if (!isInCacheRequest) {
+ if (!checkIfReady()) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
Review comment:
If the return value is true, we could cache the request in the queue.
--
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] hangc0276 commented on pull request #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#issuecomment-893229316
@merlimat @codelipenghui @eolivelli Would you please help review this PR ?
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701137506
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+
+ if (!checkIfReady() && !isInCacheRequest) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return completableFuture;
+ } else if (state == State.None) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ initPendingAckStore();
+ return completableFuture;
+ } else if (checkIfReady()) {
+
+ } else {
+ return FutureUtil.failedFuture(
+ new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));
+ }
+ }
+ }
+
+ if (!acceptQueue.isEmpty() && !isInCacheRequest) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (!acceptQueue.isEmpty()) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
Review comment:
Why we can't add to the queue if the queue is empty?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+
+ if (!checkIfReady() && !isInCacheRequest) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return completableFuture;
+ } else if (state == State.None) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ initPendingAckStore();
+ return completableFuture;
+ } else if (checkIfReady()) {
+
Review comment:
Miss something here?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java
##########
@@ -44,16 +44,19 @@ public MLPendingAckReplyCallBack(PendingAckHandleImpl pendingAckHandle) {
@Override
public void replayComplete() {
- log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
-
- if (pendingAckHandle.changeToReadyState()) {
- pendingAckHandle.completeHandleFuture();
+ synchronized (pendingAckHandle) {
log.info("Topic name : [{}], SubName : [{}] pending ack state reply success!",
pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
- } else {
- log.error("Topic name : [{}], SubName : [{}] pending ack state reply fail!",
- pendingAckHandle.getTopicName(), pendingAckHandle.getSubName());
+
+ if (pendingAckHandle.changeToReadyState()) {
+ pendingAckHandle.completeHandleFuture();
+ pendingAckHandle.handleCacheRequest();
Review comment:
We will hold the lock for a long time here.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,96 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Runnable> acceptQueue = new LinkedBlockingDeque<>();
+
public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
super(State.None);
this.topicName = persistentSubscription.getTopicName();
this.subName = persistentSubscription.getName();
this.persistentSubscription = persistentSubscription;
- TransactionPendingAckStoreProvider pendingAckStoreProvider =
- ((PersistentTopic) this.persistentSubscription.getTopic())
+ this.pendingAckStoreProvider = ((PersistentTopic) this.persistentSubscription.getTopic())
.getBrokerService().getPulsar().getTransactionPendingAckStoreProvider();
- this.pendingAckStoreFuture =
- pendingAckStoreProvider.newPendingAckStore(persistentSubscription);
-
- this.pendingAckStoreFuture.thenAccept(pendingAckStore -> {
- changeToInitializingState();
- pendingAckStore.replayAsync(this,
- ((PersistentTopic) persistentSubscription.getTopic()).getBrokerService()
- .getPulsar().getTransactionReplayExecutor());
- }).exceptionally(e -> {
- log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
- return null;
+ pendingAckStoreProvider.checkInitializedBefore(persistentSubscription).thenAccept(init -> {
+ if (init) {
+ initPendingAckStore();
+ } else {
+ completeHandleFuture();
+ }
});
}
+ 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;
+ });
+ }
+ }
+ }
+ }
+
+ 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;
+ }));
+ }
+
@Override
public CompletableFuture<Void> individualAcknowledgeMessage(TxnID txnID,
- List<MutablePair<PositionImpl, Integer>> positions) {
+ List<MutablePair<PositionImpl, Integer>> positions,
+ boolean isInCacheRequest) {
+
+ if (!checkIfReady() && !isInCacheRequest) {
+ synchronized (PendingAckHandleImpl.this) {
+ if (state == State.Initializing) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ return completableFuture;
+ } else if (state == State.None) {
+ CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+ addIndividualAcknowledgeMessageRequest(txnID, positions, completableFuture);
+ initPendingAckStore();
+ return completableFuture;
+ } else if (checkIfReady()) {
+
+ } else {
+ return FutureUtil.failedFuture(
+ new ServiceUnitNotReadyException("PendingAckHandle not replay complete!"));
+ }
+ }
+ }
+
+ if (!acceptQueue.isEmpty() && !isInCacheRequest) {
Review comment:
Is it able to refine the code here? looks like the above part has check isInCacheRequest but here check it again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on pull request #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#issuecomment-912211479
@gaoran10 thanks for you review, fix some comment, please review again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a change in pull request #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r661987304
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleState.java
##########
@@ -32,14 +32,15 @@
None,
Initializing,
Ready,
+ Error,
Review comment:
Different errors should be handled differently, but now I don’t know what error will happen, so just to be on the safe side, when error happen can't recover. unless you unload this topic
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -101,36 +103,95 @@
private final PersistentSubscription persistentSubscription;
- private final CompletableFuture<PendingAckStore> pendingAckStoreFuture;
+ private CompletableFuture<PendingAckStore> pendingAckStoreFuture;
private final CompletableFuture<PendingAckHandle> pendingAckHandleCompletableFuture = new CompletableFuture<>();
+ private final TransactionPendingAckStoreProvider pendingAckStoreProvider;
+
+ private final BlockingQueue<Accept> acceptQueue = new LinkedBlockingDeque<>();
+
+ interface Accept {
Review comment:
good idea!
--
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 #11091: [Transaction] Transaction pending ack lazy init.
Posted by GitBox <gi...@apache.org>.
congbobo184 commented on a change in pull request #11091:
URL: https://github.com/apache/pulsar/pull/11091#discussion_r701544289
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
##########
@@ -300,11 +402,36 @@ public PendingAckHandleImpl(PersistentSubscription persistentSubscription) {
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) {
+ long lowWaterMark, boolean isInCacheRequest) {
Review comment:
good catch, we should add the commit request `!isInCacheRequest` in to a queue.
--
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