You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by si...@apache.org on 2014/05/12 08:10:01 UTC
svn commit: r1593889 - in /zookeeper/bookkeeper/trunk: ./
bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/
bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/
bookkeeper-server/src/test/java/org/apache/bookkeeper/client/
Author: sijie
Date: Mon May 12 06:10:00 2014
New Revision: 1593889
URL: http://svn.apache.org/r1593889
Log:
BOOKKEEPER-747: Implement register/unregister LedgerMetadataListener in MSLedgerManagerFactory (fpj via sijie)
Added:
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MSWatchedEvent.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreWatcher.java
Modified:
zookeeper/bookkeeper/trunk/CHANGES.txt
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManagerFactory.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/InMemoryMetastoreTable.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreTable.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java
Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1593889&r1=1593888&r2=1593889&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Mon May 12 06:10:00 2014
@@ -268,6 +268,8 @@ Trunk (unreleased changes)
BOOKKEEPER-731: Missing BOOKIE_PID_DIR and BOOKIE_STOP_TIMEOUT in env variables list (rakesh via sijie)
+ BOOKKEEPER-747: Implement register/unregister LedgerMetadataListener in MSLedgerManagerFactory (fpj via sijie)
+
NEW FEATURE:
BOOKKEEPER-562: Ability to tell if a ledger is closed or not (fpj)
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java?rev=1593889&r1=1593888&r2=1593889&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManager.java Mon May 12 06:10:00 2014
@@ -100,8 +100,10 @@ abstract class AbstractZkLedgerManager i
scheduler.submit(new Runnable() {
@Override
public void run() {
- for (LedgerMetadataListener listener : listenerSet) {
- listener.onChanged(ledgerId, result);
+ synchronized(listenerSet) {
+ for (LedgerMetadataListener listener : listenerSet) {
+ listener.onChanged(ledgerId, result);
+ }
}
}
});
@@ -188,10 +190,16 @@ abstract class AbstractZkLedgerManager i
}
switch (event.getType()) {
case NodeDeleted:
- Set<LedgerMetadataListener> listenerSet = listeners.remove(ledgerId);
+ Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
if (null != listenerSet) {
- LOG.debug("Removed ledger metadata listeners on ledger {} : {}",
- ledgerId, listenerSet);
+ synchronized(listenerSet){
+ LOG.debug("Removed ledger metadata listeners on ledger {} : {}",
+ ledgerId, listenerSet);
+ for(LedgerMetadataListener l : listenerSet) {
+ unregisterLedgerMetadataListener(ledgerId, l);
+ l.onChanged( ledgerId, null );
+ }
+ }
} else {
LOG.debug("No ledger metadata listeners to remove from ledger {} after it's deleted.",
ledgerId);
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java?rev=1593889&r1=1593888&r2=1593889&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java Mon May 12 06:10:00 2014
@@ -30,7 +30,7 @@ import org.apache.zookeeper.ZooKeeper;
/**
* Flat Ledger Manager Factory
*/
-class FlatLedgerManagerFactory extends LedgerManagerFactory {
+public class FlatLedgerManagerFactory extends LedgerManagerFactory {
public static final String NAME = "flat";
public static final int CUR_VERSION = 1;
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManagerFactory.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManagerFactory.java?rev=1593889&r1=1593888&r2=1593889&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManagerFactory.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/HierarchicalLedgerManagerFactory.java Mon May 12 06:10:00 2014
@@ -30,7 +30,7 @@ import org.apache.zookeeper.ZooKeeper;
/**
* Hierarchical Ledger Manager Factory
*/
-class HierarchicalLedgerManagerFactory extends LedgerManagerFactory {
+public class HierarchicalLedgerManagerFactory extends LedgerManagerFactory {
public static final String NAME = "hierarchical";
public static final int CUR_VERSION = 1;
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java?rev=1593889&r1=1593888&r2=1593889&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MSLedgerManagerFactory.java Mon May 12 06:10:00 2014
@@ -21,18 +21,24 @@ import static org.apache.bookkeeper.meta
import static org.apache.bookkeeper.metastore.MetastoreTable.NON_FIELDS;
import java.io.IOException;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.LedgerMetadata;
import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.meta.AbstractZkLedgerManager.ReadLedgerMetadataTask;
import org.apache.bookkeeper.metastore.MSException;
+import org.apache.bookkeeper.metastore.MSWatchedEvent;
import org.apache.bookkeeper.metastore.MetaStore;
import org.apache.bookkeeper.metastore.MetastoreCallback;
import org.apache.bookkeeper.metastore.MetastoreCursor;
@@ -41,6 +47,8 @@ import org.apache.bookkeeper.metastore.M
import org.apache.bookkeeper.metastore.MetastoreFactory;
import org.apache.bookkeeper.metastore.MetastoreScannableTable;
import org.apache.bookkeeper.metastore.MetastoreTableItem;
+import org.apache.bookkeeper.metastore.MetastoreWatcher;
+import org.apache.bookkeeper.metastore.MSWatchedEvent.EventType;
import org.apache.bookkeeper.metastore.Value;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
@@ -68,6 +76,8 @@ public class MSLedgerManagerFactory exte
private final static Logger LOG = LoggerFactory.getLogger(MSLedgerManagerFactory.class);
+ static int MS_CONNECT_BACKOFF_MS = 200;
+
public static final int CUR_VERSION = 1;
public static final String TABLE_NAME = "LEDGER";
@@ -170,7 +180,7 @@ public class MSLedgerManagerFactory exte
}
}
- static class MsLedgerManager implements LedgerManager {
+ static class MsLedgerManager implements LedgerManager, MetastoreWatcher {
final ZooKeeper zk;
final AbstractConfiguration conf;
@@ -181,6 +191,10 @@ public class MSLedgerManagerFactory exte
static final String IDGEN_ZNODE = "ms-idgen";
static final String IDGENERATION_PREFIX = "/" + IDGEN_ZNODE + "/ID-";
+ // ledger metadata listeners
+ protected final ConcurrentMap<Long, Set<LedgerMetadataListener>> listeners =
+ new ConcurrentHashMap<Long, Set<LedgerMetadataListener>>();
+
// Path to generate global id
private final String idGenPath;
@@ -188,6 +202,55 @@ public class MSLedgerManagerFactory exte
// callbacks
ScheduledExecutorService scheduler;
+ protected class ReadLedgerMetadataTask implements Runnable, GenericCallback<LedgerMetadata> {
+
+ final long ledgerId;
+
+ ReadLedgerMetadataTask(long ledgerId) {
+ this.ledgerId = ledgerId;
+ }
+
+ @Override
+ public void run() {
+ if (null != listeners.get(ledgerId)) {
+ LOG.debug("Re-read ledger metadata for {}.", ledgerId);
+ readLedgerMetadata(ledgerId, ReadLedgerMetadataTask.this);
+ } else {
+ LOG.debug("Ledger metadata listener for ledger {} is already removed.", ledgerId);
+ }
+ }
+
+ @Override
+ public void operationComplete(int rc, final LedgerMetadata result) {
+ if (BKException.Code.OK == rc) {
+ final Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
+ if (null != listenerSet) {
+ LOG.debug("Ledger metadata is changed for {} : {}.", ledgerId, result);
+ scheduler.submit(new Runnable() {
+ @Override
+ public void run() {
+ synchronized(listenerSet) {
+ for (LedgerMetadataListener listener : listenerSet) {
+ listener.onChanged(ledgerId, result);
+ }
+ }
+ }
+ });
+ }
+ } else if (BKException.Code.NoSuchLedgerExistsException == rc) {
+ // the ledger is removed, do nothing
+ Set<LedgerMetadataListener> listenerSet = listeners.remove(ledgerId);
+ if (null != listenerSet) {
+ LOG.debug("Removed ledger metadata listener set on ledger {} as its ledger is deleted : {}",
+ ledgerId, listenerSet.size());
+ }
+ } else {
+ LOG.warn("Failed on read ledger metadata of ledger {} : {}", ledgerId, rc);
+ scheduler.schedule(this, MS_CONNECT_BACKOFF_MS, TimeUnit.MILLISECONDS);
+ }
+ }
+ }
+
MsLedgerManager(final AbstractConfiguration conf, final ZooKeeper zk, final MetaStore metastore) {
this.conf = conf;
this.zk = zk;
@@ -211,13 +274,65 @@ public class MSLedgerManagerFactory exte
}
@Override
+ public void process(MSWatchedEvent e){
+ long ledgerId = key2LedgerId(e.getKey());
+ switch(e.getType()) {
+ case CHANGED:
+ new ReadLedgerMetadataTask(key2LedgerId(e.getKey())).run();
+
+ break;
+ case REMOVED:
+ Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
+ if (listenerSet != null) {
+ synchronized (listenerSet) {
+ for(LedgerMetadataListener l : listenerSet){
+ unregisterLedgerMetadataListener(ledgerId, l);
+ l.onChanged( ledgerId, null );
+ }
+ }
+ }
+
+ break;
+ default:
+ LOG.warn("Unknown type: {}", e.getType());
+ break;
+ }
+ }
+
+ @Override
public void registerLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener) {
- // TODO: should provide ledger metadata listener in metadata store.
+ if (null != listener) {
+ LOG.info("Registered ledger metadata listener {} on ledger {}.", listener, ledgerId);
+ Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
+ if (listenerSet == null) {
+ Set<LedgerMetadataListener> newListenerSet = new HashSet<LedgerMetadataListener>();
+ Set<LedgerMetadataListener> oldListenerSet = listeners.putIfAbsent(ledgerId, newListenerSet);
+ if (null != oldListenerSet) {
+ listenerSet = oldListenerSet;
+ } else {
+ listenerSet = newListenerSet;
+ }
+ }
+ synchronized (listenerSet) {
+ listenerSet.add(listener);
+ }
+ new ReadLedgerMetadataTask(ledgerId).run();
+ }
}
@Override
public void unregisterLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener) {
- // TODO: should provide ledger metadata listener in metadata store.
+ Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
+ if (listenerSet != null) {
+ synchronized (listenerSet) {
+ if (listenerSet.remove(listener)) {
+ LOG.info("Unregistered ledger metadata listener {} on ledger {}.", listener, ledgerId);
+ }
+ if (listenerSet.isEmpty()) {
+ listeners.remove(ledgerId, listenerSet);
+ }
+ }
+ }
}
@Override
@@ -354,7 +469,7 @@ public class MSLedgerManagerFactory exte
readCb.operationComplete(BKException.Code.OK, metadata);
}
};
- ledgerTable.get(key, msCallback, ALL_FIELDS);
+ ledgerTable.get(key, this, msCallback, ALL_FIELDS);
}
@Override
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/InMemoryMetastoreTable.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/InMemoryMetastoreTable.java?rev=1593889&r1=1593888&r2=1593889&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/InMemoryMetastoreTable.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/InMemoryMetastoreTable.java Mon May 12 06:10:00 2014
@@ -93,10 +93,12 @@ public class InMemoryMetastoreTable impl
private String name;
private TreeMap<String, Versioned<Value>> map = null;
+ private TreeMap<String, MetastoreWatcher> watcherMap = null;
private ScheduledExecutorService scheduler;
public InMemoryMetastoreTable(InMemoryMetaStore metastore, String name) {
this.map = new TreeMap<String, Versioned<Value>>();
+ this.watcherMap = new TreeMap<String,MetastoreWatcher>();
this.name = name;
String thName = "InMemoryMetastore-Table(" + name + ")-Scheduler-%d";
ThreadFactoryBuilder tfb = new ThreadFactoryBuilder()
@@ -144,6 +146,19 @@ public class InMemoryMetastoreTable impl
}
});
}
+
+ @Override
+ public void get(final String key, final MetastoreWatcher watcher, final MetastoreCallback<Versioned<Value>> cb, final Object ctx) {
+ scheduler.submit(new Runnable() {
+ @Override
+ public void run() {
+ scheduleGet(key, ALL_FIELDS, cb, ctx);
+ synchronized(watcherMap) {
+ watcherMap.put( key, watcher );
+ }
+ }
+ });
+ }
@Override
public void get(final String key, final Set<String> fields, final MetastoreCallback<Versioned<Value>> cb,
@@ -182,10 +197,18 @@ public class InMemoryMetastoreTable impl
}
Result<Version> result = put(key, value, version);
cb.complete(result.code.getCode(), result.value, ctx);
+
+ /*
+ * If there is a watcher set for this key, we need
+ * to trigger it.
+ */
+ if(result.code == MSException.Code.OK){
+ triggerWatch(key, MSWatchedEvent.EventType.CHANGED);
+ }
}
});
}
-
+
@Override
public void remove(final String key, final Version version, final MetastoreCallback<Void> cb, final Object ctx) {
scheduler.submit(new Runnable() {
@@ -197,6 +220,10 @@ public class InMemoryMetastoreTable impl
}
Code code = remove(key, version);
cb.complete(code.getCode(), null, ctx);
+
+ if(code == MSException.Code.OK){
+ triggerWatch(key, MSWatchedEvent.EventType.REMOVED);
+ }
}
});
}
@@ -238,6 +265,16 @@ public class InMemoryMetastoreTable impl
});
}
+ private void triggerWatch(String key, MSWatchedEvent.EventType type) {
+ synchronized(watcherMap){
+ if(watcherMap.containsKey( key )) {
+ MSWatchedEvent event = new MSWatchedEvent(key, type);
+ watcherMap.get( key ).process( event );
+ watcherMap.remove( key );
+ }
+ }
+ }
+
private synchronized Versioned<Value> get(String key) {
return map.get(key);
}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MSWatchedEvent.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MSWatchedEvent.java?rev=1593889&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MSWatchedEvent.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MSWatchedEvent.java Mon May 12 06:10:00 2014
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bookkeeper.metastore;
+
+public class MSWatchedEvent {
+ public enum EventType {CHANGED, REMOVED};
+
+ String key;
+ EventType type;
+
+ public MSWatchedEvent(String key, EventType type) {
+ this.key = key;
+ this.type = type;
+ }
+
+ public EventType getType() {
+ return type;
+ }
+
+ public String getKey(){
+ return key;
+ }
+}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreTable.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreTable.java?rev=1593889&r1=1593888&r2=1593889&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreTable.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreTable.java Mon May 12 06:10:00 2014
@@ -56,6 +56,27 @@ public interface MetastoreTable {
public void get(String key, MetastoreCallback<Versioned<Value>> cb, Object ctx);
/**
+ * Get all fields of a key.
+ *
+ * <p>
+ * Return Code:<ul>
+ * <li>{@link MSException.Code.OK}: success returning the key</li>
+ * <li>{@link MSException.Code.NoKey}: no key found</li>
+ * <li>{@link MSException.Code.IllegalOp}/{@link MSException.Code.ServiceDown}: other issues</li>
+ * </ul></p>
+ *
+ * @param key
+ * Key Name
+ * @param watcher
+ * Watcher object to receive notifications
+ * @param cb
+ * Callback to return all fields of the key
+ * @param ctx
+ * Callback context
+ */
+ public void get(String key, MetastoreWatcher watcher, MetastoreCallback<Versioned<Value>> cb, Object ctx);
+
+ /**
* Get specified fields of a key.
*
* <p>
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreWatcher.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreWatcher.java?rev=1593889&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreWatcher.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/metastore/MetastoreWatcher.java Mon May 12 06:10:00 2014
@@ -0,0 +1,22 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.bookkeeper.metastore;
+
+public interface MetastoreWatcher {
+ public void process(MSWatchedEvent e);
+}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java?rev=1593889&r1=1593888&r2=1593889&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestWatchEnsembleChange.java Mon May 12 06:10:00 2014
@@ -21,24 +21,55 @@
package org.apache.bookkeeper.client;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.meta.FlatLedgerManagerFactory;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
+import org.apache.bookkeeper.meta.LedgerManager;
+import org.apache.bookkeeper.meta.LedgerManagerFactory;
+import org.apache.bookkeeper.meta.MSLedgerManagerFactory;
import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
+import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.util.ReflectionUtils;
+import org.apache.bookkeeper.versioning.Version;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+@RunWith(Parameterized.class)
public class TestWatchEnsembleChange extends BookKeeperClusterTestCase {
static Logger LOG = LoggerFactory.getLogger(TestWatchEnsembleChange.class);
final DigestType digestType;
+ final Class<? extends LedgerManagerFactory> lmFactoryCls;
- public TestWatchEnsembleChange() {
+ public TestWatchEnsembleChange(Class<? extends LedgerManagerFactory> lmFactoryCls) {
super(7);
this.digestType = DigestType.CRC32;
+ this.lmFactoryCls = lmFactoryCls;
+ baseClientConf.setLedgerManagerFactoryClass(lmFactoryCls);
+ baseConf.setLedgerManagerFactoryClass(lmFactoryCls);
+ }
+
+ @Parameters
+ public static Collection<Object[]> configs() {
+ return Arrays.asList(new Object[][] {
+ { FlatLedgerManagerFactory.class },
+ { HierarchicalLedgerManagerFactory.class },
+ { MSLedgerManagerFactory.class }
+ });
}
@Test(timeout = 60000)
@@ -68,4 +99,48 @@ public class TestWatchEnsembleChange ext
readLh.close();
lh.close();
}
+
+ @Test(timeout = 60000)
+ public void testWatchMetadataRemoval() throws Exception {
+ LedgerManagerFactory factory = ReflectionUtils.newInstance(lmFactoryCls);
+ factory.initialize(baseConf, super.zkc, factory.getCurrentVersion());
+ LedgerManager manager = factory.newLedgerManager();
+ final ByteBuffer bbLedgerId = ByteBuffer.allocate(8);
+ final CountDownLatch createLatch = new CountDownLatch(1);
+ final CountDownLatch removeLatch = new CountDownLatch(1);
+
+ manager.createLedger( new LedgerMetadata(4, 2, 2, digestType, "fpj was here".getBytes()),
+ new BookkeeperInternalCallbacks.GenericCallback<Long>(){
+
+ @Override
+ public void operationComplete(int rc, Long result) {
+ bbLedgerId.putLong(result);
+ bbLedgerId.flip();
+ createLatch.countDown();
+ }
+ });
+ assertTrue(createLatch.await(2000, TimeUnit.MILLISECONDS));
+ final long createdLid = bbLedgerId.getLong();
+
+ manager.registerLedgerMetadataListener( createdLid,
+ new LedgerMetadataListener() {
+
+ @Override
+ public void onChanged( long ledgerId, LedgerMetadata metadata ) {
+ assertEquals(ledgerId, createdLid);
+ assertEquals(metadata, null);
+ removeLatch.countDown();
+ }
+ });
+
+ manager.removeLedgerMetadata( createdLid, Version.ANY,
+ new BookkeeperInternalCallbacks.GenericCallback<Void>() {
+
+ @Override
+ public void operationComplete(int rc, Void result) {
+ assertEquals(rc, BKException.Code.OK);
+ }
+ });
+ assertTrue(removeLatch.await(2000, TimeUnit.MILLISECONDS));
+ }
}