Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

fix no response to client when handleSubscribe because PendingAckHandleImpl init fail #13655

Merged
merged 5 commits into from
Jan 14, 2022
Merged

fix no response to client when handleSubscribe because PendingAckHandleImpl init fail #13655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
265718c
fix no response to client when handleSubscribe because PendingAckHand…
wenbingshen Jan 6, 2022
e0537a9
fix checkstyle
wenbingshen Jan 6, 2022
7e41916
fix checkstyle
wenbingshen Jan 6, 2022
de15c63
PendingAckReplay also may be failed
wenbingshen Jan 7, 2022
488c191
Merge remote-tracking branch 'srp/master' into fix_Subscribe_response
wenbingshen Jan 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions ...rc/main/java/org/apache/pulsar/broker/transaction/pendingack/PendingAckReplyCallBack.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,4 +36,10 @@ public interface PendingAckReplyCallBack {
* @param pendingAckMetadataEntry {@link PendingAckMetadataEntry} the metadata entry of pending ack
*/
void handleMetadataEntry(PendingAckMetadataEntry pendingAckMetadataEntry);

/**
* Pending ack replay failed callback for pending ack store.
*/
void replayFailed(Throwable t);

}
7 changes: 7 additions & 0 deletions .../java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckReplyCallBack.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,13 @@ public void replayComplete() {
pendingAckHandle.handleCacheRequest();
}

@Override
public void replayFailed(Throwable t) {
synchronized (pendingAckHandle) {
pendingAckHandle.exceptionHandleFuture(t);
}
}

@Override
public void handleMetadataEntry(PendingAckMetadataEntry pendingAckMetadataEntry) {
TxnID txnID = new TxnID(pendingAckMetadataEntry.getTxnidMostBits(),
Expand Down
3 changes: 3 additions & 0 deletions ...src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,6 +303,8 @@ class PendingAckReplay implements Runnable {
public void run() {
try {
if (cursor.isClosed()) {
pendingAckReplyCallBack.replayFailed(new ManagedLedgerException
.CursorAlreadyClosedException("MLPendingAckStore cursor have been closed."));
log.warn("[{}] MLPendingAckStore cursor have been closed, close replay thread.",
cursor.getManagedLedger().getName());
return;
Expand Down Expand Up @@ -350,6 +352,7 @@ public void run() {
}
}
} catch (Exception e) {
pendingAckReplyCallBack.replayFailed(e);
log.error("[{}] Pending ack recover fail!", subManagedCursor.getManagedLedger().getName(), e);
return;
}
Expand Down
7 changes: 7 additions & 0 deletions .../main/java/org/apache/pulsar/broker/transaction/pendingack/impl/PendingAckHandleImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,7 @@ private void initPendingAckStore() {
}).exceptionally(e -> {
acceptQueue.clear();
changeToErrorState();
exceptionHandleFuture(e.getCause());
log.error("PendingAckHandleImpl init fail! TopicName : {}, SubName: {}", topicName, subName, e);
return null;
});
Expand Down Expand Up @@ -889,6 +890,12 @@ public synchronized void completeHandleFuture() {
}
}

public synchronized void exceptionHandleFuture(Throwable t) {
if (!this.pendingAckHandleCompletableFuture.isDone()) {
this.pendingAckHandleCompletableFuture.completeExceptionally(t);
}
}

@Override
public TransactionInPendingAckStats getTransactionInPendingAckStats(TxnID txnID) {
TransactionInPendingAckStats transactionInPendingAckStats = new TransactionInPendingAckStats();
Expand Down