[Transaction] Fix direct memory leak related to commit and abort markers
Netty ByteBufs for commit and abort markers leaked direct memory
ManagedLedgerImpl.asyncAddEntry will retain the buffer and therefore the
reference count has to be decreased with the call to release after
the asyncAddEntry call
@@ -206,6 +206,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
ByteBuf commitMarker = Markers.newTxnCommitMarker(-1L, txnID.getMostSigBits(),
txnID.getLeastSigBits());
+ try {
topic.getManagedLedger().asyncAddEntry(commitMarker, new AsyncCallbacks.AddEntryCallback() {
@Override
public void addComplete(Position position, ByteBuf entryData, Object ctx) {
@@ -224,6 +225,9 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
completableFuture.completeExceptionally(new PersistenceException(exception));
}
}, null);
+ } finally {
+ commitMarker.release();
+ }
return completableFuture;
}
@@ -235,6 +239,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
CompletableFuture<Void> completableFuture = new CompletableFuture<>();
ByteBuf abortMarker = Markers.newTxnAbortMarker(-1L, txnID.getMostSigBits(), txnID.getLeastSigBits());
+ try {
topic.getManagedLedger().asyncAddEntry(abortMarker, new AsyncCallbacks.AddEntryCallback() {
@Override
public void addComplete(Position position, ByteBuf entryData, Object ctx) {
@@ -255,6 +260,9 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
completableFuture.completeExceptionally(new PersistenceException(exception));
}
}, null);
+ } finally {
+ abortMarker.release();
+ }
return completableFuture;
}
@@ -264,6 +272,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
if (firstTxn.getMostSigBits() == txnID.getMostSigBits() && lowWaterMark >= firstTxn.getLeastSigBits()) {
ByteBuf abortMarker = Markers.newTxnAbortMarker(-1L,
txnID.getMostSigBits(), txnID.getLeastSigBits());
+ try {
topic.getManagedLedger().asyncAddEntry(abortMarker, new AsyncCallbacks.AddEntryCallback() {
@Override
public void addComplete(Position position, ByteBuf entryData, Object ctx) {
@@ -278,6 +287,9 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
log.error("Failed to abort low water mark for txn {}", txnID, exception);
}
}, null);
+ } finally {
+ abortMarker.release();
+ }
}
}
}
