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);
}
```
This method is not a thread-safe operation, currently, only this method will
change the state to close. I'm afraid if multi-threads call this method (maybe
this will not happen, but it's possible) the release operation will execute
multi times.
For example, thread-0, and thread-1 call this method at the same time, the
check `checkIfClose()` will be true, and the release operation will execute
twice.
--
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]
