You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by zh...@apache.org on 2018/01/09 12:47:04 UTC
[bookkeeper] branch master updated: ISSUE #897: un-bind zookeeper
from bookkeeper admin
This is an automated email from the ASF dual-hosted git repository.
zhaijia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 998a55f ISSUE #897: un-bind zookeeper from bookkeeper admin
998a55f is described below
commit 998a55fe2b549c93907c5774333be7f4431de3e5
Author: Jia Zhai <zh...@apache.org>
AuthorDate: Tue Jan 9 20:46:51 2018 +0800
ISSUE #897: un-bind zookeeper from bookkeeper admin
Descriptions of the changes in this PR:
- remove zookeeper from LedgerLayout to make LedgerLayout as a pure layout object,
- add LayoutManager, move zookeeper related utils into ZKLayoutManager for storing/reading/delete layout from zookeeper,
- unbind zookeeper from LedgerManagerFactory, use LayoutManager instead,
- unbind zookeeper from bookkeeperadmin.format,
- fix compile and test errors.
Master Issue: #897
Author: Jia Zhai <zh...@apache.org>
Reviewers: Enrico Olivelli <eo...@gmail.com>, Sijie Guo <si...@apache.org>
This closes #936 from zhaijack/issue_897, closes #897
---
.../java/org/apache/bookkeeper/bookie/Bookie.java | 15 +-
.../org/apache/bookkeeper/bookie/BookieShell.java | 184 ++++++++-------------
.../org/apache/bookkeeper/client/BookKeeper.java | 19 ++-
.../apache/bookkeeper/client/BookKeeperAdmin.java | 126 ++------------
.../bookkeeper/client/api/BookKeeperBuilder.java | 10 --
.../client/impl/BookKeeperBuilderImpl.java | 7 -
.../bookkeeper/discover/RegistrationClient.java | 10 +-
.../bookkeeper/discover/RegistrationManager.java | 43 +++++
.../bookkeeper/discover/ZKRegistrationClient.java | 27 ++-
.../bookkeeper/discover/ZKRegistrationManager.java | 117 +++++++++++++
.../bookkeeper/meta/FlatLedgerManagerFactory.java | 30 ++--
.../org/apache/bookkeeper/meta/LayoutManager.java | 69 ++++++++
.../org/apache/bookkeeper/meta/LedgerLayout.java | 162 +++---------------
.../bookkeeper/meta/LedgerManagerFactory.java | 65 ++++----
.../LegacyHierarchicalLedgerManagerFactory.java | 18 +-
.../bookkeeper/meta/MSLedgerManagerFactory.java | 21 ++-
.../apache/bookkeeper/meta/ZkLayoutManager.java | 107 ++++++++++++
.../org/apache/bookkeeper/replication/Auditor.java | 18 +-
.../bookkeeper/replication/ReplicationWorker.java | 17 +-
.../server/http/BKHttpServiceProvider.java | 11 +-
.../server/http/service/ExpandStorageService.java | 7 +-
.../server/http/service/GetLedgerMetaService.java | 12 +-
.../server/http/service/ListLedgerService.java | 11 +-
.../service/ListUnderReplicatedLedgerService.java | 12 +-
.../bookie/BookieInitializationTest.java | 15 +-
.../apache/bookkeeper/bookie/CompactionTest.java | 13 +-
.../org/apache/bookkeeper/bookie/CookieTest.java | 3 +-
.../apache/bookkeeper/bookie/LedgerCacheTest.java | 3 +-
.../bookkeeper/client/TestWatchEnsembleChange.java | 8 +-
.../discover/TestZkRegistrationClient.java | 73 ++++++++
.../discover/TestZkRegistrationManager.java | 73 ++++++++
.../bookkeeper/meta/LedgerManagerTestCase.java | 9 +-
.../apache/bookkeeper/meta/TestLedgerLayout.java | 66 ++++++++
.../apache/bookkeeper/meta/TestLedgerManager.java | 69 +++++---
.../bookkeeper/meta/TestZkLayoutManager.java | 79 +++++++++
...dgerLayoutTest.java => ZkLedgerLayoutTest.java} | 33 ++--
.../replication/AuditorLedgerCheckerTest.java | 10 +-
.../AuditorPeriodicBookieCheckTest.java | 8 +-
.../replication/AuditorPeriodicCheckTest.java | 16 +-
.../replication/AuditorRollingRestartTest.java | 6 +-
.../replication/BookieAutoRecoveryTest.java | 19 ++-
.../replication/BookieLedgerIndexTest.java | 10 +-
.../TestLedgerUnderreplicationManager.java | 25 ++-
.../replication/TestReplicationWorker.java | 25 ++-
.../bookkeeper/server/http/TestHttpService.java | 13 +-
.../tests/backward/TestBackwardCompat.java | 2 +-
46 files changed, 1100 insertions(+), 596 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
index 0b57d56..8eafa9f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
@@ -69,7 +69,6 @@ import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirExcepti
import org.apache.bookkeeper.common.util.Watcher;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.discover.RegistrationManager;
-import org.apache.bookkeeper.discover.ZKRegistrationManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.net.BookieSocketAddress;
@@ -92,7 +91,6 @@ import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -633,20 +631,18 @@ public class Bookie extends BookieCriticalThread {
this.registrationManager = instantiateRegistrationManager(conf);
checkEnvironment(this.registrationManager);
try {
- ZooKeeper zooKeeper = null; // ZooKeeper is null existing only for testing
if (registrationManager != null) {
- zooKeeper = ((ZKRegistrationManager) this.registrationManager).getZk();
// current the registration manager is zookeeper only
ledgerManagerFactory = LedgerManagerFactory.newLedgerManagerFactory(
conf,
- zooKeeper);
+ registrationManager.getLayoutManager());
LOG.info("instantiate ledger manager {}", ledgerManagerFactory.getClass().getName());
ledgerManager = ledgerManagerFactory.newLedgerManager();
} else {
ledgerManagerFactory = null;
ledgerManager = null;
}
- } catch (KeeperException e) {
+ } catch (IOException | InterruptedException e) {
throw new MetadataStoreException("Failed to initialize ledger manager", e);
}
stateManager = new BookieStateManager(conf, statsLogger, registrationManager, ledgerDirsManager);
@@ -1020,7 +1016,7 @@ public class Bookie extends BookieCriticalThread {
ledgerManager.close();
}
if (null != ledgerManagerFactory) {
- ledgerManagerFactory.uninitialize();
+ ledgerManagerFactory.close();
}
} catch (IOException ie) {
LOG.error("Failed to close active ledger manager : ", ie);
@@ -1274,6 +1270,11 @@ public class Bookie extends BookieCriticalThread {
return (BookieStateManager) this.stateManager;
}
+ @VisibleForTesting
+ public LedgerManagerFactory getLedgerManagerFactory() {
+ return ledgerManagerFactory;
+ }
+
// The rest of the code is test stuff
static class CounterCallback implements WriteCallback {
int count;
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 780852a..039132b 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
@@ -53,7 +53,6 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
@@ -257,9 +256,7 @@ public class BookieShell implements Tool {
boolean interactive = (!cmdLine.hasOption("n"));
boolean force = cmdLine.hasOption("f");
- ClientConfiguration adminConf = new ClientConfiguration(bkConf);
- boolean result = BookKeeperAdmin.format(adminConf, interactive,
- force);
+ boolean result = BookKeeperAdmin.format(bkConf, interactive, force);
return (result) ? 0 : 1;
}
}
@@ -686,21 +683,15 @@ public class BookieShell implements Tool {
predicate = replicasList -> !replicasList.contains(excludingBookieId);
}
- ZooKeeper zk = null;
- try {
- zk = ZooKeeperClient.newBuilder()
- .connectString(bkConf.getZkServers())
- .sessionTimeoutMs(bkConf.getZkTimeout())
- .build();
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(bkConf, zk);
- LedgerUnderreplicationManager underreplicationManager = mFactory.newLedgerUnderreplicationManager();
- Iterator<Long> iter = underreplicationManager.listLedgersToRereplicate(predicate);
- while (iter.hasNext()) {
- System.out.println(ledgerIdFormatter.formatLedgerId(iter.next()));
- }
- } finally {
- if (zk != null) {
- zk.close();
+ try (RegistrationManager rm = RegistrationManager.instantiateRegistrationManager(bkConf)) {
+ try (LedgerManagerFactory mFactory =
+ LedgerManagerFactory.newLedgerManagerFactory(bkConf, rm.getLayoutManager())) {
+ LedgerUnderreplicationManager underreplicationManager =
+ mFactory.newLedgerUnderreplicationManager();
+ Iterator<Long> iter = underreplicationManager.listLedgersToRereplicate(predicate);
+ while (iter.hasNext()) {
+ System.out.println(ledgerIdFormatter.formatLedgerId(iter.next()));
+ }
}
}
@@ -724,57 +715,40 @@ public class BookieShell implements Tool {
@Override
public int runCmd(CommandLine cmdLine) throws Exception {
- ZooKeeper zk = null;
- LedgerManagerFactory mFactory = null;
- LedgerManager m = null;
- try {
- zk = ZooKeeperClient.newBuilder()
- .connectString(bkConf.getZkServers())
- .sessionTimeoutMs(bkConf.getZkTimeout())
- .build();
- mFactory = LedgerManagerFactory.newLedgerManagerFactory(bkConf, zk);
- m = mFactory.newLedgerManager();
- LedgerRangeIterator iter = m.getLedgerRanges();
- if (cmdLine.hasOption("m")) {
- List<ReadMetadataCallback> futures = new ArrayList<ReadMetadataCallback>(LIST_BATCH_SIZE);
- while (iter.hasNext()) {
- LedgerRange r = iter.next();
- for (Long lid : r.getLedgers()) {
- ReadMetadataCallback cb = new ReadMetadataCallback(lid);
- m.readLedgerMetadata(lid, cb);
- futures.add(cb);
- }
- if (futures.size() >= LIST_BATCH_SIZE) {
- while (futures.size() > 0) {
- ReadMetadataCallback cb = futures.remove(0);
- printLedgerMetadata(cb);
+ try (LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ bkConf,
+ RegistrationManager.instantiateRegistrationManager(bkConf).getLayoutManager())) {
+ try (LedgerManager m = mFactory.newLedgerManager()) {
+ LedgerRangeIterator iter = m.getLedgerRanges();
+ if (cmdLine.hasOption("m")) {
+ List<ReadMetadataCallback> futures = new ArrayList<ReadMetadataCallback>(LIST_BATCH_SIZE);
+ while (iter.hasNext()) {
+ LedgerRange r = iter.next();
+ for (Long lid : r.getLedgers()) {
+ ReadMetadataCallback cb = new ReadMetadataCallback(lid);
+ m.readLedgerMetadata(lid, cb);
+ futures.add(cb);
+ }
+ if (futures.size() >= LIST_BATCH_SIZE) {
+ while (futures.size() > 0) {
+ ReadMetadataCallback cb = futures.remove(0);
+ printLedgerMetadata(cb);
+ }
}
}
- }
- while (futures.size() > 0) {
- ReadMetadataCallback cb = futures.remove(0);
- printLedgerMetadata(cb);
- }
- } else {
- while (iter.hasNext()) {
- LedgerRange r = iter.next();
- for (Long lid : r.getLedgers()) {
- System.out.println(ledgerIdFormatter.formatLedgerId(lid));
+ while (futures.size() > 0) {
+ ReadMetadataCallback cb = futures.remove(0);
+ printLedgerMetadata(cb);
+ }
+ } else {
+ while (iter.hasNext()) {
+ LedgerRange r = iter.next();
+ for (Long lid : r.getLedgers()) {
+ System.out.println(ledgerIdFormatter.formatLedgerId(lid));
+ }
}
}
}
- } finally {
- if (m != null) {
- try {
- m.close();
- mFactory.uninitialize();
- } catch (IOException ioe) {
- LOG.error("Failed to close ledger manager : ", ioe);
- }
- }
- if (zk != null) {
- zk.close();
- }
}
return 0;
@@ -842,31 +816,15 @@ public class BookieShell implements Tool {
return -1;
}
- ZooKeeper zk = null;
- LedgerManagerFactory mFactory = null;
- LedgerManager m = null;
- try {
- zk = ZooKeeperClient.newBuilder()
- .connectString(bkConf.getZkServers())
- .sessionTimeoutMs(bkConf.getZkTimeout())
- .build();
- mFactory = LedgerManagerFactory.newLedgerManagerFactory(bkConf, zk);
- m = mFactory.newLedgerManager();
- ReadMetadataCallback cb = new ReadMetadataCallback(lid);
- m.readLedgerMetadata(lid, cb);
- printLedgerMetadata(cb);
- } finally {
- if (m != null) {
- try {
- m.close();
- mFactory.uninitialize();
- } catch (IOException ioe) {
- LOG.error("Failed to close ledger manager : ", ioe);
+ try (RegistrationManager rm = RegistrationManager.instantiateRegistrationManager(bkConf)) {
+ try (LedgerManagerFactory mFactory =
+ LedgerManagerFactory.newLedgerManagerFactory(bkConf, rm.getLayoutManager())){
+ try (LedgerManager m = mFactory.newLedgerManager()) {
+ ReadMetadataCallback cb = new ReadMetadataCallback(lid);
+ m.readLedgerMetadata(lid, cb);
+ printLedgerMetadata(cb);
}
}
- if (zk != null) {
- zk.close();
- }
}
return 0;
@@ -1512,36 +1470,32 @@ public class BookieShell implements Tool {
printUsage();
return 1;
}
- ZooKeeper zk = null;
- try {
- zk = ZooKeeperClient.newBuilder()
- .connectString(bkConf.getZkServers())
- .sessionTimeoutMs(bkConf.getZkTimeout())
- .build();
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(bkConf, zk);
- LedgerUnderreplicationManager underreplicationManager = mFactory.newLedgerUnderreplicationManager();
- if (!enable && !disable) {
- boolean enabled = underreplicationManager.isLedgerReplicationEnabled();
- System.out.println("Autorecovery is " + (enabled ? "enabled." : "disabled."));
- } else if (enable) {
- if (underreplicationManager.isLedgerReplicationEnabled()) {
- LOG.warn("Autorecovery already enabled. Doing nothing");
- } else {
- LOG.info("Enabling autorecovery");
- underreplicationManager.enableLedgerReplication();
- }
- } else {
- if (!underreplicationManager.isLedgerReplicationEnabled()) {
- LOG.warn("Autorecovery already disabled. Doing nothing");
+
+ try (RegistrationManager rm =
+ RegistrationManager.instantiateRegistrationManager(bkConf)){
+ try (LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ bkConf,
+ rm.getLayoutManager())) {
+ LedgerUnderreplicationManager underreplicationManager = mFactory.newLedgerUnderreplicationManager();
+ if (!enable && !disable) {
+ boolean enabled = underreplicationManager.isLedgerReplicationEnabled();
+ System.out.println("Autorecovery is " + (enabled ? "enabled." : "disabled."));
+ } else if (enable) {
+ if (underreplicationManager.isLedgerReplicationEnabled()) {
+ LOG.warn("Autorecovery already enabled. Doing nothing");
+ } else {
+ LOG.info("Enabling autorecovery");
+ underreplicationManager.enableLedgerReplication();
+ }
} else {
- LOG.info("Disabling autorecovery");
- underreplicationManager.disableLedgerReplication();
+ if (!underreplicationManager.isLedgerReplicationEnabled()) {
+ LOG.warn("Autorecovery already disabled. Doing nothing");
+ } else {
+ LOG.info("Disabling autorecovery");
+ underreplicationManager.disableLedgerReplication();
+ }
}
}
- } finally {
- if (zk != null) {
- zk.close();
- }
}
return 0;
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 6dddfb1..a8c181b 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
@@ -30,7 +30,6 @@ import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.HashedWheelTimer;
import io.netty.util.concurrent.DefaultThreadFactory;
-
import java.io.IOException;
import java.util.EnumSet;
import java.util.List;
@@ -46,7 +45,6 @@ import org.apache.bookkeeper.client.AsyncCallback.CreateCallback;
import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback;
import org.apache.bookkeeper.client.AsyncCallback.IsClosedCallback;
import org.apache.bookkeeper.client.AsyncCallback.OpenCallback;
-import org.apache.bookkeeper.client.BKException.ZKException;
import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo;
import org.apache.bookkeeper.client.SyncCallbackUtils.SyncCreateAdvCallback;
import org.apache.bookkeeper.client.SyncCallbackUtils.SyncCreateCallback;
@@ -238,6 +236,7 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
* @return client builder.
* @since 4.5
*/
+ @Deprecated
public Builder zk(ZooKeeper zk) {
this.zk = zk;
return this;
@@ -444,6 +443,9 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
} catch (ConfigurationException ce) {
LOG.error("Failed to initialize registration client", ce);
throw new IOException("Failed to initialize registration client", ce);
+ } catch (BKException be) {
+ LOG.error("Failed to initialize registration client", be);
+ throw new IOException("Failed to initialize registration client", be);
}
// initialize event loop group
@@ -512,9 +514,9 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
// initialize ledger manager
try {
this.ledgerManagerFactory =
- LedgerManagerFactory.newLedgerManagerFactory(conf, ((ZKRegistrationClient) regClient).getZk());
- } catch (KeeperException ke) {
- throw new ZKException();
+ LedgerManagerFactory.newLedgerManagerFactory(conf, regClient.getLayoutManager());
+ } catch (IOException | InterruptedException e) {
+ throw e;
}
this.ledgerManager = new CleanupLedgerManager(ledgerManagerFactory.newLedgerManager());
this.ledgerIdGenerator = ledgerManagerFactory.newLedgerIdGenerator();
@@ -628,6 +630,11 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
return reorderReadSequence;
}
+ @VisibleForTesting
+ public RegistrationClient getRegClient() {
+ return regClient;
+ }
+
/**
* There are 2 digest types that can be used for verification. The CRC32 is
* cheap to compute but does not protect against byzantine bookies (i.e., a
@@ -1363,7 +1370,7 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
// which will reject any incoming metadata requests.
ledgerManager.close();
ledgerIdGenerator.close();
- ledgerManagerFactory.uninitialize();
+ ledgerManagerFactory.close();
} catch (IOException ie) {
LOG.error("Failed to close ledger manager : ", ie);
}
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 7d13100..2c5b9df 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
@@ -20,9 +20,7 @@
*/
package org.apache.bookkeeper.client;
-import static com.google.common.base.Charsets.UTF_8;
import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.bookkeeper.util.BookKeeperConstants.READONLY;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -44,7 +42,6 @@ import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.SortedMap;
-import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ExecutionException;
@@ -59,11 +56,11 @@ import org.apache.bookkeeper.client.SyncCallbackUtils.SyncReadCallback;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.discover.RegistrationClient.RegistrationListener;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManager.LedgerRangeIterator;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
-import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.MultiCallback;
@@ -75,16 +72,9 @@ import org.apache.bookkeeper.replication.ReplicationException.CompatibilityExcep
import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
-import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.bookkeeper.util.IOUtils;
-import org.apache.bookkeeper.util.ZkUtils;
-import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
import org.apache.zookeeper.AsyncCallback;
-import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZKUtil;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1128,122 +1118,40 @@ public class BookKeeperAdmin implements AutoCloseable {
* removed without prompt.
* @return Returns true if format succeeds else false.
*/
- public static boolean format(ClientConfiguration conf,
+ public static boolean format(ServerConfiguration conf,
boolean isInteractive, boolean force) throws Exception {
- ZooKeeper zkc = ZooKeeperClient.newBuilder()
- .connectString(conf.getZkServers())
- .sessionTimeoutMs(conf.getZkTimeout())
- .build();
- BookKeeper bkc = null;
- try {
- boolean ledgerRootExists = null != zkc.exists(
- conf.getZkLedgersRootPath(), false);
- boolean availableNodeExists = null != zkc.exists(
- conf.getZkAvailableBookiesPath(), false);
- List<ACL> zkAcls = ZkUtils.getACLs(conf);
- // Create ledgers root node if not exists
- if (!ledgerRootExists) {
- zkc.create(conf.getZkLedgersRootPath(), "".getBytes(UTF_8),
- zkAcls, CreateMode.PERSISTENT);
- }
- // create available bookies node if not exists
- if (!availableNodeExists) {
- zkc.create(conf.getZkAvailableBookiesPath(), "".getBytes(UTF_8),
- zkAcls, CreateMode.PERSISTENT);
- }
- // create readonly bookies node if not exists
- if (null == zkc.exists(conf.getZkAvailableBookiesPath() + "/" + READONLY, false)) {
- zkc.create(
- conf.getZkAvailableBookiesPath() + "/" + READONLY,
- new byte[0],
- zkAcls,
- CreateMode.PERSISTENT);
- }
+ try (RegistrationManager rm = RegistrationManager.instantiateRegistrationManager(conf)) {
+ boolean ledgerRootExists = rm.prepareFormat(conf);
// If old data was there then confirm with admin.
+ boolean doFormat = true;
if (ledgerRootExists) {
- boolean confirm = false;
if (!isInteractive) {
- // If non interactive and force is set, then delete old
- // data.
+ // If non interactive and force is set, then delete old data.
if (force) {
- confirm = true;
+ doFormat = true;
} else {
- confirm = false;
+ doFormat = false;
}
} else {
// Confirm with the admin.
- confirm = IOUtils
- .confirmPrompt("Ledger root already exists. "
- + "Are you sure to format bookkeeper metadata? "
- + "This may cause data loss.");
- }
- if (!confirm) {
- LOG.error("BookKeeper metadata Format aborted!!");
- return false;
- }
- }
- bkc = new BookKeeper(conf, zkc);
- // Format all ledger metadata layout
- bkc.ledgerManagerFactory.format(conf, zkc);
-
- // Clear underreplicated ledgers
- try {
- ZKUtil.deleteRecursive(zkc, ZkLedgerUnderreplicationManager.getBasePath(conf.getZkLedgersRootPath())
- + BookKeeperConstants.DEFAULT_ZK_LEDGERS_ROOT_PATH);
- } catch (KeeperException.NoNodeException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("underreplicated ledgers root path node not exists in zookeeper to delete");
- }
- }
-
- // Clear underreplicatedledger locks
- try {
- ZKUtil.deleteRecursive(zkc, ZkLedgerUnderreplicationManager.getBasePath(conf.getZkLedgersRootPath())
- + '/' + BookKeeperConstants.UNDER_REPLICATION_LOCK);
- } catch (KeeperException.NoNodeException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("underreplicatedledger locks node not exists in zookeeper to delete");
- }
- }
-
- // Clear the cookies
- try {
- ZKUtil.deleteRecursive(zkc, conf.getZkLedgersRootPath()
- + "/cookies");
- } catch (KeeperException.NoNodeException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("cookies node not exists in zookeeper to delete");
+ doFormat = IOUtils
+ .confirmPrompt("Ledger root already exists. "
+ + "Are you sure to format bookkeeper metadata? "
+ + "This may cause data loss.");
}
}
- // Clear the INSTANCEID
- try {
- zkc.delete(conf.getZkLedgersRootPath() + "/"
- + BookKeeperConstants.INSTANCEID, -1);
- } catch (KeeperException.NoNodeException e) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("INSTANCEID not exists in zookeeper to delete");
- }
+ if (!doFormat) {
+ return false;
}
- // create INSTANCEID
- String instanceId = UUID.randomUUID().toString();
- zkc.create(conf.getZkLedgersRootPath() + "/"
- + BookKeeperConstants.INSTANCEID, instanceId.getBytes(UTF_8),
- zkAcls, CreateMode.PERSISTENT);
+ BookKeeper bkc = new BookKeeper(new ClientConfiguration(conf));
+ bkc.ledgerManagerFactory.format(conf, bkc.regClient.getLayoutManager());
- LOG.info("Successfully formatted BookKeeper metadata");
- } finally {
- if (null != bkc) {
- bkc.close();
- }
- if (null != zkc) {
- zkc.close();
- }
+ return rm.format(conf);
}
- return true;
}
/**
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/BookKeeperBuilder.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/BookKeeperBuilder.java
index bc5fdb5..0e147dd 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/BookKeeperBuilder.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/BookKeeperBuilder.java
@@ -28,7 +28,6 @@ import org.apache.bookkeeper.common.annotation.InterfaceStability.Unstable;
import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.stats.StatsLogger;
-import org.apache.zookeeper.ZooKeeper;
/**
* BookKeeper Client Builder to build client instances.
@@ -49,15 +48,6 @@ public interface BookKeeperBuilder {
BookKeeperBuilder eventLoopGroup(EventLoopGroup eventLoopGroup);
/**
- * Configure the bookkeeper client with a provided ZooKeeper client.
- *
- * @param zk an external {@link ZooKeeper} client to use by the bookkeeper client.
- *
- * @return client builder.
- */
- BookKeeperBuilder zk(ZooKeeper zk);
-
- /**
* Configure the bookkeeper client with a provided {@link StatsLogger}.
*
* @param statsLogger an {@link StatsLogger} to use by the bookkeeper client to collect stats generated by the
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/BookKeeperBuilderImpl.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/BookKeeperBuilderImpl.java
index 7133c08..3a07d1b 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/BookKeeperBuilderImpl.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/BookKeeperBuilderImpl.java
@@ -30,7 +30,6 @@ import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.stats.StatsLogger;
-import org.apache.zookeeper.ZooKeeper;
/**
* Internal builder for {@link org.apache.bookkeeper.client.api.BookKeeper} client.
@@ -52,12 +51,6 @@ public class BookKeeperBuilderImpl implements BookKeeperBuilder {
}
@Override
- public BookKeeperBuilder zk(ZooKeeper zk) {
- builder.zk(zk);
- return this;
- }
-
- @Override
public BookKeeperBuilder statsLogger(StatsLogger statsLogger) {
builder.statsLogger(statsLogger);
return this;
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationClient.java
index 79dc300..b22311f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationClient.java
@@ -26,12 +26,12 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.common.annotation.InterfaceAudience.LimitedPrivate;
import org.apache.bookkeeper.common.annotation.InterfaceStability.Evolving;
import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.meta.LayoutManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.ZooKeeper;
-
/**
* A registration client, which the bookkeeper client will use to interact with registration service.
*/
@@ -116,4 +116,12 @@ public interface RegistrationClient extends AutoCloseable {
* @param listener listener to receive the topology changes of bookies.
*/
void unwatchReadOnlyBookies(RegistrationListener listener);
+
+ /**
+ * Gets layout manager.
+ *
+ * @return the layout manager
+ */
+ LayoutManager getLayoutManager();
+
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationManager.java
index 23b0139..d4a021b 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/RegistrationManager.java
@@ -22,9 +22,13 @@ import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.common.annotation.InterfaceAudience.LimitedPrivate;
import org.apache.bookkeeper.common.annotation.InterfaceStability.Evolving;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.meta.LayoutManager;
+import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.util.ReflectionUtils;
import org.apache.bookkeeper.versioning.Version;
import org.apache.bookkeeper.versioning.Versioned;
+import org.apache.commons.configuration.ConfigurationException;
/**
* Registration manager interface, which a bookie server will use to do the registration process.
@@ -34,6 +38,22 @@ import org.apache.bookkeeper.versioning.Versioned;
public interface RegistrationManager extends AutoCloseable {
/**
+ * Instantiate a RegistrationManager based on config.
+ */
+ static RegistrationManager instantiateRegistrationManager(ServerConfiguration conf) throws BookieException {
+ // Create the registration manager instance
+ Class<? extends RegistrationManager> managerCls;
+ try {
+ managerCls = conf.getRegistrationManagerClass();
+ } catch (ConfigurationException e) {
+ throw new BookieException.BookieIllegalOpException(e);
+ }
+
+ RegistrationManager manager = ReflectionUtils.newInstance(managerCls);
+ return manager.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
+ }
+
+ /**
* Registration Listener on listening the registration state.
*/
@FunctionalInterface
@@ -104,4 +124,27 @@ public interface RegistrationManager extends AutoCloseable {
* @throws BookieException when fail to remove cookie
*/
void removeCookie(String bookieId, Version version) throws BookieException;
+
+ /**
+ * Gets layout manager.
+ *
+ * @return the layout manager
+ */
+ LayoutManager getLayoutManager();
+
+ /**
+ * Prepare ledgers root node, availableNode, readonly node..
+ *
+ * @param conf the conf
+ * @return Returns true if old data exists, false if not.
+ */
+ boolean prepareFormat(ServerConfiguration conf) throws Exception;
+
+ /**
+ * Do format boolean.
+ *
+ * @param conf the conf
+ * @return Returns true if success do format, false if not.
+ */
+ boolean format(ServerConfiguration conf) throws Exception;
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
index be31520..36c38dd 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.discover;
import static org.apache.bookkeeper.util.BookKeeperConstants.READONLY;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.HashSet;
@@ -40,6 +41,8 @@ import org.apache.bookkeeper.client.BKException.ZKException;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.common.util.SafeRunnable;
import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.meta.LayoutManager;
+import org.apache.bookkeeper.meta.ZkLayoutManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.util.ZkUtils;
@@ -186,6 +189,10 @@ public class ZKRegistrationClient implements RegistrationClient {
private String bookieRegistrationPath;
private String bookieReadonlyRegistrationPath;
+ // layout manager
+ private List<ACL> acls;
+ private LayoutManager layoutManager;
+
@Override
public RegistrationClient initialize(ClientConfiguration conf,
ScheduledExecutorService scheduler,
@@ -198,6 +205,8 @@ public class ZKRegistrationClient implements RegistrationClient {
this.bookieRegistrationPath = conf.getZkAvailableBookiesPath();
this.bookieReadonlyRegistrationPath = this.bookieRegistrationPath + "/" + READONLY;
+ this.acls = ZkUtils.getACLs(conf);
+
// construct the zookeeper
if (zkOptional.isPresent()) {
this.zk = zkOptional.get();
@@ -214,10 +223,9 @@ public class ZKRegistrationClient implements RegistrationClient {
if (null == zk.exists(bookieReadonlyRegistrationPath, false)) {
try {
- List<ACL> zkAcls = ZkUtils.getACLs(conf);
zk.create(bookieReadonlyRegistrationPath,
new byte[0],
- zkAcls,
+ acls,
CreateMode.PERSISTENT);
} catch (KeeperException.NodeExistsException e) {
// this node is just now created by someone.
@@ -234,6 +242,12 @@ public class ZKRegistrationClient implements RegistrationClient {
this.ownZKHandle = true;
}
+ // layout manager
+ this.layoutManager = new ZkLayoutManager(
+ zk,
+ conf.getZkLedgersRootPath(),
+ acls);
+
return this;
}
@@ -353,5 +367,14 @@ public class ZKRegistrationClient implements RegistrationClient {
}
return newBookieAddrs;
}
+ @VisibleForTesting
+ public void setLayoutManager(LayoutManager layoutManager) {
+ this.layoutManager = layoutManager;
+ }
+
+ @Override
+ public LayoutManager getLayoutManager() {
+ return layoutManager;
+ }
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
index 172b696..724e1aa 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
@@ -25,10 +25,12 @@ import static org.apache.bookkeeper.util.BookKeeperConstants.INSTANCEID;
import static org.apache.bookkeeper.util.BookKeeperConstants.READONLY;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Charsets;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@@ -37,7 +39,11 @@ import org.apache.bookkeeper.bookie.BookieException.BookieIllegalOpException;
import org.apache.bookkeeper.bookie.BookieException.CookieNotFoundException;
import org.apache.bookkeeper.bookie.BookieException.MetadataStoreException;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.meta.LayoutManager;
+import org.apache.bookkeeper.meta.ZkLayoutManager;
+import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
@@ -52,6 +58,7 @@ import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZKUtil;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
@@ -72,6 +79,8 @@ public class ZKRegistrationManager implements RegistrationManager {
// registration paths
protected String bookieRegistrationPath;
protected String bookieReadonlyRegistrationPath;
+ // layout manager
+ private LayoutManager layoutManager;
private StatsLogger statsLogger;
@@ -98,6 +107,12 @@ public class ZKRegistrationManager implements RegistrationManager {
} catch (InterruptedException | KeeperException | IOException e) {
throw new MetadataStoreException(e);
}
+
+ this.layoutManager = new ZkLayoutManager(
+ zk,
+ conf.getZkLedgersRootPath(),
+ zkAcls);
+
return this;
}
@@ -394,4 +409,106 @@ public class ZKRegistrationManager implements RegistrationManager {
}
return instanceId;
}
+
+ @VisibleForTesting
+ public void setLayoutManager(LayoutManager layoutManager) {
+ this.layoutManager = layoutManager;
+ }
+
+ @Override
+ public LayoutManager getLayoutManager(){
+ return layoutManager;
+ }
+
+ @Override
+ public boolean prepareFormat(ServerConfiguration conf) throws Exception {
+ try (ZooKeeper zk = ZooKeeperClient.newBuilder()
+ .connectString(conf.getZkServers())
+ .sessionTimeoutMs(conf.getZkTimeout())
+ .build()) {
+ boolean ledgerRootExists = null != zk.exists(
+ conf.getZkLedgersRootPath(), false);
+ boolean availableNodeExists = null != zk.exists(
+ conf.getZkAvailableBookiesPath(), false);
+ List<ACL> zkAcls = ZkUtils.getACLs(conf);
+ // Create ledgers root node if not exists
+ if (!ledgerRootExists) {
+ zk.create(conf.getZkLedgersRootPath(), "".getBytes(Charsets.UTF_8),
+ zkAcls, CreateMode.PERSISTENT);
+ }
+ // create available bookies node if not exists
+ if (!availableNodeExists) {
+ zk.create(conf.getZkAvailableBookiesPath(), "".getBytes(Charsets.UTF_8),
+ zkAcls, CreateMode.PERSISTENT);
+ }
+
+ // create readonly bookies node if not exists
+ if (null == zk.exists(conf.getZkAvailableBookiesPath() + "/" + READONLY, false)) {
+ zk.create(
+ conf.getZkAvailableBookiesPath() + "/" + READONLY,
+ new byte[0],
+ zkAcls,
+ CreateMode.PERSISTENT);
+ }
+
+ return ledgerRootExists;
+ }
+ }
+
+ @Override
+ public boolean format(ServerConfiguration conf) throws Exception {
+ // Clear underreplicated ledgers
+ try (ZooKeeper zk = ZooKeeperClient.newBuilder()
+ .connectString(conf.getZkServers())
+ .sessionTimeoutMs(conf.getZkTimeout())
+ .build()) {
+ try {
+ ZKUtil.deleteRecursive(zk, ZkLedgerUnderreplicationManager.getBasePath(conf.getZkLedgersRootPath())
+ + BookKeeperConstants.DEFAULT_ZK_LEDGERS_ROOT_PATH);
+ } catch (KeeperException.NoNodeException e) {
+ if (log.isDebugEnabled()) {
+ log.debug("underreplicated ledgers root path node not exists in zookeeper to delete");
+ }
+ }
+
+ // Clear underreplicatedledger locks
+ try {
+ ZKUtil.deleteRecursive(zk, ZkLedgerUnderreplicationManager.getBasePath(conf.getZkLedgersRootPath())
+ + '/' + BookKeeperConstants.UNDER_REPLICATION_LOCK);
+ } catch (KeeperException.NoNodeException e) {
+ if (log.isDebugEnabled()) {
+ log.debug("underreplicatedledger locks node not exists in zookeeper to delete");
+ }
+ }
+
+ // Clear the cookies
+ try {
+ ZKUtil.deleteRecursive(zk, conf.getZkLedgersRootPath()
+ + "/cookies");
+ } catch (KeeperException.NoNodeException e) {
+ if (log.isDebugEnabled()) {
+ log.debug("cookies node not exists in zookeeper to delete");
+ }
+ }
+
+ // Clear the INSTANCEID
+ try {
+ zk.delete(conf.getZkLedgersRootPath() + "/"
+ + BookKeeperConstants.INSTANCEID, -1);
+ } catch (KeeperException.NoNodeException e) {
+ if (log.isDebugEnabled()) {
+ log.debug("INSTANCEID not exists in zookeeper to delete");
+ }
+ }
+
+ // create INSTANCEID
+ String instanceId = UUID.randomUUID().toString();
+ zk.create(conf.getZkLedgersRootPath() + "/"
+ + BookKeeperConstants.INSTANCEID, instanceId.getBytes(Charsets.UTF_8),
+ ZkUtils.getACLs(conf), CreateMode.PERSISTENT);
+
+ log.info("Successfully formatted BookKeeper metadata");
+ return true;
+ }
+ }
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java
index 1db220a..ad1c71d 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/FlatLedgerManagerFactory.java
@@ -17,9 +17,10 @@
*/
package org.apache.bookkeeper.meta;
+import static com.google.common.base.Preconditions.checkArgument;
+
import java.io.IOException;
import java.util.List;
-
import org.apache.bookkeeper.conf.AbstractConfiguration;
import org.apache.bookkeeper.replication.ReplicationException;
import org.apache.bookkeeper.util.ZkUtils;
@@ -46,20 +47,23 @@ public class FlatLedgerManagerFactory extends LedgerManagerFactory {
@Override
public LedgerManagerFactory initialize(final AbstractConfiguration conf,
- final ZooKeeper zk,
+ final LayoutManager layoutManager,
final int factoryVersion)
throws IOException {
+ checkArgument(layoutManager == null || layoutManager instanceof ZkLayoutManager);
+
if (CUR_VERSION != factoryVersion) {
throw new IOException("Incompatible layout version found : "
+ factoryVersion);
}
this.conf = conf;
- this.zk = zk;
+
+ this.zk = layoutManager == null ? null : ((ZkLayoutManager) layoutManager).getZk();
return this;
}
@Override
- public void uninitialize() throws IOException {
+ public void close() throws IOException {
// since zookeeper instance is passed from outside
// we don't need to close it here
}
@@ -82,19 +86,17 @@ public class FlatLedgerManagerFactory extends LedgerManagerFactory {
}
@Override
- public void format(AbstractConfiguration conf, ZooKeeper zk)
+ public void format(AbstractConfiguration conf, LayoutManager layoutManager)
throws InterruptedException, KeeperException, IOException {
- try (FlatLedgerManager ledgerManager = (FlatLedgerManager) newLedgerManager()) {
- String ledgersRootPath = conf.getZkLedgersRootPath();
- List<String> children = zk.getChildren(ledgersRootPath, false);
- for (String child : children) {
- if (FlatLedgerManager.isSpecialZnode(child)) {
- continue;
- }
- ZKUtil.deleteRecursive(zk, ledgersRootPath + "/" + child);
+ String ledgersRootPath = conf.getZkLedgersRootPath();
+ List<String> children = zk.getChildren(ledgersRootPath, false);
+ for (String child : children) {
+ if (FlatLedgerManager.isSpecialZnode(child)) {
+ continue;
}
+ ZKUtil.deleteRecursive(zk, ledgersRootPath + "/" + child);
}
// Delete and recreate the LAYOUT information.
- super.format(conf, zk);
+ super.format(conf, layoutManager);
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LayoutManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LayoutManager.java
new file mode 100644
index 0000000..e01c575
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LayoutManager.java
@@ -0,0 +1,69 @@
+/*
+ * 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.meta;
+
+import java.io.IOException;
+
+/**
+ * Layout manager for writing/reading/deleting ledger layout.
+ */
+public interface LayoutManager {
+
+ /**
+ * The Ledger layout exists exception.
+ */
+ class LedgerLayoutExistsException extends IOException {
+
+ public LedgerLayoutExistsException(String message) {
+ super(message);
+ }
+
+ public LedgerLayoutExistsException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public LedgerLayoutExistsException(Throwable cause) {
+ super(cause);
+ }
+ }
+
+ /**
+ * Return the ledger layout.
+ *
+ * @return the ledger layout
+ * @throws IOException when fail to read ledger layout.
+ */
+ LedgerLayout readLedgerLayout() throws IOException;
+
+ /**
+ * Store the ledger layout.
+ *
+ * @param layout ledger layout
+ * @throws IOException when fail to store ledger layout.
+ */
+ void storeLedgerLayout(LedgerLayout layout) throws IOException;
+
+ /**
+ * Delete ledger layout.
+ *
+ * @throws IOException
+ */
+ void deleteLedgerLayout() throws IOException;
+
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerLayout.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerLayout.java
index a1e2cef..eb5b705 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerLayout.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerLayout.java
@@ -1,5 +1,3 @@
-package org.apache.bookkeeper.meta;
-
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -17,70 +15,40 @@ package org.apache.bookkeeper.meta;
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.bookkeeper.meta;
import java.io.IOException;
-import java.util.List;
-
-import org.apache.bookkeeper.util.BookKeeperConstants;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.ACL;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
/**
* This class encapsulates ledger layout information that is persistently stored
- * in zookeeper. It provides parsing and serialization methods of such information.
- *
+ * in registration service. It provides parsing and serialization methods of such
+ * information.
*/
-class LedgerLayout {
- static final Logger LOG = LoggerFactory.getLogger(LedgerLayout.class);
-
+@Slf4j
+@Getter
+@EqualsAndHashCode
+@ToString
+public class LedgerLayout {
// version of compability layout version
public static final int LAYOUT_MIN_COMPAT_VERSION = 1;
// version of ledger layout metadata
public static final int LAYOUT_FORMAT_VERSION = 2;
- /**
- * Read ledger layout from zookeeper.
- *
- * @param zk ZooKeeper Client
- * @param ledgersRoot Root of the ledger namespace to check
- * @return ledger layout, or null if none set in zookeeper
- */
- public static LedgerLayout readLayout(final ZooKeeper zk, final String ledgersRoot)
- throws IOException, KeeperException {
- String ledgersLayout = ledgersRoot + "/" + BookKeeperConstants.LAYOUT_ZNODE;
-
- try {
- LedgerLayout layout;
-
- try {
- byte[] layoutData = zk.getData(ledgersLayout, false, null);
- layout = parseLayout(layoutData);
- } catch (KeeperException.NoNodeException nne) {
- return null;
- }
-
- return layout;
- } catch (InterruptedException ie) {
- throw new IOException(ie);
- }
- }
-
private static final String splitter = ":";
private static final String lSplitter = "\n";
// ledger manager factory class
- private String managerFactoryCls;
+ private final String managerFactoryClass;
// ledger manager version
- private int managerVersion;
+ private final int managerVersion;
// layout version of how to store layout information
- private int layoutFormatVersion = LAYOUT_FORMAT_VERSION;
+ private final int layoutFormatVersion;
/**
* Ledger Layout Constructor.
@@ -89,8 +57,6 @@ class LedgerLayout {
* Ledger Manager Factory Class
* @param managerVersion
* Ledger Manager Version
- * @param layoutFormatVersion
- * Ledger Layout Format Version
*/
public LedgerLayout(String managerFactoryCls, int managerVersion) {
this(managerFactoryCls, managerVersion, LAYOUT_FORMAT_VERSION);
@@ -98,78 +64,23 @@ class LedgerLayout {
LedgerLayout(String managerFactoryCls, int managerVersion,
int layoutVersion) {
- this.managerFactoryCls = managerFactoryCls;
+ this.managerFactoryClass = managerFactoryCls;
this.managerVersion = managerVersion;
this.layoutFormatVersion = layoutVersion;
}
/**
- * Get Ledger Manager Type.
- *
- * @return ledger manager type
- * @deprecated replaced by {@link #getManagerFactoryClass()}
- */
- @Deprecated
- public String getManagerType() {
- // pre V2 layout store as manager type
- return this.managerFactoryCls;
- }
-
- /**
- * Get ledger manager factory class.
- *
- * @return ledger manager factory class
- */
- public String getManagerFactoryClass() {
- return this.managerFactoryCls;
- }
-
- public int getManagerVersion() {
- return this.managerVersion;
- }
-
- /**
- * Return layout format version.
- *
- * @return layout format version
- */
- public int getLayoutFormatVersion() {
- return this.layoutFormatVersion;
- }
-
- /**
- * Store the ledger layout into zookeeper.
- */
- public void store(final ZooKeeper zk, String ledgersRoot, List<ACL> zkAcls)
- throws IOException, KeeperException, InterruptedException {
- String ledgersLayout = ledgersRoot + "/"
- + BookKeeperConstants.LAYOUT_ZNODE;
- zk.create(ledgersLayout, serialize(), zkAcls,
- CreateMode.PERSISTENT);
- }
-
- /**
- * Delete the LAYOUT from zookeeper.
- */
- public void delete(final ZooKeeper zk, String ledgersRoot)
- throws KeeperException, InterruptedException {
- String ledgersLayout = ledgersRoot + "/"
- + BookKeeperConstants.LAYOUT_ZNODE;
- zk.delete(ledgersLayout, -1);
- }
-
- /**
* Generates a byte array based on the LedgerLayout object.
*
* @return byte[]
*/
- private byte[] serialize() throws IOException {
+ public byte[] serialize() throws IOException {
String s =
new StringBuilder().append(layoutFormatVersion).append(lSplitter)
- .append(managerFactoryCls).append(splitter).append(managerVersion).toString();
+ .append(managerFactoryClass).append(splitter).append(managerVersion).toString();
- if (LOG.isDebugEnabled()) {
- LOG.debug("Serialized layout info: {}", s);
+ if (log.isDebugEnabled()) {
+ log.debug("Serialized layout info: {}", s);
}
return s.getBytes("UTF-8");
}
@@ -179,16 +90,14 @@ class LedgerLayout {
*
* @param bytes
* byte array to parse
- * @param znodeVersion
- * version of znode
* @return LedgerLayout
* @throws IOException
* if the given byte[] cannot be parsed
*/
- private static LedgerLayout parseLayout(byte[] bytes) throws IOException {
+ public static LedgerLayout parseLayout(byte[] bytes) throws IOException {
String layout = new String(bytes, "UTF-8");
- if (LOG.isDebugEnabled()) {
- LOG.debug("Parsing Layout: {}", layout);
+ if (log.isDebugEnabled()) {
+ log.debug("Parsing Layout: {}", layout);
}
String lines[] = layout.split(lSplitter);
@@ -218,29 +127,4 @@ class LedgerLayout {
}
}
- @Override
- public boolean equals(Object obj) {
- if (null == obj) {
- return false;
- }
- if (!(obj instanceof LedgerLayout)) {
- return false;
- }
- LedgerLayout other = (LedgerLayout) obj;
- return managerFactoryCls.equals(other.managerFactoryCls) && managerVersion == other.managerVersion;
- }
-
- @Override
- public int hashCode() {
- return (managerFactoryCls + managerVersion).hashCode();
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("LV").append(layoutFormatVersion).append(":")
- .append(",Type:").append(managerFactoryCls).append(":")
- .append(managerVersion);
- return sb.toString();
- }
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java
index 73db122..99df4cb 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java
@@ -18,23 +18,19 @@
package org.apache.bookkeeper.meta;
import java.io.IOException;
-import java.util.List;
-
import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.meta.LayoutManager.LedgerLayoutExistsException;
import org.apache.bookkeeper.replication.ReplicationException;
import org.apache.bookkeeper.util.ReflectionUtils;
-import org.apache.bookkeeper.util.ZkUtils;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* A factory for creating ledger managers.
*/
-public abstract class LedgerManagerFactory {
+public abstract class LedgerManagerFactory implements AutoCloseable {
static final Logger LOG = LoggerFactory.getLogger(LedgerManagerFactory.class);
// v1 layout
@@ -52,15 +48,15 @@ public abstract class LedgerManagerFactory {
*
* @param conf
* Configuration object used to initialize factory
- * @param zk
- * Available zookeeper handle for ledger manager to use.
+ * @param layoutManager
+ * Layout manager used for initialize ledger manager factory
* @param factoryVersion
* What version used to initialize factory.
* @return ledger manager factory instance
* @throws IOException when fail to initialize the factory.
*/
public abstract LedgerManagerFactory initialize(AbstractConfiguration conf,
- ZooKeeper zk,
+ LayoutManager layoutManager,
int factoryVersion)
throws IOException;
@@ -69,7 +65,9 @@ public abstract class LedgerManagerFactory {
*
* @throws IOException when fail to uninitialize the factory.
*/
- public abstract void uninitialize() throws IOException;
+ @Override
+ public void close() throws IOException {
+ }
/**
* Return the ledger id generator, which is used for global unique ledger id
@@ -103,14 +101,14 @@ public abstract class LedgerManagerFactory {
*
* @param conf
* Configuration Object.
- * @param zk
- * ZooKeeper Client Handle, talk to zk to know which ledger manager is used.
+ * @param layoutManager
+ * layout manager
* @return new ledger manager factory
* @throws IOException
*/
public static LedgerManagerFactory newLedgerManagerFactory(
- final AbstractConfiguration<?> conf, final ZooKeeper zk)
- throws IOException, KeeperException, InterruptedException {
+ final AbstractConfiguration<?> conf, LayoutManager layoutManager)
+ throws IOException, InterruptedException {
Class<? extends LedgerManagerFactory> factoryClass;
try {
factoryClass = conf.getLedgerManagerFactoryClass();
@@ -123,8 +121,8 @@ public abstract class LedgerManagerFactory {
throw new IOException("Empty Ledger Root Path.");
}
- // if zk is null, return the default ledger manager
- if (zk == null) {
+ // if layoutManager is null, return the default ledger manager
+ if (layoutManager == null) {
return new FlatLedgerManagerFactory()
.initialize(conf, null, FlatLedgerManagerFactory.CUR_VERSION);
}
@@ -133,11 +131,12 @@ public abstract class LedgerManagerFactory {
// check that the configured ledger manager is
// compatible with the existing layout
- LedgerLayout layout = LedgerLayout.readLayout(zk, ledgerRootPath);
+ LedgerLayout layout = layoutManager.readLedgerLayout();
+
if (layout == null) { // no existing layout
- lmFactory = createNewLMFactory(conf, zk, factoryClass);
+ lmFactory = createNewLMFactory(conf, layoutManager, factoryClass);
return lmFactory
- .initialize(conf, zk, lmFactory.getCurrentVersion());
+ .initialize(conf, layoutManager, lmFactory.getCurrentVersion());
}
if (LOG.isDebugEnabled()) {
LOG.debug("read ledger layout {}", layout);
@@ -162,7 +161,7 @@ public abstract class LedgerManagerFactory {
} else {
throw new IOException("Unknown ledger manager type: " + lmType);
}
- return lmFactory.initialize(conf, zk, layout.getManagerVersion());
+ return lmFactory.initialize(conf, layoutManager, layout.getManagerVersion());
}
// handle V2 layout case
@@ -188,7 +187,7 @@ public abstract class LedgerManagerFactory {
}
// instantiate a factory
lmFactory = ReflectionUtils.newInstance(factoryClass);
- return lmFactory.initialize(conf, zk, layout.getManagerVersion());
+ return lmFactory.initialize(conf, layoutManager, layout.getManagerVersion());
}
/**
@@ -196,11 +195,10 @@ public abstract class LedgerManagerFactory {
* LedgerManagerFactory instance.
*/
private static LedgerManagerFactory createNewLMFactory(
- final AbstractConfiguration conf, final ZooKeeper zk,
+ final AbstractConfiguration conf, final LayoutManager layoutManager,
Class<? extends LedgerManagerFactory> factoryClass)
- throws IOException, KeeperException, InterruptedException {
+ throws IOException, InterruptedException {
- String ledgerRootPath = conf.getZkLedgersRootPath();
LedgerManagerFactory lmFactory;
LedgerLayout layout;
// use default ledger manager factory if no one provided
@@ -228,11 +226,10 @@ public abstract class LedgerManagerFactory {
layout = new LedgerLayout(factoryClass.getName(),
lmFactory.getCurrentVersion());
- List<ACL> zkAcls = ZkUtils.getACLs(conf);
try {
- layout.store(zk, ledgerRootPath, zkAcls);
- } catch (KeeperException.NodeExistsException nee) {
- LedgerLayout layout2 = LedgerLayout.readLayout(zk, ledgerRootPath);
+ layoutManager.storeLedgerLayout(layout);
+ } catch (LedgerLayoutExistsException e) {
+ LedgerLayout layout2 = layoutManager.readLedgerLayout();
if (!layout2.equals(layout)) {
throw new IOException(
"Contention writing to layout to zookeeper, "
@@ -249,10 +246,10 @@ public abstract class LedgerManagerFactory {
*
* @param conf
* Configuration instance
- * @param zk
- * Zookeeper instance
+ * @param lm
+ * Layout manager
*/
- public void format(final AbstractConfiguration<?> conf, final ZooKeeper zk)
+ public void format(final AbstractConfiguration<?> conf, final LayoutManager lm)
throws InterruptedException, KeeperException, IOException {
Class<? extends LedgerManagerFactory> factoryClass;
@@ -262,10 +259,8 @@ public abstract class LedgerManagerFactory {
throw new IOException("Failed to get ledger manager factory class from configuration : ", e);
}
- LedgerLayout layout = LedgerLayout.readLayout(zk,
- conf.getZkLedgersRootPath());
- layout.delete(zk, conf.getZkLedgersRootPath());
+ lm.deleteLedgerLayout();
// Create new layout information again.
- createNewLMFactory(conf, zk, factoryClass);
+ createNewLMFactory(conf, lm, factoryClass);
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LegacyHierarchicalLedgerManagerFactory.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LegacyHierarchicalLedgerManagerFactory.java
index 1d82d95..70c1abd 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LegacyHierarchicalLedgerManagerFactory.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LegacyHierarchicalLedgerManagerFactory.java
@@ -17,6 +17,8 @@
*/
package org.apache.bookkeeper.meta;
+import static com.google.common.base.Preconditions.checkArgument;
+
import java.io.IOException;
import java.util.List;
@@ -46,20 +48,24 @@ public class LegacyHierarchicalLedgerManagerFactory extends LedgerManagerFactory
@Override
public LedgerManagerFactory initialize(final AbstractConfiguration conf,
- final ZooKeeper zk,
+ final LayoutManager lm,
final int factoryVersion)
- throws IOException {
+ throws IOException {
+ checkArgument(lm instanceof ZkLayoutManager);
+
+ ZkLayoutManager zkLayoutManager = (ZkLayoutManager) lm;
+
if (CUR_VERSION != factoryVersion) {
throw new IOException("Incompatible layout version found : "
+ factoryVersion);
}
this.conf = conf;
- this.zk = zk;
+ this.zk = zkLayoutManager.getZk();
return this;
}
@Override
- public void uninitialize() throws IOException {
+ public void close() throws IOException {
// since zookeeper instance is passed from outside
// we don't need to close it here
}
@@ -83,7 +89,7 @@ public class LegacyHierarchicalLedgerManagerFactory extends LedgerManagerFactory
}
@Override
- public void format(AbstractConfiguration conf, ZooKeeper zk)
+ public void format(AbstractConfiguration conf, LayoutManager layoutManager)
throws InterruptedException, KeeperException, IOException {
String ledgersRootPath = conf.getZkLedgersRootPath();
List<String> children = zk.getChildren(ledgersRootPath, false);
@@ -94,7 +100,7 @@ public class LegacyHierarchicalLedgerManagerFactory extends LedgerManagerFactory
ZKUtil.deleteRecursive(zk, ledgersRootPath + "/" + child);
}
// Delete and recreate the LAYOUT information.
- super.format(conf, zk);
+ super.format(conf, layoutManager);
}
}
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 47149d6..0e69385 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
@@ -17,12 +17,12 @@
*/
package org.apache.bookkeeper.meta;
+import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.bookkeeper.metastore.MetastoreTable.ALL_FIELDS;
import static org.apache.bookkeeper.metastore.MetastoreTable.NON_FIELDS;
import com.google.common.base.Optional;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
@@ -36,7 +36,6 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.LedgerMetadata;
import org.apache.bookkeeper.conf.AbstractConfiguration;
@@ -94,13 +93,17 @@ public class MSLedgerManagerFactory extends LedgerManagerFactory {
}
@Override
- public LedgerManagerFactory initialize(final AbstractConfiguration conf, final ZooKeeper zk,
- final int factoryVersion) throws IOException {
+ public LedgerManagerFactory initialize(final AbstractConfiguration conf,
+ final LayoutManager layoutManager,
+ final int factoryVersion) throws IOException {
+ checkArgument(layoutManager instanceof ZkLayoutManager);
+ ZkLayoutManager zkLayoutManager = (ZkLayoutManager) layoutManager;
+
if (CUR_VERSION != factoryVersion) {
throw new IOException("Incompatible layout version found : " + factoryVersion);
}
this.conf = conf;
- this.zk = zk;
+ this.zk = zkLayoutManager.getZk();
// load metadata store
String msName = conf.getMetastoreImplClass();
@@ -119,7 +122,7 @@ public class MSLedgerManagerFactory extends LedgerManagerFactory {
}
@Override
- public void uninitialize() throws IOException {
+ public void close() throws IOException {
metastore.close();
}
@@ -698,8 +701,8 @@ public class MSLedgerManagerFactory extends LedgerManagerFactory {
}
@Override
- public void format(AbstractConfiguration conf, ZooKeeper zk) throws InterruptedException,
- KeeperException, IOException {
+ public void format(AbstractConfiguration conf, LayoutManager layoutManager)
+ throws InterruptedException, KeeperException, IOException {
MetastoreTable ledgerTable;
try {
ledgerTable = metastore.createScannableTable(TABLE_NAME);
@@ -714,7 +717,7 @@ public class MSLedgerManagerFactory extends LedgerManagerFactory {
}
LOG.info("Finished cleaning up table {}.", TABLE_NAME);
// Delete and recreate the LAYOUT information.
- super.format(conf, zk);
+ super.format(conf, layoutManager);
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLayoutManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLayoutManager.java
new file mode 100644
index 0000000..aa61a3c
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLayoutManager.java
@@ -0,0 +1,107 @@
+/*
+ * 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.meta;
+
+import static org.apache.bookkeeper.util.BookKeeperConstants.LAYOUT_ZNODE;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import java.util.List;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.KeeperException.NodeExistsException;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+
+/**
+ * Provide utils for writing/reading/deleting layout in zookeeper.
+ */
+public class ZkLayoutManager implements LayoutManager {
+
+ private final ZooKeeper zk;
+ private final String ledgersLayout;
+ private final List<ACL> acls;
+
+ public ZkLayoutManager(ZooKeeper zk,
+ String ledgersRoot,
+ List<ACL> acls) {
+ this.zk = zk;
+ this.ledgersLayout = ledgersRoot + "/" + LAYOUT_ZNODE;
+ this.acls = acls;
+ }
+
+ @VisibleForTesting
+ public ZooKeeper getZk() {
+ return zk;
+ }
+
+ /**
+ * Read ledger layout from zookeeper.
+ *
+ * @return ledger layout, or null if none set in zookeeper
+ */
+ @Override
+ public LedgerLayout readLedgerLayout() throws IOException {
+ try {
+ byte[] layoutData = zk.getData(ledgersLayout, false, null);
+ return LedgerLayout.parseLayout(layoutData);
+ } catch (NoNodeException nne) {
+ return null;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ } catch (KeeperException e) {
+ throw new IOException(e);
+ }
+ }
+
+ /**
+ * Store the ledger layout to zookeeper.
+ */
+ @Override
+ public void storeLedgerLayout(LedgerLayout layout) throws IOException {
+ try {
+ zk.create(ledgersLayout, layout.serialize(), acls,
+ CreateMode.PERSISTENT);
+ } catch (NodeExistsException e) {
+ throw new LedgerLayoutExistsException(e);
+ } catch (KeeperException e) {
+ throw new IOException(e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+
+ /**
+ * Delete the ledger layout from zookeeper.
+ */
+ @Override
+ public void deleteLedgerLayout() throws IOException {
+ try {
+ zk.delete(ledgersLayout, -1);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ } catch (KeeperException e) {
+ throw new IOException(e);
+ }
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
index 778ce8a..453afdc 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/Auditor.java
@@ -24,7 +24,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.SettableFuture;
-
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
@@ -38,7 +37,6 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeperAdmin;
@@ -133,19 +131,21 @@ public class Auditor {
private void initialize(ServerConfiguration conf, ZooKeeper zkc)
throws UnavailableException {
try {
+ ClientConfiguration clientConfiguration = new ClientConfiguration(conf);
+ clientConfiguration.setClientRole(ClientConfiguration.CLIENT_ROLE_SYSTEM);
+ LOG.info("AuthProvider used by the Auditor is {}",
+ clientConfiguration.getClientAuthProviderFactoryClass());
+ this.bkc = new BookKeeper(clientConfiguration, zkc);
+
LedgerManagerFactory ledgerManagerFactory = LedgerManagerFactory
- .newLedgerManagerFactory(conf, zkc);
+ .newLedgerManagerFactory(
+ conf,
+ bkc.getRegClient().getLayoutManager());
ledgerManager = ledgerManagerFactory.newLedgerManager();
this.bookieLedgerIndexer = new BookieLedgerIndexer(ledgerManager);
this.ledgerUnderreplicationManager = ledgerManagerFactory
.newLedgerUnderreplicationManager();
-
- ClientConfiguration clientConfiguration = new ClientConfiguration(conf);
- clientConfiguration.setClientRole(ClientConfiguration.CLIENT_ROLE_SYSTEM);
- LOG.info("AuthProvider used by the Auditor is {}",
- clientConfiguration.getClientAuthProviderFactoryClass());
- this.bkc = new BookKeeper(clientConfiguration, zkc);
this.admin = new BookKeeperAdmin(bkc, statsLogger);
if (this.ledgerUnderreplicationManager
.initializeLostBookieRecoveryDelay(conf.getLostBookieRecoveryDelay())) {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java
index b3adc0a..014e73c 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java
@@ -25,7 +25,6 @@ import static org.apache.bookkeeper.replication.ReplicationStats.REPLICATE_EXCEP
import static org.apache.bookkeeper.replication.ReplicationStats.REREPLICATE_OP;
import com.google.common.base.Stopwatch;
-
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
@@ -37,7 +36,6 @@ import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-
import org.apache.bookkeeper.bookie.BookieThread;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BKException.BKNoSuchLedgerExistsException;
@@ -126,18 +124,19 @@ public class ReplicationWorker implements Runnable {
InterruptedException, IOException {
this.zkc = zkc;
this.conf = conf;
- LedgerManagerFactory mFactory = LedgerManagerFactory
- .newLedgerManagerFactory(this.conf, this.zkc);
- this.underreplicationManager = mFactory
- .newLedgerUnderreplicationManager();
try {
this.bkc = BookKeeper.forConfig(new ClientConfiguration(conf))
- .zk(zkc)
- .statsLogger(statsLogger.scope(BK_CLIENT_SCOPE))
- .build();
+ .statsLogger(statsLogger.scope(BK_CLIENT_SCOPE))
+ .build();
} catch (BKException e) {
throw new IOException("Failed to instantiate replication worker", e);
}
+ LedgerManagerFactory mFactory = LedgerManagerFactory
+ .newLedgerManagerFactory(
+ this.conf,
+ bkc.getRegClient().getLayoutManager());
+ this.underreplicationManager = mFactory
+ .newLedgerUnderreplicationManager();
this.admin = new BookKeeperAdmin(bkc, statsLogger);
this.ledgerChecker = new LedgerChecker(bkc);
this.workerThread = new BookieThread(this, "ReplicationWorker");
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/BKHttpServiceProvider.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/BKHttpServiceProvider.java
index 461c520..c6b85d2 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/BKHttpServiceProvider.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/BKHttpServiceProvider.java
@@ -33,6 +33,7 @@ import org.apache.bookkeeper.http.HttpServiceProvider;
import org.apache.bookkeeper.http.service.ErrorHttpService;
import org.apache.bookkeeper.http.service.HeartbeatService;
import org.apache.bookkeeper.http.service.HttpEndpointService;
+import org.apache.bookkeeper.meta.LayoutManager;
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.replication.Auditor;
import org.apache.bookkeeper.replication.AutoRecoveryMain;
@@ -67,6 +68,7 @@ public class BKHttpServiceProvider implements HttpServiceProvider {
private final BookieServer bookieServer;
private final AutoRecoveryMain autoRecovery;
private final ServerConfiguration serverConf;
+ private final LayoutManager layoutManager;
private final ZooKeeper zk;
private final BookKeeperAdmin bka;
private final ExecutorService executor;
@@ -79,6 +81,7 @@ public class BKHttpServiceProvider implements HttpServiceProvider {
this.bookieServer = bookieServer;
this.autoRecovery = autoRecovery;
this.serverConf = serverConf;
+ this.layoutManager = bookieServer.getBookie().getRegistrationManager().getLayoutManager();
this.zk = ZooKeeperClient.newBuilder()
.connectString(serverConf.getZkServers())
.sessionTimeoutMs(serverConf.getZkTimeout())
@@ -171,9 +174,9 @@ public class BKHttpServiceProvider implements HttpServiceProvider {
case DELETE_LEDGER:
return new DeleteLedgerService(configuration);
case LIST_LEDGER:
- return new ListLedgerService(configuration, zk);
+ return new ListLedgerService(configuration, bookieServer);
case GET_LEDGER_META:
- return new GetLedgerMetaService(configuration, zk);
+ return new GetLedgerMetaService(configuration, bookieServer);
case READ_LEDGER_ENTRY:
return new ReadLedgerEntryService(configuration, bka);
@@ -187,13 +190,13 @@ public class BKHttpServiceProvider implements HttpServiceProvider {
case LIST_DISK_FILE:
return new ListDiskFilesService(configuration);
case EXPAND_STORAGE:
- return new ExpandStorageService(configuration, zk);
+ return new ExpandStorageService(configuration);
// autorecovery
case RECOVERY_BOOKIE:
return new RecoveryBookieService(configuration, bka, executor);
case LIST_UNDER_REPLICATED_LEDGER:
- return new ListUnderReplicatedLedgerService(configuration, zk);
+ return new ListUnderReplicatedLedgerService(configuration, bookieServer);
case WHO_IS_AUDITOR:
return new WhoIsAuditorService(configuration, zk);
case TRIGGER_AUDIT:
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ExpandStorageService.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ExpandStorageService.java
index b79fb41..4e7dc16 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ExpandStorageService.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ExpandStorageService.java
@@ -21,11 +21,9 @@ package org.apache.bookkeeper.server.http.service;
import static com.google.common.base.Preconditions.checkNotNull;
import com.google.common.collect.Lists;
-
import java.io.File;
import java.util.Arrays;
import java.util.List;
-
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -36,7 +34,6 @@ import org.apache.bookkeeper.http.service.HttpEndpointService;
import org.apache.bookkeeper.http.service.HttpServiceRequest;
import org.apache.bookkeeper.http.service.HttpServiceResponse;
import org.apache.bookkeeper.stats.NullStatsLogger;
-import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,12 +48,10 @@ public class ExpandStorageService implements HttpEndpointService {
static final Logger LOG = LoggerFactory.getLogger(ExpandStorageService.class);
protected ServerConfiguration conf;
- private ZooKeeper zk;
- public ExpandStorageService(ServerConfiguration conf, ZooKeeper zk) {
+ public ExpandStorageService(ServerConfiguration conf) {
checkNotNull(conf);
this.conf = conf;
- this.zk = zk;
}
/*
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/GetLedgerMetaService.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/GetLedgerMetaService.java
index c92ad81..5a06faf 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/GetLedgerMetaService.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/GetLedgerMetaService.java
@@ -31,8 +31,8 @@ import org.apache.bookkeeper.http.service.HttpServiceRequest;
import org.apache.bookkeeper.http.service.HttpServiceResponse;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
+import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.util.JsonUtil;
-import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,12 +45,11 @@ public class GetLedgerMetaService implements HttpEndpointService {
static final Logger LOG = LoggerFactory.getLogger(GetLedgerMetaService.class);
protected ServerConfiguration conf;
- protected ZooKeeper zk;
-
- public GetLedgerMetaService(ServerConfiguration conf, ZooKeeper zk) {
+ protected BookieServer bookieServer;
+ public GetLedgerMetaService(ServerConfiguration conf, BookieServer bookieServer) {
checkNotNull(conf);
this.conf = conf;
- this.zk = zk;
+ this.bookieServer = bookieServer;
}
@Override
@@ -61,7 +60,7 @@ public class GetLedgerMetaService implements HttpEndpointService {
if (HttpServer.Method.GET == request.getMethod() && (params != null) && params.containsKey("ledger_id")) {
Long ledgerId = Long.parseLong(params.get("ledger_id"));
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(conf, zk);
+ LedgerManagerFactory mFactory = bookieServer.getBookie().getLedgerManagerFactory();
LedgerManager manager = mFactory.newLedgerManager();
// output <ledgerId: ledgerMetadata>
@@ -73,7 +72,6 @@ public class GetLedgerMetaService implements HttpEndpointService {
output.put(ledgerId.toString(), new String(md.serialize(), UTF_8));
manager.close();
- mFactory.uninitialize();
String jsonResponse = JsonUtil.toJson(output);
LOG.debug("output body:" + jsonResponse);
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 2062507..1023d6b 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
@@ -36,9 +36,9 @@ import org.apache.bookkeeper.http.service.HttpServiceRequest;
import org.apache.bookkeeper.http.service.HttpServiceResponse;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
+import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
import org.apache.bookkeeper.util.JsonUtil;
-import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,12 +53,12 @@ public class ListLedgerService implements HttpEndpointService {
static final Logger LOG = LoggerFactory.getLogger(ListLedgerService.class);
protected ServerConfiguration conf;
- protected ZooKeeper zk;
+ protected BookieServer bookieServer;
- public ListLedgerService(ServerConfiguration conf, ZooKeeper zk) {
+ public ListLedgerService(ServerConfiguration conf, BookieServer bookieServer) {
checkNotNull(conf);
this.conf = conf;
- this.zk = zk;
+ this.bookieServer = bookieServer;
}
// Number of LedgerMetadata contains in each page
@@ -109,7 +109,7 @@ public class ListLedgerService implements HttpEndpointService {
int pageIndex = (printMeta && params.containsKey("page"))
? Integer.parseInt(params.get("page")) : -1;
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(conf, zk);
+ LedgerManagerFactory mFactory = bookieServer.getBookie().getLedgerManagerFactory();
LedgerManager manager = mFactory.newLedgerManager();
LedgerManager.LedgerRangeIterator iter = manager.getLedgerRanges();
@@ -162,7 +162,6 @@ public class ListLedgerService implements HttpEndpointService {
}
manager.close();
- mFactory.uninitialize();
String jsonResponse = JsonUtil.toJson(output);
LOG.debug("output body:" + jsonResponse);
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ListUnderReplicatedLedgerService.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ListUnderReplicatedLedgerService.java
index a76d509..3a07ef6 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ListUnderReplicatedLedgerService.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/ListUnderReplicatedLedgerService.java
@@ -21,12 +21,10 @@ package org.apache.bookkeeper.server.http.service;
import static com.google.common.base.Preconditions.checkNotNull;
import com.google.common.collect.Lists;
-
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;
-
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.http.HttpServer;
import org.apache.bookkeeper.http.service.HttpEndpointService;
@@ -34,9 +32,9 @@ import org.apache.bookkeeper.http.service.HttpServiceRequest;
import org.apache.bookkeeper.http.service.HttpServiceResponse;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
+import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.util.JsonUtil;
import org.apache.commons.lang.StringUtils;
-import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,12 +49,12 @@ public class ListUnderReplicatedLedgerService implements HttpEndpointService {
static final Logger LOG = LoggerFactory.getLogger(ListUnderReplicatedLedgerService.class);
protected ServerConfiguration conf;
- protected ZooKeeper zk;
+ protected BookieServer bookieServer;
- public ListUnderReplicatedLedgerService(ServerConfiguration conf, ZooKeeper zk) {
+ public ListUnderReplicatedLedgerService(ServerConfiguration conf, BookieServer bookieServer) {
checkNotNull(conf);
this.conf = conf;
- this.zk = zk;
+ this.bookieServer = bookieServer;
}
/*
@@ -93,7 +91,7 @@ public class ListUnderReplicatedLedgerService implements HttpEndpointService {
try {
List<Long> outputLedgers = Lists.newArrayList();
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(conf, zk);
+ LedgerManagerFactory mFactory = bookieServer.getBookie().getLedgerManagerFactory();
LedgerUnderreplicationManager underreplicationManager = mFactory.newLedgerUnderreplicationManager();
Iterator<Long> iter = underreplicationManager.listLedgersToRereplicate(predicate);
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
index 2e42b05..412e20d 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
@@ -252,9 +252,9 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
final String bkRegPath = conf.getZkAvailableBookiesPath() + "/"
+ InetAddress.getLocalHost().getCanonicalHostName() + ":" + conf.getBookiePort();
+ conf.setZkServers(zkUtil.getZooKeeperConnectString());
MockBookie bWithFQDNHostname = new MockBookie(conf);
- conf.setZkServers(zkUtil.getZooKeeperConnectString());
rm.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
bWithFQDNHostname.registrationManager = rm;
@@ -273,9 +273,9 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
final String bkRegPath = conf.getZkAvailableBookiesPath() + "/"
+ (InetAddress.getLocalHost().getCanonicalHostName().split("\\.", 2)[0]) + ":" + conf.getBookiePort();
+ conf.setZkServers(zkUtil.getZooKeeperConnectString());
MockBookie bWithShortHostname = new MockBookie(conf);
- conf.setZkServers(zkUtil.getZooKeeperConnectString());
rm.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
bWithShortHostname.registrationManager = rm;
@@ -296,7 +296,7 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
conf.setJournalDirName(tmpDir.getPath())
.setLedgerDirNames(new String[] { tmpDir.getPath() })
- .setZkServers(null);
+ .setZkServers(zkUtil.getZooKeeperConnectString());
String bkRegPath = conf.getZkAvailableBookiesPath() + "/"
+ InetAddress.getLocalHost().getHostAddress() + ":"
@@ -304,7 +304,6 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
MockBookie b = new MockBookie(conf);
- conf.setZkServers(zkUtil.getZooKeeperConnectString());
rm.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
b.setRegistrationManager(rm);
b.testRegisterBookie(conf);
@@ -474,10 +473,10 @@ public class BookieInitializationTest extends BookKeeperClusterTestCase {
} catch (Exception e) {
// shouldn't be able to start
}
- ClientConfiguration clientConf = new ClientConfiguration();
- clientConf.setZkServers(zkUtil.getZooKeeperConnectString());
- clientConf.setZkLedgersRootPath(zkRoot);
- BookKeeperAdmin.format(clientConf, false, false);
+ ServerConfiguration adminConf = new ServerConfiguration();
+ adminConf.setZkServers(zkUtil.getZooKeeperConnectString());
+ adminConf.setZkLedgersRootPath(zkRoot);
+ BookKeeperAdmin.format(adminConf, false, false);
Bookie b = new Bookie(conf);
b.shutdown();
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 8fb7980..56c09fd 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
@@ -45,6 +45,7 @@ import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.LedgerMetadata;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.proto.BookieServer;
@@ -241,7 +242,11 @@ public abstract class CompactionTest extends BookKeeperClusterTestCase {
InterleavedLedgerStorage storage = new InterleavedLedgerStorage();
storage.initialize(
conf,
- LedgerManagerFactory.newLedgerManagerFactory(conf, zkc).newLedgerManager(),
+ LedgerManagerFactory
+ .newLedgerManagerFactory(
+ conf,
+ RegistrationManager.instantiateRegistrationManager(conf).getLayoutManager())
+ .newLedgerManager(),
dirManager,
dirManager,
null,
@@ -840,7 +845,11 @@ public abstract class CompactionTest extends BookKeeperClusterTestCase {
InterleavedLedgerStorage storage = new InterleavedLedgerStorage();
storage.initialize(
conf,
- LedgerManagerFactory.newLedgerManagerFactory(conf, zkc).newLedgerManager(),
+ LedgerManagerFactory
+ .newLedgerManagerFactory(
+ conf,
+ RegistrationManager.instantiateRegistrationManager(conf).getLayoutManager())
+ .newLedgerManager(),
dirManager,
dirManager,
null,
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java
index f14744f..5a518e1 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CookieTest.java
@@ -37,7 +37,6 @@ import java.util.List;
import java.util.Set;
import org.apache.bookkeeper.bookie.BookieException.InvalidCookieException;
import org.apache.bookkeeper.client.BookKeeperAdmin;
-import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
import org.apache.bookkeeper.discover.RegistrationManager;
@@ -505,7 +504,7 @@ public class CookieTest extends BookKeeperClusterTestCase {
*/
@Test
public void testVerifyCookieWithFormat() throws Exception {
- ClientConfiguration adminConf = new ClientConfiguration();
+ ServerConfiguration adminConf = new ServerConfiguration();
adminConf.setZkServers(zkUtil.getZooKeeperConnectString());
adminConf.setProperty("bookkeeper.format", true);
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerCacheTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerCacheTest.java
index 72e7d27..6ae9410 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerCacheTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerCacheTest.java
@@ -100,7 +100,7 @@ public class LedgerCacheTest {
flushThread.join();
}
bookie.ledgerStorage.shutdown();
- ledgerManagerFactory.uninitialize();
+ ledgerManagerFactory.close();
FileUtils.deleteDirectory(txnDir);
FileUtils.deleteDirectory(ledgerDir);
for (File dir : tempDirs) {
@@ -530,7 +530,6 @@ public class LedgerCacheTest {
ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
conf.setGcWaitTime(gcWaitTime);
conf.setLedgerDirNames(new String[] { tmpDir.toString() });
- conf.setJournalDirName(tmpDir.toString());
conf.setLedgerStorageClass(FlushTestSortedLedgerStorage.class.getName());
Bookie bookie = new Bookie(conf);
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 7bab6f6..f79c27e 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
@@ -29,8 +29,8 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-
import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.FlatLedgerManagerFactory;
import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerIdGenerator;
@@ -112,7 +112,11 @@ public class TestWatchEnsembleChange extends BookKeeperClusterTestCase {
@Test
public void testWatchMetadataRemoval() throws Exception {
LedgerManagerFactory factory = ReflectionUtils.newInstance(lmFactoryCls);
- factory.initialize(baseConf, super.zkc, factory.getCurrentVersion());
+ baseConf.setZkServers(zkUtil.getZooKeeperConnectString());
+ factory.initialize(baseConf,
+ RegistrationManager.instantiateRegistrationManager(baseConf).getLayoutManager(),
+ factory.getCurrentVersion());
+
final LedgerManager manager = factory.newLedgerManager();
LedgerIdGenerator idGenerator = factory.newLedgerIdGenerator();
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClient.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClient.java
new file mode 100644
index 0000000..385d802
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationClient.java
@@ -0,0 +1,73 @@
+/*
+ * 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.discover;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.bookkeeper.meta.LayoutManager;
+import org.apache.bookkeeper.meta.LedgerLayout;
+import org.junit.Test;
+
+/**
+ * Unit test of {@link RegistrationClient}.
+ */
+public class TestZkRegistrationClient {
+ private final LedgerLayout ledgerLayout;
+ private final LayoutManager layoutManager;
+ private final ZKRegistrationClient zkRegistrationClient;
+
+ public TestZkRegistrationClient() {
+ this.ledgerLayout = mock(LedgerLayout.class);
+ this.layoutManager = mock(LayoutManager.class);
+ this.zkRegistrationClient = new ZKRegistrationClient();
+ zkRegistrationClient.setLayoutManager(layoutManager);
+ }
+
+ @Test
+ public void testGetLayoutManager() throws Exception {
+ assertEquals(layoutManager, zkRegistrationClient.getLayoutManager());
+ }
+
+ @Test
+ public void testReadLedgerLayout() throws Exception {
+ when(layoutManager.readLedgerLayout()).thenReturn(ledgerLayout);
+ assertEquals(ledgerLayout, zkRegistrationClient.getLayoutManager().readLedgerLayout());
+ }
+
+ @Test
+ public void testStoreLedgerLayout() throws Exception {
+ zkRegistrationClient.getLayoutManager().storeLedgerLayout(ledgerLayout);
+
+ verify(layoutManager, times(1))
+ .storeLedgerLayout(eq(ledgerLayout));
+ }
+
+ @Test
+ public void testDeleteLedgerLayout() throws Exception {
+ zkRegistrationClient.getLayoutManager().deleteLedgerLayout();
+
+ verify(layoutManager, times(1))
+ .deleteLedgerLayout();
+ }
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationManager.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationManager.java
new file mode 100644
index 0000000..284e4e9
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/discover/TestZkRegistrationManager.java
@@ -0,0 +1,73 @@
+/*
+ * 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.discover;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.bookkeeper.meta.LayoutManager;
+import org.apache.bookkeeper.meta.LedgerLayout;
+import org.junit.Test;
+
+/**
+ * Unit test of {@link RegistrationManager}.
+ */
+public class TestZkRegistrationManager {
+ private final LedgerLayout ledgerLayout;
+ private final LayoutManager layoutManager;
+ private final ZKRegistrationManager zkRegistrationManager;
+
+ public TestZkRegistrationManager() {
+ this.ledgerLayout = mock(LedgerLayout.class);
+ this.layoutManager = mock(LayoutManager.class);
+ this.zkRegistrationManager = new ZKRegistrationManager();
+ zkRegistrationManager.setLayoutManager(layoutManager);
+ }
+
+ @Test
+ public void testGetLayoutManager() throws Exception {
+ assertEquals(layoutManager, zkRegistrationManager.getLayoutManager());
+ }
+
+ @Test
+ public void testReadLedgerLayout() throws Exception {
+ when(layoutManager.readLedgerLayout()).thenReturn(ledgerLayout);
+ assertEquals(ledgerLayout, zkRegistrationManager.getLayoutManager().readLedgerLayout());
+ }
+
+ @Test
+ public void testStoreLedgerLayout() throws Exception {
+ zkRegistrationManager.getLayoutManager().storeLedgerLayout(ledgerLayout);
+
+ verify(layoutManager, times(1))
+ .storeLedgerLayout(eq(ledgerLayout));
+ }
+
+ @Test
+ public void testDeleteLedgerLayout() throws Exception {
+ zkRegistrationManager.getLayoutManager().deleteLedgerLayout();
+
+ verify(layoutManager, times(1))
+ .deleteLedgerLayout();
+ }
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java
index 32be5fc..4148575 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerManagerTestCase.java
@@ -39,6 +39,7 @@ import org.apache.bookkeeper.bookie.LedgerDirsManager;
import org.apache.bookkeeper.bookie.StateManager;
import org.apache.bookkeeper.common.util.Watcher;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.bookkeeper.util.SnapshotMap;
@@ -103,7 +104,11 @@ public abstract class LedgerManagerTestCase extends BookKeeperClusterTestCase {
@Override
public void setUp() throws Exception {
super.setUp();
- ledgerManagerFactory = LedgerManagerFactory.newLedgerManagerFactory(baseConf, zkc);
+ baseConf.setZkServers(zkUtil.getZooKeeperConnectString());
+ ledgerManagerFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ baseConf,
+ RegistrationManager
+ .instantiateRegistrationManager(baseConf).getLayoutManager());
}
@After
@@ -112,7 +117,7 @@ public abstract class LedgerManagerTestCase extends BookKeeperClusterTestCase {
if (null != ledgerManager) {
ledgerManager.close();
}
- ledgerManagerFactory.uninitialize();
+ ledgerManagerFactory.close();
super.tearDown();
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerLayout.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerLayout.java
new file mode 100644
index 0000000..655713e
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerLayout.java
@@ -0,0 +1,66 @@
+/*
+ * 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.meta;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+import org.junit.Test;
+
+/**
+ * Unit test of {@link LedgerLayout} class itself.
+ */
+public class TestLedgerLayout {
+
+ private static final LedgerLayout hierarchical1 =
+ new LedgerLayout(
+ HierarchicalLedgerManagerFactory.class.getName(),
+ 1);
+
+ private static final LedgerLayout hierarchical2 =
+ new LedgerLayout(
+ HierarchicalLedgerManagerFactory.class.getName(),
+ 2);
+
+ private static final LedgerLayout flat1 =
+ new LedgerLayout(
+ FlatLedgerManagerFactory.class.getName(),
+ 1);
+
+ @Test
+ public void testEquals() {
+ assertEquals(hierarchical1, hierarchical1);
+ assertNotEquals(hierarchical1, hierarchical2);
+ assertNotEquals(hierarchical1, flat1);
+ }
+
+ @Test
+ public void testGetters() {
+ assertEquals(
+ HierarchicalLedgerManagerFactory.class.getName(),
+ hierarchical1.getManagerFactoryClass());
+ assertEquals(
+ 1,
+ hierarchical1.getManagerVersion());
+ assertEquals(
+ LedgerLayout.LAYOUT_FORMAT_VERSION,
+ hierarchical1.getLayoutFormatVersion());
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java
index 8a44c73..744b9af 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestLedgerManager.java
@@ -28,9 +28,11 @@ import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
-
import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
@@ -60,7 +62,8 @@ public class TestLedgerManager extends BookKeeperClusterTestCase {
f.setAccessible(true);
f.set(layout, layoutVersion);
- layout.store(zkc, ledgersRootPath, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(zkc, ledgersRootPath, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ zkLayoutManager.storeLedgerLayout(layout);
}
/**
@@ -76,15 +79,22 @@ public class TestLedgerManager extends BookKeeperClusterTestCase {
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
conf.setZkLedgersRootPath(root0);
- LedgerManagerFactory m = LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(
+ zkc,
+ conf.getZkLedgersRootPath(),
+ ZkUtils.getACLs(conf));
+
+ LedgerManagerFactory m = LedgerManagerFactory.newLedgerManagerFactory(
+ conf,
+ zkLayoutManager);
assertTrue("Ledger manager is unexpected type",
(m instanceof FlatLedgerManagerFactory));
- m.uninitialize();
+ m.close();
// mismatching conf
conf.setLedgerManagerFactoryClass(HierarchicalLedgerManagerFactory.class);
try {
- LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
+ LedgerManagerFactory.newLedgerManagerFactory(conf, zkLayoutManager);
fail("Shouldn't reach here");
} catch (Exception e) {
LOG.error("Received exception", e);
@@ -100,7 +110,7 @@ public class TestLedgerManager extends BookKeeperClusterTestCase {
conf.setLedgerManagerFactoryClassName("DoesNotExist");
try {
- LedgerManagerFactory f = LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
+ LedgerManagerFactory f = LedgerManagerFactory.newLedgerManagerFactory(conf, zkLayoutManager);
fail("Shouldn't reach here");
} catch (Exception e) {
LOG.error("Received exception", e);
@@ -127,22 +137,30 @@ public class TestLedgerManager extends BookKeeperClusterTestCase {
conf.setLedgerManagerFactoryClass(FlatLedgerManagerFactory.class);
- LedgerManagerFactory m = LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(
+ zkc,
+ conf.getZkLedgersRootPath(),
+ ZkUtils.getACLs(conf));
+
+ LedgerManagerFactory m = LedgerManagerFactory.newLedgerManagerFactory(
+ conf,
+ zkLayoutManager);
+
assertTrue("Ledger manager is unexpected type",
(m instanceof FlatLedgerManagerFactory));
- m.uninitialize();
+ m.close();
// v2 setting doesn't effect v1
conf.setLedgerManagerFactoryClass(HierarchicalLedgerManagerFactory.class);
- m = LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
+ m = LedgerManagerFactory.newLedgerManagerFactory(conf, zkLayoutManager);
assertTrue("Ledger manager is unexpected type",
(m instanceof FlatLedgerManagerFactory));
- m.uninitialize();
+ m.close();
// mismatching conf
conf.setLedgerManagerType(HierarchicalLedgerManagerFactory.NAME);
try {
- LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
+ LedgerManagerFactory.newLedgerManagerFactory(conf, zkLayoutManager);
fail("Shouldn't reach here");
} catch (Exception e) {
LOG.error("Received exception", e);
@@ -164,11 +182,14 @@ public class TestLedgerManager extends BookKeeperClusterTestCase {
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
conf.setZkLedgersRootPath(root0);
- new LedgerLayout("DoesNotExist",
- 0xdeadbeef).store(zkc, root0, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ LedgerLayout layout = new LedgerLayout("DoesNotExist",
+ 0xdeadbeef);
+
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(zkc, root0, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ zkLayoutManager.storeLedgerLayout(layout);
try {
- LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
+ LedgerManagerFactory.newLedgerManagerFactory(conf, zkLayoutManager);
fail("Shouldn't reach here");
} catch (Exception e) {
LOG.error("Received exception", e);
@@ -182,11 +203,13 @@ public class TestLedgerManager extends BookKeeperClusterTestCase {
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
conf.setZkLedgersRootPath(root1);
- new LedgerLayout(FlatLedgerManagerFactory.class.getName(),
- 0xdeadbeef).store(zkc, root1, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ LedgerLayout layout1 = new LedgerLayout(FlatLedgerManagerFactory.class.getName(),
+ 0xdeadbeef);
+ ZkLayoutManager zkLayoutManager1 = new ZkLayoutManager(zkc, root1, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ zkLayoutManager1.storeLedgerLayout(layout1);
try {
- LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
+ LedgerManagerFactory.newLedgerManagerFactory(conf, zkLayoutManager1);
fail("Shouldn't reach here");
} catch (Exception e) {
LOG.error("Received exception", e);
@@ -201,6 +224,7 @@ public class TestLedgerManager extends BookKeeperClusterTestCase {
private final String root;
private final CyclicBarrier barrier;
private ZooKeeper zkc;
+ private ClientConfiguration conf;
CreateLMThread(String zkConnectString, String root,
String factoryCls, CyclicBarrier barrier) throws Exception {
@@ -210,17 +234,20 @@ public class TestLedgerManager extends BookKeeperClusterTestCase {
zkc = ZooKeeperClient.newBuilder()
.connectString(zkConnectString)
.build();
+ this.conf = new ClientConfiguration();
+ conf.setZkServers(zkConnectString);
}
public void run() {
- ClientConfiguration conf = new ClientConfiguration();
conf.setLedgerManagerFactoryClassName(factoryCls);
try {
barrier.await();
- LedgerManagerFactory factory =
- LedgerManagerFactory.newLedgerManagerFactory(conf, zkc);
- factory.uninitialize();
+ LedgerManagerFactory factory = LedgerManagerFactory.newLedgerManagerFactory(
+ conf,
+ RegistrationManager
+ .instantiateRegistrationManager(new ServerConfiguration(conf)).getLayoutManager());
+ factory.close();
success = true;
} catch (Exception e) {
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestZkLayoutManager.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestZkLayoutManager.java
new file mode 100644
index 0000000..ab09f61
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/TestZkLayoutManager.java
@@ -0,0 +1,79 @@
+/*
+ * 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.meta;
+
+import static org.apache.bookkeeper.util.BookKeeperConstants.LAYOUT_ZNODE;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.junit.Test;
+
+/**
+ * Unit test of {@link ZkLayoutManager}.
+ */
+public class TestZkLayoutManager {
+
+ private static final String ledgersRootPath = "/path/to/ledgers/root";
+ private static final String layoutPath = ledgersRootPath + "/" + LAYOUT_ZNODE;
+ private static final int managerVersion = 78;
+
+ private final ZooKeeper zk;
+ private final LedgerLayout layout;
+ private final ZkLayoutManager zkLayoutManager;
+
+ public TestZkLayoutManager() {
+ this.zk = mock(ZooKeeper.class);
+ this.layout = new LedgerLayout(
+ HierarchicalLedgerManagerFactory.class.getName(),
+ managerVersion);
+ this.zkLayoutManager = new ZkLayoutManager(zk, ledgersRootPath, Ids.OPEN_ACL_UNSAFE);
+ }
+
+ @Test
+ public void testReadLayout() throws Exception {
+ when(zk.getData(eq(layoutPath), eq(false), eq(null)))
+ .thenReturn(layout.serialize());
+
+ assertEquals(layout, zkLayoutManager.readLedgerLayout());
+ }
+
+ @Test
+ public void testStoreLayout() throws Exception {
+ zkLayoutManager.storeLedgerLayout(layout);
+
+ verify(zk, times(1))
+ .create(eq(layoutPath), eq(layout.serialize()), eq(Ids.OPEN_ACL_UNSAFE), eq(CreateMode.PERSISTENT));
+ }
+
+ @Test
+ public void testDeleteLayout() throws Exception {
+ zkLayoutManager.deleteLedgerLayout();
+
+ verify(zk, times(1))
+ .delete(eq(layoutPath), eq(-1));
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerLayoutTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/ZkLedgerLayoutTest.java
similarity index 81%
rename from bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerLayoutTest.java
rename to bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/ZkLedgerLayoutTest.java
index 75c26d7..cbf89ac 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerLayoutTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/ZkLedgerLayoutTest.java
@@ -28,21 +28,19 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.lang.reflect.Field;
-
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooDefs.Ids;
import org.junit.Test;
/**
- * Test the ledger layout.
+ * Test store/read/delete ledger layout operations on zookeeper.
*/
-public class LedgerLayoutTest extends BookKeeperClusterTestCase {
+public class ZkLedgerLayoutTest extends BookKeeperClusterTestCase {
- public LedgerLayoutTest() {
+ public ZkLedgerLayoutTest() {
super(0);
}
@@ -51,20 +49,20 @@ public class LedgerLayoutTest extends BookKeeperClusterTestCase {
ClientConfiguration conf = new ClientConfiguration();
conf.setLedgerManagerFactoryClass(HierarchicalLedgerManagerFactory.class);
String ledgerRootPath = "/testLedgerLayout";
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(zkc, ledgerRootPath, Ids.OPEN_ACL_UNSAFE);
zkc.create(ledgerRootPath, new byte[0],
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
- LedgerLayout layout = LedgerLayout.readLayout(zkc, ledgerRootPath);
- assertTrue("Layout should be null", layout == null);
+ assertEquals(null, zkLayoutManager.readLedgerLayout());
String testName = "foobar";
int testVersion = 0xdeadbeef;
// use layout defined in configuration also create it in zookeeper
LedgerLayout layout2 = new LedgerLayout(testName, testVersion);
- layout2.store(zkc, ledgerRootPath, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ zkLayoutManager.storeLedgerLayout(layout2);
- layout = LedgerLayout.readLayout(zkc, ledgerRootPath);
+ LedgerLayout layout = zkLayoutManager.readLedgerLayout();
assertEquals(testName, layout.getManagerFactoryClass());
assertEquals(testVersion, layout.getManagerVersion());
}
@@ -79,8 +77,9 @@ public class LedgerLayoutTest extends BookKeeperClusterTestCase {
Field f = LedgerLayout.class.getDeclaredField("layoutFormatVersion");
f.setAccessible(true);
f.set(layout, layoutVersion);
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(zkc, ledgersRootPath, Ids.OPEN_ACL_UNSAFE);
- layout.store(zkc, ledgersRootPath, ZooDefs.Ids.OPEN_ACL_UNSAFE);
+ zkLayoutManager.storeLedgerLayout(layout);
}
@Test
@@ -92,8 +91,10 @@ public class LedgerLayoutTest extends BookKeeperClusterTestCase {
FlatLedgerManagerFactory.CUR_VERSION,
LedgerLayout.LAYOUT_FORMAT_VERSION + 1);
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(zkc, conf.getZkLedgersRootPath(), Ids.OPEN_ACL_UNSAFE);
+
try {
- LedgerLayout.readLayout(zkc, conf.getZkLedgersRootPath());
+ zkLayoutManager.readLedgerLayout();
fail("Shouldn't reach here!");
} catch (IOException ie) {
assertTrue("Invalid exception", ie.getMessage().contains("version not compatible"));
@@ -110,9 +111,10 @@ public class LedgerLayoutTest extends BookKeeperClusterTestCase {
sb.append(LedgerLayout.LAYOUT_FORMAT_VERSION).append("\n");
zkc.create(ledgersLayout, sb.toString().getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(zkc, conf.getZkLedgersRootPath(), Ids.OPEN_ACL_UNSAFE);
try {
- LedgerLayout.readLayout(zkc, conf.getZkLedgersRootPath());
+ zkLayoutManager.readLedgerLayout();
fail("Shouldn't reach here!");
} catch (IOException ie) {
assertTrue("Invalid exception", ie.getMessage().contains("version absent from"));
@@ -131,9 +133,10 @@ public class LedgerLayoutTest extends BookKeeperClusterTestCase {
.append(FlatLedgerManagerFactory.class.getName());
zkc.create(ledgersLayout, sb.toString().getBytes(),
Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(zkc, rootPath, Ids.OPEN_ACL_UNSAFE);
try {
- LedgerLayout.readLayout(zkc, rootPath);
+ zkLayoutManager.readLedgerLayout();
fail("Shouldn't reach here!");
} catch (IOException ie) {
assertTrue("Invalid exception", ie.getMessage().contains("Invalid Ledger Manager"));
@@ -147,8 +150,10 @@ public class LedgerLayoutTest extends BookKeeperClusterTestCase {
writeLedgerLayout(conf.getZkLedgersRootPath(),
FlatLedgerManagerFactory.NAME,
FlatLedgerManagerFactory.CUR_VERSION, 1);
+ ZkLayoutManager zkLayoutManager = new ZkLayoutManager(zkc, conf.getZkLedgersRootPath(), Ids.OPEN_ACL_UNSAFE);
+
+ LedgerLayout layout = zkLayoutManager.readLedgerLayout();
- LedgerLayout layout = LedgerLayout.readLayout(zkc, conf.getZkLedgersRootPath());
assertNotNull("Should not be null", layout);
assertEquals(FlatLedgerManagerFactory.NAME, layout.getManagerFactoryClass());
assertEquals(FlatLedgerManagerFactory.CUR_VERSION, layout.getManagerVersion());
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java
index c2e3f3e..b3ea6a8 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorLedgerCheckerTest.java
@@ -48,6 +48,7 @@ import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.LedgerMetadata;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
@@ -125,6 +126,7 @@ public class AuditorLedgerCheckerTest extends BookKeeperClusterTestCase {
rng = new Random(System.currentTimeMillis()); // Initialize the Random
urLedgerList = new HashSet<Long>();
ledgerList = new ArrayList<Long>(2);
+ baseClientConf.setZkServers(zkUtil.getZooKeeperConnectString());
}
@Override
@@ -531,8 +533,12 @@ public class AuditorLedgerCheckerTest extends BookKeeperClusterTestCase {
ensemble.add(new BookieSocketAddress("11.11.11.11:1111"));
ensemble.add(new BookieSocketAddress("88.88.88.88:8888"));
metadata.addEnsemble(0, ensemble);
- LedgerManager ledgerManager = LedgerManagerFactory.newLedgerManagerFactory(baseClientConf, zkc)
- .newLedgerManager();
+ LedgerManager ledgerManager = LedgerManagerFactory.newLedgerManagerFactory(
+ baseClientConf,
+ RegistrationManager
+ .instantiateRegistrationManager(new ServerConfiguration(baseClientConf))
+ .getLayoutManager()).newLedgerManager();
+
MutableInt ledgerCreateRC = new MutableInt(-1);
CountDownLatch latch = new CountDownLatch(1);
long ledgerId = (Math.abs(rand.nextLong())) % 100000000;
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 0fb33f1..bfd88e3 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,13 +23,13 @@ package org.apache.bookkeeper.replication;
import static org.junit.Assert.assertEquals;
import java.util.List;
-
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.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
@@ -95,7 +95,11 @@ public class AuditorPeriodicBookieCheckTest extends BookKeeperClusterTestCase {
*/
@Test
public void testPeriodicBookieCheckInterval() throws Exception {
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(bsConfs.get(0), zkc);
+ bsConfs.get(0).setZkServers(zkUtil.getZooKeeperConnectString());
+ LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ bsConfs.get(0),
+ RegistrationManager.instantiateRegistrationManager(bsConfs.get(0)).getLayoutManager());
+
LedgerManager ledgerManager = mFactory.newLedgerManager();
final LedgerUnderreplicationManager underReplicationManager = mFactory.newLedgerUnderreplicationManager();
final int numLedgers = 1;
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicCheckTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicCheckTest.java
index a612120..6baa5d9 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicCheckTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPeriodicCheckTest.java
@@ -25,7 +25,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import io.netty.buffer.ByteBuf;
-
import java.io.File;
import java.io.FileOutputStream;
import java.io.FilenameFilter;
@@ -38,7 +37,6 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieAccessor;
import org.apache.bookkeeper.bookie.IndexPersistenceMgr;
@@ -47,6 +45,7 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
import org.apache.bookkeeper.stats.NullStatsLogger;
@@ -121,7 +120,9 @@ public class AuditorPeriodicCheckTest extends BookKeeperClusterTestCase {
*/
@Test
public void testEntryLogCorruption() throws Exception {
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(bsConfs.get(0), zkc);
+ LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ bsConfs.get(0),
+ RegistrationManager.instantiateRegistrationManager(bsConfs.get(0)).getLayoutManager());
LedgerUnderreplicationManager underReplicationManager = mFactory.newLedgerUnderreplicationManager();
underReplicationManager.disableLedgerReplication();
@@ -170,7 +171,10 @@ public class AuditorPeriodicCheckTest extends BookKeeperClusterTestCase {
*/
@Test
public void testIndexCorruption() throws Exception {
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(bsConfs.get(0), zkc);
+ LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ bsConfs.get(0),
+ RegistrationManager.instantiateRegistrationManager(bsConfs.get(0)).getLayoutManager());
+
LedgerUnderreplicationManager underReplicationManager = mFactory.newLedgerUnderreplicationManager();
LedgerHandle lh = bkc.createLedger(3, 3, DigestType.CRC32, "passwd".getBytes());
@@ -217,7 +221,9 @@ public class AuditorPeriodicCheckTest extends BookKeeperClusterTestCase {
*/
@Test
public void testPeriodicCheckWhenDisabled() throws Exception {
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(bsConfs.get(0), zkc);
+ LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ bsConfs.get(0),
+ RegistrationManager.instantiateRegistrationManager(bsConfs.get(0)).getLayoutManager());
final LedgerUnderreplicationManager underReplicationManager = mFactory.newLedgerUnderreplicationManager();
final int numLedgers = 10;
final int numMsgs = 2;
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorRollingRestartTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorRollingRestartTest.java
index 932de79..9b967e9 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorRollingRestartTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorRollingRestartTest.java
@@ -25,6 +25,7 @@ import static org.junit.Assert.assertEquals;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
@@ -48,7 +49,10 @@ public class AuditorRollingRestartTest extends BookKeeperClusterTestCase {
*/
@Test
public void testAuditingDuringRollingRestart() throws Exception {
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(bsConfs.get(0), zkc);
+ LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ bsConfs.get(0),
+ RegistrationManager.instantiateRegistrationManager(bsConfs.get(0)).getLayoutManager());
+
final LedgerUnderreplicationManager underReplicationManager = mFactory.newLedgerUnderreplicationManager();
LedgerHandle lh = bkc.createLedger(3, 3, DigestType.CRC32, "passwd".getBytes());
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieAutoRecoveryTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieAutoRecoveryTest.java
index 291e1cf..7c758e4 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieAutoRecoveryTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieAutoRecoveryTest.java
@@ -30,13 +30,14 @@ import java.util.List;
import java.util.SortedMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.BookKeeperTestClient;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.LedgerHandleAdapter;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
+import org.apache.bookkeeper.meta.LayoutManager;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
@@ -93,12 +94,20 @@ public class BookieAutoRecoveryTest extends BookKeeperClusterTestCase {
public void setUp() throws Exception {
super.setUp();
baseConf.setZkServers(zkUtil.getZooKeeperConnectString());
+ baseClientConf.setZkServers(zkUtil.getZooKeeperConnectString());
// initialize urReplicationManager
- mFactory = LedgerManagerFactory.newLedgerManagerFactory(baseClientConf,
- zkc);
+ LayoutManager layoutManager = RegistrationManager
+ .instantiateRegistrationManager(new ServerConfiguration(baseClientConf)).getLayoutManager();
+
+ mFactory = LedgerManagerFactory
+ .newLedgerManagerFactory(
+ baseClientConf,
+ layoutManager);
underReplicationManager = mFactory.newLedgerUnderreplicationManager();
LedgerManagerFactory newLedgerManagerFactory = LedgerManagerFactory
- .newLedgerManagerFactory(baseClientConf, zkc);
+ .newLedgerManagerFactory(
+ baseClientConf,
+ layoutManager);
ledgerManager = newLedgerManagerFactory.newLedgerManager();
}
@@ -106,7 +115,7 @@ public class BookieAutoRecoveryTest extends BookKeeperClusterTestCase {
public void tearDown() throws Exception {
super.tearDown();
if (null != mFactory) {
- mFactory.uninitialize();
+ mFactory.close();
mFactory = null;
}
if (null != underReplicationManager) {
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieLedgerIndexTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieLedgerIndexTest.java
index 6fb0c88..43933fb 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieLedgerIndexTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieLedgerIndexTest.java
@@ -30,15 +30,16 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
-
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.meta.LedgerManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.MSLedgerManagerFactory;
+import org.apache.bookkeeper.meta.ZkLayoutManager;
import org.apache.bookkeeper.replication.ReplicationException.BKAuditException;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.util.ZkUtils;
import org.apache.commons.io.FileUtils;
import org.apache.zookeeper.KeeperException;
import org.junit.After;
@@ -84,13 +85,16 @@ public class BookieLedgerIndexTest extends BookKeeperClusterTestCase {
@Before
public void setUp() throws Exception {
super.setUp();
+ baseConf.setZkServers(zkUtil.getZooKeeperConnectString());
rng = new Random(System.currentTimeMillis()); // Initialize the Random
// Number Generator
entries = new ArrayList<byte[]>(); // initialize the entries list
ledgerList = new ArrayList<Long>(3);
// initialize ledger manager
newLedgerManagerFactory = LedgerManagerFactory.newLedgerManagerFactory(
- baseConf, zkc);
+ baseConf,
+ new ZkLayoutManager(zkc, baseConf.getZkLedgersRootPath(), ZkUtils.getACLs(baseConf)));
+
ledgerManager = newLedgerManagerFactory.newLedgerManager();
}
@@ -98,7 +102,7 @@ public class BookieLedgerIndexTest extends BookKeeperClusterTestCase {
public void tearDown() throws Exception {
super.tearDown();
if (null != newLedgerManagerFactory) {
- newLedgerManagerFactory.uninitialize();
+ newLedgerManagerFactory.close();
newLedgerManagerFactory = null;
}
if (null != ledgerManager) {
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestLedgerUnderreplicationManager.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestLedgerUnderreplicationManager.java
index 25d381f..322c6ea 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestLedgerUnderreplicationManager.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestLedgerUnderreplicationManager.java
@@ -27,7 +27,6 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import com.google.protobuf.TextFormat;
-
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Collection;
@@ -44,17 +43,18 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
+import org.apache.bookkeeper.meta.ZkLayoutManager;
import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
import org.apache.bookkeeper.proto.DataFormats.UnderreplicatedLedgerFormat;
import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException;
import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
import org.apache.bookkeeper.test.ZooKeeperUtil;
import org.apache.bookkeeper.util.BookKeeperConstants;
+import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.KeeperException;
@@ -105,12 +105,25 @@ public class TestLedgerUnderreplicationManager {
.connectString(zkUtil.getZooKeeperConnectString())
.sessionTimeoutMs(10000)
.build();
- lmf1 = LedgerManagerFactory.newLedgerManagerFactory(conf, zkc1);
- lmf2 = LedgerManagerFactory.newLedgerManagerFactory(conf, zkc2);
+
basePath = conf.getZkLedgersRootPath() + '/'
+ BookKeeperConstants.UNDER_REPLICATION_NODE;
urLedgerPath = basePath
+ BookKeeperConstants.DEFAULT_ZK_LEDGERS_ROOT_PATH;
+
+ lmf1 = LedgerManagerFactory.newLedgerManagerFactory(
+ conf,
+ new ZkLayoutManager(
+ zkc1,
+ conf.getZkLedgersRootPath(),
+ ZkUtils.getACLs(conf)));
+ lmf2 = LedgerManagerFactory.newLedgerManagerFactory(
+ conf,
+ new ZkLayoutManager(
+ zkc2,
+ conf.getZkLedgersRootPath(),
+ ZkUtils.getACLs(conf)));
+
}
@After
@@ -131,11 +144,11 @@ public class TestLedgerUnderreplicationManager {
zkc2 = null;
}
if (lmf1 != null) {
- lmf1.uninitialize();
+ lmf1.close();
lmf1 = null;
}
if (lmf2 != null) {
- lmf2.uninitialize();
+ lmf2.close();
lmf2 = null;
}
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java
index 61e363e..b60d3f6 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java
@@ -35,6 +35,7 @@ import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.LedgerHandleAdapter;
import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
import org.apache.bookkeeper.net.BookieSocketAddress;
@@ -84,9 +85,12 @@ public class TestReplicationWorker extends BookKeeperClusterTestCase {
@Override
public void setUp() throws Exception {
super.setUp();
+ baseConf.setZkServers(zkUtil.getZooKeeperConnectString());
// initialize urReplicationManager
- mFactory = LedgerManagerFactory.newLedgerManagerFactory(baseClientConf,
- zkc);
+ mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ baseClientConf,
+ RegistrationManager
+ .instantiateRegistrationManager(new ServerConfiguration(baseClientConf)).getLayoutManager());
underReplicationManager = mFactory.newLedgerUnderreplicationManager();
}
@@ -94,7 +98,7 @@ public class TestReplicationWorker extends BookKeeperClusterTestCase {
public void tearDown() throws Exception {
super.tearDown();
if (null != mFactory){
- mFactory.uninitialize();
+ mFactory.close();
mFactory = null;
}
if (null != underReplicationManager){
@@ -413,7 +417,12 @@ public class TestReplicationWorker extends BookKeeperClusterTestCase {
ReplicationWorker rw = new ReplicationWorker(zkc, baseConf);
LedgerManagerFactory mFactory = LedgerManagerFactory
- .newLedgerManagerFactory(baseClientConf, zkc);
+ .newLedgerManagerFactory(
+ baseClientConf,
+ RegistrationManager
+ .instantiateRegistrationManager(new ServerConfiguration(baseClientConf))
+ .getLayoutManager());
+
LedgerUnderreplicationManager underReplicationManager = mFactory
.newLedgerUnderreplicationManager();
rw.start();
@@ -473,8 +482,14 @@ public class TestReplicationWorker extends BookKeeperClusterTestCase {
ReplicationWorker rw = new ReplicationWorker(zkc, baseConf);
+ baseClientConf.setZkServers(zkUtil.getZooKeeperConnectString());
LedgerManagerFactory mFactory = LedgerManagerFactory
- .newLedgerManagerFactory(baseClientConf, zkc);
+ .newLedgerManagerFactory(
+ baseClientConf,
+ RegistrationManager
+ .instantiateRegistrationManager(new ServerConfiguration(baseClientConf))
+ .getLayoutManager());
+
LedgerUnderreplicationManager underReplicationManager = mFactory
.newLedgerUnderreplicationManager();
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 4ab51a4..467d6f2 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,19 +23,18 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import com.google.common.collect.Maps;
-
import java.io.File;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.LedgerHandleAdapter;
import org.apache.bookkeeper.client.LedgerMetadata;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.http.HttpServer;
import org.apache.bookkeeper.http.service.HttpEndpointService;
import org.apache.bookkeeper.http.service.HttpServiceRequest;
@@ -83,8 +82,9 @@ public class TestHttpService extends BookKeeperClusterTestCase {
super.setUp();
baseConf.setZkServers(zkUtil.getZooKeeperConnectString());
this.bkHttpServiceProvider = new BKHttpServiceProvider.Builder()
- .setServerConfiguration(baseConf)
- .build();
+ .setBookieServer(bs.get(numberOfBookies - 1))
+ .setServerConfiguration(baseConf)
+ .build();
}
@Test
@@ -663,7 +663,10 @@ public class TestHttpService extends BookKeeperClusterTestCase {
//2, GET, should return success.
// first put ledger into rereplicate. then use api to list ur ledger.
- LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(bsConfs.get(0), zkc);
+ LedgerManagerFactory mFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ bsConfs.get(0),
+ RegistrationManager.instantiateRegistrationManager(bsConfs.get(0)).getLayoutManager());
+
LedgerManager ledgerManager = mFactory.newLedgerManager();
final LedgerUnderreplicationManager underReplicationManager = mFactory.newLedgerUnderreplicationManager();
diff --git a/tests/backward/src/test/java/org/apache/bookkeeper/tests/backward/TestBackwardCompat.java b/tests/backward/src/test/java/org/apache/bookkeeper/tests/backward/TestBackwardCompat.java
index 4101c0c..c6a904e 100644
--- a/tests/backward/src/test/java/org/apache/bookkeeper/tests/backward/TestBackwardCompat.java
+++ b/tests/backward/src/test/java/org/apache/bookkeeper/tests/backward/TestBackwardCompat.java
@@ -412,7 +412,7 @@ public class TestBackwardCompat {
// Format the metadata using current version
ServerCurrent currentServer = new ServerCurrent(journalDir, ledgerDir,
port, false);
- BookKeeperAdmin.format(new ClientConfiguration(currentServer.conf),
+ BookKeeperAdmin.format(currentServer.conf,
false, true);
// start the current version server with old version cookie
try {
--
To stop receiving notification emails like this one, please contact
['"commits@bookkeeper.apache.org" <co...@bookkeeper.apache.org>'].