You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2018/11/12 20:12:02 UTC
[GitHub] ivankelly closed pull request #1797: Move version out of
LedgerMetadata
ivankelly closed pull request #1797: Move version out of LedgerMetadata
URL: https://github.com/apache/bookkeeper/pull/1797
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java
index 18ce8c4392..64c4fa80b9 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java
@@ -1005,33 +1005,30 @@ public void process(Long ledgerId, VoidCallback cb) {
printLedgerMetadata(ledgerId, null, false);
cb.processResult(BKException.Code.OK, null, null);
} else {
- GenericCallback<LedgerMetadata> gencb = new GenericCallback<LedgerMetadata>() {
- @Override
- public void operationComplete(int rc, LedgerMetadata ledgerMetadata) {
- if (rc == BKException.Code.OK) {
- if ((bookieAddress == null)
- || BookKeeperAdmin.areEntriesOfLedgerStoredInTheBookie(ledgerId,
- bookieAddress, ledgerMetadata)) {
- /*
- * the print method has to be in
- * synchronized scope, otherwise
- * output of printLedgerMetadata
- * could interleave since this
- * callback for different
- * ledgers can happen in
- * different threads.
- */
- synchronized (BookieShell.this) {
- printLedgerMetadata(ledgerId, ledgerMetadata, printMeta);
- }
+ GenericCallback<Versioned<LedgerMetadata>> gencb = (rc, ledgerMetadata) -> {
+ if (rc == BKException.Code.OK) {
+ if ((bookieAddress == null)
+ || BookKeeperAdmin.areEntriesOfLedgerStoredInTheBookie(
+ ledgerId, bookieAddress, ledgerMetadata.getValue())) {
+ /*
+ * the print method has to be in
+ * synchronized scope, otherwise
+ * output of printLedgerMetadata
+ * could interleave since this
+ * callback for different
+ * ledgers can happen in
+ * different threads.
+ */
+ synchronized (BookieShell.this) {
+ printLedgerMetadata(ledgerId, ledgerMetadata.getValue(), printMeta);
}
- } else if (rc == BKException.Code.NoSuchLedgerExistsException) {
- rc = BKException.Code.OK;
- } else {
- LOG.error("Unable to read the ledger: " + ledgerId + " information");
}
- cb.processResult(rc, null, null);
+ } else if (rc == BKException.Code.NoSuchLedgerExistsException) {
+ rc = BKException.Code.OK;
+ } else {
+ LOG.error("Unable to read the ledger: " + ledgerId + " information");
}
+ cb.processResult(rc, null, null);
};
ledgerManager.readLedgerMetadata(ledgerId, gencb);
}
@@ -1113,21 +1110,21 @@ public int runCmd(CommandLine cmdLine) throws Exception {
runFunctionWithLedgerManagerFactory(bkConf, mFactory -> {
try (LedgerManager m = mFactory.newLedgerManager()) {
if (cmdLine.hasOption("dumptofile")) {
- GenericCallbackFuture<LedgerMetadata> cb = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> cb = new GenericCallbackFuture<>();
m.readLedgerMetadata(lid, cb);
Files.write(FileSystems.getDefault().getPath(cmdLine.getOptionValue("dumptofile")),
- cb.join().serialize());
+ cb.join().getValue().serialize());
} else if (cmdLine.hasOption("restorefromfile")) {
byte[] serialized = Files.readAllBytes(
FileSystems.getDefault().getPath(cmdLine.getOptionValue("restorefromfile")));
- LedgerMetadata md = LedgerMetadata.parseConfig(serialized, Version.NEW, Optional.absent());
- GenericCallbackFuture<LedgerMetadata> cb = new GenericCallbackFuture<>();
+ LedgerMetadata md = LedgerMetadata.parseConfig(serialized, Optional.absent());
+ GenericCallbackFuture<Versioned<LedgerMetadata>> cb = new GenericCallbackFuture<>();
m.createLedgerMetadata(lid, md, cb);
cb.join();
} else {
- GenericCallbackFuture<LedgerMetadata> cb = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> cb = new GenericCallbackFuture<>();
m.readLedgerMetadata(lid, cb);
- printLedgerMetadata(lid, cb.get(), true);
+ printLedgerMetadata(lid, cb.get().getValue(), true);
}
} catch (Exception e) {
throw new UncheckedExecutionException(e);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ScanAndCompareGarbageCollector.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ScanAndCompareGarbageCollector.java
index 33f10e901a..19d630ff75 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ScanAndCompareGarbageCollector.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ScanAndCompareGarbageCollector.java
@@ -49,6 +49,7 @@
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.util.MathUtils;
import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
@@ -183,7 +184,7 @@ public void gc(GarbageCleaner garbageCleaner) {
if (verifyMetadataOnGc) {
CountDownLatch latch = new CountDownLatch(1);
final AtomicInteger metaRC = new AtomicInteger(0);
- ledgerManager.readLedgerMetadata(bkLid, (int rc, LedgerMetadata x) -> {
+ ledgerManager.readLedgerMetadata(bkLid, (rc, ignore) -> {
metaRC.set(rc);
latch.countDown();
});
@@ -236,19 +237,19 @@ public void gc(GarbageCleaner garbageCleaner) {
ZkLedgerUnderreplicationManager.acquireUnderreplicatedLedgerLock(zk, zkLedgersRootPath, ledgerId,
zkAcls);
semaphore.acquire();
- ledgerManager.readLedgerMetadata(ledgerId, new GenericCallback<LedgerMetadata>() {
+ ledgerManager.readLedgerMetadata(ledgerId, new GenericCallback<Versioned<LedgerMetadata>>() {
@Override
- public void operationComplete(int rc, LedgerMetadata ledgerMetadata) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> ledgerMetadata) {
if (rc == BKException.Code.OK) {
// do not delete a ledger that is not closed, since the ensemble might change again and
// include the current bookie while we are deleting it
- if (!ledgerMetadata.isClosed()) {
+ if (!ledgerMetadata.getValue().isClosed()) {
release();
return;
}
SortedMap<Long, ? extends List<BookieSocketAddress>> ensembles =
- ledgerMetadata.getAllEnsembles();
+ ledgerMetadata.getValue().getAllEnsembles();
for (List<BookieSocketAddress> ensemble : ensembles.values()) {
// check if this bookie is supposed to have this ledger
if (ensemble.contains(selfBookieAddress)) {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
index d434ffa39e..4d0fb32ffc 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
@@ -79,6 +79,7 @@
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.util.ReflectionUtils;
import org.apache.bookkeeper.util.SafeRunnable;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.lang.SystemUtils;
import org.apache.zookeeper.KeeperException;
@@ -1299,11 +1300,11 @@ public void deleteLedger(long lId) throws InterruptedException, BKException {
* @param cb callback method
*/
public void asyncIsClosed(long lId, final IsClosedCallback cb, final Object ctx){
- ledgerManager.readLedgerMetadata(lId, new GenericCallback<LedgerMetadata>(){
+ ledgerManager.readLedgerMetadata(lId, new GenericCallback<Versioned<LedgerMetadata>>(){
@Override
- public void operationComplete(int rc, LedgerMetadata lm){
+ public void operationComplete(int rc, Versioned<LedgerMetadata> lm){
if (rc == BKException.Code.OK) {
- cb.isClosedComplete(rc, lm.isClosed(), ctx);
+ cb.isClosedComplete(rc, lm.getValue().isClosed(), ctx);
} else {
cb.isClosedComplete(rc, false, ctx);
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
index f637889be7..5aa20867c4 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
@@ -82,6 +82,7 @@
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.util.IOUtils;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
@@ -490,9 +491,9 @@ public void asyncGetLedgersContainBookies(final Set<BookieSocketAddress> bookies
bkc.getLedgerManager().asyncProcessLedgers(new Processor<Long>() {
@Override
public void process(final Long lid, final AsyncCallback.VoidCallback cb) {
- bkc.getLedgerManager().readLedgerMetadata(lid, new GenericCallback<LedgerMetadata>() {
+ bkc.getLedgerManager().readLedgerMetadata(lid, new GenericCallback<Versioned<LedgerMetadata>>() {
@Override
- public void operationComplete(int rc, LedgerMetadata metadata) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> metadata) {
if (BKException.Code.NoSuchLedgerExistsException == rc) {
// the ledger was deleted during this iteration.
cb.processResult(BKException.Code.OK, null, null);
@@ -501,11 +502,11 @@ public void operationComplete(int rc, LedgerMetadata metadata) {
cb.processResult(rc, null, null);
return;
}
- Set<BookieSocketAddress> bookiesInLedger = metadata.getBookiesInThisLedger();
+ Set<BookieSocketAddress> bookiesInLedger = metadata.getValue().getBookiesInThisLedger();
Sets.SetView<BookieSocketAddress> intersection =
Sets.intersection(bookiesInLedger, bookies);
if (!intersection.isEmpty()) {
- ledgers.put(lid, metadata);
+ ledgers.put(lid, metadata.getValue());
}
cb.processResult(BKException.Code.OK, null, null);
}
@@ -1637,7 +1638,7 @@ private static boolean areEntriesOfSegmentStoredInTheBookie(LedgerMetadata ledge
}
static class ReadMetadataCallback extends AbstractFuture<LedgerMetadata>
- implements GenericCallback<LedgerMetadata> {
+ implements GenericCallback<Versioned<LedgerMetadata>> {
final long ledgerId;
ReadMetadataCallback(long ledgerId) {
@@ -1648,11 +1649,11 @@ long getLedgerId() {
return ledgerId;
}
- public void operationComplete(int rc, LedgerMetadata result) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> result) {
if (rc != 0) {
setException(BKException.create(rc));
} else {
- set(result);
+ set(result.getValue());
}
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
index e39b2a19b0..b55e7ba9ca 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java
@@ -46,6 +46,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.util.MathUtils;
+import org.apache.bookkeeper.versioning.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,7 +54,7 @@
* Encapsulates asynchronous ledger create operation.
*
*/
-class LedgerCreateOp implements GenericCallback<LedgerMetadata> {
+class LedgerCreateOp implements GenericCallback<Versioned<LedgerMetadata>> {
static final Logger LOG = LoggerFactory.getLogger(LedgerCreateOp.class);
@@ -187,7 +188,7 @@ public void initiateAdv(final long ledgerId) {
* Callback when created ledger.
*/
@Override
- public void operationComplete(int rc, LedgerMetadata writtenMetadata) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> writtenMetadata) {
if (this.generateLedgerId && (BKException.Code.LedgerExistException == rc)) {
// retry to generate a new ledger id
generateLedgerIdAndCreateLedger();
@@ -199,9 +200,9 @@ public void operationComplete(int rc, LedgerMetadata writtenMetadata) {
try {
if (adv) {
- lh = new LedgerHandleAdv(bk.getClientCtx(), ledgerId, metadata, digestType, passwd, writeFlags);
+ lh = new LedgerHandleAdv(bk.getClientCtx(), ledgerId, writtenMetadata, digestType, passwd, writeFlags);
} else {
- lh = new LedgerHandle(bk.getClientCtx(), ledgerId, metadata, digestType, passwd, writeFlags);
+ lh = new LedgerHandle(bk.getClientCtx(), ledgerId, writtenMetadata, digestType, passwd, writeFlags);
}
} catch (GeneralSecurityException e) {
LOG.error("Security exception while creating ledger: " + ledgerId, e);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
index 76a9178db6..28929cf9a8 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragmentReplicator.java
@@ -375,7 +375,7 @@ private static void updateEnsembleInfo(
MetadataUpdateLoop updateLoop = new MetadataUpdateLoop(
ledgerManager,
lh.getId(),
- lh::getLedgerMetadata,
+ lh::getVersionedLedgerMetadata,
(metadata) -> {
// returns true if any of old bookies exist in ensemble
List<BookieSocketAddress> ensemble = metadata.getAllEnsembles().get(fragmentStartId);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
index 6125a3d1b9..b5f0043dde 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
@@ -75,12 +75,12 @@
import org.apache.bookkeeper.common.util.MathUtils;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookieProtocol;
-import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.proto.checksum.DigestManager;
import org.apache.bookkeeper.stats.Counter;
import org.apache.bookkeeper.stats.Gauge;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.util.SafeRunnable;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.commons.collections4.IteratorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -95,7 +95,7 @@
final ClientContext clientCtx;
final byte[] ledgerKey;
- private LedgerMetadata metadata;
+ private Versioned<LedgerMetadata> versionedMetadata;
final long ledgerId;
long lastAddPushed;
@@ -156,15 +156,17 @@
private final OpStatsLogger clientChannelWriteWaitStats;
LedgerHandle(ClientContext clientCtx,
- long ledgerId, LedgerMetadata metadata,
+ long ledgerId, Versioned<LedgerMetadata> versionedMetadata,
BookKeeper.DigestType digestType, byte[] password,
EnumSet<WriteFlag> writeFlags)
throws GeneralSecurityException, NumberFormatException {
this.clientCtx = clientCtx;
- this.metadata = metadata;
+ this.versionedMetadata = versionedMetadata;
this.pendingAddOps = new ConcurrentLinkedQueue<PendingAddOp>();
this.writeFlags = writeFlags;
+
+ LedgerMetadata metadata = versionedMetadata.getValue();
if (metadata.isClosed()) {
lastAddConfirmed = lastAddPushed = metadata.getLastEntryId();
length = metadata.getLength();
@@ -190,7 +192,9 @@
// password, so that the bookie can avoid processing the keys for each entry
this.ledgerKey = DigestManager.generateMasterKey(password);
distributionSchedule = new RoundRobinDistributionSchedule(
- metadata.getWriteQuorumSize(), metadata.getAckQuorumSize(), metadata.getEnsembleSize());
+ metadata.getWriteQuorumSize(),
+ metadata.getAckQuorumSize(),
+ metadata.getEnsembleSize());
this.bookieFailureHistory = CacheBuilder.newBuilder()
.expireAfterWrite(clientCtx.getConf().bookieFailureHistoryExpirationMSec, TimeUnit.MILLISECONDS)
.build(new CacheLoader<BookieSocketAddress, Long>() {
@@ -310,14 +314,19 @@ public synchronized long getLastAddPushed() {
*/
@Override
public LedgerMetadata getLedgerMetadata() {
- return metadata;
+ return versionedMetadata.getValue();
+ }
+
+ Versioned<LedgerMetadata> getVersionedLedgerMetadata() {
+ return versionedMetadata;
}
- boolean setLedgerMetadata(LedgerMetadata expected, LedgerMetadata newMetadata) {
+ boolean setLedgerMetadata(Versioned<LedgerMetadata> expected, Versioned<LedgerMetadata> newMetadata) {
synchronized (this) {
// ensure that we only update the metadata if it is the object we expect it to be
- if (metadata == expected) {
- metadata = newMetadata;
+ if (versionedMetadata == expected) {
+ versionedMetadata = newMetadata;
+ LedgerMetadata metadata = versionedMetadata.getValue();
if (metadata.isClosed()) {
lastAddConfirmed = lastAddPushed = metadata.getLastEntryId();
length = metadata.getLength();
@@ -419,14 +428,6 @@ BookiesHealthInfo getBookiesHealthInfo() {
return bookiesHealthInfo;
}
- void writeLedgerConfig(GenericCallback<LedgerMetadata> writeCb) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Writing metadata to ledger manager: {}, {}", this.ledgerId, getLedgerMetadata().getVersion());
- }
-
- clientCtx.getLedgerManager().writeLedgerMetadata(ledgerId, getLedgerMetadata(), writeCb);
- }
-
/**
* {@inheritDoc}
*/
@@ -532,7 +533,7 @@ public void safeRun() {
tearDownWriteHandleState();
new MetadataUpdateLoop(
clientCtx.getLedgerManager(), getId(),
- LedgerHandle.this::getLedgerMetadata,
+ LedgerHandle.this::getVersionedLedgerMetadata,
(metadata) -> {
if (metadata.isClosed()) {
/* If the ledger has been closed with the same lastEntry
@@ -1844,7 +1845,7 @@ void ensembleChangeLoop(List<BookieSocketAddress> origEnsemble, Map<Integer, Boo
AtomicInteger attempts = new AtomicInteger(0);
new MetadataUpdateLoop(
clientCtx.getLedgerManager(), getId(),
- this::getLedgerMetadata,
+ this::getVersionedLedgerMetadata,
(metadata) -> !metadata.isClosed() && !metadata.isInRecovery()
&& failedBookies.entrySet().stream().anyMatch(
(e) -> metadata.getLastEnsembleValue().get(e.getKey()).equals(e.getValue())),
@@ -1875,13 +1876,13 @@ void ensembleChangeLoop(List<BookieSocketAddress> origEnsemble, Map<Integer, Boo
if (ex != null) {
LOG.warn("{}[attempt:{}] Exception changing ensemble", logContext, attempts.get(), ex);
handleUnrecoverableErrorDuringAdd(BKException.getExceptionCode(ex, WriteException));
- } else if (metadata.isClosed()) {
+ } else if (metadata.getValue().isClosed()) {
if (LOG.isDebugEnabled()) {
LOG.debug("{}[attempt:{}] Metadata closed during attempt to replace bookie."
+ " Another client must have recovered the ledger.", logContext, attempts.get());
}
handleUnrecoverableErrorDuringAdd(BKException.Code.LedgerClosedException);
- } else if (metadata.isInRecovery()) {
+ } else if (metadata.getValue().isInRecovery()) {
if (LOG.isDebugEnabled()) {
LOG.debug("{}[attempt:{}] Metadata marked as in-recovery during attempt to replace bookie."
+ " Another client must be recovering the ledger.", logContext, attempts.get());
@@ -1919,10 +1920,6 @@ void unsetSuccessAndSendWriteRequest(List<BookieSocketAddress> ensemble, final S
}
}
- void rereadMetadata(final GenericCallback<LedgerMetadata> cb) {
- clientCtx.getLedgerManager().readLedgerMetadata(ledgerId, cb);
- }
-
void registerOperationFailureOnBookie(BookieSocketAddress bookie, long entryId) {
if (clientCtx.getConf().enableBookieFailureTracking) {
bookieFailureHistory.put(bookie, entryId);
@@ -1958,6 +1955,6 @@ public void closeComplete(int rc, LedgerHandle lh, Object ctx) {
// Getting current ensemble from the metadata is only a temporary
// thing until metadata is immutable. At that point, current ensemble
// becomes a property of the LedgerHandle itself.
- return metadata.getCurrentEnsemble();
+ return versionedMetadata.getValue().getCurrentEnsemble();
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
index bd8beae48b..6f38b8e678 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java
@@ -37,6 +37,7 @@
import org.apache.bookkeeper.client.api.WriteAdvHandle;
import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.util.SafeRunnable;
+import org.apache.bookkeeper.versioning.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,7 +57,7 @@ public int compare(PendingAddOp o1, PendingAddOp o2) {
}
LedgerHandleAdv(ClientContext clientCtx,
- long ledgerId, LedgerMetadata metadata,
+ long ledgerId, Versioned<LedgerMetadata> metadata,
BookKeeper.DigestType digestType, byte[] password, EnumSet<WriteFlag> writeFlags)
throws GeneralSecurityException, NumberFormatException {
super(clientCtx, ledgerId, metadata, digestType, password, writeFlags);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadata.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadata.java
index d8e1c1be37..620a798019 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadata.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadata.java
@@ -45,7 +45,6 @@
import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat;
-import org.apache.bookkeeper.versioning.Version;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -85,7 +84,6 @@
private LedgerMetadataFormat.State state;
private TreeMap<Long, ImmutableList<BookieSocketAddress>> ensembles = new TreeMap<>();
private List<BookieSocketAddress> currentEnsemble;
- volatile Version version = Version.NEW;
private boolean hasPassword = false;
private LedgerMetadataFormat.DigestType digestType;
@@ -140,8 +138,7 @@ public LedgerMetadata(int ensembleSize,
DigestType digestType,
java.util.Optional<byte[]> password,
java.util.Optional<Long> ctime,
- Map<String, byte[]> customMetadata,
- Version version) {
+ Map<String, byte[]> customMetadata) {
checkArgument(ensembles.size() > 0, "There must be at least one ensemble in the ledger");
this.ensembleSize = ensembleSize;
@@ -174,7 +171,6 @@ public LedgerMetadata(int ensembleSize,
});
this.customMetadata.putAll(customMetadata);
- this.version = version;
}
/**
@@ -197,7 +193,6 @@ public LedgerMetadata(int ensembleSize, int writeQuorumSize, int ackQuorumSize,
this.lastEntryId = other.lastEntryId;
this.metadataFormatVersion = other.metadataFormatVersion;
this.state = other.state;
- this.version = other.version;
this.hasPassword = other.hasPassword;
this.digestType = other.digestType;
this.ctime = other.ctime;
@@ -472,17 +467,14 @@ LedgerMetadataFormat buildProtoFormat(boolean withPassword) {
*
* @param bytes
* byte array to parse
- * @param version
- * version of the ledger metadata
* @param msCtime
* metadata store creation time, used for legacy ledgers
* @return LedgerConfig
* @throws IOException
* if the given byte[] cannot be parsed
*/
- public static LedgerMetadata parseConfig(byte[] bytes, Version version, Optional<Long> msCtime) throws IOException {
+ public static LedgerMetadata parseConfig(byte[] bytes, Optional<Long> msCtime) throws IOException {
LedgerMetadata lc = new LedgerMetadata();
- lc.version = version;
String config = new String(bytes, UTF_8);
@@ -608,37 +600,6 @@ static LedgerMetadata parseVersion1Config(LedgerMetadata lc,
return lc;
}
- /**
- * Updates the version of this metadata.
- *
- * @param v Version
- */
- public void setVersion(Version v) {
- this.version = v;
- }
-
- /**
- * Returns the last version.
- *
- * @return version
- */
- public Version getVersion() {
- return this.version;
- }
-
- /**
- * Is the metadata newer than given <i>newMeta</i>.
- *
- * @param newMeta the metadata to compare
- * @return true if <i>this</i> is newer than <i>newMeta</i>, false otherwise
- */
- boolean isNewerThan(LedgerMetadata newMeta) {
- if (null == version) {
- return false;
- }
- return Version.Occurred.AFTER == version.compare(newMeta.version);
- }
-
/**
* Routine to compare two {@code Map<String, byte[]>}; Since the values in the map are {@code byte[]}, we can't use
* {@code Map.equals}.
@@ -689,8 +650,7 @@ public String toSafeString() {
private String toStringRepresentation(boolean withPassword) {
StringBuilder sb = new StringBuilder();
- sb.append("(meta:").append(new String(serialize(withPassword), UTF_8)).append(", version:").append(version)
- .append(")");
+ sb.append("(meta:").append(new String(serialize(withPassword), UTF_8)).append(")");
return sb.toString();
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java
index b1b9dbdd3a..8206438b71 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerMetadataBuilder.java
@@ -20,6 +20,7 @@
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import java.util.Arrays;
@@ -30,11 +31,18 @@
import java.util.TreeMap;
import org.apache.bookkeeper.client.api.DigestType;
+import org.apache.bookkeeper.common.annotation.InterfaceAudience.LimitedPrivate;
+import org.apache.bookkeeper.common.annotation.InterfaceStability.Unstable;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat;
-import org.apache.bookkeeper.versioning.Version;
-class LedgerMetadataBuilder {
+/**
+ * Builder for building LedgerMetadata objects.
+ */
+@LimitedPrivate
+@Unstable
+@VisibleForTesting
+public class LedgerMetadataBuilder {
private int ensembleSize = 3;
private int writeQuorumSize = 3;
private int ackQuorumSize = 2;
@@ -51,13 +59,11 @@
private Optional<Long> ctime = Optional.empty();
private Map<String, byte[]> customMetadata = Collections.emptyMap();
- private Version version = Version.NEW;
-
- static LedgerMetadataBuilder create() {
+ public static LedgerMetadataBuilder create() {
return new LedgerMetadataBuilder();
}
- static LedgerMetadataBuilder from(LedgerMetadata other) {
+ public static LedgerMetadataBuilder from(LedgerMetadata other) {
LedgerMetadataBuilder builder = new LedgerMetadataBuilder();
builder.ensembleSize = other.getEnsembleSize();
builder.writeQuorumSize = other.getWriteQuorumSize();
@@ -86,36 +92,39 @@ static LedgerMetadataBuilder from(LedgerMetadata other) {
}
builder.customMetadata = ImmutableMap.copyOf(other.getCustomMetadata());
- builder.version = other.getVersion();
-
return builder;
}
- LedgerMetadataBuilder withPassword(byte[] password) {
+ public LedgerMetadataBuilder withPassword(byte[] password) {
this.password = Optional.of(Arrays.copyOf(password, password.length));
return this;
}
- LedgerMetadataBuilder withEnsembleSize(int ensembleSize) {
+ public LedgerMetadataBuilder withDigestType(DigestType digestType) {
+ this.digestType = digestType;
+ return this;
+ }
+
+ public LedgerMetadataBuilder withEnsembleSize(int ensembleSize) {
checkState(ensembles.size() == 0, "Can only set ensemble size before adding ensembles to the builder");
this.ensembleSize = ensembleSize;
return this;
}
- LedgerMetadataBuilder withWriteQuorumSize(int writeQuorumSize) {
+ public LedgerMetadataBuilder withWriteQuorumSize(int writeQuorumSize) {
checkArgument(ensembleSize >= writeQuorumSize, "Write quorum must be less or equal to ensemble size");
checkArgument(writeQuorumSize >= ackQuorumSize, "Write quorum must be greater or equal to ack quorum");
this.writeQuorumSize = writeQuorumSize;
return this;
}
- LedgerMetadataBuilder withAckQuorumSize(int ackQuorumSize) {
+ public LedgerMetadataBuilder withAckQuorumSize(int ackQuorumSize) {
checkArgument(writeQuorumSize >= ackQuorumSize, "Ack quorum must be less or equal to write quorum");
this.ackQuorumSize = ackQuorumSize;
return this;
}
- LedgerMetadataBuilder newEnsembleEntry(long firstEntry, List<BookieSocketAddress> ensemble) {
+ public LedgerMetadataBuilder newEnsembleEntry(long firstEntry, List<BookieSocketAddress> ensemble) {
checkArgument(ensemble.size() == ensembleSize,
"Size of passed in ensemble must match the ensembleSize of the builder");
checkArgument(ensembles.isEmpty() || firstEntry > ensembles.lastKey(),
@@ -124,7 +133,7 @@ LedgerMetadataBuilder newEnsembleEntry(long firstEntry, List<BookieSocketAddress
return this;
}
- LedgerMetadataBuilder replaceEnsembleEntry(long firstEntry, List<BookieSocketAddress> ensemble) {
+ public LedgerMetadataBuilder replaceEnsembleEntry(long firstEntry, List<BookieSocketAddress> ensemble) {
checkArgument(ensemble.size() == ensembleSize,
"Size of passed in ensemble must match the ensembleSize of the builder");
checkArgument(ensembles.containsKey(firstEntry),
@@ -133,23 +142,22 @@ LedgerMetadataBuilder replaceEnsembleEntry(long firstEntry, List<BookieSocketAdd
return this;
}
- LedgerMetadataBuilder withInRecoveryState() {
+ public LedgerMetadataBuilder withInRecoveryState() {
this.state = LedgerMetadataFormat.State.IN_RECOVERY;
return this;
}
- LedgerMetadataBuilder closingAt(long lastEntryId, long length) {
+ public LedgerMetadataBuilder closingAt(long lastEntryId, long length) {
this.lastEntryId = Optional.of(lastEntryId);
this.length = Optional.of(length);
this.state = LedgerMetadataFormat.State.CLOSED;
return this;
}
- LedgerMetadata build() {
+ public LedgerMetadata build() {
return new LedgerMetadata(ensembleSize, writeQuorumSize, ackQuorumSize,
state, lastEntryId, length, ensembles,
- digestType, password, ctime, customMetadata,
- version);
+ digestType, password, ctime, customMetadata);
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java
index 7f169ab1ba..16d7bd59b3 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java
@@ -40,6 +40,7 @@
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.util.MathUtils;
import org.apache.bookkeeper.util.OrderedGenericCallback;
+import org.apache.bookkeeper.versioning.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,7 +48,7 @@
* Encapsulates the ledger open operation.
*
*/
-class LedgerOpenOp implements GenericCallback<LedgerMetadata> {
+class LedgerOpenOp implements GenericCallback<Versioned<LedgerMetadata>> {
static final Logger LOG = LoggerFactory.getLogger(LedgerOpenOp.class);
final BookKeeper bk;
@@ -125,21 +126,25 @@ public void initiateWithoutRecovery() {
* Implements Open Ledger Callback.
*/
@Override
- public void operationComplete(int rc, LedgerMetadata metadata) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> versionedMetadata) {
if (BKException.Code.OK != rc) {
// open ledger failed.
openComplete(rc, null);
return;
}
+ LedgerMetadata metadata = versionedMetadata.getValue();
final byte[] passwd;
// we should use digest type from metadata *ONLY* when:
// 1) digest type is stored in metadata
// 2) `autodetection` is enabled
- DigestType digestType = enableDigestAutodetection && metadata.hasPassword()
- ? fromApiDigestType(metadata.getDigestType())
- : suggestedDigestType;
+ DigestType digestType;
+ if (enableDigestAutodetection && metadata.hasPassword()) {
+ digestType = fromApiDigestType(metadata.getDigestType());
+ } else {
+ digestType = suggestedDigestType;
+ }
/* For an administrative open, the default passwords
* are read from the configuration, but if the metadata
@@ -168,7 +173,7 @@ public void operationComplete(int rc, LedgerMetadata metadata) {
// get the ledger metadata back
try {
- lh = new ReadOnlyLedgerHandle(bk.getClientCtx(), ledgerId, metadata, digestType,
+ lh = new ReadOnlyLedgerHandle(bk.getClientCtx(), ledgerId, versionedMetadata, digestType,
passwd, !doRecovery);
} catch (GeneralSecurityException e) {
LOG.error("Security exception while opening ledger: " + ledgerId, e);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/MetadataUpdateLoop.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/MetadataUpdateLoop.java
index 9cd2afd1c9..821a4dac17 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/MetadataUpdateLoop.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/MetadataUpdateLoop.java
@@ -25,6 +25,7 @@
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,7 +51,7 @@
private final LedgerManager lm;
private final long ledgerId;
- private final Supplier<LedgerMetadata> currentLocalValue;
+ private final Supplier<Versioned<LedgerMetadata>> currentLocalValue;
private final NeedsUpdatePredicate needsTransformation;
private final MetadataTransform transform;
private final LocalValueUpdater updateLocalValue;
@@ -69,7 +70,7 @@
}
interface LocalValueUpdater {
- boolean updateValue(LedgerMetadata oldValue, LedgerMetadata newValue);
+ boolean updateValue(Versioned<LedgerMetadata> oldValue, Versioned<LedgerMetadata> newValue);
}
/**
@@ -88,7 +89,7 @@
*/
MetadataUpdateLoop(LedgerManager lm,
long ledgerId,
- Supplier<LedgerMetadata> currentLocalValue,
+ Supplier<Versioned<LedgerMetadata>> currentLocalValue,
NeedsUpdatePredicate needsTransformation,
MetadataTransform transform,
LocalValueUpdater updateLocalValue) {
@@ -103,22 +104,23 @@
ledgerId, System.identityHashCode(this));
}
- CompletableFuture<LedgerMetadata> run() {
- CompletableFuture<LedgerMetadata> promise = new CompletableFuture<>();
+ CompletableFuture<Versioned<LedgerMetadata>> run() {
+ CompletableFuture<Versioned<LedgerMetadata>> promise = new CompletableFuture<>();
writeLoop(currentLocalValue.get(), promise);
return promise;
}
- private void writeLoop(LedgerMetadata currentLocal, CompletableFuture<LedgerMetadata> promise) {
+ private void writeLoop(Versioned<LedgerMetadata> currentLocal,
+ CompletableFuture<Versioned<LedgerMetadata>> promise) {
LOG.debug("{} starting write loop iteration, attempt {}",
logContext, WRITE_LOOP_COUNT_UPDATER.incrementAndGet(this));
try {
- if (needsTransformation.needsUpdate(currentLocal)) {
- LedgerMetadata transformed = transform.transform(currentLocal);
+ if (needsTransformation.needsUpdate(currentLocal.getValue())) {
+ LedgerMetadata transformed = transform.transform(currentLocal.getValue());
- writeToStore(ledgerId, transformed)
+ writeToStore(ledgerId, transformed, currentLocal.getVersion())
.whenComplete((writtenMetadata, ex) -> {
if (ex == null) {
if (updateLocalValue.updateValue(currentLocal, writtenMetadata)) {
@@ -153,16 +155,16 @@ private void writeLoop(LedgerMetadata currentLocal, CompletableFuture<LedgerMeta
}
}
- private CompletableFuture<LedgerMetadata> updateLocalValueFromStore(long ledgerId) {
- CompletableFuture<LedgerMetadata> promise = new CompletableFuture<>();
+ private CompletableFuture<Versioned<LedgerMetadata>> updateLocalValueFromStore(long ledgerId) {
+ CompletableFuture<Versioned<LedgerMetadata>> promise = new CompletableFuture<>();
readLoop(ledgerId, promise);
return promise;
}
- private void readLoop(long ledgerId, CompletableFuture<LedgerMetadata> promise) {
- LedgerMetadata current = currentLocalValue.get();
+ private void readLoop(long ledgerId, CompletableFuture<Versioned<LedgerMetadata>> promise) {
+ Versioned<LedgerMetadata> current = currentLocalValue.get();
lm.readLedgerMetadata(ledgerId,
(rc, read) -> {
@@ -185,10 +187,11 @@ private void readLoop(long ledgerId, CompletableFuture<LedgerMetadata> promise)
});
}
- private CompletableFuture<LedgerMetadata> writeToStore(long ledgerId, LedgerMetadata toWrite) {
- CompletableFuture<LedgerMetadata> promise = new CompletableFuture<>();
+ private CompletableFuture<Versioned<LedgerMetadata>> writeToStore(long ledgerId, LedgerMetadata toWrite,
+ Version currentVersion) {
+ CompletableFuture<Versioned<LedgerMetadata>> promise = new CompletableFuture<>();
- lm.writeLedgerMetadata(ledgerId, toWrite,
+ lm.writeLedgerMetadata(ledgerId, toWrite, currentVersion,
(rc, written) -> {
if (rc != BKException.Code.OK) {
promise.completeExceptionally(BKException.create(rc));
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
index e4794de575..9c2901ccee 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java
@@ -45,6 +45,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.TimedGenericCallback;
import org.apache.bookkeeper.util.SafeRunnable;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,25 +64,22 @@
class MetadataUpdater extends SafeRunnable {
- final LedgerMetadata newMetadata;
+ final Versioned<LedgerMetadata> newMetadata;
- MetadataUpdater(LedgerMetadata metadata) {
+ MetadataUpdater(Versioned<LedgerMetadata> metadata) {
this.newMetadata = metadata;
}
@Override
public void safeRun() {
while (true) {
- LedgerMetadata currentMetadata = getLedgerMetadata();
+ Versioned<LedgerMetadata> currentMetadata = getVersionedLedgerMetadata();
Version.Occurred occurred = currentMetadata.getVersion().compare(newMetadata.getVersion());
if (Version.Occurred.BEFORE == occurred) {
- LOG.info("Updated ledger metadata for ledger {} to {}.", ledgerId, newMetadata.toSafeString());
synchronized (ReadOnlyLedgerHandle.this) {
- if (newMetadata.isClosed()) {
- ReadOnlyLedgerHandle.this.lastAddConfirmed = newMetadata.getLastEntryId();
- ReadOnlyLedgerHandle.this.length = newMetadata.getLength();
- }
if (setLedgerMetadata(currentMetadata, newMetadata)) {
+ LOG.info("Updated ledger metadata for ledger {} to {}, version {}.",
+ ledgerId, newMetadata.getValue().toSafeString(), newMetadata.getVersion());
break;
}
}
@@ -98,7 +96,7 @@ public String toString() {
}
ReadOnlyLedgerHandle(ClientContext clientCtx,
- long ledgerId, LedgerMetadata metadata,
+ long ledgerId, Versioned<LedgerMetadata> metadata,
BookKeeper.DigestType digestType, byte[] password,
boolean watch)
throws GeneralSecurityException, NumberFormatException {
@@ -147,7 +145,7 @@ public void asyncAddEntry(final byte[] data, final int offset, final int length,
}
@Override
- public void onChanged(long lid, LedgerMetadata newMetadata) {
+ public void onChanged(long lid, Versioned<LedgerMetadata> newMetadata) {
if (LOG.isDebugEnabled()) {
LOG.debug("Received ledger metadata update on {} : {}", lid, newMetadata);
}
@@ -157,7 +155,7 @@ public void onChanged(long lid, LedgerMetadata newMetadata) {
if (null == newMetadata) {
return;
}
- LedgerMetadata currentMetadata = getLedgerMetadata();
+ Versioned<LedgerMetadata> currentMetadata = getVersionedLedgerMetadata();
Version.Occurred occurred = currentMetadata.getVersion().compare(newMetadata.getVersion());
if (LOG.isDebugEnabled()) {
LOG.debug("Try to update metadata from {} to {} : {}",
@@ -276,13 +274,13 @@ void recover(GenericCallback<Void> finalCb,
}
new MetadataUpdateLoop(
clientCtx.getLedgerManager(), getId(),
- this::getLedgerMetadata,
+ this::getVersionedLedgerMetadata,
needsUpdate,
(metadata) -> LedgerMetadataBuilder.from(metadata).withInRecoveryState().build(),
this::setLedgerMetadata)
.run()
.thenCompose((metadata) -> {
- if (metadata.isClosed()) {
+ if (metadata.getValue().isClosed()) {
return CompletableFuture.completedFuture(ReadOnlyLedgerHandle.this);
} else {
return new LedgerRecoveryOp(ReadOnlyLedgerHandle.this, clientCtx)
@@ -301,16 +299,16 @@ void recover(GenericCallback<Void> finalCb,
});
}
- CompletableFuture<LedgerMetadata> closeRecovered() {
+ CompletableFuture<Versioned<LedgerMetadata>> closeRecovered() {
long lac, len;
synchronized (this) {
lac = lastAddConfirmed;
len = length;
}
LOG.info("Closing recovered ledger {} at entry {}", getId(), lac);
- CompletableFuture<LedgerMetadata> f = new MetadataUpdateLoop(
+ CompletableFuture<Versioned<LedgerMetadata>> f = new MetadataUpdateLoop(
clientCtx.getLedgerManager(), getId(),
- this::getLedgerMetadata,
+ this::getVersionedLedgerMetadata,
(metadata) -> metadata.isInRecovery(),
(metadata) -> {
LedgerMetadataBuilder builder = LedgerMetadataBuilder.from(metadata);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/UpdateLedgerOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/UpdateLedgerOp.java
index 37f8b096d6..fec6e58af4 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/UpdateLedgerOp.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/UpdateLedgerOp.java
@@ -38,6 +38,7 @@
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallbackFuture;
+import org.apache.bookkeeper.versioning.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,10 +95,10 @@ public void updateBookieIdInLedgers(final BookieSocketAddress oldBookieId, final
final long ledgerId = ledgerItr.next();
issuedLedgerCnt.incrementAndGet();
- GenericCallbackFuture<LedgerMetadata> readPromise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> readPromise = new GenericCallbackFuture<>();
lm.readLedgerMetadata(ledgerId, readPromise);
- CompletableFuture<LedgerMetadata> writePromise = readPromise.thenCompose((readMetadata) -> {
- AtomicReference<LedgerMetadata> ref = new AtomicReference<>(readMetadata);
+ CompletableFuture<Versioned<LedgerMetadata>> writePromise = readPromise.thenCompose((readMetadata) -> {
+ AtomicReference<Versioned<LedgerMetadata>> ref = new AtomicReference<>(readMetadata);
return new MetadataUpdateLoop(
lm, ledgerId,
ref::get,
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java
index ccfbb52ecf..b0af1ba849 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java
@@ -46,6 +46,7 @@
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
@@ -85,7 +86,7 @@
/**
* ReadLedgerMetadataTask class.
*/
- protected class ReadLedgerMetadataTask implements Runnable, GenericCallback<LedgerMetadata> {
+ protected class ReadLedgerMetadataTask implements Runnable, GenericCallback<Versioned<LedgerMetadata>> {
final long ledgerId;
@@ -108,7 +109,7 @@ public void run() {
}
@Override
- public void operationComplete(int rc, final LedgerMetadata result) {
+ public void operationComplete(int rc, final Versioned<LedgerMetadata> result) {
if (BKException.Code.OK == rc) {
final Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
if (null != listenerSet) {
@@ -245,15 +246,13 @@ public void process(WatchedEvent event) {
@Override
public void createLedgerMetadata(final long ledgerId, final LedgerMetadata metadata,
- final GenericCallback<LedgerMetadata> ledgerCb) {
+ final GenericCallback<Versioned<LedgerMetadata>> ledgerCb) {
String ledgerPath = getLedgerPath(ledgerId);
StringCallback scb = new StringCallback() {
@Override
public void processResult(int rc, String path, Object ctx, String name) {
if (rc == Code.OK.intValue()) {
- // update version
- metadata.setVersion(new LongVersion(0));
- ledgerCb.operationComplete(BKException.Code.OK, metadata);
+ ledgerCb.operationComplete(BKException.Code.OK, new Versioned<>(metadata, new LongVersion(0)));
} else if (rc == Code.NODEEXISTS.intValue()) {
LOG.warn("Failed to create ledger metadata for {} which already exist", ledgerId);
ledgerCb.operationComplete(BKException.Code.LedgerExistException, null);
@@ -381,11 +380,11 @@ public void unregisterLedgerMetadataListener(long ledgerId, LedgerMetadataListen
}
@Override
- public void readLedgerMetadata(final long ledgerId, final GenericCallback<LedgerMetadata> readCb) {
+ public void readLedgerMetadata(final long ledgerId, final GenericCallback<Versioned<LedgerMetadata>> readCb) {
readLedgerMetadata(ledgerId, readCb, null);
}
- protected void readLedgerMetadata(final long ledgerId, final GenericCallback<LedgerMetadata> readCb,
+ protected void readLedgerMetadata(final long ledgerId, final GenericCallback<Versioned<LedgerMetadata>> readCb,
Watcher watcher) {
zk.getData(getLedgerPath(ledgerId), watcher, new DataCallback() {
@Override
@@ -409,29 +408,28 @@ public void processResult(int rc, String path, Object ctx, byte[] data, Stat sta
readCb.operationComplete(BKException.Code.ZKException, null);
return;
}
- LedgerMetadata metadata;
+
try {
- metadata = LedgerMetadata.parseConfig(data, new LongVersion(stat.getVersion()),
- Optional.of(stat.getCtime()));
+ LongVersion version = new LongVersion(stat.getVersion());
+ LedgerMetadata metadata = LedgerMetadata.parseConfig(data, Optional.of(stat.getCtime()));
+ readCb.operationComplete(BKException.Code.OK, new Versioned<>(metadata, version));
} catch (IOException e) {
LOG.error("Could not parse ledger metadata for ledger: " + ledgerId, e);
readCb.operationComplete(BKException.Code.ZKException, null);
- return;
}
- readCb.operationComplete(BKException.Code.OK, metadata);
+
}
}, null);
}
@Override
- public void writeLedgerMetadata(final long ledgerId, final LedgerMetadata metadata,
- final GenericCallback<LedgerMetadata> cb) {
- Version v = metadata.getVersion();
- if (!(v instanceof LongVersion)) {
+ public void writeLedgerMetadata(final long ledgerId, final LedgerMetadata metadata, final Version currentVersion,
+ final GenericCallback<Versioned<LedgerMetadata>> cb) {
+ if (!(currentVersion instanceof LongVersion)) {
cb.operationComplete(BKException.Code.MetadataVersionException, null);
return;
}
- final LongVersion zv = (LongVersion) v;
+ final LongVersion zv = (LongVersion) currentVersion;
zk.setData(getLedgerPath(ledgerId),
metadata.serialize(), (int) zv.getLongVersion(),
new StatCallback() {
@@ -441,8 +439,8 @@ public void processResult(int rc, String path, Object ctx, Stat stat) {
cb.operationComplete(BKException.Code.MetadataVersionException, null);
} else if (KeeperException.Code.OK.intValue() == rc) {
// update metadata version
- metadata.setVersion(zv.setLongVersion(stat.getVersion()));
- cb.operationComplete(BKException.Code.OK, metadata);
+ cb.operationComplete(BKException.Code.OK,
+ new Versioned<>(metadata, new LongVersion(stat.getVersion())));
} else if (KeeperException.Code.NONODE.intValue() == rc) {
LOG.warn("Ledger node does not exist in ZooKeeper: ledgerId={}", ledgerId);
cb.operationComplete(BKException.Code.NoSuchLedgerExistsException, null);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java
index 36f9d8c741..6a87cf2823 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java
@@ -32,6 +32,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.Processor;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback;
/**
@@ -109,14 +110,14 @@ private GenericCallback removeCallback(GenericCallback callback) {
@Override
public void createLedgerMetadata(long lid, LedgerMetadata metadata,
- GenericCallback<LedgerMetadata> cb) {
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
closeLock.readLock().lock();
try {
if (closed) {
cb.operationComplete(BKException.Code.ClientClosedException, null);
return;
}
- underlying.createLedgerMetadata(lid, metadata, new CleanupGenericCallback<LedgerMetadata>(cb));
+ underlying.createLedgerMetadata(lid, metadata, new CleanupGenericCallback<>(cb));
} finally {
closeLock.readLock().unlock();
}
@@ -140,14 +141,14 @@ public void removeLedgerMetadata(long ledgerId, Version version,
@Override
public void readLedgerMetadata(long ledgerId,
- GenericCallback<LedgerMetadata> readCb) {
+ GenericCallback<Versioned<LedgerMetadata>> readCb) {
closeLock.readLock().lock();
try {
if (closed) {
readCb.operationComplete(BKException.Code.ClientClosedException, null);
return;
}
- underlying.readLedgerMetadata(ledgerId, new CleanupGenericCallback<LedgerMetadata>(readCb));
+ underlying.readLedgerMetadata(ledgerId, new CleanupGenericCallback<>(readCb));
} finally {
closeLock.readLock().unlock();
}
@@ -155,15 +156,16 @@ public void readLedgerMetadata(long ledgerId,
@Override
public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
- GenericCallback<LedgerMetadata> cb) {
+ Version currentVersion,
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
closeLock.readLock().lock();
try {
if (closed) {
cb.operationComplete(BKException.Code.ClientClosedException, null);
return;
}
- underlying.writeLedgerMetadata(ledgerId, metadata,
- new CleanupGenericCallback<LedgerMetadata>(cb));
+ underlying.writeLedgerMetadata(ledgerId, metadata, currentVersion,
+ new CleanupGenericCallback<>(cb));
} finally {
closeLock.readLock().unlock();
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManager.java
index 84b9cb69ab..1ad2eb1c64 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManager.java
@@ -28,6 +28,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.Processor;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback;
/**
@@ -55,7 +56,7 @@
* for other issue</li>
* </ul>
*/
- void createLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericCallback<LedgerMetadata> cb);
+ void createLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericCallback<Versioned<LedgerMetadata>> cb);
/**
* Remove a specified ledger metadata by ledgerId and version.
@@ -86,7 +87,7 @@
* <li>{@link BKException.Code.ZKException} for other issue</li>
* </ul>
*/
- void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> readCb);
+ void readLedgerMetadata(long ledgerId, GenericCallback<Versioned<LedgerMetadata>> readCb);
/**
* Write ledger metadata.
@@ -95,6 +96,8 @@
* Ledger Id
* @param metadata
* Ledger Metadata to write
+ * @param currentVersion
+ * The version of the metadata we expect to be overwriting.
* @param cb
* Callback when finished writing ledger metadata, returning the written metadata.
* Return code:<ul>
@@ -103,7 +106,9 @@
* <li>{@link BKException.Code.ZKException} for other issue</li>
* </ul>
*/
- void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericCallback<LedgerMetadata> cb);
+ void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
+ Version currentVersion,
+ GenericCallback<Versioned<LedgerMetadata>> cb);
/**
* Register the ledger metadata <i>listener</i> on <i>ledgerId</i>.
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java
index 5b28a0be48..d8fc4f2b5f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java
@@ -224,7 +224,7 @@ public LedgerIdGenerator newLedgerIdGenerator() {
// callbacks
ScheduledExecutorService scheduler;
- protected class ReadLedgerMetadataTask implements Runnable, GenericCallback<LedgerMetadata> {
+ protected class ReadLedgerMetadataTask implements Runnable, GenericCallback<Versioned<LedgerMetadata>> {
final long ledgerId;
@@ -247,12 +247,12 @@ public void run() {
}
@Override
- public void operationComplete(int rc, final LedgerMetadata result) {
+ public void operationComplete(int rc, final Versioned<LedgerMetadata> result) {
if (BKException.Code.OK == rc) {
final Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
if (null != listenerSet) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Ledger metadata is changed for {} : {}.", ledgerId, result);
+ LOG.debug("Ledger metadata is changed for {} : {}.", ledgerId, result.getValue());
}
scheduler.submit(new Runnable() {
@Override
@@ -376,7 +376,7 @@ public void close() {
@Override
public void createLedgerMetadata(final long lid, final LedgerMetadata metadata,
- final GenericCallback<LedgerMetadata> ledgerCb) {
+ final GenericCallback<Versioned<LedgerMetadata>> ledgerCb) {
MetastoreCallback<Version> msCallback = new MetastoreCallback<Version>() {
@Override
public void complete(int rc, Version version, Object ctx) {
@@ -391,9 +391,7 @@ public void complete(int rc, Version version, Object ctx) {
if (LOG.isDebugEnabled()) {
LOG.debug("Create ledger {} with version {} successfully.", lid, version);
}
- // update version
- metadata.setVersion(version);
- ledgerCb.operationComplete(BKException.Code.OK, metadata);
+ ledgerCb.operationComplete(BKException.Code.OK, new Versioned<>(metadata, version));
}
};
@@ -423,7 +421,7 @@ public void complete(int rc, Void value, Object ctx) {
}
@Override
- public void readLedgerMetadata(final long ledgerId, final GenericCallback<LedgerMetadata> readCb) {
+ public void readLedgerMetadata(long ledgerId, GenericCallback<Versioned<LedgerMetadata>> readCb) {
final String key = ledgerId2Key(ledgerId);
MetastoreCallback<Versioned<Value>> msCallback = new MetastoreCallback<Versioned<Value>>() {
@Override
@@ -440,54 +438,49 @@ public void complete(int rc, Versioned<Value> value, Object ctx) {
readCb.operationComplete(BKException.Code.MetaStoreException, null);
return;
}
- LedgerMetadata metadata;
try {
- metadata = LedgerMetadata.parseConfig(value.getValue().getField(META_FIELD),
- value.getVersion(), Optional.<Long>absent());
+ LedgerMetadata metadata = LedgerMetadata.parseConfig(
+ value.getValue().getField(META_FIELD), Optional.<Long>absent());
+ readCb.operationComplete(BKException.Code.OK, new Versioned<>(metadata, value.getVersion()));
} catch (IOException e) {
LOG.error("Could not parse ledger metadata for ledger " + ledgerId + " : ", e);
readCb.operationComplete(BKException.Code.MetaStoreException, null);
- return;
}
- readCb.operationComplete(BKException.Code.OK, metadata);
}
};
ledgerTable.get(key, this, msCallback, ALL_FIELDS);
}
@Override
- public void writeLedgerMetadata(final long ledgerId, final LedgerMetadata metadata,
- final GenericCallback<LedgerMetadata> cb) {
+ public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
+ Version currentVersion,
+ final GenericCallback<Versioned<LedgerMetadata>> cb) {
Value data = new Value().setField(META_FIELD, metadata.serialize());
if (LOG.isDebugEnabled()) {
- LOG.debug("Writing ledger {} metadata, version {}", new Object[] { ledgerId, metadata.getVersion() });
+ LOG.debug("Writing ledger {} metadata, version {}", new Object[] { ledgerId, currentVersion });
}
final String key = ledgerId2Key(ledgerId);
MetastoreCallback<Version> msCallback = new MetastoreCallback<Version>() {
@Override
public void complete(int rc, Version version, Object ctx) {
- int bkRc;
if (MSException.Code.BadVersion.getCode() == rc) {
LOG.info("Bad version provided to updat metadata for ledger {}", ledgerId);
- bkRc = BKException.Code.MetadataVersionException;
+ cb.operationComplete(BKException.Code.MetadataVersionException, null);
} else if (MSException.Code.NoKey.getCode() == rc) {
LOG.warn("Ledger {} doesn't exist when writing its ledger metadata.", ledgerId);
- bkRc = BKException.Code.NoSuchLedgerExistsException;
+ cb.operationComplete(BKException.Code.NoSuchLedgerExistsException, null);
} else if (MSException.Code.OK.getCode() == rc) {
- metadata.setVersion(version);
- bkRc = BKException.Code.OK;
+ cb.operationComplete(BKException.Code.OK, new Versioned<>(metadata, version));
} else {
LOG.warn("Conditional update ledger metadata failed: ",
MSException.create(MSException.Code.get(rc), "Failed to put key " + key));
- bkRc = BKException.Code.MetaStoreException;
+ cb.operationComplete(BKException.Code.MetaStoreException, null);
}
-
- cb.operationComplete(bkRc, metadata);
}
};
- ledgerTable.put(key, data, metadata.getVersion(), msCallback, null);
+ ledgerTable.put(key, data, currentVersion, msCallback, null);
}
@Override
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java
index d5ed5aeb8d..8412e06f42 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java
@@ -38,6 +38,7 @@
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.util.MathUtils;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,7 +69,7 @@
* @param metadata
* new ledger metadata.
*/
- void onChanged(long ledgerId, LedgerMetadata metadata);
+ void onChanged(long ledgerId, Versioned<LedgerMetadata> metadata);
}
/**
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/BookieLedgerIndexer.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/BookieLedgerIndexer.java
index 80bc1a5c9c..5227176b5b 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/BookieLedgerIndexer.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/BookieLedgerIndexer.java
@@ -33,6 +33,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.Processor;
import org.apache.bookkeeper.replication.ReplicationException.BKAuditException;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,32 +67,24 @@ public BookieLedgerIndexer(LedgerManager ledgerManager) {
Processor<Long> ledgerProcessor = new Processor<Long>() {
@Override
- public void process(final Long ledgerId,
- final AsyncCallback.VoidCallback iterCallback) {
- GenericCallback<LedgerMetadata> genericCallback = new GenericCallback<LedgerMetadata>() {
- @Override
- public void operationComplete(int rc,
- LedgerMetadata ledgerMetadata) {
- if (rc == BKException.Code.OK) {
- for (Map.Entry<Long, ? extends List<BookieSocketAddress>> ensemble : ledgerMetadata
- .getAllEnsembles().entrySet()) {
- for (BookieSocketAddress bookie : ensemble
- .getValue()) {
- putLedger(bookie2ledgersMap,
- bookie.toString(),
- ledgerId);
- }
+ public void process(Long ledgerId, AsyncCallback.VoidCallback iterCallback) {
+ GenericCallback<Versioned<LedgerMetadata>> genericCallback = (rc, ledgerMetadata) -> {
+ if (rc == BKException.Code.OK) {
+ for (Map.Entry<Long, ? extends List<BookieSocketAddress>> ensemble
+ : ledgerMetadata.getValue().getAllEnsembles().entrySet()) {
+ for (BookieSocketAddress bookie : ensemble.getValue()) {
+ putLedger(bookie2ledgersMap, bookie.toString(), ledgerId);
}
- } else if (rc == BKException.Code.NoSuchLedgerExistsException) {
- LOG.info("Ignoring replication of already deleted ledger {}",
- ledgerId);
- rc = BKException.Code.OK;
- } else {
- LOG.warn("Unable to read the ledger:" + ledgerId
- + " information");
}
- iterCallback.processResult(rc, null, null);
+ } else if (rc == BKException.Code.NoSuchLedgerExistsException) {
+ LOG.info("Ignoring replication of already deleted ledger {}",
+ ledgerId);
+ rc = BKException.Code.OK;
+ } else {
+ LOG.warn("Unable to read the ledger:" + ledgerId
+ + " information");
}
+ iterCallback.processResult(rc, null, null);
};
ledgerManager.readLedgerMetadata(ledgerId, genericCallback);
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ListLedgerService.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ListLedgerService.java
index efd00e7341..928b33a266 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ListLedgerService.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ListLedgerService.java
@@ -39,6 +39,7 @@
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
+import org.apache.bookkeeper.versioning.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,7 +69,7 @@ public ListLedgerService(ServerConfiguration conf, BookieServer bookieServer) {
* Callback for reading ledger metadata.
*/
public static class ReadLedgerMetadataCallback extends AbstractFuture<LedgerMetadata>
- implements BookkeeperInternalCallbacks.GenericCallback<LedgerMetadata> {
+ implements BookkeeperInternalCallbacks.GenericCallback<Versioned<LedgerMetadata>> {
final long ledgerId;
ReadLedgerMetadataCallback(long ledgerId) {
@@ -79,11 +80,11 @@ long getLedgerId() {
return ledgerId;
}
- public void operationComplete(int rc, LedgerMetadata result) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> result) {
if (rc != 0) {
setException(BKException.create(rc));
} else {
- set(result);
+ set(result.getValue());
}
}
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java
index dbf09d9814..6e7b23da2f 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java
@@ -73,6 +73,7 @@
import org.apache.bookkeeper.util.MathUtils;
import org.apache.bookkeeper.util.TestUtils;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback;
import org.junit.Before;
import org.junit.Test;
@@ -918,7 +919,7 @@ private LedgerManager getLedgerManager(final Set<Long> ledgers) {
LedgerManager manager = new LedgerManager() {
@Override
public void createLedgerMetadata(long lid, LedgerMetadata metadata,
- GenericCallback<LedgerMetadata> cb) {
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
unsupported();
}
@Override
@@ -927,12 +928,13 @@ public void removeLedgerMetadata(long ledgerId, Version version,
unsupported();
}
@Override
- public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> readCb) {
+ public void readLedgerMetadata(long ledgerId, GenericCallback<Versioned<LedgerMetadata>> readCb) {
unsupported();
}
@Override
public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
- GenericCallback<LedgerMetadata> cb) {
+ Version currentVersion,
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
unsupported();
}
@Override
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/TestGcOverreplicatedLedger.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/TestGcOverreplicatedLedger.java
index 1796319988..2fd686dee9 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/TestGcOverreplicatedLedger.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/TestGcOverreplicatedLedger.java
@@ -48,6 +48,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.util.SnapshotMap;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.ZooDefs;
import org.junit.Assert;
import org.junit.Before;
@@ -86,12 +87,12 @@ public void testGcOverreplicatedLedger() throws Exception {
final AtomicReference<LedgerMetadata> newLedgerMetadata = new AtomicReference<>(null);
final CountDownLatch latch = new CountDownLatch(1);
- ledgerManager.readLedgerMetadata(lh.getId(), new GenericCallback<LedgerMetadata>() {
+ ledgerManager.readLedgerMetadata(lh.getId(), new GenericCallback<Versioned<LedgerMetadata>>() {
@Override
- public void operationComplete(int rc, LedgerMetadata result) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> result) {
if (rc == BKException.Code.OK) {
- newLedgerMetadata.set(result);
+ newLedgerMetadata.set(result.getValue());
}
latch.countDown();
}
@@ -133,12 +134,12 @@ public void testNoGcOfLedger() throws Exception {
final AtomicReference<LedgerMetadata> newLedgerMetadata = new AtomicReference<>(null);
final CountDownLatch latch = new CountDownLatch(1);
- ledgerManager.readLedgerMetadata(lh.getId(), new GenericCallback<LedgerMetadata>() {
+ ledgerManager.readLedgerMetadata(lh.getId(), new GenericCallback<Versioned<LedgerMetadata>>() {
@Override
- public void operationComplete(int rc, LedgerMetadata result) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> result) {
if (rc == BKException.Code.OK) {
- newLedgerMetadata.set(result);
+ newLedgerMetadata.set(result.getValue());
}
latch.countDown();
}
@@ -184,12 +185,12 @@ public void testNoGcIfLedgerBeingReplicated() throws Exception {
final AtomicReference<LedgerMetadata> newLedgerMetadata = new AtomicReference<>(null);
final CountDownLatch latch = new CountDownLatch(1);
- ledgerManager.readLedgerMetadata(lh.getId(), new GenericCallback<LedgerMetadata>() {
+ ledgerManager.readLedgerMetadata(lh.getId(), new GenericCallback<Versioned<LedgerMetadata>>() {
@Override
- public void operationComplete(int rc, LedgerMetadata result) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> result) {
if (rc == BKException.Code.OK) {
- newLedgerMetadata.set(result);
+ newLedgerMetadata.set(result.getValue());
}
latch.countDown();
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
index e657195e44..d9bd8882b3 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookieRecoveryTest.java
@@ -49,6 +49,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.ReadEntryCallback;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.versioning.Versioned;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -559,13 +560,13 @@ public SyncLedgerMetaObject() {
private LedgerMetadata getLedgerMetadata(LedgerHandle lh) throws Exception {
final SyncLedgerMetaObject syncObj = new SyncLedgerMetaObject();
- bkc.getLedgerManager().readLedgerMetadata(lh.getId(), new GenericCallback<LedgerMetadata>() {
+ bkc.getLedgerManager().readLedgerMetadata(lh.getId(), new GenericCallback<Versioned<LedgerMetadata>>() {
@Override
- public void operationComplete(int rc, LedgerMetadata result) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> result) {
synchronized (syncObj) {
syncObj.rc = rc;
- syncObj.meta = result;
+ syncObj.meta = result.getValue();
syncObj.value = true;
syncObj.notify();
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ClientUtil.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ClientUtil.java
index 5add60f62c..e00ec5b954 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ClientUtil.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ClientUtil.java
@@ -25,15 +25,19 @@
import java.security.GeneralSecurityException;
import java.util.function.Function;
+import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallbackFuture;
import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.DigestType;
import org.apache.bookkeeper.proto.checksum.DigestManager;
import org.apache.bookkeeper.util.ByteBufList;
+import org.apache.bookkeeper.versioning.Versioned;
/**
* Client utilities.
*/
public class ClientUtil {
+ public static final org.apache.bookkeeper.client.api.DigestType DIGEST_TYPE =
+ org.apache.bookkeeper.client.api.DigestType.CRC32C;
public static final byte[] PASSWD = "foobar".getBytes(UTF_8);
public static ByteBuf generatePacket(long ledgerId, long entryId, long lastAddConfirmed,
@@ -55,21 +59,34 @@ public static boolean isLedgerOpen(LedgerHandle handle) {
return !handle.getLedgerMetadata().isClosed();
}
- public static LedgerMetadata setupLedger(ClientContext clientCtx, long ledgerId,
- LedgerMetadataBuilder builder) throws Exception {
- LedgerMetadata md = builder.withPassword(PASSWD).build();
- GenericCallbackFuture<LedgerMetadata> mdPromise = new GenericCallbackFuture<>();
- clientCtx.getLedgerManager().createLedgerMetadata(ledgerId, md, mdPromise);
+ public static Versioned<LedgerMetadata> setupLedger(ClientContext clientCtx, long ledgerId,
+ LedgerMetadataBuilder builder) throws Exception {
+ return setupLedger(clientCtx.getLedgerManager(), ledgerId, builder);
+ }
+
+ public static Versioned<LedgerMetadata> setupLedger(LedgerManager ledgerManager, long ledgerId,
+ LedgerMetadataBuilder builder) throws Exception {
+ LedgerMetadata md = builder.withPassword(PASSWD).withDigestType(DIGEST_TYPE).build();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> mdPromise = new GenericCallbackFuture<>();
+ ledgerManager.createLedgerMetadata(ledgerId, md, mdPromise);
return mdPromise.get();
}
- public static LedgerMetadata transformMetadata(ClientContext clientCtx, long ledgerId,
- Function<LedgerMetadata, LedgerMetadata> transform)
+ public static Versioned<LedgerMetadata> transformMetadata(ClientContext clientCtx, long ledgerId,
+ Function<LedgerMetadata, LedgerMetadata> transform)
+ throws Exception {
+ return transformMetadata(clientCtx.getLedgerManager(), ledgerId, transform);
+ }
+
+ public static Versioned<LedgerMetadata> transformMetadata(LedgerManager ledgerManager, long ledgerId,
+ Function<LedgerMetadata, LedgerMetadata> transform)
throws Exception {
- GenericCallbackFuture<LedgerMetadata> readPromise = new GenericCallbackFuture<>();
- GenericCallbackFuture<LedgerMetadata> writePromise = new GenericCallbackFuture<>();
- clientCtx.getLedgerManager().readLedgerMetadata(ledgerId, readPromise);
- clientCtx.getLedgerManager().writeLedgerMetadata(ledgerId, transform.apply(readPromise.get()), writePromise);
+ GenericCallbackFuture<Versioned<LedgerMetadata>> readPromise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> writePromise = new GenericCallbackFuture<>();
+ ledgerManager.readLedgerMetadata(ledgerId, readPromise);
+ ledgerManager.writeLedgerMetadata(ledgerId, transform.apply(readPromise.get().getValue()),
+ readPromise.get().getVersion(), writePromise);
return writePromise.get();
}
+
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/HandleFailuresTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/HandleFailuresTest.java
index ce26680987..3a547b9f07 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/HandleFailuresTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/HandleFailuresTest.java
@@ -36,6 +36,7 @@
import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.versioning.Versioned;
import org.junit.Assert;
import org.junit.Test;
@@ -58,7 +59,7 @@
@Test
public void testChangeTriggeredOneTimeForOneFailure() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create().newEnsembleEntry(
0L, Lists.newArrayList(b1, b2, b3)));
@@ -73,7 +74,7 @@ public void testChangeTriggeredOneTimeForOneFailure() throws Exception {
lh.appendAsync("entry4".getBytes());
lh.appendAsync("entry5".getBytes()).get();
- verify(clientCtx.getLedgerManager(), times(1)).writeLedgerMetadata(anyLong(), any(), any());
+ verify(clientCtx.getLedgerManager(), times(1)).writeLedgerMetadata(anyLong(), any(), any(), any());
Assert.assertEquals(lh.getLedgerMetadata().getAllEnsembles().size(), 1);
Assert.assertEquals(lh.getLedgerMetadata().getAllEnsembles().get(0L), Lists.newArrayList(b4, b2, b3));
}
@@ -81,7 +82,7 @@ public void testChangeTriggeredOneTimeForOneFailure() throws Exception {
@Test
public void testSecondFailureOccursWhileFirstBeingHandled() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -119,7 +120,7 @@ public void testSecondFailureOccursWhileFirstBeingHandled() throws Exception {
metadataBlocker.complete(null);
future.get();
- verify(clientCtx.getLedgerManager(), times(2)).writeLedgerMetadata(anyLong(), any(), any());
+ verify(clientCtx.getLedgerManager(), times(2)).writeLedgerMetadata(anyLong(), any(), any(), any());
Assert.assertEquals(lh.getLedgerMetadata().getAllEnsembles().size(), 1);
Assert.assertTrue(lh.getLedgerMetadata().getAllEnsembles().get(0L).contains(b3));
Assert.assertTrue(lh.getLedgerMetadata().getAllEnsembles().get(0L).contains(b4));
@@ -129,7 +130,7 @@ public void testSecondFailureOccursWhileFirstBeingHandled() throws Exception {
@Test
public void testHandlingFailuresOneBookieFailsImmediately() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -149,7 +150,7 @@ public void testHandlingFailuresOneBookieFailsImmediately() throws Exception {
@Test
public void testHandlingFailuresOneBookieFailsAfterOneEntry() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -172,7 +173,7 @@ public void testHandlingFailuresOneBookieFailsAfterOneEntry() throws Exception {
@Test
public void testHandlingFailuresMultipleBookieFailImmediatelyNotEnoughToReplace() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -198,7 +199,7 @@ public void testHandlingFailuresMultipleBookieFailImmediatelyNotEnoughToReplace(
@Test
public void testHandlingFailuresMultipleBookieFailAfterOneEntryNotEnoughToReplace() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -226,7 +227,7 @@ public void testHandlingFailuresMultipleBookieFailAfterOneEntryNotEnoughToReplac
@Test
public void testClientClosesWhileFailureHandlerInProgress() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -267,8 +268,8 @@ public void testClientClosesWhileFailureHandlerInProgress() throws Exception {
@Test
public void testMetadataSetToClosedDuringFailureHandler() throws Exception {
- MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ MockClientContext clientCtx = MockClientContext.create();
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -310,8 +311,8 @@ public void testMetadataSetToClosedDuringFailureHandler() throws Exception {
@Test
public void testMetadataSetToInRecoveryDuringFailureHandler() throws Exception {
- MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ MockClientContext clientCtx = MockClientContext.create();
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -353,7 +354,7 @@ public void testMetadataSetToInRecoveryDuringFailureHandler() throws Exception {
@Test
public void testOldEnsembleChangedDuringFailureHandler() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -406,7 +407,7 @@ public void testOldEnsembleChangedDuringFailureHandler() throws Exception {
@Test
public void testNoAddsAreCompletedWhileFailureHandlingInProgress() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerClose2Test.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerClose2Test.java
index f2db821dfd..93846bb68f 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerClose2Test.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerClose2Test.java
@@ -23,6 +23,7 @@
import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.versioning.Versioned;
import org.junit.Assert;
import org.junit.Test;
@@ -47,9 +48,8 @@ public void testTryAddAfterCloseHasBeenCalled() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
for (int i = 0; i < 1000; i++) {
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, i,
- LedgerMetadataBuilder.create().newEnsembleEntry(
- 0L, Lists.newArrayList(b1, b2, b3)));
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, i,
+ LedgerMetadataBuilder.create().newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
LedgerHandle lh = new LedgerHandle(clientCtx, i, md, BookKeeper.DigestType.CRC32C,
ClientUtil.PASSWD, WriteFlag.NONE);
CompletableFuture<?> closeFuture = lh.closeAsync();
@@ -71,7 +71,7 @@ public void testTryAddAfterCloseHasBeenCalled() throws Exception {
@Test
public void testMetadataChangedDuringClose() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -115,7 +115,7 @@ public void testMetadataChangedDuringClose() throws Exception {
@Test
public void testMetadataCloseWithCorrectLengthDuringClose() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -156,7 +156,7 @@ public void testMetadataCloseWithCorrectLengthDuringClose() throws Exception {
@Test
public void testMetadataCloseWithDifferentLengthDuringClose() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -197,7 +197,7 @@ public void testMetadataCloseWithDifferentLengthDuringClose() throws Exception {
@Test
public void testMetadataCloseMarkedInRecoveryWhileClosing() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
@@ -237,7 +237,7 @@ public void testMetadataCloseMarkedInRecoveryWhileClosing() throws Exception {
@Test
public void testCloseWhileAddInProgress() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = ClientUtil.setupLedger(clientCtx, 10L,
+ Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 10L,
LedgerMetadataBuilder.create()
.withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b1, b2, b3)));
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerRecovery2Test.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerRecovery2Test.java
index 2dba8c3212..4f483a6df5 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerRecovery2Test.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/LedgerRecovery2Test.java
@@ -28,6 +28,7 @@
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallbackFuture;
+import org.apache.bookkeeper.versioning.Versioned;
import org.junit.Assert;
import org.junit.Test;
@@ -47,12 +48,12 @@
private static final BookieSocketAddress b4 = new BookieSocketAddress("b4", 3181);
private static final BookieSocketAddress b5 = new BookieSocketAddress("b5", 3181);
- private static LedgerMetadata setupLedger(ClientContext clientCtx, long ledgerId,
+ private static Versioned<LedgerMetadata> setupLedger(ClientContext clientCtx, long ledgerId,
List<BookieSocketAddress> bookies) throws Exception {
LedgerMetadata md = LedgerMetadataBuilder.create()
.withPassword(PASSWD)
.newEnsembleEntry(0, bookies).build();
- GenericCallbackFuture<LedgerMetadata> mdPromise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> mdPromise = new GenericCallbackFuture<>();
clientCtx.getLedgerManager().createLedgerMetadata(1L, md, mdPromise);
return mdPromise.get();
}
@@ -61,7 +62,7 @@ private static LedgerMetadata setupLedger(ClientContext clientCtx, long ledgerId
public void testCantRecoverAllDown() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = setupLedger(clientCtx, 1L, Lists.newArrayList(b1, b2, b3));
+ Versioned<LedgerMetadata> md = setupLedger(clientCtx, 1L, Lists.newArrayList(b1, b2, b3));
clientCtx.getMockBookieClient().errorBookies(b1, b2, b3);
@@ -81,7 +82,7 @@ public void testCantRecoverAllDown() throws Exception {
public void testCanReadLacButCantWrite() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
+ Versioned<LedgerMetadata> md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
clientCtx.getMockBookieClient().seedEntries(b1, 1L, 0L, -1L);
clientCtx.getMockBookieClient().setPreWriteHook(
@@ -103,7 +104,7 @@ public void testCanReadLacButCantWrite() throws Exception {
public void testMetadataClosedDuringRecovery() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
- LedgerMetadata md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
+ Versioned<LedgerMetadata> md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
CompletableFuture<Void> writingBack = new CompletableFuture<>();
CompletableFuture<Void> blocker = new CompletableFuture<>();
@@ -123,12 +124,8 @@ public void testMetadataClosedDuringRecovery() throws Exception {
writingBack.get(10, TimeUnit.SECONDS);
- GenericCallbackFuture<LedgerMetadata> readPromise = new GenericCallbackFuture<>();
- clientCtx.getLedgerManager().readLedgerMetadata(1L, readPromise);
- LedgerMetadataBuilder builder = LedgerMetadataBuilder.from(readPromise.get());
- GenericCallbackFuture<LedgerMetadata> writePromise = new GenericCallbackFuture<>();
- clientCtx.getLedgerManager().writeLedgerMetadata(1L, builder.closingAt(-1, 0).build(), writePromise);
- writePromise.get();
+ ClientUtil.transformMetadata(clientCtx, 1L,
+ (metadata) -> LedgerMetadataBuilder.from(metadata).closingAt(-1, 0).build());
// allow recovery to continue
blocker.complete(null);
@@ -144,7 +141,7 @@ public void testNewEnsembleAddedDuringRecovery() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
clientCtx.getMockRegistrationClient().addBookies(b4).get();
- LedgerMetadata md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
+ Versioned<LedgerMetadata> md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
CompletableFuture<Void> writingBack = new CompletableFuture<>();
CompletableFuture<Void> blocker = new CompletableFuture<>();
@@ -169,13 +166,9 @@ public void testNewEnsembleAddedDuringRecovery() throws Exception {
writingBack.get(10, TimeUnit.SECONDS);
- GenericCallbackFuture<LedgerMetadata> readPromise = new GenericCallbackFuture<>();
- clientCtx.getLedgerManager().readLedgerMetadata(1L, readPromise);
- LedgerMetadata newMeta = LedgerMetadataBuilder.from(readPromise.get())
- .newEnsembleEntry(1L, Lists.newArrayList(b1, b2, b4)).build();
- GenericCallbackFuture<LedgerMetadata> writePromise = new GenericCallbackFuture<>();
- clientCtx.getLedgerManager().writeLedgerMetadata(1L, newMeta, writePromise);
- writePromise.get();
+ ClientUtil.transformMetadata(clientCtx, 1L,
+ (metadata) -> LedgerMetadataBuilder.from(metadata).newEnsembleEntry(1L, Lists.newArrayList(b1, b2, b4))
+ .build());
// allow recovery to continue
failing.completeExceptionally(new BKException.BKWriteException());
@@ -194,7 +187,7 @@ public void testRecoveryBookieFailedAtStart() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
clientCtx.getMockRegistrationClient().addBookies(b4).get();
- LedgerMetadata md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
+ Versioned<LedgerMetadata> md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
CompletableFuture<Void> writingBack = new CompletableFuture<>();
CompletableFuture<Void> blocker = new CompletableFuture<>();
@@ -219,7 +212,7 @@ public void testRecoveryOneBookieFailsDuring() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
clientCtx.getMockRegistrationClient().addBookies(b4).get();
- LedgerMetadata md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
+ Versioned<LedgerMetadata> md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
clientCtx.getMockBookieClient().seedEntries(b1, 1L, 0L, -1L);
clientCtx.getMockBookieClient().seedEntries(b3, 1L, 1L, -1L);
clientCtx.getMockBookieClient().setPreWriteHook(
@@ -251,7 +244,7 @@ public void testRecoveryTwoBookiesFailOnSameEntry() throws Exception {
MockClientContext clientCtx = MockClientContext.create();
clientCtx.getMockRegistrationClient().addBookies(b4, b5).get();
- LedgerMetadata md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
+ Versioned<LedgerMetadata> md = setupLedger(clientCtx, 1, Lists.newArrayList(b1, b2, b3));
clientCtx.getMockBookieClient().seedEntries(b1, 1L, 0L, -1L);
clientCtx.getMockBookieClient().setPreWriteHook(
(bookie, ledgerId, entryId) -> {
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MetadataUpdateLoopTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MetadataUpdateLoopTest.java
index e70a7d31e1..7154d94239 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MetadataUpdateLoopTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MetadataUpdateLoopTest.java
@@ -39,12 +39,16 @@
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.MockLedgerManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallbackFuture;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.commons.lang3.tuple.Triple;
import org.junit.Assert;
@@ -72,12 +76,12 @@ public void testBasicUpdate() throws Exception {
new BookieSocketAddress("0.0.0.2:3181"),
new BookieSocketAddress("0.0.0.3:3181"),
new BookieSocketAddress("0.0.0.4:3181"))).build();
- GenericCallbackFuture<LedgerMetadata> promise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> promise = new GenericCallbackFuture<>();
long ledgerId = 1234L;
lm.createLedgerMetadata(ledgerId, initMeta, promise);
- LedgerMetadata writtenMetadata = promise.get();
+ Versioned<LedgerMetadata> writtenMetadata = promise.get();
- AtomicReference<LedgerMetadata> reference = new AtomicReference<>(writtenMetadata);
+ AtomicReference<Versioned<LedgerMetadata>> reference = new AtomicReference<>(writtenMetadata);
BookieSocketAddress newAddress = new BookieSocketAddress("0.0.0.5:3181");
MetadataUpdateLoop loop = new MetadataUpdateLoop(
@@ -94,7 +98,7 @@ public void testBasicUpdate() throws Exception {
loop.run().get();
Assert.assertNotEquals(reference.get(), writtenMetadata);
- Assert.assertEquals(reference.get().getEnsemble(0L).get(0), newAddress);
+ Assert.assertEquals(reference.get().getValue().getEnsemble(0L).get(0), newAddress);
}
}
@@ -115,12 +119,12 @@ public void testConflictOnWrite() throws Exception {
LedgerMetadata initMeta = LedgerMetadataBuilder.create().withEnsembleSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b0, b1)).build();
- GenericCallbackFuture<LedgerMetadata> promise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> promise = new GenericCallbackFuture<>();
lm.createLedgerMetadata(ledgerId, initMeta, promise);
- LedgerMetadata writtenMetadata = promise.get();
+ Versioned<LedgerMetadata> writtenMetadata = promise.get();
- AtomicReference<LedgerMetadata> reference1 = new AtomicReference<>(writtenMetadata);
- CompletableFuture<LedgerMetadata> loop1 = new MetadataUpdateLoop(
+ AtomicReference<Versioned<LedgerMetadata>> reference1 = new AtomicReference<>(writtenMetadata);
+ CompletableFuture<Versioned<LedgerMetadata>> loop1 = new MetadataUpdateLoop(
lm,
ledgerId,
reference1::get,
@@ -132,8 +136,8 @@ public void testConflictOnWrite() throws Exception {
},
reference1::compareAndSet).run();
- AtomicReference<LedgerMetadata> reference2 = new AtomicReference<>(writtenMetadata);
- CompletableFuture<LedgerMetadata> loop2 = new MetadataUpdateLoop(
+ AtomicReference<Versioned<LedgerMetadata>> reference2 = new AtomicReference<>(writtenMetadata);
+ CompletableFuture<Versioned<LedgerMetadata>> loop2 = new MetadataUpdateLoop(
lm,
ledgerId,
reference2::get,
@@ -147,21 +151,21 @@ public void testConflictOnWrite() throws Exception {
lm.releaseWrites();
- LedgerMetadata l1meta = loop1.get();
- LedgerMetadata l2meta = loop2.get();
+ Versioned<LedgerMetadata> l1meta = loop1.get();
+ Versioned<LedgerMetadata> l2meta = loop2.get();
Assert.assertEquals(l1meta, reference1.get());
Assert.assertEquals(l2meta, reference2.get());
Assert.assertEquals(l1meta.getVersion().compare(l2meta.getVersion()), Version.Occurred.BEFORE);
- Assert.assertEquals(l1meta.getEnsemble(0L).get(0), b2);
- Assert.assertEquals(l1meta.getEnsemble(0L).get(1), b1);
+ Assert.assertEquals(l1meta.getValue().getEnsemble(0L).get(0), b2);
+ Assert.assertEquals(l1meta.getValue().getEnsemble(0L).get(1), b1);
- Assert.assertEquals(l2meta.getEnsemble(0L).get(0), b2);
- Assert.assertEquals(l2meta.getEnsemble(0L).get(1), b3);
+ Assert.assertEquals(l2meta.getValue().getEnsemble(0L).get(0), b2);
+ Assert.assertEquals(l2meta.getValue().getEnsemble(0L).get(1), b3);
- verify(lm, times(3)).writeLedgerMetadata(anyLong(), any(), any());
+ verify(lm, times(3)).writeLedgerMetadata(anyLong(), any(), any(), any());
}
}
@@ -182,13 +186,13 @@ public void testConflictOnWriteBothWritingSame() throws Exception {
LedgerMetadata initMeta = LedgerMetadataBuilder.create().withEnsembleSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b0, b1)).build();
- GenericCallbackFuture<LedgerMetadata> promise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> promise = new GenericCallbackFuture<>();
lm.createLedgerMetadata(ledgerId, initMeta, promise);
- LedgerMetadata writtenMetadata = promise.get();
+ Versioned<LedgerMetadata> writtenMetadata = promise.get();
- AtomicReference<LedgerMetadata> reference = new AtomicReference<>(writtenMetadata);
+ AtomicReference<Versioned<LedgerMetadata>> reference = new AtomicReference<>(writtenMetadata);
- CompletableFuture<LedgerMetadata> loop1 = new MetadataUpdateLoop(
+ CompletableFuture<Versioned<LedgerMetadata>> loop1 = new MetadataUpdateLoop(
lm,
ledgerId,
reference::get,
@@ -199,7 +203,7 @@ public void testConflictOnWriteBothWritingSame() throws Exception {
return LedgerMetadataBuilder.from(currentMetadata).replaceEnsembleEntry(0L, ensemble).build();
},
reference::compareAndSet).run();
- CompletableFuture<LedgerMetadata> loop2 = new MetadataUpdateLoop(
+ CompletableFuture<Versioned<LedgerMetadata>> loop2 = new MetadataUpdateLoop(
lm,
ledgerId,
reference::get,
@@ -216,10 +220,10 @@ public void testConflictOnWriteBothWritingSame() throws Exception {
Assert.assertEquals(loop1.get(), loop2.get());
Assert.assertEquals(loop1.get(), reference.get());
- Assert.assertEquals(reference.get().getEnsemble(0L).get(0), b2);
- Assert.assertEquals(reference.get().getEnsemble(0L).get(1), b1);
+ Assert.assertEquals(reference.get().getValue().getEnsemble(0L).get(0), b2);
+ Assert.assertEquals(reference.get().getValue().getEnsemble(0L).get(1), b1);
- verify(lm, times(2)).writeLedgerMetadata(anyLong(), any(), any());
+ verify(lm, times(2)).writeLedgerMetadata(anyLong(), any(), any(), any());
}
}
@@ -238,13 +242,13 @@ public void testConflictOnLocalUpdate() throws Exception {
LedgerMetadata initMeta = LedgerMetadataBuilder.create().withEnsembleSize(2)
.newEnsembleEntry(0L, Lists.newArrayList(b0, b1)).build();
- GenericCallbackFuture<LedgerMetadata> promise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> promise = new GenericCallbackFuture<>();
lm.createLedgerMetadata(ledgerId, initMeta, promise);
- LedgerMetadata writtenMetadata = promise.get();
+ Versioned<LedgerMetadata> writtenMetadata = promise.get();
- AtomicReference<LedgerMetadata> reference = new AtomicReference<>(writtenMetadata);
+ AtomicReference<Versioned<LedgerMetadata>> reference = new AtomicReference<>(writtenMetadata);
- CompletableFuture<LedgerMetadata> loop1 = new MetadataUpdateLoop(
+ CompletableFuture<Versioned<LedgerMetadata>> loop1 = new MetadataUpdateLoop(
lm,
ledgerId,
reference::get,
@@ -257,7 +261,7 @@ public void testConflictOnLocalUpdate() throws Exception {
reference::compareAndSet).run();
lm.waitForWriteCount(1);
- CompletableFuture<LedgerMetadata> loop2 = new MetadataUpdateLoop(
+ CompletableFuture<Versioned<LedgerMetadata>> loop2 = new MetadataUpdateLoop(
lm,
ledgerId,
reference::get,
@@ -274,10 +278,10 @@ public void testConflictOnLocalUpdate() throws Exception {
Assert.assertEquals(loop1.get(), reference.get());
- Assert.assertEquals(reference.get().getEnsemble(0L).get(0), b2);
- Assert.assertEquals(reference.get().getEnsemble(0L).get(1), b3);
+ Assert.assertEquals(reference.get().getValue().getEnsemble(0L).get(0), b2);
+ Assert.assertEquals(reference.get().getValue().getEnsemble(0L).get(1), b3);
- verify(lm, times(3)).writeLedgerMetadata(anyLong(), any(), any());
+ verify(lm, times(3)).writeLedgerMetadata(anyLong(), any(), any(), any());
}
}
@@ -306,17 +310,17 @@ public void testHammer() throws Exception {
LedgerMetadata initMeta = LedgerMetadataBuilder.create().withEnsembleSize(ensembleSize)
.newEnsembleEntry(0L, initialEnsemble).build();
- GenericCallbackFuture<LedgerMetadata> promise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> promise = new GenericCallbackFuture<>();
lm.createLedgerMetadata(ledgerId, initMeta, promise);
- LedgerMetadata writtenMetadata = promise.get();
+ Versioned<LedgerMetadata> writtenMetadata = promise.get();
- AtomicReference<LedgerMetadata> reference = new AtomicReference<>(writtenMetadata);
+ AtomicReference<Versioned<LedgerMetadata>> reference = new AtomicReference<>(writtenMetadata);
List<BookieSocketAddress> replacementBookies = IntStream.range(0, ensembleSize)
.mapToObj((i) -> address(String.format("0.0.%d.1:3181", i)))
.collect(Collectors.toList());
- List<CompletableFuture<LedgerMetadata>> loops = IntStream.range(0, ensembleSize)
+ List<CompletableFuture<Versioned<LedgerMetadata>>> loops = IntStream.range(0, ensembleSize)
.mapToObj((i) -> new MetadataUpdateLoop(
lm,
ledgerId,
@@ -332,7 +336,7 @@ public void testHammer() throws Exception {
loops.forEach((l) -> l.join());
- Assert.assertEquals(reference.get().getEnsemble(0L), replacementBookies);
+ Assert.assertEquals(reference.get().getValue().getEnsemble(0L), replacementBookies);
}
}
@@ -351,19 +355,19 @@ public void testNewestValueCannotBeUsedAfterReadBack() throws Exception {
LedgerMetadata initMeta = LedgerMetadataBuilder.create().withEnsembleSize(1)
.newEnsembleEntry(0L, Lists.newArrayList(b0)).build();
- GenericCallbackFuture<LedgerMetadata> promise = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> promise = new GenericCallbackFuture<>();
lm.createLedgerMetadata(ledgerId, initMeta, promise);
- LedgerMetadata writtenMetadata = promise.get();
+ Versioned<LedgerMetadata> writtenMetadata = promise.get();
- AtomicReference<LedgerMetadata> reference = new AtomicReference<>(writtenMetadata);
- CompletableFuture<LedgerMetadata> loop1 = new MetadataUpdateLoop(
+ AtomicReference<Versioned<LedgerMetadata>> reference = new AtomicReference<>(writtenMetadata);
+ CompletableFuture<Versioned<LedgerMetadata>> loop1 = new MetadataUpdateLoop(
lm,
ledgerId,
reference::get,
(currentMetadata) -> !currentMetadata.isClosed(),
(currentMetadata) -> LedgerMetadataBuilder.from(currentMetadata).closingAt(10L, 100L).build(),
reference::compareAndSet).run();
- CompletableFuture<LedgerMetadata> loop2 = new MetadataUpdateLoop(
+ CompletableFuture<Versioned<LedgerMetadata>> loop2 = new MetadataUpdateLoop(
lm,
ledgerId,
reference::get,
@@ -382,7 +386,7 @@ public void testNewestValueCannotBeUsedAfterReadBack() throws Exception {
reference::compareAndSet).run();
lm.releaseWrites();
- LedgerMetadata l1meta = loop1.get();
+ Versioned<LedgerMetadata> l1meta = loop1.get();
try {
loop2.get();
Assert.fail("Update loop should have failed");
@@ -390,10 +394,10 @@ public void testNewestValueCannotBeUsedAfterReadBack() throws Exception {
Assert.assertEquals(ee.getCause().getClass(), BKException.BKLedgerClosedException.class);
}
Assert.assertEquals(l1meta, reference.get());
- Assert.assertEquals(l1meta.getEnsemble(0L).get(0), b0);
- Assert.assertTrue(l1meta.isClosed());
+ Assert.assertEquals(l1meta.getValue().getEnsemble(0L).get(0), b0);
+ Assert.assertTrue(l1meta.getValue().isClosed());
- verify(lm, times(2)).writeLedgerMetadata(anyLong(), any(), any());
+ verify(lm, times(2)).writeLedgerMetadata(anyLong(), any(), any(), any());
}
}
@@ -416,7 +420,8 @@ public void close() {
static class DeferCallbacksMockLedgerManager extends MockLedgerManager {
int writeCount = 0;
final int numToDefer;
- List<Triple<GenericCallback<LedgerMetadata>, Integer, LedgerMetadata>> deferred = Lists.newArrayList();
+ List<Triple<GenericCallback<Versioned<LedgerMetadata>>, Integer, Versioned<LedgerMetadata>>> deferred =
+ Lists.newArrayList();
DeferCallbacksMockLedgerManager(int numToDefer) {
this.numToDefer = numToDefer;
@@ -434,8 +439,9 @@ synchronized void waitForWriteCount(int count) throws Exception {
@Override
public synchronized void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
- GenericCallback<LedgerMetadata> cb) {
- super.writeLedgerMetadata(ledgerId, metadata,
+ Version currentVersion,
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
+ super.writeLedgerMetadata(ledgerId, metadata, currentVersion,
(rc, written) -> {
synchronized (DeferCallbacksMockLedgerManager.this) {
if (writeCount++ < numToDefer) {
@@ -451,9 +457,18 @@ public synchronized void writeLedgerMetadata(long ledgerId, LedgerMetadata metad
};
}
+ @Data
+ @AllArgsConstructor
+ static class DeferredUpdate {
+ final long ledgerId;
+ final LedgerMetadata metadata;
+ final Version currentVersion;
+ final GenericCallback<Versioned<LedgerMetadata>> cb;
+ }
+
static class BlockableMockLedgerManager extends MockLedgerManager {
boolean blocking = false;
- List<Triple<Long, LedgerMetadata, GenericCallback<LedgerMetadata>>> reqs = Lists.newArrayList();
+ List<DeferredUpdate> reqs = Lists.newArrayList();
synchronized void blockWrites() {
blocking = true;
@@ -461,16 +476,18 @@ synchronized void blockWrites() {
synchronized void releaseWrites() {
blocking = false;
- reqs.forEach((r) -> super.writeLedgerMetadata(r.getLeft(), r.getMiddle(), r.getRight()));
+ reqs.forEach((r) -> super.writeLedgerMetadata(r.getLedgerId(), r.getMetadata(),
+ r.getCurrentVersion(), r.getCb()));
}
@Override
public synchronized void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
- GenericCallback<LedgerMetadata> cb) {
+ Version currentVersion,
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
if (blocking) {
- reqs.add(Triple.of(ledgerId, metadata, cb));
+ reqs.add(new DeferredUpdate(ledgerId, metadata, currentVersion, cb));
} else {
- super.writeLedgerMetadata(ledgerId, metadata, cb);
+ super.writeLedgerMetadata(ledgerId, metadata, currentVersion, cb);
}
};
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
index c290a7e01b..7d71837a5c 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookKeeperTestCase.java
@@ -65,6 +65,9 @@
import org.apache.bookkeeper.proto.checksum.DigestManager;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.util.ByteBufList;
+import org.apache.bookkeeper.versioning.LongVersion;
+import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.junit.After;
import org.junit.Before;
import org.mockito.invocation.InvocationOnMock;
@@ -368,7 +371,8 @@ private void setupReadLedgerMetadata() {
if (ledgerMetadata == null) {
cb.operationComplete(BKException.Code.NoSuchLedgerExistsException, null);
} else {
- cb.operationComplete(BKException.Code.OK, new LedgerMetadata(ledgerMetadata));
+ cb.operationComplete(BKException.Code.OK,
+ new Versioned<>(new LedgerMetadata(ledgerMetadata), new LongVersion(1)));
}
});
return null;
@@ -421,7 +425,7 @@ private void setupCreateLedgerMetadata() {
executor.executeOrdered(ledgerId, () -> {
LedgerMetadata ledgerMetadata = (LedgerMetadata) args[1];
mockLedgerMetadataRegistry.put(ledgerId, new LedgerMetadata(ledgerMetadata));
- cb.operationComplete(BKException.Code.OK, null);
+ cb.operationComplete(BKException.Code.OK, new Versioned<>(ledgerMetadata, new LongVersion(1)));
});
return null;
}).when(ledgerManager).createLedgerMetadata(anyLong(), any(), any());
@@ -433,14 +437,15 @@ private void setupWriteLedgerMetadata() {
Object[] args = invocation.getArguments();
Long ledgerId = (Long) args[0];
LedgerMetadata metadata = (LedgerMetadata) args[1];
- BookkeeperInternalCallbacks.GenericCallback cb = (BookkeeperInternalCallbacks.GenericCallback) args[2];
+ Version currentVersion = (Version) args[2];
+ BookkeeperInternalCallbacks.GenericCallback cb = (BookkeeperInternalCallbacks.GenericCallback) args[3];
executor.executeOrdered(ledgerId, () -> {
LedgerMetadata newMetadata = LedgerMetadataBuilder.from(metadata).build();
mockLedgerMetadataRegistry.put(ledgerId, newMetadata);
- cb.operationComplete(BKException.Code.OK, newMetadata);
+ cb.operationComplete(BKException.Code.OK, new Versioned<>(newMetadata, new LongVersion(1234)));
});
return null;
- }).when(ledgerManager).writeLedgerMetadata(anyLong(), any(), any());
+ }).when(ledgerManager).writeLedgerMetadata(anyLong(), any(), any(), any());
}
@SuppressWarnings("unchecked")
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockLedgerHandle.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockLedgerHandle.java
index 47350208dc..c2235c25fc 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockLedgerHandle.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockLedgerHandle.java
@@ -41,6 +41,8 @@
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
+import org.apache.bookkeeper.versioning.LongVersion;
+import org.apache.bookkeeper.versioning.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,8 +62,8 @@
MockLedgerHandle(MockBookKeeper bk, long id, DigestType digest, byte[] passwd) throws GeneralSecurityException {
super(bk.getClientCtx(), id,
- new LedgerMetadata(3, 3, 2, DigestType.MAC, "".getBytes()), DigestType.MAC, "".getBytes(),
- WriteFlag.NONE);
+ new Versioned<>(new LedgerMetadata(3, 3, 2, DigestType.MAC, "".getBytes()), new LongVersion(0L)),
+ DigestType.MAC, "".getBytes(), WriteFlag.NONE);
this.bk = bk;
this.id = id;
this.digest = digest;
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java
index 9bdbfacce3..fab13edbac 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java
@@ -63,6 +63,7 @@
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.bookkeeper.util.ByteBufList;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.KeeperException;
import org.junit.After;
@@ -93,7 +94,8 @@ void setLatch(CountDownLatch waitLatch) {
}
@Override
- public void createLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericCallback<LedgerMetadata> cb) {
+ public void createLedgerMetadata(long ledgerId, LedgerMetadata metadata,
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
lm.createLedgerMetadata(ledgerId, metadata, cb);
}
@@ -103,7 +105,7 @@ public void removeLedgerMetadata(long ledgerId, Version version, GenericCallback
}
@Override
- public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> readCb) {
+ public void readLedgerMetadata(long ledgerId, GenericCallback<Versioned<LedgerMetadata>> readCb) {
lm.readLedgerMetadata(ledgerId, readCb);
}
@@ -113,8 +115,9 @@ public LedgerRangeIterator getLedgerRanges() {
}
@Override
- public void writeLedgerMetadata(final long ledgerId, final LedgerMetadata metadata,
- final GenericCallback<LedgerMetadata> cb) {
+ public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
+ Version currentVersion,
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
final CountDownLatch cdl = waitLatch;
if (null != cdl) {
executorService.submit(new Runnable() {
@@ -126,11 +129,11 @@ public void run() {
Thread.currentThread().interrupt();
LOG.error("Interrupted on waiting latch : ", e);
}
- lm.writeLedgerMetadata(ledgerId, metadata, cb);
+ lm.writeLedgerMetadata(ledgerId, metadata, currentVersion, cb);
}
});
} else {
- lm.writeLedgerMetadata(ledgerId, metadata, cb);
+ lm.writeLedgerMetadata(ledgerId, metadata, currentVersion, cb);
}
}
@@ -364,19 +367,11 @@ public void operationComplete(int rc, Void result) {
LedgerHandle newRecoverLh = newBk.openLedgerNoRecovery(lh.getId(), digestType, "".getBytes());
assertEquals(BookieProtocol.INVALID_ENTRY_ID, newRecoverLh.getLastAddPushed());
assertEquals(BookieProtocol.INVALID_ENTRY_ID, newRecoverLh.getLastAddConfirmed());
+
// mark the ledger as in recovery to update version.
- newRecoverLh.getLedgerMetadata().markLedgerInRecovery();
- final CountDownLatch updateLatch = new CountDownLatch(1);
- final AtomicInteger updateResult = new AtomicInteger(0x12345);
- newRecoverLh.writeLedgerConfig(new GenericCallback<LedgerMetadata>() {
- @Override
- public void operationComplete(int rc, LedgerMetadata result) {
- updateResult.set(rc);
- updateLatch.countDown();
- }
- });
- updateLatch.await();
- assertEquals(BKException.Code.OK, updateResult.get());
+ ClientUtil.transformMetadata(newBk.getClientCtx(), newRecoverLh.getId(),
+ (metadata) -> LedgerMetadataBuilder.from(metadata).withInRecoveryState().build());
+
newRecoverLh.close();
LOG.info("Updated ledger manager {}.", newRecoverLh.getLedgerMetadata());
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerFragmentReplication.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerFragmentReplication.java
index 6af5ed9d9a..f407c3571e 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerFragmentReplication.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestLedgerFragmentReplication.java
@@ -37,6 +37,8 @@
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.versioning.LongVersion;
+import org.apache.bookkeeper.versioning.Versioned;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -244,7 +246,9 @@ public boolean isClosed() {
return true;
}
};
- LedgerHandle lh = new LedgerHandle(bkc.getClientCtx(), 0, metadata, TEST_DIGEST_TYPE,
+ LedgerHandle lh = new LedgerHandle(bkc.getClientCtx(), 0,
+ new Versioned<>(metadata, new LongVersion(0L)),
+ TEST_DIGEST_TYPE,
TEST_PSSWD, WriteFlag.NONE);
testSplitIntoSubFragments(10, 21, -1, 1, lh);
testSplitIntoSubFragments(10, 21, 20, 1, lh);
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSequenceRead.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSequenceRead.java
index 7bfba0c407..e51368fd38 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSequenceRead.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestSequenceRead.java
@@ -22,16 +22,9 @@
import static org.junit.Assert.assertEquals;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.SortedMap;
-import java.util.TreeMap;
-import java.util.concurrent.CountDownLatch;
+import com.google.common.collect.Lists;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
-import org.apache.bookkeeper.net.BookieSocketAddress;
-import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.junit.Test;
import org.slf4j.Logger;
@@ -44,53 +37,29 @@
private static final Logger logger = LoggerFactory.getLogger(TestSequenceRead.class);
- final DigestType digestType;
- final byte[] passwd = "sequence-read".getBytes();
-
public TestSequenceRead() {
super(5);
- this.digestType = DigestType.CRC32;
}
- private LedgerHandle createLedgerWithDuplicatedBookies() throws Exception {
- final LedgerHandle lh = bkc.createLedger(3, 3, 3, digestType, passwd);
+ private long createLedgerWithDuplicatedBookies() throws Exception {
+ long ledgerId = 12345L;
// introduce duplicated bookies in an ensemble.
- SortedMap<Long, ? extends List<BookieSocketAddress>> ensembles = lh.getLedgerMetadata().getAllEnsembles();
- TreeMap<Long, List<BookieSocketAddress>> newEnsembles = new TreeMap<>();
- for (Map.Entry<Long, ? extends List<BookieSocketAddress>> entry : ensembles.entrySet()) {
- List<BookieSocketAddress> newList = new ArrayList<BookieSocketAddress>(entry.getValue().size());
- BookieSocketAddress firstBookie = entry.getValue().get(0);
- for (BookieSocketAddress ignored : entry.getValue()) {
- newList.add(firstBookie);
- }
- newEnsembles.put(entry.getKey(), newList);
- }
- lh.getLedgerMetadata().setEnsembles(newEnsembles);
- // update the ledger metadata with duplicated bookies
- final CountDownLatch latch = new CountDownLatch(1);
- bkc.getLedgerManager().writeLedgerMetadata(lh.getId(), lh.getLedgerMetadata(),
- new BookkeeperInternalCallbacks.GenericCallback<LedgerMetadata>() {
- @Override
- public void operationComplete(int rc, LedgerMetadata result) {
- if (BKException.Code.OK == rc) {
- latch.countDown();
- } else {
- logger.error("Error on writing ledger metadata for ledger {} : ", lh.getId(),
- BKException.getMessage(rc));
- }
- }
- });
- latch.await();
- logger.info("Update ledger metadata with duplicated bookies for ledger {}.", lh.getId());
- return lh;
+ LedgerMetadataBuilder builder = LedgerMetadataBuilder.create()
+ .withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
+ .newEnsembleEntry(0L, Lists.newArrayList(getBookie(0), getBookie(0), getBookie(0)));
+ ClientUtil.setupLedger(bkc.getLedgerManager(), ledgerId, builder);
+
+ logger.info("Update ledger metadata with duplicated bookies for ledger {}.", ledgerId);
+ return ledgerId;
}
@Test
public void testSequenceReadOnDuplicatedBookies() throws Exception {
- final LedgerHandle lh = createLedgerWithDuplicatedBookies();
+ final long ledgerId = createLedgerWithDuplicatedBookies();
// should be able to open the ledger even it has duplicated bookies
- final LedgerHandle readLh = bkc.openLedger(lh.getId(), digestType, passwd);
+ final LedgerHandle readLh = bkc.openLedger(
+ ledgerId, DigestType.fromApiDigestType(ClientUtil.DIGEST_TYPE), ClientUtil.PASSWD);
assertEquals(LedgerHandle.INVALID_ENTRY_ID, readLh.getLastAddConfirmed());
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java
index 456839a632..43acabde5a 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java
@@ -44,6 +44,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -134,10 +135,10 @@ private void testWatchMetadataRemoval(LedgerManagerFactory factory) throws Excep
@Override
public void operationComplete(int rc, final Long lid) {
manager.createLedgerMetadata(lid, new LedgerMetadata(4, 2, 2, digestType, "fpj was here".getBytes()),
- new BookkeeperInternalCallbacks.GenericCallback<LedgerMetadata>(){
+ new BookkeeperInternalCallbacks.GenericCallback<Versioned<LedgerMetadata>>(){
@Override
- public void operationComplete(int rc, LedgerMetadata result) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> result) {
bbLedgerId.putLong(lid);
bbLedgerId.flip();
createLatch.countDown();
@@ -154,7 +155,7 @@ public void operationComplete(int rc, LedgerMetadata result) {
new LedgerMetadataListener() {
@Override
- public void onChanged(long ledgerId, LedgerMetadata metadata) {
+ public void onChanged(long ledgerId, Versioned<LedgerMetadata> metadata) {
assertEquals(ledgerId, createdLid);
assertEquals(metadata, null);
removeLatch.countDown();
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/AbstractZkLedgerManagerTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/AbstractZkLedgerManagerTest.java
index b2cf68d25c..290af7bdad 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/AbstractZkLedgerManagerTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/AbstractZkLedgerManagerTest.java
@@ -59,6 +59,7 @@
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.bookkeeper.zookeeper.MockZooKeeperTestCase;
import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
@@ -154,13 +155,10 @@ public void testCreateLedgerMetadataSuccess() throws Exception {
KeeperException.Code.OK.intValue(), ledgerStr
);
- assertEquals(Version.NEW, metadata.getVersion());
-
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
ledgerManager.createLedgerMetadata(ledgerId, metadata, callbackFuture);
- callbackFuture.get();
- assertEquals(new LongVersion(0), metadata.getVersion());
+ assertEquals(new LongVersion(0), callbackFuture.get().getVersion());
}
@Test
@@ -171,9 +169,7 @@ public void testCreateLedgerMetadataNodeExists() throws Exception {
ledgerStr, CreateMode.PERSISTENT,
KeeperException.Code.NODEEXISTS.intValue(), null);
- assertEquals(Version.NEW, metadata.getVersion());
-
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
ledgerManager.createLedgerMetadata(ledgerId, metadata, callbackFuture);
try {
result(callbackFuture);
@@ -183,9 +179,6 @@ public void testCreateLedgerMetadataNodeExists() throws Exception {
BKException bke = (BKException) e;
assertEquals(Code.LedgerExistException, bke.getCode());
}
-
- // creation failed, so metadata should not be modified
- assertEquals(Version.NEW, metadata.getVersion());
}
@Test
@@ -196,9 +189,7 @@ public void testCreateLedgerMetadataException() throws Exception {
ledgerStr, CreateMode.PERSISTENT,
KeeperException.Code.CONNECTIONLOSS.intValue(), null);
- assertEquals(Version.NEW, metadata.getVersion());
-
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
ledgerManager.createLedgerMetadata(ledgerId, metadata, callbackFuture);
try {
result(callbackFuture);
@@ -208,9 +199,6 @@ public void testCreateLedgerMetadataException() throws Exception {
BKException bke = (BKException) e;
assertEquals(Code.ZKException, bke.getCode());
}
-
- // creation failed, so metadata should not be modified
- assertEquals(Version.NEW, metadata.getVersion());
}
@Test
@@ -355,7 +343,6 @@ public void testReadLedgerMetadataSuccess() throws Exception {
long ledgerId = System.currentTimeMillis();
String ledgerStr = String.valueOf(ledgerId);
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1234);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -363,10 +350,11 @@ public void testReadLedgerMetadataSuccess() throws Exception {
ledgerStr, false,
KeeperException.Code.OK.intValue(), metadata.serialize(), stat);
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
ledgerManager.readLedgerMetadata(ledgerId, callbackFuture);
- LedgerMetadata readMetadata = result(callbackFuture);
- assertEquals(metadata, readMetadata);
+ Versioned<LedgerMetadata> readMetadata = result(callbackFuture);
+ assertEquals(metadata, readMetadata.getValue());
+ assertEquals(new LongVersion(1234), readMetadata.getVersion());
verify(mockZk, times(1))
.getData(eq(ledgerStr), eq(null), any(DataCallback.class), any());
@@ -381,7 +369,7 @@ public void testReadLedgerMetadataNoNode() throws Exception {
ledgerStr, false,
KeeperException.Code.NONODE.intValue(), null, null);
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
ledgerManager.readLedgerMetadata(ledgerId, callbackFuture);
try {
result(callbackFuture);
@@ -403,7 +391,7 @@ public void testReadLedgerMetadataException() throws Exception {
ledgerStr, false,
KeeperException.Code.CONNECTIONLOSS.intValue(), null, null);
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
ledgerManager.readLedgerMetadata(ledgerId, callbackFuture);
try {
result(callbackFuture);
@@ -425,7 +413,7 @@ public void testReadLedgerMetadataStatMissing() throws Exception {
ledgerStr, false,
KeeperException.Code.OK.intValue(), metadata.serialize(), null);
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
ledgerManager.readLedgerMetadata(ledgerId, callbackFuture);
try {
result(callbackFuture);
@@ -443,7 +431,6 @@ public void testReadLedgerMetadataDataCorrupted() throws Exception {
long ledgerId = System.currentTimeMillis();
String ledgerStr = String.valueOf(ledgerId);
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1234);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -451,7 +438,7 @@ public void testReadLedgerMetadataDataCorrupted() throws Exception {
ledgerStr, false,
KeeperException.Code.OK.intValue(), new byte[0], stat);
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
ledgerManager.readLedgerMetadata(ledgerId, callbackFuture);
try {
result(callbackFuture);
@@ -469,7 +456,6 @@ public void testWriteLedgerMetadataSuccess() throws Exception {
long ledgerId = System.currentTimeMillis();
String ledgerStr = String.valueOf(ledgerId);
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1235);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -477,13 +463,10 @@ public void testWriteLedgerMetadataSuccess() throws Exception {
ledgerStr, metadata.serialize(), 1234,
KeeperException.Code.OK.intValue(), stat);
- assertEquals(new LongVersion(1234L), metadata.getVersion());
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
+ ledgerManager.writeLedgerMetadata(ledgerId, metadata, new LongVersion(1234L), callbackFuture);
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
- ledgerManager.writeLedgerMetadata(ledgerId, metadata, callbackFuture);
- result(callbackFuture);
-
- assertEquals(new LongVersion(1235L), metadata.getVersion());
+ assertEquals(new LongVersion(1235L), result(callbackFuture).getVersion());
verify(mockZk, times(1))
.setData(eq(ledgerStr), any(byte[].class), eq(1234), any(StatCallback.class), any());
@@ -494,15 +477,12 @@ public void testWriteLedgerMetadataBadVersion() throws Exception {
long ledgerId = System.currentTimeMillis();
String ledgerStr = String.valueOf(ledgerId);
- metadata.setVersion(new LongVersion(1234L));
mockZkSetData(
ledgerStr, metadata.serialize(), 1234,
KeeperException.Code.BADVERSION.intValue(), null);
- assertEquals(new LongVersion(1234L), metadata.getVersion());
-
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
- ledgerManager.writeLedgerMetadata(ledgerId, metadata, callbackFuture);
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
+ ledgerManager.writeLedgerMetadata(ledgerId, metadata, new LongVersion(1234L), callbackFuture);
try {
result(callbackFuture);
fail("Should fail on writing ledger metadata if encountering bad version");
@@ -510,9 +490,6 @@ public void testWriteLedgerMetadataBadVersion() throws Exception {
assertEquals(Code.MetadataVersionException, bke.getCode());
}
- // version remain unchanged
- assertEquals(new LongVersion(1234L), metadata.getVersion());
-
verify(mockZk, times(1))
.setData(eq(ledgerStr), any(byte[].class), eq(1234), any(StatCallback.class), any());
}
@@ -522,15 +499,12 @@ public void testWriteLedgerMetadataException() throws Exception {
long ledgerId = System.currentTimeMillis();
String ledgerStr = String.valueOf(ledgerId);
- metadata.setVersion(new LongVersion(1234L));
mockZkSetData(
ledgerStr, metadata.serialize(), 1234,
KeeperException.Code.CONNECTIONLOSS.intValue(), null);
- assertEquals(new LongVersion(1234L), metadata.getVersion());
-
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
- ledgerManager.writeLedgerMetadata(ledgerId, metadata, callbackFuture);
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
+ ledgerManager.writeLedgerMetadata(ledgerId, metadata, new LongVersion(1234L), callbackFuture);
try {
result(callbackFuture);
fail("Should fail on writing ledger metadata if encountering zookeeper exceptions");
@@ -538,8 +512,6 @@ public void testWriteLedgerMetadataException() throws Exception {
assertEquals(Code.ZKException, bke.getCode());
}
- // version remain unchanged
- assertEquals(new LongVersion(1234L), metadata.getVersion());
verify(mockZk, times(1))
.setData(eq(ledgerStr), any(byte[].class), eq(1234), any(StatCallback.class), any());
@@ -560,10 +532,8 @@ public void testWriteLedgerMetadataInvalidVersion() throws Exception {
private void testWriteLedgerMetadataInvalidVersion(Version invalidVersion) throws Exception {
long ledgerId = System.currentTimeMillis();
- metadata.setVersion(invalidVersion);
-
- GenericCallbackFuture<LedgerMetadata> callbackFuture = new GenericCallbackFuture<>();
- ledgerManager.writeLedgerMetadata(ledgerId, metadata, callbackFuture);
+ GenericCallbackFuture<Versioned<LedgerMetadata>> callbackFuture = new GenericCallbackFuture<>();
+ ledgerManager.writeLedgerMetadata(ledgerId, metadata, invalidVersion, callbackFuture);
try {
result(callbackFuture);
fail("Should fail on writing ledger metadata if an invalid version is provided.");
@@ -581,9 +551,8 @@ public void testLedgerMetadataListener() throws Exception {
String ledgerStr = String.valueOf(ledgerId);
LinkedBlockingQueue<LedgerMetadata> changes = new LinkedBlockingQueue<>();
- LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata);
+ LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata.getValue());
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1234);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -606,7 +575,6 @@ public void testLedgerMetadataListener() throws Exception {
Watcher registeredWatcher1 = watcherSet1.stream().findFirst().get();
// mock get data to return an updated metadata
- metadata.setVersion(new LongVersion(1235L));
when(stat.getVersion()).thenReturn(1235);
mockZkGetData(
ledgerStr, true,
@@ -644,9 +612,9 @@ public void testLedgerMetadataListenerOnLedgerDeleted() throws Exception {
String ledgerStr = String.valueOf(ledgerId);
LinkedBlockingQueue<Optional<LedgerMetadata>> changes = new LinkedBlockingQueue<>();
- LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(Optional.ofNullable(metadata));
+ LedgerMetadataListener listener =
+ (ledgerId1, metadata) -> changes.add(Optional.ofNullable(metadata != null ? metadata.getValue() : null));
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1234);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -695,9 +663,10 @@ public void testLedgerMetadataListenerOnLedgerDeletedEvent() throws Exception {
String ledgerStr = String.valueOf(ledgerId);
LinkedBlockingQueue<Optional<LedgerMetadata>> changes = new LinkedBlockingQueue<>();
- LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(Optional.ofNullable(metadata));
+ LedgerMetadataListener listener =
+ (ledgerId1, metadata) -> changes.add(
+ Optional.ofNullable(metadata != null ? metadata.getValue() : null));
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1234);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -737,9 +706,8 @@ public void testLedgerMetadataListenerOnRetries() throws Exception {
String ledgerStr = String.valueOf(ledgerId);
LinkedBlockingQueue<LedgerMetadata> changes = new LinkedBlockingQueue<>();
- LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata);
+ LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata.getValue());
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1234);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -786,9 +754,8 @@ public void testLedgerMetadataListenerOnSessionExpired() throws Exception {
String ledgerStr = String.valueOf(ledgerId);
LinkedBlockingQueue<LedgerMetadata> changes = new LinkedBlockingQueue<>();
- LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata);
+ LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata.getValue());
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1234);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -835,9 +802,8 @@ public void testUnregisterLedgerMetadataListener() throws Exception {
String ledgerStr = String.valueOf(ledgerId);
LinkedBlockingQueue<LedgerMetadata> changes = new LinkedBlockingQueue<>();
- LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata);
+ LedgerMetadataListener listener = (ledgerId1, metadata) -> changes.add(metadata.getValue());
- metadata.setVersion(new LongVersion(1234L));
Stat stat = mock(Stat.class);
when(stat.getVersion()).thenReturn(1234);
when(stat.getCtime()).thenReturn(metadata.getCtime());
@@ -861,7 +827,6 @@ public void testUnregisterLedgerMetadataListener() throws Exception {
Watcher registeredWatcher1 = watcherSet1.stream().findFirst().get();
// mock get data to return an updated metadata
- metadata.setVersion(new LongVersion(1235L));
when(stat.getVersion()).thenReturn(1235);
mockZkGetData(
ledgerStr, true,
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/GcLedgersTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/GcLedgersTest.java
index 1459500c2f..10192cbeda 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/GcLedgersTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/GcLedgersTest.java
@@ -72,6 +72,7 @@
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.test.TestStatsProvider;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -107,9 +108,9 @@ public void operationComplete(int rc, final Long ledgerId) {
getLedgerManager().createLedgerMetadata(ledgerId,
new LedgerMetadata(1, 1, 1, DigestType.MAC, "".getBytes()),
- new GenericCallback<LedgerMetadata>() {
+ new GenericCallback<Versioned<LedgerMetadata>>() {
@Override
- public void operationComplete(int rc, LedgerMetadata writtenMetadata) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> writtenMetadata) {
if (rc == BKException.Code.OK) {
activeLedgers.put(ledgerId, true);
createdLedgers.add(ledgerId);
@@ -505,7 +506,7 @@ public void testGcLedgersIfReadLedgerMetadataSaysNoSuchLedger() throws Exception
LedgerManager mockLedgerManager = new CleanupLedgerManager(getLedgerManager()) {
@Override
- public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> readCb) {
+ public void readLedgerMetadata(long ledgerId, GenericCallback<Versioned<LedgerMetadata>> readCb) {
readCb.operationComplete(BKException.Code.NoSuchLedgerExistsException, null);
}
};
@@ -547,7 +548,7 @@ public void testGcLedgersIfReadLedgerMetadataFailsForDeletedLedgers() throws Exc
LedgerManager mockLedgerManager = new CleanupLedgerManager(getLedgerManager()) {
@Override
- public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> readCb) {
+ public void readLedgerMetadata(long ledgerId, GenericCallback<Versioned<LedgerMetadata>> readCb) {
readCb.operationComplete(BKException.Code.ZKException, null);
}
};
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerIteratorTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerIteratorTest.java
index ac8223bf00..477afe4937 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerIteratorTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerIteratorTest.java
@@ -53,6 +53,7 @@
import org.apache.bookkeeper.util.MathUtils;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.junit.After;
@@ -87,7 +88,7 @@ public void throwAsyncErrors() throws Throwable {
}
}
- class RCCheckCB implements GenericCallback<LedgerMetadata> {
+ class RCCheckCB implements GenericCallback<Versioned<LedgerMetadata>> {
private final String opType;
private final CountDownLatch latch;
private final Optional<Integer> rcExpected;
@@ -101,7 +102,7 @@ public RCCheckCB(String opType, CountDownLatch latch, Optional<Integer> rcExpect
}
@Override
- public void operationComplete(int rc, LedgerMetadata writtenMetadata) {
+ public void operationComplete(int rc, Versioned<LedgerMetadata> writtenMetadata) {
safeWrapper(() -> {
try {
rcExpected.map((Integer expected) -> {
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/MockLedgerManager.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/MockLedgerManager.java
index ea04192f88..075b8f1b9a 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/MockLedgerManager.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/MockLedgerManager.java
@@ -37,6 +37,7 @@
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.zookeeper.AsyncCallback;
@@ -77,12 +78,12 @@ public MockLedgerManager newClient() {
return new MockLedgerManager(metadataMap, executor, false);
}
- private LedgerMetadata readMetadata(long ledgerId) throws Exception {
+ private Versioned<LedgerMetadata> readMetadata(long ledgerId) throws Exception {
Pair<LongVersion, byte[]> pair = metadataMap.get(ledgerId);
if (pair == null) {
return null;
} else {
- return LedgerMetadata.parseConfig(pair.getRight(), pair.getLeft(), Optional.absent());
+ return new Versioned<>(LedgerMetadata.parseConfig(pair.getRight(), Optional.absent()), pair.getLeft());
}
}
@@ -95,14 +96,15 @@ public void executeCallback(Runnable r) {
}
@Override
- public void createLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericCallback<LedgerMetadata> cb) {
+ public void createLedgerMetadata(long ledgerId, LedgerMetadata metadata,
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
executor.submit(() -> {
if (metadataMap.containsKey(ledgerId)) {
executeCallback(() -> cb.operationComplete(BKException.Code.LedgerExistException, null));
} else {
metadataMap.put(ledgerId, Pair.of(new LongVersion(0L), metadata.serialize()));
try {
- LedgerMetadata readBack = readMetadata(ledgerId);
+ Versioned<LedgerMetadata> readBack = readMetadata(ledgerId);
executeCallback(() -> cb.operationComplete(BKException.Code.OK, readBack));
} catch (Exception e) {
LOG.error("Error reading back written metadata", e);
@@ -116,10 +118,10 @@ public void createLedgerMetadata(long ledgerId, LedgerMetadata metadata, Generic
public void removeLedgerMetadata(long ledgerId, Version version, GenericCallback<Void> cb) {}
@Override
- public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> cb) {
+ public void readLedgerMetadata(long ledgerId, GenericCallback<Versioned<LedgerMetadata>> cb) {
executor.submit(() -> {
try {
- LedgerMetadata metadata = readMetadata(ledgerId);
+ Versioned<LedgerMetadata> metadata = readMetadata(ledgerId);
if (metadata == null) {
executeCallback(
() -> cb.operationComplete(BKException.Code.NoSuchLedgerExistsException, null));
@@ -134,20 +136,21 @@ public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> cb
}
@Override
- public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericCallback<LedgerMetadata> cb) {
+ public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
+ Version currentVersion, GenericCallback<Versioned<LedgerMetadata>> cb) {
preWriteHook.runHook(ledgerId, metadata)
.thenComposeAsync((ignore) -> {
try {
- LedgerMetadata oldMetadata = readMetadata(ledgerId);
+ Versioned<LedgerMetadata> oldMetadata = readMetadata(ledgerId);
if (oldMetadata == null) {
return FutureUtils.exception(new BKException.BKNoSuchLedgerExistsException());
- } else if (!oldMetadata.getVersion().equals(metadata.getVersion())) {
+ } else if (!oldMetadata.getVersion().equals(currentVersion)) {
return FutureUtils.exception(new BKException.BKMetadataVersionException());
} else {
LongVersion oldVersion = (LongVersion) oldMetadata.getVersion();
metadataMap.put(ledgerId, Pair.of(new LongVersion(oldVersion.getLongVersion() + 1),
metadata.serialize()));
- LedgerMetadata readBack = readMetadata(ledgerId);
+ Versioned<LedgerMetadata> readBack = readMetadata(ledgerId);
return FutureUtils.value(readBack);
}
} catch (Exception e) {
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicBookieCheckTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicBookieCheckTest.java
index 81bfc3d5f3..06df18b755 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicBookieCheckTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicBookieCheckTest.java
@@ -23,20 +23,17 @@
import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithLedgerManagerFactory;
import static org.junit.Assert.assertEquals;
+import com.google.common.collect.Lists;
import com.google.common.util.concurrent.UncheckedExecutionException;
-import java.util.ArrayList;
-import java.util.List;
+
import lombok.Cleanup;
-import org.apache.bookkeeper.client.BookKeeper.DigestType;
-import org.apache.bookkeeper.client.LedgerHandle;
-import org.apache.bookkeeper.client.LedgerHandleAdapter;
-import org.apache.bookkeeper.client.LedgerMetadata;
+import org.apache.bookkeeper.client.ClientUtil;
+import org.apache.bookkeeper.client.LedgerMetadataBuilder;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
-import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallbackFuture;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.junit.After;
import org.junit.Before;
@@ -93,18 +90,13 @@ public void testPeriodicBookieCheckInterval() throws Exception {
try (LedgerManager ledgerManager = mFactory.newLedgerManager()) {
@Cleanup final LedgerUnderreplicationManager underReplicationManager =
mFactory.newLedgerUnderreplicationManager();
-
- LedgerHandle lh = bkc.createLedger(3, 3, DigestType.CRC32, "passwd".getBytes());
- LedgerMetadata md = LedgerHandleAdapter.getLedgerMetadata(lh);
- List<BookieSocketAddress> ensemble = new ArrayList<>(md.getAllEnsembles().get(0L));
- ensemble.set(0, new BookieSocketAddress("1.1.1.1", 1000));
- md.updateEnsemble(0L, ensemble);
-
- GenericCallbackFuture<LedgerMetadata> cb =
- new GenericCallbackFuture<LedgerMetadata>();
- ledgerManager.writeLedgerMetadata(lh.getId(), md, cb);
- cb.get();
-
+ long ledgerId = 12345L;
+ ClientUtil.setupLedger(bkc.getLedgerManager(), ledgerId,
+ LedgerMetadataBuilder.create().withEnsembleSize(3)
+ .withWriteQuorumSize(3).withAckQuorumSize(3)
+ .newEnsembleEntry(0L, Lists.newArrayList(
+ new BookieSocketAddress("1.1.1.1", 1000),
+ getBookie(0), getBookie(1))));
long underReplicatedLedger = -1;
for (int i = 0; i < 10; i++) {
underReplicatedLedger = underReplicationManager.pollLedgerToRereplicate();
@@ -113,7 +105,7 @@ public void testPeriodicBookieCheckInterval() throws Exception {
}
Thread.sleep(CHECK_INTERVAL * 1000);
}
- assertEquals("Ledger should be under replicated", lh.getId(), underReplicatedLedger);
+ assertEquals("Ledger should be under replicated", ledgerId, underReplicatedLedger);
} catch (Exception e) {
throw new UncheckedExecutionException(e.getMessage(), e);
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/TestHttpService.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/TestHttpService.java
index e9feada8b1..079e08f422 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/TestHttpService.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/TestHttpService.java
@@ -23,20 +23,19 @@
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.UncheckedExecutionException;
import java.io.File;
-import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import lombok.Cleanup;
import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.ClientUtil;
import org.apache.bookkeeper.client.LedgerHandle;
-import org.apache.bookkeeper.client.LedgerHandleAdapter;
-import org.apache.bookkeeper.client.LedgerMetadata;
+import org.apache.bookkeeper.client.LedgerMetadataBuilder;
import org.apache.bookkeeper.common.util.JsonUtil;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
@@ -48,9 +47,9 @@
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
-import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallbackFuture;
import org.apache.bookkeeper.replication.AuditorElector;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
@@ -664,25 +663,27 @@ private void testListUnderReplicatedLedgerService(LedgerManagerFactory mFactory)
@Cleanup final LedgerUnderreplicationManager underReplicationManager =
mFactory.newLedgerUnderreplicationManager();
- LedgerHandle lh = bkc.createLedger(3, 3, BookKeeper.DigestType.CRC32, "passwd".getBytes());
- LedgerMetadata md = LedgerHandleAdapter.getLedgerMetadata(lh);
- List<BookieSocketAddress> ensemble = new ArrayList<>(md.getAllEnsembles().get(0L));
- ensemble.set(0, new BookieSocketAddress("1.1.1.1", 1000));
- md.updateEnsemble(0L, ensemble);
-
- GenericCallbackFuture<LedgerMetadata> cb =
- new GenericCallbackFuture<LedgerMetadata>();
- ledgerManager.writeLedgerMetadata(lh.getId(), md, cb);
- cb.get();
-
+ // 192.0.2.0/24 is reserved TEST-NET range
+ LedgerMetadataBuilder metadata = LedgerMetadataBuilder.create()
+ .withEnsembleSize(3).withWriteQuorumSize(3).withAckQuorumSize(3)
+ .newEnsembleEntry(0L, Lists.newArrayList(
+ new BookieSocketAddress("192.0.2.1", 1000),
+ getBookie(0),
+ getBookie(1)));
+ ClientUtil.setupLedger(ledgerManager, 1L, metadata);
+
+ // wait for up to two minutes to complete.
+ // if the metadata was created just before checkAllLedgers ran, then we need to wait for the timeout
long underReplicatedLedger = -1;
- for (int i = 0; i < 10; i++) {
+ for (int i = 0; i < 120; i++) {
underReplicatedLedger = underReplicationManager.pollLedgerToRereplicate();
if (underReplicatedLedger != -1) {
+ LOG.info("Underreplicated ledgers found, breaking out of loop");
break;
}
Thread.sleep(1000);
}
+ assertTrue(underReplicatedLedger != -1);
HttpServiceRequest request2 = new HttpServiceRequest(null, HttpServer.Method.GET, null);
HttpServiceResponse response2 = listUnderReplicatedLedgerService.handle(request2);
diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManager.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManager.java
index 1dc4439845..7df7038f96 100644
--- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManager.java
+++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManager.java
@@ -54,6 +54,7 @@
import org.apache.bookkeeper.util.collections.ConcurrentLongHashMap;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
/**
@@ -66,7 +67,6 @@
try {
return LedgerMetadata.parseConfig(
bs.getBytes(),
- Version.ANY,
Optional.absent()
);
} catch (IOException ioe) {
@@ -105,7 +105,7 @@ private boolean isClosed() {
@Override
public void createLedgerMetadata(long ledgerId,
LedgerMetadata metadata,
- GenericCallback<LedgerMetadata> cb) {
+ GenericCallback<Versioned<LedgerMetadata>> cb) {
String ledgerKey = EtcdUtils.getLedgerKey(scope, ledgerId);
ByteSequence ledgerKeyBs = ByteSequence.fromString(ledgerKey);
log.info("Create ledger metadata under key {}", ledgerKey);
@@ -135,8 +135,8 @@ public void createLedgerMetadata(long ledgerId,
cb.operationComplete(Code.LedgerExistException, null);
}
} else {
- metadata.setVersion(new LongVersion(resp.getHeader().getRevision()));
- cb.operationComplete(Code.OK, metadata);
+ cb.operationComplete(Code.OK, new Versioned<>(metadata,
+ new LongVersion(resp.getHeader().getRevision())));
}
})
.exceptionally(cause -> {
@@ -211,7 +211,7 @@ public void removeLedgerMetadata(long ledgerId,
}
@Override
- public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> readCb) {
+ public void readLedgerMetadata(long ledgerId, GenericCallback<Versioned<LedgerMetadata>> readCb) {
String ledgerKey = EtcdUtils.getLedgerKey(scope, ledgerId);
ByteSequence ledgerKeyBs = ByteSequence.fromString(ledgerKey);
log.info("read ledger metadata under key {}", ledgerKey);
@@ -220,14 +220,10 @@ public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> re
if (getResp.getCount() > 0) {
KeyValue kv = getResp.getKvs().get(0);
byte[] data = kv.getValue().getBytes();
- LedgerMetadata metadata;
try {
- metadata = LedgerMetadata.parseConfig(
- data,
- new LongVersion(kv.getModRevision()),
- Optional.absent()
- );
- readCb.operationComplete(Code.OK, metadata);
+ LedgerMetadata metadata = LedgerMetadata.parseConfig(data, Optional.absent());
+ readCb.operationComplete(
+ Code.OK, new Versioned<>(metadata, new LongVersion(kv.getModRevision())));
} catch (IOException ioe) {
log.error("Could not parse ledger metadata for ledger : {}", ledgerId, ioe);
readCb.operationComplete(Code.MetaStoreException, null);
@@ -244,13 +240,13 @@ public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> re
}
@Override
- public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericCallback<LedgerMetadata> cb) {
- Version v = metadata.getVersion();
- if (Version.NEW == v || !(v instanceof LongVersion)) {
+ public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata,
+ Version currentVersion, GenericCallback<Versioned<LedgerMetadata>> cb) {
+ if (Version.NEW == currentVersion || !(currentVersion instanceof LongVersion)) {
cb.operationComplete(Code.MetadataVersionException, null);
return;
}
- final LongVersion lv = (LongVersion) v;
+ final LongVersion lv = (LongVersion) currentVersion;
String ledgerKey = EtcdUtils.getLedgerKey(scope, ledgerId);
ByteSequence ledgerKeyBs = ByteSequence.fromString(ledgerKey);
kvClient.txn()
@@ -268,8 +264,8 @@ public void writeLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericC
.commit()
.thenAccept(resp -> {
if (resp.isSucceeded()) {
- metadata.setVersion(new LongVersion(resp.getHeader().getRevision()));
- cb.operationComplete(Code.OK, metadata);
+ cb.operationComplete(
+ Code.OK, new Versioned<>(metadata, new LongVersion(resp.getHeader().getRevision())));
} else {
GetResponse getResp = resp.getGetResponses().get(0);
if (getResp.getCount() > 0) {
diff --git a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/LedgerMetadataConsumer.java b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/LedgerMetadataConsumer.java
index a201b25381..d466b340d7 100644
--- a/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/LedgerMetadataConsumer.java
+++ b/metadata-drivers/etcd/src/main/java/org/apache/bookkeeper/metadata/etcd/LedgerMetadataConsumer.java
@@ -61,12 +61,10 @@ public boolean equals(Object obj) {
@Override
public void accept(Versioned<LedgerMetadata> ledgerMetadataVersioned) {
- LedgerMetadata lm = ledgerMetadataVersioned.getValue();
- if (null == lm) {
+ if (null == ledgerMetadataVersioned.getValue()) {
onDeletedConsumer.accept(ledgerId);
} else {
- lm.setVersion(ledgerMetadataVersioned.getVersion());
- listener.onChanged(ledgerId, lm);
+ listener.onChanged(ledgerId, ledgerMetadataVersioned);
}
}
}
diff --git a/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManagerTest.java b/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManagerTest.java
index 226fd48a83..e44678034d 100644
--- a/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManagerTest.java
+++ b/metadata-drivers/etcd/src/test/java/org/apache/bookkeeper/metadata/etcd/EtcdLedgerManagerTest.java
@@ -50,6 +50,7 @@
import org.apache.bookkeeper.client.BKException.Code;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerMetadata;
+import org.apache.bookkeeper.client.LedgerMetadataBuilder;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.meta.LedgerManager.LedgerRange;
import org.apache.bookkeeper.meta.LedgerManager.LedgerRangeIterator;
@@ -60,6 +61,7 @@
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
+import org.apache.bookkeeper.versioning.Versioned;
import org.apache.commons.lang.RandomStringUtils;
import org.junit.After;
import org.junit.Before;
@@ -102,7 +104,7 @@ public void testLedgerCRUD() throws Exception {
// ledger doesn't exist: read
- GenericCallbackFuture<LedgerMetadata> readFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> readFuture = new GenericCallbackFuture<>();
lm.readLedgerMetadata(ledgerId, readFuture);
try {
result(readFuture);
@@ -124,9 +126,8 @@ public void testLedgerCRUD() throws Exception {
// ledger doesn't exist : write
- GenericCallbackFuture<LedgerMetadata> writeFuture = new GenericCallbackFuture<>();
- metadata.setVersion(new LongVersion(999L));
- lm.writeLedgerMetadata(ledgerId, metadata, writeFuture);
+ GenericCallbackFuture<Versioned<LedgerMetadata>> writeFuture = new GenericCallbackFuture<>();
+ lm.writeLedgerMetadata(ledgerId, metadata, new LongVersion(999L), writeFuture);
try {
result(deleteFuture);
fail("Should fail on updating ledger metadata if the ledger doesn't exist");
@@ -136,10 +137,10 @@ public void testLedgerCRUD() throws Exception {
// ledger doesn't exist : create
- GenericCallbackFuture<LedgerMetadata> createFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> createFuture = new GenericCallbackFuture<>();
lm.createLedgerMetadata(ledgerId, metadata, createFuture);
- LedgerMetadata writtenMetadata = result(createFuture);
- assertSame(metadata, writtenMetadata);
+ Versioned<LedgerMetadata> writtenMetadata = result(createFuture);
+ assertSame(metadata, writtenMetadata.getValue());
Version version = writtenMetadata.getVersion();
assertNotNull(version);
assertTrue(version instanceof LongVersion);
@@ -161,14 +162,12 @@ public void testLedgerCRUD() throws Exception {
readFuture = new GenericCallbackFuture<>();
lm.readLedgerMetadata(ledgerId, readFuture);
- LedgerMetadata readMetadata = result(readFuture);
- assertEquals(metadata, readMetadata);
+ Versioned<LedgerMetadata> readMetadata = result(readFuture);
+ assertEquals(metadata, readMetadata.getValue());
// ledger exists: update metadata with wrong version
-
- readMetadata.setVersion(new LongVersion(Long.MAX_VALUE));
writeFuture = new GenericCallbackFuture<>();
- lm.writeLedgerMetadata(ledgerId, readMetadata, writeFuture);
+ lm.writeLedgerMetadata(ledgerId, readMetadata.getValue(), new LongVersion(Long.MAX_VALUE), writeFuture);
try {
result(writeFuture);
fail("Should fail to write metadata using a wrong version");
@@ -178,7 +177,7 @@ public void testLedgerCRUD() throws Exception {
readFuture = new GenericCallbackFuture<>();
lm.readLedgerMetadata(ledgerId, readFuture);
readMetadata = result(readFuture);
- assertEquals(metadata, readMetadata);
+ assertEquals(metadata, readMetadata.getValue());
// ledger exists: delete metadata with wrong version
@@ -193,13 +192,13 @@ public void testLedgerCRUD() throws Exception {
readFuture = new GenericCallbackFuture<>();
lm.readLedgerMetadata(ledgerId, readFuture);
readMetadata = result(readFuture);
- assertEquals(metadata, readMetadata);
+ assertEquals(metadata, readMetadata.getValue());
// ledger exists: update metadata with the right version
LongVersion curVersion = (LongVersion) readMetadata.getVersion();
writeFuture = new GenericCallbackFuture<>();
- lm.writeLedgerMetadata(ledgerId, readMetadata, writeFuture);
+ lm.writeLedgerMetadata(ledgerId, readMetadata.getValue(), curVersion, writeFuture);
writtenMetadata = result(writeFuture);
LongVersion newVersion = (LongVersion) writtenMetadata.getVersion();
assertTrue(curVersion.getLongVersion() < newVersion.getLongVersion());
@@ -266,9 +265,9 @@ public void testLedgerRangeIterator() throws Exception {
}
private void createNumLedgers(int numLedgers) throws Exception {
- List<CompletableFuture<LedgerMetadata>> createFutures = new ArrayList<>(numLedgers);
+ List<CompletableFuture<Versioned<LedgerMetadata>>> createFutures = new ArrayList<>(numLedgers);
for (int i = 0; i < numLedgers; i++) {
- GenericCallbackFuture<LedgerMetadata> createFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> createFuture = new GenericCallbackFuture<>();
createFutures.add(createFuture);
LedgerMetadata metadata = new LedgerMetadata(
3, 3, 2,
@@ -291,15 +290,15 @@ public void testRegisterLedgerMetadataListener() throws Exception {
"test-password".getBytes(UTF_8)
);
metadata.addEnsemble(0L, createNumBookies(3));
- GenericCallbackFuture<LedgerMetadata> createFuture = new GenericCallbackFuture<>();
+ GenericCallbackFuture<Versioned<LedgerMetadata>> createFuture = new GenericCallbackFuture<>();
lm.createLedgerMetadata(ledgerId, metadata, createFuture);
result(createFuture);
- LedgerMetadata readMetadata = readLedgerMetadata(ledgerId);
- log.info("Create ledger metadata : {}", readMetadata);
+ Versioned<LedgerMetadata> readMetadata = readLedgerMetadata(ledgerId);
+ log.info("Create ledger metadata : {}", readMetadata.getValue());
// register first listener
- LinkedBlockingQueue<LedgerMetadata> metadataQueue1 = new LinkedBlockingQueue<>();
+ LinkedBlockingQueue<Versioned<LedgerMetadata>> metadataQueue1 = new LinkedBlockingQueue<>();
LedgerMetadataListener listener1 = (lid, m) -> {
log.info("[listener1] Received ledger {} metadata : {}", lid, m);
metadataQueue1.add(m);
@@ -307,7 +306,7 @@ public void testRegisterLedgerMetadataListener() throws Exception {
log.info("Registered first listener for ledger {}", ledgerId);
lm.registerLedgerMetadataListener(ledgerId, listener1);
// we should receive a metadata notification when a ledger is created
- LedgerMetadata notifiedMetadata = metadataQueue1.take();
+ Versioned<LedgerMetadata> notifiedMetadata = metadataQueue1.take();
assertEquals(readMetadata, notifiedMetadata);
ValueStream<LedgerMetadata> lms = lm.getLedgerMetadataStream(ledgerId);
assertNotNull(lms.waitUntilWatched());
@@ -315,21 +314,21 @@ public void testRegisterLedgerMetadataListener() throws Exception {
// register second listener
- LinkedBlockingQueue<LedgerMetadata> metadataQueue2 = new LinkedBlockingQueue<>();
+ LinkedBlockingQueue<Versioned<LedgerMetadata>> metadataQueue2 = new LinkedBlockingQueue<>();
LedgerMetadataListener listener2 = (lid, m) -> {
log.info("[listener2] Received ledger {} metadata : {}", lid, m);
metadataQueue2.add(m);
};
log.info("Registered second listener for ledger {}", ledgerId);
lm.registerLedgerMetadataListener(ledgerId, listener2);
- LedgerMetadata notifiedMetadata2 = metadataQueue2.take();
+ Versioned<LedgerMetadata> notifiedMetadata2 = metadataQueue2.take();
assertEquals(readMetadata, notifiedMetadata2);
assertNotNull(lm.getLedgerMetadataStream(ledgerId));
// update the metadata
- metadata.setVersion(notifiedMetadata.getVersion());
- metadata.addEnsemble(10L, createNumBookies(3));
- writeLedgerMetadata(ledgerId, metadata);
+ writeLedgerMetadata(ledgerId,
+ LedgerMetadataBuilder.from(metadata).newEnsembleEntry(10L, createNumBookies(3)).build(),
+ notifiedMetadata.getVersion());
readMetadata = readLedgerMetadata(ledgerId);
assertEquals(readMetadata, metadataQueue1.take());
assertEquals(readMetadata, metadataQueue2.take());
@@ -344,9 +343,9 @@ public void testRegisterLedgerMetadataListener() throws Exception {
assertEquals(1, lms.getNumConsumers());
// update the metadata again
- metadata.setVersion(readMetadata.getVersion());
- metadata.addEnsemble(20L, createNumBookies(3));
- writeLedgerMetadata(ledgerId, metadata);
+ writeLedgerMetadata(ledgerId,
+ LedgerMetadataBuilder.from(metadata).newEnsembleEntry(20L, createNumBookies(3)).build(),
+ readMetadata.getVersion());
readMetadata = readLedgerMetadata(ledgerId);
assertEquals(readMetadata, metadataQueue1.take());
assertNull(metadataQueue2.poll());
@@ -360,9 +359,9 @@ public void testRegisterLedgerMetadataListener() throws Exception {
assertEquals(0, lms.getNumConsumers());
// update the metadata again
- metadata.setVersion(readMetadata.getVersion());
- metadata.addEnsemble(30L, createNumBookies(3));
- writeLedgerMetadata(ledgerId, metadata);
+ writeLedgerMetadata(ledgerId,
+ LedgerMetadataBuilder.from(metadata).newEnsembleEntry(30L, createNumBookies(3)).build(),
+ readMetadata.getVersion());
readMetadata = readLedgerMetadata(ledgerId);
assertNull(metadataQueue1.poll());
assertNull(metadataQueue2.poll());
@@ -386,15 +385,15 @@ public void testRegisterLedgerMetadataListener() throws Exception {
assertNull(metadataQueue2.poll());
}
- LedgerMetadata readLedgerMetadata(long lid) throws Exception {
- GenericCallbackFuture<LedgerMetadata> readFuture = new GenericCallbackFuture<>();
+ Versioned<LedgerMetadata> readLedgerMetadata(long lid) throws Exception {
+ GenericCallbackFuture<Versioned<LedgerMetadata>> readFuture = new GenericCallbackFuture<>();
lm.readLedgerMetadata(lid, readFuture);
return result(readFuture);
}
- void writeLedgerMetadata(long lid, LedgerMetadata metadata) throws Exception {
- GenericCallbackFuture<LedgerMetadata> writeFuture = new GenericCallbackFuture<>();
- lm.writeLedgerMetadata(lid, metadata, writeFuture);
+ void writeLedgerMetadata(long lid, LedgerMetadata metadata, Version version) throws Exception {
+ GenericCallbackFuture<Versioned<LedgerMetadata>> writeFuture = new GenericCallbackFuture<>();
+ lm.writeLedgerMetadata(lid, metadata, version, writeFuture);
result(writeFuture);
}
diff --git a/tests/backward-compat/recovery-no-password/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatRecoveryNoPassword.groovy b/tests/backward-compat/recovery-no-password/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatRecoveryNoPassword.groovy
index 0818b2aa54..a4858e9a8d 100644
--- a/tests/backward-compat/recovery-no-password/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatRecoveryNoPassword.groovy
+++ b/tests/backward-compat/recovery-no-password/src/test/groovy/org/apache/bookkeeper/tests/backwardcompat/TestCompatRecoveryNoPassword.groovy
@@ -38,6 +38,7 @@ import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.ReadEntryCallback
import org.apache.bookkeeper.tests.integration.utils.BookKeeperClusterUtils
import org.apache.bookkeeper.tests.integration.utils.DockerUtils
import org.apache.bookkeeper.tests.integration.utils.MavenClassLoader
+import org.apache.bookkeeper.versioning.Versioned
import org.jboss.arquillian.junit.Arquillian
import org.jboss.arquillian.test.api.ArquillianResource
@@ -58,7 +59,7 @@ class TestCompatRecoveryNoPassword {
DockerClient docker
private LedgerMetadata getLedgerMetadata(BookKeeper bookkeeper, long ledgerId) throws Exception {
- CompletableFuture<LedgerMetadata> future = new CompletableFuture<LedgerMetadata>()
+ CompletableFuture<Versioned<LedgerMetadata>> future = new CompletableFuture<>()
bookkeeper.getLedgerManager().readLedgerMetadata(
ledgerId, { rc, result ->
if (rc != BKException.Code.OK) {
@@ -67,7 +68,7 @@ class TestCompatRecoveryNoPassword {
future.complete(result)
}
})
- return future.get()
+ return future.get().getValue()
}
private static class ReplicationVerificationCallback implements ReadEntryCallback {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services