codelipenghui commented on code in PR #21406:
URL: https://github.com/apache/pulsar/pull/21406#discussion_r1387450051
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -255,6 +231,42 @@ public long getCommittedTxnCount() {
@Override
public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long
sequenceId, ByteBuf buffer) {
+ buffer.retain();
Review Comment:
When will this buffer be released?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -255,6 +231,42 @@ public long getCommittedTxnCount() {
@Override
public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long
sequenceId, ByteBuf buffer) {
+ buffer.retain();
+ CompletableFuture<Position> future = publishFuture.thenCompose(ignore
-> {
+ if (checkIfNoSnapshot()) {
+ CompletableFuture<Void> completableFuture = new
CompletableFuture<>();
+
snapshotAbortedTxnProcessor.takeAbortedTxnsSnapshot(maxReadPosition).thenRun(()
-> {
+ if (changeToReadyStateFromNoSnapshot()) {
+ timer.newTimeout(TopicTransactionBuffer.this,
+ takeSnapshotIntervalTime,
TimeUnit.MILLISECONDS);
+ completableFuture.complete(null);
+ } else {
+ //This case should not happen.
+ log.error("[{}]Failed to change state of transaction
buffer to Ready from NoSnapshot",
+ topic.getName());
+ completableFuture.completeExceptionally(new
BrokerServiceException.ServiceUnitNotReadyException(
+ "Transaction Buffer take first snapshot
failed, the current state is: " + getState()));
+ }
+ }).exceptionally(exception -> {
+ log.error("Topic {} failed to take snapshot",
this.topic.getName());
+ completableFuture.completeExceptionally(exception);
+ return null;
+ });
+ buffer.retain();
Review Comment:
Why do we need to retain the buffer again?
https://github.com/apache/pulsar/pull/21406/files#diff-ecd728301a585f256e8a649b5e65b28c166194477355b3a1eefc198d014c25d3R234
has retained the buffer.
It looks like the buffer retain and release have been messed up.
--
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]
