liangyepianzhou commented on code in PR #21406:
URL: https://github.com/apache/pulsar/pull/21406#discussion_r1379592964
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -255,6 +230,40 @@ public long getCommittedTxnCount() {
@Override
public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long
sequenceId, ByteBuf buffer) {
+ buffer.retain();
+ return transactionBufferFuture.thenCompose(ignore -> {
Review Comment:
I introduce a new attribute `publishFuture` to resolve this problem.
First, initialize `future` to `transactionBufferFuture.thenApply(__ ->
PositionImpl.EARLIEST)`. Then, in each request, we add a new operation using
the `thenCompose` method and update the result to a new `CompletableFuture`. In
this way, each added operation will depend on the previous operation, ensuring
that they are executed in order.
--
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]
