gaoran10 commented on code in PR #15015:
URL: https://github.com/apache/pulsar/pull/15015#discussion_r852158043
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -465,17 +467,22 @@ public CompletableFuture<Void> purgeTxns(List<Long>
dataLedgers) {
@Override
public CompletableFuture<Void> clearSnapshot() {
- return this.takeSnapshotWriter.thenCompose(writer -> {
+ return this.takeSnapshotWriter.getFuture().thenCompose(writer -> {
TransactionBufferSnapshot snapshot = new
TransactionBufferSnapshot();
snapshot.setTopicName(topic.getName());
return writer.deleteAsync(snapshot);
}).thenCompose(__ -> CompletableFuture.completedFuture(null));
}
@Override
- public CompletableFuture<Void> closeAsync() {
+ public synchronized CompletableFuture<Void> closeAsync() {
Review Comment:
```
@Override
public synchronized CompletableFuture<Void> closeAsync() {
if (!checkIfClose()) {
// make sure the snapshot counted writer only release one time.
this.topic.getBrokerService().getPulsar().getTransactionBufferSnapshotService()
.releaseReferenceWriter(this.takeSnapshotWriter);
}
changeToCloseState();
return CompletableFuture.completedFuture(null);
}
```
I'm afraid this method is not a thread-safe operation. If multi-threads call
this method (maybe this will not happen, but it's possible) the release
operation will execute multi times.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
