liangyepianzhou commented on code in PR #21406:
URL: https://github.com/apache/pulsar/pull/21406#discussion_r1381917773
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -231,7 +232,7 @@ public long getCommittedTxnCount() {
@Override
public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long
sequenceId, ByteBuf buffer) {
buffer.retain();
- return transactionBufferFuture.thenCompose(ignore -> {
+ CompletableFuture<Position> future = publishFuture.thenCompose(ignore
-> {
Review Comment:
The actual situation is different from this example.
We have a global variable `transactionBufferFuture` whose generic type is
Void.
```java
private final CompletableFuture<Void> transactionBufferFuture = new
CompletableFuture<>();
```
But the method generic type of the method `appendBufferToTxn` is Position.
```java
public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long
sequenceId, ByteBuf buffer)
```
So, we inevitably introduce a new future within this method. And since
`sequencer.sequential` involves thread-switching operations, we need to perform
an additional `buffer.retain();`. Later, if there is any optimization for
`sequencer.sequential`, we also need to consider its impact on
`buffer.retain();`. This will introduce some complexity.
```
public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long
sequenceId, ByteBuf buffer) {
buffer.retain();
final CompletableFuture<Position> publishFuture = new
CompletableFuture<>();
sequencer.sequential(() -> {
buffer.retain();
return transactionBufferFuture.thenCompose(ignore -> {})
.thenAccept(publishFuture::complete);
}).exceptionally(ex -> {
publishFuture.completeExceptionally(ex);
return null;
});
return publishFuture;
}
```
--
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]
