[Transaction][buffer] Add basic operation of transaction #4738
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
@sijie PTAL |
sijie
requested changes
Jul 16, 2019
| import org.apache.bookkeeper.mledger.Position; | ||
| import org.apache.pulsar.transaction.impl.common.TxnID; | ||
|
|
||
| public interface TransactionCursor { |
There was a problem hiding this comment.
add comment for this interface.
also please add comments for the @param in following methods.
There was a problem hiding this comment.
Thanks for your reminder. I will add later.
|
|
||
| @Override | ||
| public CompletableFuture<SortedMap<Long, Position>> readEntries(int num, long startSequenceId) { | ||
| return null; |
There was a problem hiding this comment.
return CompletableFuture.failedFuture(new UnsupportedOperationException());
| @@ -99,6 +99,8 @@ | |||
| long committedAtLedgerId, | |||
| long committedAtEntryId); | |||
|
|
|||
| CompletableFuture<Void> commitTxn(TxnID txnID, long committedAtLedgerId, long committedAtEntryId, ByteBuf marker); | |||
There was a problem hiding this comment.
I am not sure why do you need marker in commitTxn and abortTxn. The caller of this interface doesn't provide the marker. The marker should be part of the implementation not the interface.
There was a problem hiding this comment.
So the marker in data ledgers is different from the transaction log?
|
|
||
| @Override | ||
| public CompletableFuture<TransactionMeta> getTransactionMeta(TxnID txnID) { | ||
| CompletableFuture<TransactionMeta> getFuture = new CompletableFuture<>(); |
There was a problem hiding this comment.
I don't think you need additional getFuture here.
it should just be :
return txnCursor.getTxnMeta(txnID, false);
| } | ||
|
|
||
| @Override | ||
| public CompletableFuture<Void> appendBufferToTxn(TxnID txnId, long sequenceId, ByteBuf buffer) { |
There was a problem hiding this comment.
It seems that sequenceId is not used in publishMessage. sequenceId should be used for de-duplication.
There was a problem hiding this comment.
Ah. I should add it to TxnCtx.
| ByteBuf marker) { | ||
| CompletableFuture<Void> commitFuture = new CompletableFuture<>(); | ||
|
|
||
| publishMessage(marker, new TxnCtx() { |
There was a problem hiding this comment.
The marker should be generated within this implementation.
|
|
||
| CompletableFuture<Void> abortFuture = new CompletableFuture<>(); | ||
|
|
||
| publishMessage(marker, new TxnCtx() { |
There was a problem hiding this comment.
marker should be prepared in this implementation.
| import org.apache.pulsar.transaction.buffer.exceptions.TransactionNotFoundException; | ||
| import org.apache.pulsar.transaction.impl.common.TxnID; | ||
|
|
||
| public class TransactionCursorImpl implements TransactionCursor { |
There was a problem hiding this comment.
the transaction cursor is used for persistent transaction buffer. but you are using an in-memory implementation of transaction metadata. this doesn't seem to be right.
zymap
changed the title
sijie
requested changes
Jul 22, 2019
| @Override | ||
| void close(); | ||
|
|
||
| CompletableFuture<Void> closeBuffer(); |
| @@ -125,7 +123,5 @@ | |||
| /** | |||
| * {@inheritDoc} | |||
There was a problem hiding this comment.
change the comment. since it doesn't inherit from AutoCloseable anymore.
| * @param ledgerId the transaction committed ledger id | ||
| * @return | ||
| */ | ||
| CompletableFuture<Set<TxnID>> getRemoveTxns(long ledgerId); |
| * @param ledgerId the remove transaction id | ||
| * @return | ||
| */ | ||
| CompletableFuture<Void> removeCommittedLedger(long ledgerId); |
| CompletableFuture<Set<TxnID>> getRemoveTxns(long ledgerId); | ||
|
|
||
| /** | ||
| * Remove transaction from index. |
| } | ||
|
|
||
|
|
||
| dataLedgers.forEach(dataLedger -> { |
There was a problem hiding this comment.
purgeFuture need to wait all dataLedgers are deleted. but the implementation here doesn't guarantee it.
| private CompletableFuture<Void> deleteTxn(TxnID txnID) { | ||
| CompletableFuture<Void> deleteFuture = new CompletableFuture<>(); | ||
|
|
||
| txnCursor.getTxnMeta(txnID, false).thenCompose(meta -> { |
There was a problem hiding this comment.
return txnCursor.getTxnMeta(txnID, false)
.thenCompose(meta -> meta.readEntries(...))
.thenCompose(positions -> {
...
});
| Marker commitMarker = Marker.builder().txnID(txnID).status(TxnStatus.COMMITTED).build(); | ||
| ByteBuf marker = Unpooled.wrappedBuffer(commitMarker.serialize()); | ||
|
|
||
| txnCursor.getTxnMeta(txnID, false).thenCompose(meta -> { |
There was a problem hiding this comment.
return txnCursor.getTxnMeta(txnID, false)
.thenCompose(meta -> publishMarker())
.thenCompose(position -> txnCursor.commitTxn(...));
|
|
||
| txnCursor.getTxnMeta(txnID, false).thenCompose(meta -> { | ||
| meta.readEntries(meta.numEntries(), -1L).thenCompose(longPositionSortedMap -> { | ||
| longPositionSortedMap.values().forEach(position -> { |
There was a problem hiding this comment.
CompletableFuture<Void> asyncMarkDelete(position);
List<CompletableFuture<Void>> deleteFutures = longPositionSortedMap.values()
.stream()
.map(position -> asyncMarkDelete())
.collect(Collectors.toList());
return CompletableFuture.allOf(deleteFutures).thenApply(ignored -> null);
| public CompletableFuture<List<TransactionEntry>> readNext(int numEntries) { | ||
| CompletableFuture<List<TransactionEntry>> readFuture = new CompletableFuture<>(); | ||
|
|
||
| meta.readEntries(numEntries, currentSeuquenceId).thenCompose(entries -> { |
There was a problem hiding this comment.
return meta.readEntries(numEntries, currentSequenceId)
.thenCompose(entries -> readEntry(entries);
meta.readEntries(...).whenComplete((entries, cause) -> {
if (null != cause) {
///
} else {
....
}
});
sijie
requested changes
Jul 23, 2019
| @@ -123,9 +121,9 @@ | |||
| CompletableFuture<Void> purgeTxns(List<Long> dataLedgers); | |||
|
|
|||
| /** | |||
| * {@inheritDoc} | |||
| * Close the buffer asynchronous. | |||
There was a problem hiding this comment.
Suggested change
| * Close the buffer asynchronous. | |
| * Close the buffer asynchronously. |
| import org.apache.pulsar.transaction.impl.common.TxnID; | ||
|
|
||
| /** | ||
| * The transaction Cursor maintained the index of all transaction. |
There was a problem hiding this comment.
Suggested change
| * The transaction Cursor maintained the index of all transaction. | |
| * The transaction Cursor maintains the index of all transactions. |
| package org.apache.pulsar.transaction.buffer.exceptions; | ||
|
|
||
| /** | ||
| * Exception when get a non exist committed ledger. |
There was a problem hiding this comment.
Exception is thrown when no transactions found committed at a given ledger.
|
|
||
|
|
||
| @Override | ||
| public String getProducerName() { |
There was a problem hiding this comment.
The producer name should be transaction id. Because pulsar is using producer name for de-duplications. You need to guarantee the messages produced by a transaction is not duplicated when we produce them to the transaction buffer.
| return FutureUtil.waitForAll(removeFutures); | ||
| } | ||
|
|
||
| private CompletableFuture<Void> deleteLedger(long dataledger) { |
There was a problem hiding this comment.
I don't think we should call it deleteLedger
There was a problem hiding this comment.
How about cleanTxnsOnLedger() ?
| public CompletableFuture<Void> removeTxnsCommittedAtLedger(long ledgerId) { | ||
|
|
||
| synchronized (committedTxnIndex) { | ||
| Set<TxnID> txnIDS = committedTxnIndex.remove(ledgerId); |
| if (startSequenceId != -1L) { | ||
| readEntries = entries.tailMap(startSequenceId); | ||
| } | ||
| if (startSequenceId == -2L) { |
| } | ||
|
|
||
| if (num != 0) { | ||
| result.put(-2L, PositionImpl.earliest); |
| } | ||
|
|
||
| @Override | ||
| public CompletableFuture<TransactionMeta> commitTxn(long committedAtLedgerId, long committedAtEntryId) { |
There was a problem hiding this comment.
You need to synchronize the method when you modifying the txnStatus
| return abortFuture; | ||
| } | ||
|
|
||
| this.txnStatus = TxnStatus.ABORTED; |
|
@sijie This pr is ready for review. Please take a look when you have time. Thanks. |
sijie
requested changes
Jul 26, 2019
...fer/src/main/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBuffer.java
Outdated
Show resolved
Hide resolved
...fer/src/main/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBuffer.java
Outdated
Show resolved
Hide resolved
...fer/src/main/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBuffer.java
Outdated
Show resolved
Hide resolved
...fer/src/main/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBuffer.java
Show resolved
Hide resolved
...fer/src/main/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBuffer.java
Outdated
Show resolved
Hide resolved
...src/test/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBufferTest.java
Outdated
Show resolved
Hide resolved
...src/test/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBufferTest.java
Outdated
Show resolved
Hide resolved
...src/test/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBufferTest.java
Outdated
Show resolved
Hide resolved
...src/test/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBufferTest.java
Outdated
Show resolved
Hide resolved
...src/test/java/org/apache/pulsar/transaction/buffer/impl/PersistentTransactionBufferTest.java
Outdated
Show resolved
Hide resolved
|
run java8 tests |
--- *Modifications* Add primary operation of transaction. Keep all actions persistently. Describe the modifications you've done. - add commit operation - add abort operation - add openreader operation *TODO* - add purge txn operation - add tests
--- (If this PR fixes a github issue, please add `Fixes #<xyz>`) Fixes #<xyz> (or if this PR is one task of a github issue, please add `Master Issue: #<xyz>` to link to the master issue) Master Issue: #<xyz> *Motivation* Describe here the context, and why you're making that change. What is the problem you're trying to solve. *Modifications* Describe the modifications you've done. *Verify this change* Please pick either of following options. - This change is a trivial rework / code cleanup without any test coverage. - This change is already covered by existing tests, such as *(please describe tests)*. - This change added tests and can be verified as follows: *(example:)* - *Added integration tests for end-to-end deployment* - *Extended integration test for recovery after broker failure*
|
run java8 tests |
|
run integration tests |
|
run java8 tests |
3 similar comments
|
run java8 tests |
|
run java8 tests |
|
run java8 tests |
|
run cpp tests |
sijie
approved these changes
Aug 1, 2019
|
run java8 tests |
4 similar comments
|
run java8 tests |
|
run java8 tests |
|
run java8 tests |
|
run java8 tests |
|
ci error |
|
run java8 tests |
50 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Modifications
Add primary operation of transaction. Keep all actions persistently. (except the index persistent, the persistent logic will add in next pull request)
Describe the modifications you've done.