Skip to content

pulsar
pulsar

apache / pulsar

通知 129
Star 40
Fork 3
pulsar
pulsar

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 f0fbb4da 编写于 作者: M Matteo Merli 提交者： GitHub
浏览文件
操作

Extending ZK version wrapper to include ctime and mtime (#288)

上级 d1cb591d
隐藏空白更改
内联 并排
Showing with 106 addition and 79 deletion
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
浏览文件 @ f0fbb4da
......@@ -18,7 +18,6 @@ package org.apache.bookkeeper.mledger.impl;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun;
import static org.apache.bookkeeper.mledger.impl.MetaStoreImplZookeeper.ZNodeProtobufFormat;
import java.util.ArrayDeque;
import java.util.Collections;
......@@ -56,7 +55,8 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.PositionBound;
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.Version;
import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
import org.apache.bookkeeper.mledger.impl.MetaStoreImplZookeeper.ZNodeProtobufFormat;
import org.apache.bookkeeper.mledger.proto.MLDataFormats;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
......@@ -105,8 +105,8 @@ public class ManagedCursorImpl implements ManagedCursor {
// Current ledger used to append the mark-delete position
private volatile LedgerHandle cursorLedger;
// Version of the cursor z-node
private volatile Version cursorLedgerVersion;
// Stat of the cursor z-node
private volatile Stat cursorLedgerStat;
private final RangeSet<PositionImpl> individualDeletedMessages = TreeRangeSet.create();
private final ReadWriteLock lock = new ReentrantReadWriteLock();
......@@ -188,9 +188,9 @@ public class ManagedCursorImpl implements ManagedCursor {
log.info("[{}] Recovering from bookkeeper ledger cursor: {}", ledger.getName(), name);
ledger.getStore().asyncGetCursorInfo(ledger.getName(), name, new MetaStoreCallback<ManagedCursorInfo>() {
@Override
public void operationComplete(ManagedCursorInfo info, Version version) {
public void operationComplete(ManagedCursorInfo info, Stat stat) {
cursorLedgerVersion = version;
cursorLedgerStat = stat;
if (info.getCursorsLedgerId() == -1L) {
// There is no cursor ledger to read the last position from. It means the cursor has been properly
......@@ -765,10 +765,10 @@ public class ManagedCursorImpl implements ManagedCursor {
if (cursorLedger == null) {
persistPositionMetaStore(-1, newPosition, new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat stat) {
log.info("[{}] Updated cursor {} with ledger id {} md-position={} rd-position={}",
ledger.getName(), name, -1, markDeletePosition, readPosition);
cursorLedgerVersion = version;
cursorLedgerStat = stat;
finalCallback.operationComplete();
}
......@@ -1684,11 +1684,11 @@ public class ManagedCursorImpl implements ManagedCursor {
log.debug("[{}][{}] Closing cursor at md-position: {}", ledger.getName(), name, markDeletePosition);
}
ledger.getStore().asyncUpdateCursorInfo(ledger.getName(), name, info.build(), cursorLedgerVersion,
ledger.getStore().asyncUpdateCursorInfo(ledger.getName(), name, info.build(), cursorLedgerStat,
new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
callback.operationComplete(result, version);
public void operationComplete(Void result, Stat stat) {
callback.operationComplete(result, stat);
}
@Override
......@@ -1739,7 +1739,7 @@ public class ManagedCursorImpl implements ManagedCursor {
persistPositionMetaStore(-1, markDeletePosition, new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat stat) {
log.info("[{}][{}] Closed cursor at md-position={}", ledger.getName(), name,
markDeletePosition);
......@@ -1974,12 +1974,12 @@ public class ManagedCursorImpl implements ManagedCursor {
}
persistPositionMetaStore(lh.getId(), markDeletePosition, new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat stat) {
log.info("[{}] Updated cursor {} with ledger id {} md-position={} rd-position={}",
ledger.getName(), name, lh.getId(), markDeletePosition, readPosition);
final LedgerHandle oldLedger = cursorLedger;
cursorLedger = lh;
cursorLedgerVersion = version;
cursorLedgerStat = stat;
// At this point the position had already been safely markdeleted
callback.operationComplete();
......
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
浏览文件 @ f0fbb4da
......@@ -56,7 +56,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerMXBean;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.VoidCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.Version;
import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
import org.apache.bookkeeper.mledger.util.CallbackMutex;
......@@ -95,7 +95,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
private final ConcurrentLongHashMap<CompletableFuture<LedgerHandle>> ledgerCache = new ConcurrentLongHashMap<>();
private final NavigableMap<Long, LedgerInfo> ledgers = new ConcurrentSkipListMap<>();
private Version ledgersVersion;
private volatile Stat ledgersStat;
private final ManagedCursorContainer cursors = new ManagedCursorContainer();
private final ManagedCursorContainer activeCursors = new ManagedCursorContainer();
......@@ -195,7 +195,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
NUMBER_OF_ENTRIES_UPDATER.set(this, 0);
ENTRIES_ADDED_COUNTER_UPDATER.set(this, 0);
STATE_UPDATER.set(this, State.None);
this.ledgersVersion = null;
this.ledgersStat = null;
this.mbean = new ManagedLedgerMBeanImpl(this);
this.entryCache = factory.getEntryCacheManager().getEntryCache(this);
this.waitingCursors = Queues.newConcurrentLinkedQueue();
......@@ -212,8 +212,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// Fetch the list of existing ledgers in the managed ledger
store.getManagedLedgerInfo(name, new MetaStoreCallback<ManagedLedgerInfo>() {
@Override
public void operationComplete(ManagedLedgerInfo mlInfo, Version version) {
ledgersVersion = version;
public void operationComplete(ManagedLedgerInfo mlInfo, Stat stat) {
ledgersStat = stat;
for (LedgerInfo ls : mlInfo.getLedgerInfoList()) {
ledgers.put(ls.getLedgerId(), ls);
}
......@@ -286,8 +286,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
final MetaStoreCallback<Void> storeLedgersCb = new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void v, Version version) {
ledgersVersion = version;
public void operationComplete(Void v, Stat stat) {
ledgersStat = stat;
initializeCursors(callback);
}
......@@ -320,7 +320,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
ManagedLedgerInfo mlInfo = ManagedLedgerInfo.newBuilder().addAllLedgerInfo(ledgers.values())
.build();
store.asyncUpdateLedgerIds(name, mlInfo, ledgersVersion, storeLedgersCb);
store.asyncUpdateLedgerIds(name, mlInfo, ledgersStat, storeLedgersCb);
}));
}, null);
}
......@@ -331,7 +331,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
store.getCursors(name, new MetaStoreCallback<List<String>>() {
@Override
public void operationComplete(List<String> consumers, Version v) {
public void operationComplete(List<String> consumers, Stat s) {
// Load existing cursors
final AtomicInteger cursorCount = new AtomicInteger(consumers.size());
if (log.isDebugEnabled()) {
......@@ -621,7 +621,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
// ledger from BK) don't, we end up having a loose ledger leaked but the state will be consistent.
store.asyncRemoveCursor(ManagedLedgerImpl.this.name, consumerName, new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat stat) {
cursor.asyncDeleteCursorLedger();
cursors.removeCursor(consumerName);
......@@ -951,13 +951,13 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
final MetaStoreCallback<Void> cb = new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void v, Version version) {
public void operationComplete(Void v, Stat stat) {
if (log.isDebugEnabled()) {
log.debug("[{}] Updating of ledgers list after create complete. version={}", name, version);
log.debug("[{}] Updating of ledgers list after create complete. version={}", name, stat);
}
ledgersVersion = version;
ledgersStat = stat;
ledgersListMutex.unlock();
updateLedgersIdsComplete(version);
updateLedgersIdsComplete(stat);
synchronized (ManagedLedgerImpl.this) {
mbean.addLedgerSwitchLatencySample(System.nanoTime() - lastLedgerCreationInitiationTimestamp,
TimeUnit.NANOSECONDS);
......@@ -1013,12 +1013,12 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
ManagedLedgerInfo mlInfo = ManagedLedgerInfo.newBuilder().addAllLedgerInfo(ledgers.values()).build();
if (log.isDebugEnabled()) {
log.debug("[{}] Updating ledgers ids with new ledger. version={}", name, ledgersVersion);
log.debug("[{}] Updating ledgers ids with new ledger. version={}", name, ledgersStat);
}
store.asyncUpdateLedgerIds(name, mlInfo, ledgersVersion, callback);
store.asyncUpdateLedgerIds(name, mlInfo, ledgersStat, callback);
}
public synchronized void updateLedgersIdsComplete(Version version) {
public synchronized void updateLedgersIdsComplete(Stat stat) {
STATE_UPDATER.set(this, State.LedgerOpened);
lastLedgerCreatedTimestamp = System.currentTimeMillis();
......@@ -1436,12 +1436,12 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
log.debug("[{}] Updating of ledgers list after trimming", name);
}
ManagedLedgerInfo mlInfo = ManagedLedgerInfo.newBuilder().addAllLedgerInfo(ledgers.values()).build();
store.asyncUpdateLedgerIds(name, mlInfo, ledgersVersion, new MetaStoreCallback<Void>() {
store.asyncUpdateLedgerIds(name, mlInfo, ledgersStat, new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat stat) {
log.info("[{}] End TrimConsumedLedgers. ledgers={} totalSize={}", name, ledgers.size(),
TOTAL_SIZE_UPDATER.get(ManagedLedgerImpl.this));
ledgersVersion = version;
ledgersStat = stat;
ledgersListMutex.unlock();
trimmerMutex.unlock();
......@@ -1593,7 +1593,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
private void deleteMetadata(DeleteLedgerCallback callback, Object ctx) {
store.removeManagedLedger(name, new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat stat) {
log.info("[{}] Successfully deleted managed ledger", name);
factory.close(ManagedLedgerImpl.this);
callback.deleteLedgerComplete(ctx);
......
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerOfflineBacklog.java
浏览文件 @ f0fbb4da
......@@ -144,7 +144,7 @@ public class ManagedLedgerOfflineBacklog {
store.getManagedLedgerInfo(managedLedgerName,
new MetaStore.MetaStoreCallback<MLDataFormats.ManagedLedgerInfo>() {
@Override
public void operationComplete(MLDataFormats.ManagedLedgerInfo mlInfo, MetaStore.Version version) {
public void operationComplete(MLDataFormats.ManagedLedgerInfo mlInfo, MetaStore.Stat version) {
for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ls : mlInfo.getLedgerInfoList()) {
ledgers.put(ls.getLedgerId(), ls);
}
......@@ -227,7 +227,7 @@ public class ManagedLedgerOfflineBacklog {
store.getCursors(managedLedgerName, new MetaStore.MetaStoreCallback<List<String>>() {
@Override
public void operationComplete(List<String> cursors, MetaStore.Version v) {
public void operationComplete(List<String> cursors, MetaStore.Stat v) {
// Load existing cursors
if (log.isDebugEnabled()) {
log.debug("[{}] Found {} cursors", managedLedgerName, cursors.size());
......@@ -333,7 +333,7 @@ public class ManagedLedgerOfflineBacklog {
new MetaStore.MetaStoreCallback<MLDataFormats.ManagedCursorInfo>() {
@Override
public void operationComplete(MLDataFormats.ManagedCursorInfo info,
MetaStore.Version version) {
MetaStore.Stat version) {
long cursorLedgerId = info.getCursorsLedgerId();
if (log.isDebugEnabled()) {
log.debug("[{}] Cursor {} meta-data read ledger id {}", managedLedgerName,
......
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStore.java
浏览文件 @ f0fbb4da
......@@ -26,15 +26,18 @@ import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo;
*/
public interface MetaStore {
public static interface Version {
public static interface Stat {
int getVersion();
long getCreationTimestamp();
long getModificationTimestamp();
}
public static interface UpdateLedgersIdsCallback {
void updateLedgersIdsComplete(MetaStoreException status, Version version);
void updateLedgersIdsComplete(MetaStoreException status, Stat stat);
}
public static interface MetaStoreCallback<T> {
void operationComplete(T result, Version version);
void operationComplete(T result, Stat stat);
void operationFailed(MetaStoreException e);
}
......@@ -63,7 +66,7 @@ public interface MetaStore {
* @param ctx
* opaque context object
*/
void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Version version,
void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Stat version,
MetaStoreCallback<Void> callback);
/**
......@@ -98,7 +101,7 @@ public interface MetaStore {
* the callback
* @throws MetaStoreException
*/
void asyncUpdateCursorInfo(String ledgerName, String cursorName, ManagedCursorInfo info, Version version,
void asyncUpdateCursorInfo(String ledgerName, String cursorName, ManagedCursorInfo info, Stat version,
MetaStoreCallback<Void> callback);
/**
......
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplZookeeper.java
浏览文件 @ f0fbb4da
......@@ -58,11 +58,36 @@ public class MetaStoreImplZookeeper implements MetaStore {
private final ZNodeProtobufFormat protobufFormat;
private final OrderedSafeExecutor executor;
private static class ZKVersion implements Version {
final int version;
private static class ZKStat implements Stat {
private final int version;
private final long creationTimestamp;
private final long modificationTimestamp;
ZKStat(org.apache.zookeeper.data.Stat stat) {
this.version = stat.getVersion();
this.creationTimestamp = stat.getCtime();
this.modificationTimestamp = stat.getMtime();
}
ZKStat() {
this.version = 0;
this.creationTimestamp = System.currentTimeMillis();
this.modificationTimestamp = System.currentTimeMillis();
}
@Override
public int getVersion() {
return version;
}
@Override
public long getCreationTimestamp() {
return creationTimestamp;
}
ZKVersion(int version) {
this.version = version;
@Override
public long getModificationTimestamp() {
return modificationTimestamp;
}
}
......@@ -113,7 +138,7 @@ public class MetaStoreImplZookeeper implements MetaStore {
try {
ManagedLedgerInfo info = parseManagedLedgerInfo(readData);
info = updateMLInfoTimestamp(info);
callback.operationComplete(info, new ZKVersion(stat.getVersion()));
callback.operationComplete(info, new ZKStat(stat));
} catch (ParseException | InvalidProtocolBufferException e) {
callback.operationFailed(new MetaStoreException(e));
}
......@@ -123,7 +148,7 @@ public class MetaStoreImplZookeeper implements MetaStore {
StringCallback createcb = (rc1, path1, ctx1, name) -> {
if (rc1 == Code.OK.intValue()) {
ManagedLedgerInfo info = ManagedLedgerInfo.getDefaultInstance();
callback.operationComplete(info, new ZKVersion(0));
callback.operationComplete(info, new ZKStat());
} else {
callback.operationFailed(new MetaStoreException(KeeperException.create(Code.get(rc1))));
}
......@@ -138,20 +163,20 @@ public class MetaStoreImplZookeeper implements MetaStore {
}
@Override
public void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Version version,
public void asyncUpdateLedgerIds(String ledgerName, ManagedLedgerInfo mlInfo, Stat stat,
final MetaStoreCallback<Void> callback) {
ZKVersion zkVersion = (ZKVersion) version;
ZKStat zkStat = (ZKStat) stat;
if (log.isDebugEnabled()) {
log.debug("[{}] Updating metadata version={} with content={}", ledgerName, zkVersion.version, mlInfo);
log.debug("[{}] Updating metadata version={} with content={}", ledgerName, zkStat.version, mlInfo);
}
byte[] serializedMlInfo = protobufFormat == ZNodeProtobufFormat.Text ? //
mlInfo.toString().getBytes(Encoding) : // Text format
mlInfo.toByteArray(); // Binary format
zk.setData(prefix + ledgerName, serializedMlInfo, zkVersion.version,
(rc, path, zkCtx, stat) -> executor.submit(safeRun(() -> {
zk.setData(prefix + ledgerName, serializedMlInfo, zkStat.getVersion(),
(rc, path, zkCtx, stat1) -> executor.submit(safeRun(() -> {
if (log.isDebugEnabled()) {
log.debug("[{}] UpdateLedgersIdsCallback.processResult rc={} newVersion={}", ledgerName,
Code.get(rc), stat != null ? stat.getVersion() : "null");
......@@ -165,7 +190,7 @@ public class MetaStoreImplZookeeper implements MetaStore {
status = new MetaStoreException(KeeperException.create(Code.get(rc)));
callback.operationFailed(status);
} else {
callback.operationComplete(null, new ZKVersion(stat.getVersion()));
callback.operationComplete(null, new ZKStat(stat1));
}
})), null);
}
......@@ -187,8 +212,7 @@ public class MetaStoreImplZookeeper implements MetaStore {
if (log.isDebugEnabled()) {
log.debug("[{}] Get childrend completed version={}", ledgerName, stat.getVersion());
}
ZKVersion version = new ZKVersion(stat.getVersion());
callback.operationComplete(children, version);
callback.operationComplete(children, new ZKStat(stat));
})), null);
}
......@@ -206,7 +230,7 @@ public class MetaStoreImplZookeeper implements MetaStore {
} else {
try {
ManagedCursorInfo info = parseManagedCursorInfo(data);
callback.operationComplete(info, new ZKVersion(stat.getVersion()));
callback.operationComplete(info, new ZKStat(stat));
} catch (ParseException | InvalidProtocolBufferException e) {
callback.operationFailed(new MetaStoreException(e));
}
......@@ -220,7 +244,7 @@ public class MetaStoreImplZookeeper implements MetaStore {
@Override
public void asyncUpdateCursorInfo(final String ledgerName, final String cursorName, final ManagedCursorInfo info,
Version version, final MetaStoreCallback<Void> callback) {
Stat stat, final MetaStoreCallback<Void> callback) {
log.info("[{}] [{}] Updating cursor info ledgerId={} mark-delete={}:{}", ledgerName, cursorName,
info.getCursorsLedgerId(), info.getMarkDeleteLedgerId(), info.getMarkDeleteEntryId());
......@@ -229,7 +253,7 @@ public class MetaStoreImplZookeeper implements MetaStore {
info.toString().getBytes(Encoding) : // Text format
info.toByteArray(); // Binary format
if (version == null) {
if (stat == null) {
if (log.isDebugEnabled()) {
log.debug("[{}] Creating consumer {} on meta-data store with {}", ledgerName, cursorName, info);
}
......@@ -244,21 +268,21 @@ public class MetaStoreImplZookeeper implements MetaStore {
log.debug("[{}] Created consumer {} on meta-data store with {}", ledgerName, cursorName,
info);
}
callback.operationComplete(null, new ZKVersion(0));
callback.operationComplete(null, new ZKStat());
}
})), null);
} else {
ZKVersion zkVersion = (ZKVersion) version;
ZKStat zkStat = (ZKStat) stat;
if (log.isDebugEnabled()) {
log.debug("[{}] Updating consumer {} on meta-data store with {}", ledgerName, cursorName, info);
}
zk.setData(path, content, zkVersion.version, (rc, path1, ctx, stat) -> executor.submit(safeRun(() -> {
zk.setData(path, content, zkStat.getVersion(), (rc, path1, ctx, stat1) -> executor.submit(safeRun(() -> {
if (rc == Code.BADVERSION.intValue()) {
callback.operationFailed(new BadVersionException(KeeperException.create(Code.get(rc))));
} else if (rc != Code.OK.intValue()) {
callback.operationFailed(new MetaStoreException(KeeperException.create(Code.get(rc))));
} else {
callback.operationComplete(null, new ZKVersion(stat.getVersion()));
callback.operationComplete(null, new ZKStat(stat1));
}
})), null);
}
......
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
浏览文件 @ f0fbb4da
......@@ -58,7 +58,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.Version;
import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
import org.apache.bookkeeper.mledger.impl.MetaStoreImplZookeeper.ZNodeProtobufFormat;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
......@@ -1473,7 +1473,7 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
final MetaStore store = factory.getMetaStore();
store.getManagedLedgerInfo("my_test_ledger", new MetaStoreCallback<ManagedLedgerInfo>() {
@Override
public void operationComplete(ManagedLedgerInfo result, Version version) {
public void operationComplete(ManagedLedgerInfo result, Stat version) {
// Update the list
ManagedLedgerInfo.Builder info = ManagedLedgerInfo.newBuilder(result);
info.clearLedgerInfo();
......@@ -1482,7 +1482,7 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
store.asyncUpdateLedgerIds("my_test_ledger", info.build(), version, new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat version) {
counter.countDown();
}
......@@ -1725,7 +1725,7 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
@Test
public void testBackwardCompatiblityForMeta() throws Exception {
final ManagedLedgerInfo[] storedMLInfo = new ManagedLedgerInfo[3];
final Version[] versions = new Version[1];
final Stat[] versions = new Stat[1];
ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(bkc, bkc.getZkHandle());
ManagedLedgerConfig conf = new ManagedLedgerConfig();
......@@ -1745,7 +1745,7 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
// obtain the ledger info
store.getManagedLedgerInfo("backward_test_ledger", new MetaStoreCallback<ManagedLedgerInfo>() {
@Override
public void operationComplete(ManagedLedgerInfo result, Version version) {
public void operationComplete(ManagedLedgerInfo result, Stat version) {
storedMLInfo[0] = result;
versions[0] = version;
l1.countDown();
......@@ -1774,7 +1774,7 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
CountDownLatch l2 = new CountDownLatch(1);
store.asyncUpdateLedgerIds("backward_test_ledger", storedMLInfo[1], versions[0], new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat version) {
l2.countDown();
}
......
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/MetaStoreImplZookeeperTest.java
浏览文件 @ f0fbb4da
......@@ -23,7 +23,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
import org.apache.bookkeeper.mledger.impl.MetaStore.Version;
import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
import org.apache.bookkeeper.mledger.proto.MLDataFormats;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo;
......@@ -58,7 +58,7 @@ public class MetaStoreImplZookeeperTest extends MockedBookKeeperTestCase {
store.removeManagedLedger("non-existing", new MetaStoreCallback<Void>() {
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat version) {
counter.countDown();
}
......@@ -89,7 +89,7 @@ public class MetaStoreImplZookeeperTest extends MockedBookKeeperTestCase {
latch.countDown();
}
public void operationComplete(ManagedLedgerInfo result, Version version) {
public void operationComplete(ManagedLedgerInfo result, Stat version) {
fail("Operation should have failed");
}
});
......@@ -113,7 +113,7 @@ public class MetaStoreImplZookeeperTest extends MockedBookKeeperTestCase {
latch.countDown();
}
public void operationComplete(ManagedCursorInfo result, Version version) {
public void operationComplete(ManagedCursorInfo result, Stat version) {
fail("Operation should have failed");
}
});
......@@ -135,7 +135,7 @@ public class MetaStoreImplZookeeperTest extends MockedBookKeeperTestCase {
latch.countDown();
}
public void operationComplete(ManagedLedgerInfo result, Version version) {
public void operationComplete(ManagedLedgerInfo result, Stat version) {
fail("Operation should have failed");
}
});
......@@ -157,7 +157,7 @@ public class MetaStoreImplZookeeperTest extends MockedBookKeeperTestCase {
fail("should have succeeded");
}
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat version) {
// Update again using the version
zkc.failNow(Code.CONNECTIONLOSS);
......@@ -169,7 +169,7 @@ public class MetaStoreImplZookeeperTest extends MockedBookKeeperTestCase {
}
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat version) {
fail("should have failed");
}
});
......@@ -192,7 +192,7 @@ public class MetaStoreImplZookeeperTest extends MockedBookKeeperTestCase {
fail("should have succeeded");
}
public void operationComplete(ManagedLedgerInfo mlInfo, Version version) {
public void operationComplete(ManagedLedgerInfo mlInfo, Stat version) {
// Update again using the version
zkc.failNow(Code.BADVERSION);
......@@ -203,7 +203,7 @@ public class MetaStoreImplZookeeperTest extends MockedBookKeeperTestCase {
}
@Override
public void operationComplete(Void result, Version version) {
public void operationComplete(Void result, Stat version) {
fail("should have failed");
}
});
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或