ISSUE #2567: Save last revision in rocksdb
Descriptions of the changes in this PR:
Save latest revision in rocksdb.
### Motivation
Currently while restoring the complete transaction log is replayed.
As the transaction log builds up over time, it takes longer and longer to restore the storage containers on a new pod.
### Changes
Instead we will save the revision in the rocksdb. We will use the revision retrieved after
restoring the last checkpoint to skip ahead in the journal logs and only replay the entries
that are not included in the checkpoint
Master Issue: #2567
Reviewers: Enrico Olivelli <eolivelli@gmail.com>, Andrey Yegorov <andrea.yegorov@datastax.com>, Matteo Merli <mmerli@apache.org>
This closes #2568 from sursingh/save-last-version, closes #2567
diff --git a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVStore.java b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVStore.java
index fabf1e4..0205c48 100644
--- a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVStore.java
+++ b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVStore.java
@@ -217,6 +217,22 @@
}
}
+ protected void updateLastRevision(WriteBatch batch, long revision) {
+ if (revision >= 0) { // k/v comes from log stream
+ if (getLastRevision() >= revision) { // these k/v pairs are duplicates
+ return;
+ }
+ try {
+ // update revision
+ setLastRevision(revision);
+ batch.put(metaCfHandle, LAST_REVISION, lastRevisionBytes);
+ } catch (RocksDBException e) {
+ throw new StateStoreRuntimeException(
+ "Error while updating last revision " + revision + " from store " + name, e);
+ }
+ }
+ }
+
@Override
@SuppressWarnings("unchecked")
public synchronized void init(StateStoreSpec spec) throws StateStoreException {
diff --git a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/mvcc/MVCCStoreImpl.java b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/mvcc/MVCCStoreImpl.java
index fa595dc..5c9a83d 100644
--- a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/mvcc/MVCCStoreImpl.java
+++ b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/mvcc/MVCCStoreImpl.java
@@ -365,6 +365,7 @@
IncrementResult<K, V> result = null;
try {
result = increment(revision, batch, op);
+ updateLastRevision(batch, revision);
executeBatch(batch);
return result;
} catch (StateStoreRuntimeException e) {
@@ -463,6 +464,7 @@
PutResult<K, V> result = null;
try {
result = put(revision, batch, op);
+ updateLastRevision(batch, revision);
executeBatch(batch);
return result;
} catch (StateStoreRuntimeException e) {
@@ -576,6 +578,7 @@
DeleteResult<K, V> result = null;
try {
result = delete(revision, batch, op, true);
+ updateLastRevision(batch, revision);
executeBatch(batch);
return result;
} catch (StateStoreRuntimeException e) {
@@ -740,6 +743,7 @@
for (Op<K, V> o : operations) {
results.add(executeOp(revision, batch, o));
}
+ updateLastRevision(batch, revision);
executeBatch(batch);
// 4. repare the result
