You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2023/09/01 08:04:26 UTC
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20559: [optimize][txn] Implement lazy loading for transaction buffer
poorbarcode commented on code in PR #20559:
URL: https://github.com/apache/pulsar/pull/20559#discussion_r1312688627
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/TransactionBufferProvider.java:
##########
@@ -51,5 +52,5 @@ static TransactionBufferProvider newProvider(String providerClassName) throws IO
* @param originTopic
* @return
*/
- TransactionBuffer newTransactionBuffer(Topic originTopic);
+ TransactionBuffer newTransactionBuffer(Topic originTopic, PositionImpl startUsedPosition);
Review Comment:
Since the `TransactionBufferProvider` is a public API and users might use the config `transactionBufferProviderClassName` to change the implementation to use, we can't modify the original method; adding a new default method is better.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -250,7 +231,54 @@ public long getCommittedTxnCount() {
}
@Override
+/**
+ * If it's the first time using the transaction buffer, the method records the max read position of the topic
+ * to metadata of the topic at the time of the first write. If the transaction buffer has been used before,
+ * it writes the message after recovering the transaction buffer.
+ */
public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long sequenceId, ByteBuf buffer) {
+ ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger();
+ if (checkIfReady()) {
+ // Transaction buffer is ready, directly append the buffer
+ return internalAppendBufferToTxn(txnId, sequenceId, buffer);
+ } else if (changeToInitializingState()) {
+ // Transition to initializing state and record max read position if it's the first usage
+ Position firstMaxReadPosition = topic.getManagedLedger().getLastConfirmedEntry();
+ if (managedLedger.getProperties().get(MAX_READ_POSITION) == null) {
+ HashMap<String, String> hashMap = new HashMap<>();
+ hashMap.put(MAX_READ_POSITION, firstMaxReadPosition.toString());
+ managedLedger.asyncSetProperties(hashMap, new AsyncCallbacks.UpdatePropertiesCallback() {
+ @Override
+ public void updatePropertiesComplete(Map<String, String> properties, Object ctx) {
+ changeToReadyState();
+ transactionBufferFuture.complete(null);
+ }
+
+ @Override
+ public void updatePropertiesFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("Failed to set first max read position to topic {}", topic.getName(), exception);
+ changeToCloseState();
+ transactionBufferFuture.completeExceptionally(exception);
+ }
+ }, null);
+ }
+ }
+ // Return a CompletableFuture that will complete after recovering the transaction buffer,
+ // then append the buffer
+ return transactionBufferFuture.thenCompose(ignore -> internalAppendBufferToTxn(txnId, sequenceId, buffer));
+ }
+
+ private PositionImpl getPositionFromString(String positionStr) {
Review Comment:
The method `getPositionFromString` is useless, should we remove it?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -1459,61 +1459,48 @@ protected void handleProducer(final CommandProducer cmdProducer) {
});
schemaVersionFuture.thenAccept(schemaVersion -> {
- topic.checkIfTransactionBufferRecoverCompletely(isTxnEnabled).thenAccept(future -> {
Review Comment:
The method `checkIfTransactionBufferRecoverCompletely` triggers an event to persist a snapshot once, but you did not persist a snapshot when the first sending with the transition.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -66,6 +69,8 @@
@Slf4j
public class TopicTransactionBuffer extends TopicTransactionBufferState implements TransactionBuffer, TimerTask {
+ public static final String MAX_READ_POSITION = "maxReadPosition";
Review Comment:
The prop added does not mean the max read position of the topic, right? It is the first position that sending with the transaction.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java:
##########
@@ -250,7 +231,54 @@ public long getCommittedTxnCount() {
}
@Override
+/**
+ * If it's the first time using the transaction buffer, the method records the max read position of the topic
+ * to metadata of the topic at the time of the first write. If the transaction buffer has been used before,
+ * it writes the message after recovering the transaction buffer.
+ */
public CompletableFuture<Position> appendBufferToTxn(TxnID txnId, long sequenceId, ByteBuf buffer) {
+ ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) topic.getManagedLedger();
+ if (checkIfReady()) {
+ // Transaction buffer is ready, directly append the buffer
+ return internalAppendBufferToTxn(txnId, sequenceId, buffer);
+ } else if (changeToInitializingState()) {
+ // Transition to initializing state and record max read position if it's the first usage
+ Position firstMaxReadPosition = topic.getManagedLedger().getLastConfirmedEntry();
+ if (managedLedger.getProperties().get(MAX_READ_POSITION) == null) {
+ HashMap<String, String> hashMap = new HashMap<>();
+ hashMap.put(MAX_READ_POSITION, firstMaxReadPosition.toString());
+ managedLedger.asyncSetProperties(hashMap, new AsyncCallbacks.UpdatePropertiesCallback() {
+ @Override
+ public void updatePropertiesComplete(Map<String, String> properties, Object ctx) {
+ changeToReadyState();
Review Comment:
Should we start the scheduled task to persist the snapshot here? And add a test for this case.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org