You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2018/02/17 06:36:32 UTC
[GitHub] sijie closed pull request #1127: BP-29 (task 1) : Introduce `metadataServiceUrl`
sijie closed pull request #1127: BP-29 (task 1) : Introduce `metadataServiceUrl`
URL: https://github.com/apache/bookkeeper/pull/1127
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
index e772c7917..04dd6be1f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
@@ -25,7 +25,10 @@
import javax.net.ssl.SSLEngine;
import org.apache.bookkeeper.feature.Feature;
+import org.apache.bookkeeper.meta.AbstractZkLedgerManagerFactory;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
import org.apache.bookkeeper.meta.LedgerManagerFactory;
+import org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory;
import org.apache.bookkeeper.util.EntryFormatter;
import org.apache.bookkeeper.util.LedgerIdFormatter;
import org.apache.bookkeeper.util.ReflectionUtils;
@@ -65,6 +68,7 @@
// Ledger Manager
protected static final String LEDGER_MANAGER_TYPE = "ledgerManagerType";
protected static final String LEDGER_MANAGER_FACTORY_CLASS = "ledgerManagerFactoryClass";
+ protected static final String METADATA_SERVICE_URI = "metadataServiceUri";
protected static final String ZK_LEDGERS_ROOT_PATH = "zkLedgersRootPath";
protected static final String ZK_REQUEST_RATE_LIMIT = "zkRequestRateLimit";
protected static final String AVAILABLE_NODE = "available";
@@ -175,6 +179,58 @@ public void loadConf(CompositeConfiguration baseConf) {
}
}
+ /**
+ * Get metadata service uri.
+ *
+ * @return metadata service uri.
+ */
+ @SuppressWarnings("deprecation")
+ public String getMetadataServiceUri() throws ConfigurationException {
+ String serviceUri = getString(METADATA_SERVICE_URI);
+ if (null == serviceUri) {
+ // no service uri is defined, fallback to old settings
+ String ledgerManagerType;
+ Class<? extends LedgerManagerFactory> factoryClass = getLedgerManagerFactoryClass();
+ if (factoryClass == null) {
+ ledgerManagerType = HierarchicalLedgerManagerFactory.NAME;
+ } else {
+ if (!AbstractZkLedgerManagerFactory.class.isAssignableFrom(factoryClass)) {
+ // this is a non-zk implementation
+ throw new UnsupportedOperationException("metadata service uri is not supported for "
+ + factoryClass);
+ }
+ if (factoryClass == HierarchicalLedgerManagerFactory.class) {
+ ledgerManagerType = HierarchicalLedgerManagerFactory.NAME;
+ } else if (factoryClass == org.apache.bookkeeper.meta.FlatLedgerManagerFactory.class) {
+ ledgerManagerType = org.apache.bookkeeper.meta.FlatLedgerManagerFactory.NAME;
+ } else if (factoryClass == LongHierarchicalLedgerManagerFactory.class) {
+ ledgerManagerType = LongHierarchicalLedgerManagerFactory.NAME;
+ } else {
+ throw new IllegalArgumentException("Unknown zookeeper based ledger manager factory : "
+ + factoryClass);
+ }
+ }
+ serviceUri = String.format(
+ "zk+%s://%s%s",
+ ledgerManagerType,
+ getZkServers(),
+ getZkLedgersRootPath());
+ }
+ return serviceUri;
+ }
+
+ /**
+ * Set the metadata service uri.
+ *
+ * @param serviceUri the metadata service uri.
+ * @return the configuration object.
+ * @throws ConfigurationException
+ */
+ public T setMetadataServiceUri(String serviceUri) throws ConfigurationException {
+ setProperty(METADATA_SERVICE_URI, serviceUri);
+ return getThis();
+ }
+
/**
* Get zookeeper servers to connect.
*
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataBookieDriver.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataBookieDriver.java
new file mode 100644
index 000000000..c88a9c46f
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataBookieDriver.java
@@ -0,0 +1,71 @@
+/*
+ * 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 org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
+import org.apache.bookkeeper.discover.RegistrationManager.RegistrationListener;
+import org.apache.bookkeeper.meta.exceptions.MetadataException;
+import org.apache.bookkeeper.stats.StatsLogger;
+
+/**
+ * Driver to manage all the metadata managers required by a bookie server.
+ */
+public interface MetadataBookieDriver extends AutoCloseable {
+
+ /**
+ * Initialize the metadata driver.
+ *
+ * @param conf configuration
+ * @param listener registration listener listening on registration state changes.
+ * @param statsLogger stats logger
+ * @return metadata driver
+ */
+ MetadataBookieDriver initialize(ServerConfiguration conf,
+ RegistrationListener listener,
+ StatsLogger statsLogger)
+ throws MetadataException;
+
+ /**
+ * Get the scheme of the metadata driver.
+ *
+ * @return the scheme of the metadata driver.
+ */
+ String getScheme();
+
+ /**
+ * Return the registration manager used for registering/unregistering bookies.
+ *
+ * @return the registration manager used for registering/unregistering bookies.
+ */
+ RegistrationManager getRegistrationManager()
+ throws MetadataException;
+
+ /**
+ * Return the ledger manager factory used for accessing ledger metadata.
+ *
+ * @return the ledger manager factory used for accessing ledger metadata.
+ */
+ LedgerManagerFactory getLedgerManagerFactory()
+ throws MetadataException;
+
+ @Override
+ void close();
+
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataClientDriver.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataClientDriver.java
new file mode 100644
index 000000000..2fd302c09
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataClientDriver.java
@@ -0,0 +1,81 @@
+/*
+ * 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.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
+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.discover.RegistrationClient;
+import org.apache.bookkeeper.meta.exceptions.MetadataException;
+import org.apache.bookkeeper.stats.StatsLogger;
+
+/**
+ * Driver to manage all the metadata managers required by a bookkeeper client.
+ */
+@LimitedPrivate
+@Evolving
+public interface MetadataClientDriver extends AutoCloseable {
+
+ /**
+ * Initialize the metadata driver.
+ *
+ * @param conf configuration
+ * @param scheduler scheduler
+ * @param statsLogger stats logger
+ * @param ctx optional context object passed in for initialization.
+ * currently it is an external zookeeper instance, which can
+ * be used for zookeeper based metadata implementation.
+ * @return metadata driver
+ * @throws MetadataException when fail to initialize the client driver.
+ */
+ MetadataClientDriver initialize(ClientConfiguration conf,
+ ScheduledExecutorService scheduler,
+ StatsLogger statsLogger,
+ Optional<Object> ctx)
+ throws MetadataException;
+
+ /**
+ * Get the scheme of the metadata driver.
+ *
+ * @return the scheme of the metadata driver.
+ */
+ String getScheme();
+
+ /**
+ * Return the registration client used for discovering registered bookies.
+ *
+ * @return the registration client used for discovering registered bookies.
+ */
+ RegistrationClient getRegistrationClient()
+ throws MetadataException;
+
+ /**
+ * Return the ledger manager factory used for accessing ledger metadata.
+ *
+ * @return the ledger manager factory used for accessing ledger metadata.
+ */
+ LedgerManagerFactory getLedgerManagerFactory()
+ throws MetadataException;
+
+ @Override
+ void close();
+
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataDrivers.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataDrivers.java
new file mode 100644
index 000000000..8a657f26f
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/MetadataDrivers.java
@@ -0,0 +1,290 @@
+/*
+ * 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 com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Sets;
+import java.net.URI;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.util.ReflectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * A driver manager for managing a set of metadata drivers.
+ *
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Slf4j
+public final class MetadataDrivers {
+
+ static final String ZK_CLIENT_DRIVER_CLASS = "org.apache.bookkeeper.meta.zk.ZKMetadataClientDriver";
+ static final String ZK_BOOKIE_DRIVER_CLASS = "org.apache.bookkeeper.meta.zk.ZKMetadataBookieDriver";
+ static final String BK_METADATA_CLIENT_DRIVERS_PROPERTY = "bookkeeper.metadata.client.drivers";
+ static final String BK_METADATA_BOOKIE_DRIVERS_PROPERTY = "bookkeeper.metadata.bookie.drivers";
+
+ @ToString
+ static class MetadataClientDriverInfo {
+
+ final Class<? extends MetadataClientDriver> driverClass;
+ final String driverClassName;
+
+ MetadataClientDriverInfo(Class<? extends MetadataClientDriver> driverClass) {
+ this.driverClass = driverClass;
+ this.driverClassName = driverClass.getName();
+ }
+
+ }
+
+ @ToString
+ static class MetadataBookieDriverInfo {
+
+ final Class<? extends MetadataBookieDriver> driverClass;
+ final String driverClassName;
+
+ MetadataBookieDriverInfo(Class<? extends MetadataBookieDriver> driverClass) {
+ this.driverClass = driverClass;
+ this.driverClassName = driverClass.getName();
+ }
+
+ }
+
+ @Getter(AccessLevel.PACKAGE)
+ private static final ConcurrentMap<String, MetadataClientDriverInfo> clientDrivers;
+ @Getter(AccessLevel.PACKAGE)
+ private static final ConcurrentMap<String, MetadataBookieDriverInfo> bookieDrivers;
+ private static boolean initialized = false;
+
+ static {
+ clientDrivers = new ConcurrentHashMap<>();
+ bookieDrivers = new ConcurrentHashMap<>();
+ initialize();
+ }
+
+ static void initialize() {
+ if (initialized) {
+ return;
+ }
+ loadInitialDrivers();
+ initialized = true;
+ log.info("BookKeeper metadata driver manager initialized");
+ }
+
+ @VisibleForTesting
+ static void loadInitialDrivers() {
+ loadInitialClientDrivers();
+ loadInitialBookieDrivers();
+ }
+
+ private static void loadInitialClientDrivers() {
+ Set<String> driverList = Sets.newHashSet();
+
+ // add default zookeeper based driver
+ driverList.add(ZK_CLIENT_DRIVER_CLASS);
+
+ // load drivers from system property
+ String driversStr = System.getProperty(BK_METADATA_CLIENT_DRIVERS_PROPERTY);
+ if (null != driversStr) {
+ String[] driversArray = StringUtils.split(driversStr, ':');
+ for (String driver : driversArray) {
+ driverList.add(driver);
+ }
+ }
+
+ // initialize the drivers
+ for (String driverClsName : driverList) {
+ try {
+ MetadataClientDriver driver =
+ ReflectionUtils.newInstance(driverClsName, MetadataClientDriver.class);
+ MetadataClientDriverInfo driverInfo =
+ new MetadataClientDriverInfo(driver.getClass());
+ clientDrivers.put(driver.getScheme().toLowerCase(), driverInfo);
+ } catch (Exception e) {
+ log.warn("Failed to load metadata client driver {}", driverClsName, e);
+ }
+ }
+ }
+
+ private static void loadInitialBookieDrivers() {
+ Set<String> driverList = Sets.newHashSet();
+
+ // add default zookeeper based driver
+ driverList.add(ZK_BOOKIE_DRIVER_CLASS);
+
+ // load drivers from system property
+ String driversStr = System.getProperty(BK_METADATA_BOOKIE_DRIVERS_PROPERTY);
+ if (null != driversStr) {
+ String[] driversArray = StringUtils.split(driversStr, ':');
+ for (String driver : driversArray) {
+ driverList.add(driver);
+ }
+ }
+
+ // initialize the drivers
+ for (String driverClsName : driverList) {
+ try {
+ MetadataBookieDriver driver =
+ ReflectionUtils.newInstance(driverClsName, MetadataBookieDriver.class);
+ MetadataBookieDriverInfo driverInfo =
+ new MetadataBookieDriverInfo(driver.getClass());
+ bookieDrivers.put(driver.getScheme().toLowerCase(), driverInfo);
+ } catch (Exception e) {
+ log.warn("Failed to load metadata bookie driver {}", driverClsName, e);
+ }
+ }
+ }
+
+ /**
+ * Register the metadata client {@code driver}.
+ *
+ * @param metadataBackendScheme scheme of metadata backend.
+ * @param driver metadata client driver
+ */
+ public static void registerClientDriver(String metadataBackendScheme,
+ Class<? extends MetadataClientDriver> driver) {
+ if (!initialized) {
+ initialize();
+ }
+
+ String scheme = metadataBackendScheme.toLowerCase();
+ MetadataClientDriverInfo oldDriverInfo = clientDrivers.get(scheme);
+ if (null != oldDriverInfo) {
+ return;
+ }
+ MetadataClientDriverInfo newDriverInfo = new MetadataClientDriverInfo(driver);
+ oldDriverInfo = clientDrivers.putIfAbsent(scheme, newDriverInfo);
+ if (null != oldDriverInfo) {
+ log.debug("Metadata client driver for {} is already there.", scheme);
+ }
+ }
+
+ /**
+ * Register the metadata bookie {@code driver}.
+ *
+ * @param metadataBackendScheme scheme of metadata backend.
+ * @param driver metadata bookie driver
+ */
+ public static void registerBookieDriver(String metadataBackendScheme,
+ Class<? extends MetadataBookieDriver> driver) {
+ if (!initialized) {
+ initialize();
+ }
+
+ String scheme = metadataBackendScheme.toLowerCase();
+ MetadataBookieDriverInfo oldDriverInfo = bookieDrivers.get(scheme);
+ if (null != oldDriverInfo) {
+ return;
+ }
+ MetadataBookieDriverInfo newDriverInfo = new MetadataBookieDriverInfo(driver);
+ oldDriverInfo = bookieDrivers.putIfAbsent(scheme, newDriverInfo);
+ if (null != oldDriverInfo) {
+ log.debug("Metadata bookie driver for {} is already there.", scheme);
+ }
+ }
+
+ /**
+ * Retrieve the client driver for {@code scheme}.
+ *
+ * @param scheme the scheme for the metadata client driver
+ * @return the metadata client driver
+ * @throws NullPointerException when scheme is null
+ */
+ public static MetadataClientDriver getClientDriver(String scheme) {
+ checkNotNull(scheme, "Client Driver Scheme is null");
+ if (!initialized) {
+ initialize();
+ }
+ MetadataClientDriverInfo driverInfo = clientDrivers.get(scheme.toLowerCase());
+ if (null == driverInfo) {
+ throw new IllegalArgumentException("Unknown backend " + scheme);
+ }
+ return ReflectionUtils.newInstance(driverInfo.driverClass);
+ }
+
+ /**
+ * Retrieve the client driver for {@code uri}.
+ *
+ * @param uri the metadata service uri
+ * @return the metadata client driver for {@code uri}
+ * @throws NullPointerException if the metadata service {@code uri} is null or doesn't have scheme
+ * or there is no namespace driver registered for the scheme
+ * @throws IllegalArgumentException if the metadata service {@code uri} scheme is illegal
+ */
+ public static MetadataClientDriver getClientDriver(URI uri) {
+ // Validate the uri and load the backend according to scheme
+ checkNotNull(uri, "Metadata service uri is null");
+ String scheme = uri.getScheme();
+ checkNotNull(scheme, "Invalid metadata service uri : " + uri);
+ scheme = scheme.toLowerCase();
+ String[] schemeParts = StringUtils.split(scheme, '+');
+ checkArgument(schemeParts.length > 0,
+ "Invalid metadata service scheme found : " + uri);
+ return getClientDriver(schemeParts[0]);
+ }
+
+ /**
+ * Retrieve the bookie driver for {@code scheme}.
+ *
+ * @param scheme the scheme for the metadata bookie driver
+ * @return the metadata bookie driver
+ * @throws NullPointerException when scheme is null
+ */
+ public static MetadataBookieDriver getBookieDriver(String scheme) {
+ checkNotNull(scheme, "Bookie Driver Scheme is null");
+ if (!initialized) {
+ initialize();
+ }
+ MetadataBookieDriverInfo driverInfo = bookieDrivers.get(scheme.toLowerCase());
+ if (null == driverInfo) {
+ throw new IllegalArgumentException("Unknown backend " + scheme);
+ }
+ return ReflectionUtils.newInstance(driverInfo.driverClass);
+ }
+
+ /**
+ * Retrieve the bookie driver for {@code uri}.
+ *
+ * @param uri the metadata service uri
+ * @return the metadata bookie driver for {@code uri}
+ * @throws NullPointerException if the metadata service {@code uri} is null or doesn't have scheme
+ * or there is no namespace driver registered for the scheme
+ * @throws IllegalArgumentException if the metadata service {@code uri} scheme is illegal
+ */
+ public static MetadataBookieDriver getBookieDriver(URI uri) {
+ // Validate the uri and load the backend according to scheme
+ checkNotNull(uri, "Metadata service uri is null");
+ String scheme = uri.getScheme();
+ checkNotNull(scheme, "Invalid metadata service uri : " + uri);
+ scheme = scheme.toLowerCase();
+ String[] schemeParts = StringUtils.split(scheme, '+');
+ checkArgument(schemeParts.length > 0,
+ "Invalid metadata service scheme found : " + uri);
+ return getBookieDriver(schemeParts[0]);
+ }
+
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/Code.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/Code.java
new file mode 100644
index 000000000..8c936aa5a
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/Code.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.meta.exceptions;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Metadata Exception Code.
+ */
+@RequiredArgsConstructor
+@Getter
+public enum Code {
+
+ // the passed in metadata service uri is invalid
+ INVALID_METADATA_SERVICE_URI(-1),
+ // encountered service error when talking to the metadata service
+ METADATA_SERVICE_ERROR(-2);
+
+ private final int code;
+
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/MetadataException.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/MetadataException.java
new file mode 100644
index 000000000..c61a3dfd8
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/MetadataException.java
@@ -0,0 +1,47 @@
+/*
+ * 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.exceptions;
+
+import lombok.Getter;
+
+/**
+ * Metadata Exception.
+ */
+@Getter
+public class MetadataException extends Exception {
+
+ private static final long serialVersionUID = -7833758146152388503L;
+
+ private final Code code;
+
+ public MetadataException(Code code, String message) {
+ super(message);
+ this.code = code;
+ }
+
+ public MetadataException(Code code, String message, Throwable cause) {
+ super(message, cause);
+ this.code = code;
+ }
+
+ public MetadataException(Code code, Throwable cause) {
+ super(cause);
+ this.code = code;
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/package-info.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/package-info.java
new file mode 100644
index 000000000..3d69edb6a
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/exceptions/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+/**
+ * Exceptions for metadata service.
+ */
+package org.apache.bookkeeper.meta.exceptions;
\ No newline at end of file
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataBookieDriver.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataBookieDriver.java
new file mode 100644
index 000000000..219ae2d3a
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataBookieDriver.java
@@ -0,0 +1,94 @@
+/*
+ * 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.zk;
+
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
+import org.apache.bookkeeper.discover.RegistrationManager.RegistrationListener;
+import org.apache.bookkeeper.discover.ZKRegistrationManager;
+import org.apache.bookkeeper.meta.MetadataBookieDriver;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.meta.exceptions.Code;
+import org.apache.bookkeeper.meta.exceptions.MetadataException;
+import org.apache.bookkeeper.stats.StatsLogger;
+
+/**
+ * ZooKeeper based metadata bookie driver.
+ */
+@Slf4j
+public class ZKMetadataBookieDriver
+ extends ZKMetadataDriverBase
+ implements MetadataBookieDriver {
+
+ // register myself
+ static {
+ MetadataDrivers.registerBookieDriver(
+ SCHEME,
+ ZKMetadataBookieDriver.class);
+ }
+
+ ServerConfiguration serverConf;
+ RegistrationManager regManager;
+ RegistrationListener listener;
+
+ @Override
+ public synchronized MetadataBookieDriver initialize(ServerConfiguration conf,
+ RegistrationListener listener,
+ StatsLogger statsLogger)
+ throws MetadataException {
+ super.initialize(
+ conf, statsLogger, Optional.empty());
+ this.statsLogger = statsLogger;
+ this.serverConf = conf;
+ this.listener = listener;
+ return this;
+ }
+
+ @Override
+ public synchronized RegistrationManager getRegistrationManager()
+ throws MetadataException {
+ if (null == regManager) {
+ regManager = new ZKRegistrationManager();
+ try {
+ regManager.initialize(
+ serverConf,
+ listener,
+ statsLogger);
+ } catch (BookieException e) {
+ throw new MetadataException(
+ Code.METADATA_SERVICE_ERROR,
+ "Failed to initialize registration manager",
+ e);
+ }
+ }
+ return regManager;
+ }
+
+ @Override
+ public void close() {
+ if (null != regManager) {
+ regManager.close();
+ regManager = null;
+ }
+ super.close();
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java
new file mode 100644
index 000000000..3fec50cb7
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriver.java
@@ -0,0 +1,96 @@
+/*
+ * 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.zk;
+
+import java.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.discover.RegistrationClient;
+import org.apache.bookkeeper.discover.ZKRegistrationClient;
+import org.apache.bookkeeper.meta.MetadataClientDriver;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.meta.exceptions.Code;
+import org.apache.bookkeeper.meta.exceptions.MetadataException;
+import org.apache.bookkeeper.stats.StatsLogger;
+
+/**
+ * ZooKeeper based metadata client driver.
+ */
+@Slf4j
+public class ZKMetadataClientDriver
+ extends ZKMetadataDriverBase
+ implements MetadataClientDriver {
+
+ // register myself to driver manager
+ static {
+ MetadataDrivers.registerClientDriver(
+ SCHEME, ZKMetadataClientDriver.class);
+ }
+
+ ClientConfiguration clientConf;
+ ScheduledExecutorService scheduler;
+ RegistrationClient regClient;
+
+ @Override
+ public synchronized MetadataClientDriver initialize(ClientConfiguration conf,
+ ScheduledExecutorService scheduler,
+ StatsLogger statsLogger,
+ Optional<Object> optionalCtx)
+ throws MetadataException {
+ super.initialize(
+ conf,
+ statsLogger,
+ optionalCtx);
+ this.statsLogger = statsLogger;
+ this.clientConf = conf;
+ this.scheduler = scheduler;
+ return this;
+ }
+
+ @Override
+ public synchronized RegistrationClient getRegistrationClient() throws MetadataException {
+ if (null == regClient) {
+ regClient = new ZKRegistrationClient();
+ try {
+ regClient.initialize(
+ clientConf,
+ scheduler,
+ statsLogger,
+ Optional.of(zk));
+ } catch (BKException e) {
+ throw new MetadataException(
+ Code.METADATA_SERVICE_ERROR,
+ "Failed to initialize registration client",
+ e);
+ }
+ }
+ return regClient;
+ }
+
+ @Override
+ public synchronized void close() {
+ if (null != regClient) {
+ regClient.close();
+ regClient = null;
+ }
+ super.close();
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBase.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBase.java
new file mode 100644
index 000000000..1a6034978
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBase.java
@@ -0,0 +1,222 @@
+/*
+ * 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.zk;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Optional;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
+import org.apache.bookkeeper.meta.LayoutManager;
+import org.apache.bookkeeper.meta.LedgerManagerFactory;
+import org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory;
+import org.apache.bookkeeper.meta.ZkLayoutManager;
+import org.apache.bookkeeper.meta.exceptions.Code;
+import org.apache.bookkeeper.meta.exceptions.MetadataException;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.bookkeeper.zookeeper.BoundExponentialBackoffRetryPolicy;
+import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+
+/**
+ * This is a mixin class for supporting zookeeper based metadata driver.
+ */
+@Slf4j
+public class ZKMetadataDriverBase implements AutoCloseable {
+
+ protected static final String SCHEME = "zk";
+
+ protected static String getZKServersFromServiceUri(URI uri) {
+ return uri.getAuthority().replace(";", ",");
+ }
+
+ @SuppressWarnings("deprecation")
+ protected static Class<? extends LedgerManagerFactory> resolveLedgerManagerFactory(URI metadataServiceUri) {
+ checkNotNull(metadataServiceUri, "Metadata service uri is null");
+ String scheme = metadataServiceUri.getScheme();
+ checkNotNull(scheme, "Invalid metadata service : " + metadataServiceUri);
+ String[] schemeParts = StringUtils.split(scheme.toLowerCase(), '+');
+ checkArgument(SCHEME.equals(schemeParts[0]), "Unknown metadata service scheme found : "
+ + schemeParts[0]);
+ Class<? extends LedgerManagerFactory> ledgerManagerFactoryClass;
+ if (schemeParts.length > 1) {
+ switch (schemeParts[1]) {
+ case org.apache.bookkeeper.meta.FlatLedgerManagerFactory.NAME:
+ ledgerManagerFactoryClass = org.apache.bookkeeper.meta.FlatLedgerManagerFactory.class;
+ break;
+ case HierarchicalLedgerManagerFactory.NAME:
+ ledgerManagerFactoryClass = HierarchicalLedgerManagerFactory.class;
+ break;
+ case LongHierarchicalLedgerManagerFactory.NAME:
+ ledgerManagerFactoryClass = LongHierarchicalLedgerManagerFactory.class;
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown ledger manager type found '"
+ + schemeParts[1] + "' at uri : " + metadataServiceUri);
+ }
+ } else {
+ ledgerManagerFactoryClass = HierarchicalLedgerManagerFactory.class;
+ }
+ return ledgerManagerFactoryClass;
+ }
+
+ // URI
+ protected URI metadataServiceUri;
+ protected AbstractConfiguration<?> conf;
+ protected StatsLogger statsLogger;
+
+ // zookeeper related variables
+ protected List<ACL> acls;
+ protected ZooKeeper zk = null;
+ // whether the zk handle is one we created, or is owned by whoever
+ // instantiated us
+ protected boolean ownZKHandle = false;
+
+ // ledgers root path
+ protected String ledgersRootPath;
+
+ // managers
+ protected LayoutManager layoutManager;
+ protected LedgerManagerFactory lmFactory;
+ protected Class<? extends LedgerManagerFactory> ledgerManagerFactoryClass = null;
+
+ public String getScheme() {
+ return SCHEME;
+ }
+
+ @SneakyThrows(InterruptedException.class)
+ protected void initialize(AbstractConfiguration<?> conf,
+ StatsLogger statsLogger,
+ Optional<Object> optionalCtx) throws MetadataException {
+ this.conf = conf;
+
+ String metadataServiceUriStr;
+ try {
+ metadataServiceUriStr = conf.getMetadataServiceUri();
+ } catch (ConfigurationException e) {
+ log.error("Failed to retrieve metadata service uri from configuration", e);
+ throw new MetadataException(
+ Code.INVALID_METADATA_SERVICE_URI, e);
+ }
+ this.metadataServiceUri = URI.create(metadataServiceUriStr);
+ ledgerManagerFactoryClass = resolveLedgerManagerFactory(metadataServiceUri);
+
+ // get the initialize root path
+ this.ledgersRootPath = metadataServiceUri.getPath();
+ String bookieReadonlyRegistrationPath = ledgersRootPath + "/" + AVAILABLE_NODE;
+
+ // construct the zookeeper
+ String zkServers = getZKServersFromServiceUri(metadataServiceUri);
+ this.acls = ZkUtils.getACLs(conf);
+ if (optionalCtx.isPresent()
+ && optionalCtx.get() instanceof ZooKeeper) {
+ // if an external zookeeper is added, use the zookeeper instance
+ this.zk = (ZooKeeper) (optionalCtx.get());
+ this.ownZKHandle = false;
+ } else {
+ try {
+ this.zk = ZooKeeperClient.newBuilder()
+ .connectString(zkServers)
+ .sessionTimeoutMs(conf.getZkTimeout())
+ .operationRetryPolicy(new BoundExponentialBackoffRetryPolicy(conf.getZkTimeout(),
+ conf.getZkTimeout(), 0))
+ .statsLogger(statsLogger)
+ .build();
+
+ if (null == zk.exists(bookieReadonlyRegistrationPath, false)) {
+ try {
+ zk.create(bookieReadonlyRegistrationPath,
+ new byte[0],
+ acls,
+ CreateMode.PERSISTENT);
+ } catch (KeeperException.NodeExistsException e) {
+ // this node is just now created by someone.
+ }
+ }
+ } catch (IOException | KeeperException e) {
+ log.error("Failed to create zookeeper client to {}", zkServers, e);
+ MetadataException me = new MetadataException(
+ Code.METADATA_SERVICE_ERROR,
+ "Failed to create zookeeper client to " + zkServers);
+ me.fillInStackTrace();
+ throw me;
+ }
+ this.ownZKHandle = true;
+ }
+
+ // once created the zookeeper client, create the layout manager and registration client
+ this.layoutManager = new ZkLayoutManager(
+ zk,
+ ledgersRootPath,
+ acls);
+ }
+
+ @SneakyThrows
+ public synchronized LedgerManagerFactory getLedgerManagerFactory()
+ throws MetadataException {
+ if (null == lmFactory) {
+ try {
+ lmFactory = LedgerManagerFactory.newLedgerManagerFactory(
+ conf,
+ layoutManager);
+ } catch (IOException e) {
+ throw new MetadataException(
+ Code.METADATA_SERVICE_ERROR, "Failed to initialized ledger manager factory", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ }
+ return lmFactory;
+ }
+
+ @Override
+ public void close() {
+ if (null != lmFactory) {
+ try {
+ lmFactory.close();
+ } catch (IOException e) {
+ log.warn("Failed to close zookeeper based ledger manager", e);
+ }
+ lmFactory = null;
+ }
+ if (ownZKHandle && null != zk) {
+ try {
+ zk.close();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("Interrupted on closing zookeeper client", e);
+ }
+ zk = null;
+ }
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/package-info.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/package-info.java
new file mode 100644
index 000000000..cd62b1792
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/zk/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+/**
+ * ZooKeeper based metadata implementation.
+ */
+package org.apache.bookkeeper.meta.zk;
\ No newline at end of file
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/conf/AbstractConfigurationTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/conf/AbstractConfigurationTest.java
new file mode 100644
index 000000000..70b9cb3f2
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/conf/AbstractConfigurationTest.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.conf;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+
+import org.apache.bookkeeper.meta.AbstractZkLedgerManagerFactory;
+import org.apache.bookkeeper.meta.LedgerManagerFactory;
+import org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Unit test of {@link AbstractConfiguration}.
+ */
+public class AbstractConfigurationTest {
+
+ private static final String DEFAULT_METADATA_SERVICE_URI =
+ "zk+hierarchical://127.0.0.1/path/to/ledgers";
+ private static final String FLAT_METADATA_SERVICE_URI =
+ "zk+flat://127.0.0.1/path/to/ledgers";
+ private static final String LONGHIERARCHICAL_METADATA_SERVICE_URI =
+ "zk+longhierarchical://127.0.0.1/path/to/ledgers";
+
+ private AbstractConfiguration conf;
+
+ @Before
+ public void setup() {
+ this.conf = new ClientConfiguration();
+ this.conf.setZkServers("127.0.0.1");
+ this.conf.setZkLedgersRootPath("/path/to/ledgers");
+ }
+
+ @Test
+ public void testDefaultServiceUri() throws Exception {
+ assertEquals(
+ DEFAULT_METADATA_SERVICE_URI,
+ conf.getMetadataServiceUri());
+ }
+
+ @Test
+ public void testSetMetadataServiceUri() throws Exception {
+ assertEquals(
+ DEFAULT_METADATA_SERVICE_URI,
+ conf.getMetadataServiceUri());
+ String serviceUri = "etcd://128.0.0.1/key/prefix";
+ conf.setMetadataServiceUri(serviceUri);
+ assertEquals(
+ "Service URI should be changed to " + serviceUri,
+ serviceUri,
+ conf.getMetadataServiceUri());
+ }
+
+ @SuppressWarnings({ "unchecked" })
+ @Test(expected = UnsupportedOperationException.class)
+ public void testUnsupportedLedgerManagerFactory() throws Exception {
+ LedgerManagerFactory mockFactory = mock(LedgerManagerFactory.class, CALLS_REAL_METHODS);
+ conf.setLedgerManagerFactoryClass(mockFactory.getClass());
+ conf.getMetadataServiceUri();
+ }
+
+ @SuppressWarnings({ "deprecation", "unchecked" })
+ @Test
+ public void testFlatLedgerManagerUri() throws Exception {
+ conf.setLedgerManagerFactoryClass(org.apache.bookkeeper.meta.FlatLedgerManagerFactory.class);
+ assertEquals(
+ FLAT_METADATA_SERVICE_URI,
+ conf.getMetadataServiceUri());
+ }
+
+ @SuppressWarnings({ "unchecked" })
+ @Test
+ public void testLongHierarchicalLedgerManagerUri() throws Exception {
+ conf.setLedgerManagerFactoryClass(LongHierarchicalLedgerManagerFactory.class);
+ assertEquals(
+ LONGHIERARCHICAL_METADATA_SERVICE_URI,
+ conf.getMetadataServiceUri());
+ }
+
+ @SuppressWarnings({ "unchecked" })
+ @Test(expected = IllegalArgumentException.class)
+ public void testUnknownZkLedgerManagerFactory() throws Exception {
+ AbstractZkLedgerManagerFactory mockZkFactory =
+ mock(AbstractZkLedgerManagerFactory.class, CALLS_REAL_METHODS);
+ conf.setLedgerManagerFactoryClass(mockZkFactory.getClass());
+ conf.getMetadataServiceUri();
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/MetadataDriversTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/MetadataDriversTest.java
new file mode 100644
index 000000000..bd89fb3df
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/MetadataDriversTest.java
@@ -0,0 +1,372 @@
+/*
+ * 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.meta.MetadataDrivers.BK_METADATA_BOOKIE_DRIVERS_PROPERTY;
+import static org.apache.bookkeeper.meta.MetadataDrivers.BK_METADATA_CLIENT_DRIVERS_PROPERTY;
+import static org.apache.bookkeeper.meta.MetadataDrivers.ZK_BOOKIE_DRIVER_CLASS;
+import static org.apache.bookkeeper.meta.MetadataDrivers.ZK_CLIENT_DRIVER_CLASS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.Maps;
+import java.net.URI;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationClient;
+import org.apache.bookkeeper.discover.RegistrationManager;
+import org.apache.bookkeeper.discover.RegistrationManager.RegistrationListener;
+import org.apache.bookkeeper.meta.MetadataDrivers.MetadataBookieDriverInfo;
+import org.apache.bookkeeper.meta.MetadataDrivers.MetadataClientDriverInfo;
+import org.apache.bookkeeper.meta.exceptions.MetadataException;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.commons.lang3.StringUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Unit test of {@link MetadataDrivers}.
+ */
+public class MetadataDriversTest {
+
+ static class ClientDriver1 implements MetadataClientDriver {
+
+ @Override
+ public MetadataClientDriver initialize(ClientConfiguration conf,
+ ScheduledExecutorService scheduler,
+ StatsLogger statsLogger,
+ Optional<Object> ctx) throws MetadataException {
+ return this;
+ }
+
+ @Override
+ public String getScheme() {
+ return "driver1";
+ }
+
+ @Override
+ public RegistrationClient getRegistrationClient() throws MetadataException {
+ return mock(RegistrationClient.class);
+ }
+
+ @Override
+ public LedgerManagerFactory getLedgerManagerFactory() throws MetadataException {
+ return mock(LedgerManagerFactory.class);
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ static class ClientDriver2 implements MetadataClientDriver {
+
+ @Override
+ public MetadataClientDriver initialize(ClientConfiguration conf,
+ ScheduledExecutorService scheduler,
+ StatsLogger statsLogger,
+ Optional<Object> ctx) throws MetadataException {
+ return this;
+ }
+
+ @Override
+ public String getScheme() {
+ return "driver2";
+ }
+
+ @Override
+ public RegistrationClient getRegistrationClient() throws MetadataException {
+ return mock(RegistrationClient.class);
+ }
+
+ @Override
+ public LedgerManagerFactory getLedgerManagerFactory() throws MetadataException {
+ return mock(LedgerManagerFactory.class);
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ static class BookieDriver1 implements MetadataBookieDriver {
+
+ @Override
+ public MetadataBookieDriver initialize(ServerConfiguration conf,
+ RegistrationListener listener,
+ StatsLogger statsLogger) throws MetadataException {
+ return this;
+ }
+
+ @Override
+ public String getScheme() {
+ return "driver1";
+ }
+
+ @Override
+ public RegistrationManager getRegistrationManager() throws MetadataException {
+ return mock(RegistrationManager.class);
+ }
+
+ @Override
+ public LedgerManagerFactory getLedgerManagerFactory() throws MetadataException {
+ return mock(LedgerManagerFactory.class);
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ static class BookieDriver2 implements MetadataBookieDriver {
+
+ @Override
+ public MetadataBookieDriver initialize(ServerConfiguration conf,
+ RegistrationListener listener,
+ StatsLogger statsLogger) throws MetadataException {
+ return this;
+ }
+
+ @Override
+ public String getScheme() {
+ return "driver2";
+ }
+
+ @Override
+ public RegistrationManager getRegistrationManager() throws MetadataException {
+ return mock(RegistrationManager.class);
+ }
+
+ @Override
+ public LedgerManagerFactory getLedgerManagerFactory() throws MetadataException {
+ return mock(LedgerManagerFactory.class);
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ private Map<String, MetadataClientDriverInfo> savedClientDrivers;
+ private Map<String, MetadataBookieDriverInfo> savedBookieDrivers;
+
+ @Before
+ public void setup() {
+ savedClientDrivers = Maps.newHashMap(MetadataDrivers.getClientDrivers());
+ savedBookieDrivers = Maps.newHashMap(MetadataDrivers.getBookieDrivers());
+ }
+
+ @After
+ public void teardown() {
+ MetadataDrivers.getClientDrivers().clear();
+ MetadataDrivers.getClientDrivers().putAll(savedClientDrivers);
+ MetadataDrivers.getBookieDrivers().clear();
+ MetadataDrivers.getBookieDrivers().putAll(savedBookieDrivers);
+ }
+
+ @Test
+ public void testDefaultDrivers() {
+ MetadataClientDriver clientDriver = MetadataDrivers.getClientDriver("zk");
+ assertEquals(
+ ZK_CLIENT_DRIVER_CLASS,
+ clientDriver.getClass().getName());
+ clientDriver = MetadataDrivers.getClientDriver(URI.create("zk+hierarchical://127.0.0.1/ledgers"));
+ assertEquals(
+ ZK_CLIENT_DRIVER_CLASS,
+ clientDriver.getClass().getName());
+
+ MetadataBookieDriver bookieDriver = MetadataDrivers.getBookieDriver("zk");
+ assertEquals(
+ ZK_BOOKIE_DRIVER_CLASS,
+ bookieDriver.getClass().getName());
+ bookieDriver = MetadataDrivers.getBookieDriver(URI.create("zk+hierarchical://127.0.0.1/ledgers"));
+ assertEquals(
+ ZK_BOOKIE_DRIVER_CLASS,
+ bookieDriver.getClass().getName());
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testClientDriverNullScheme() {
+ MetadataDrivers.getClientDriver((String) null);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testBookieDriverNullScheme() {
+ MetadataDrivers.getBookieDriver((String) null);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testClientDriverNullURI() {
+ MetadataDrivers.getClientDriver((URI) null);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testBookieDriverNullURI() {
+ MetadataDrivers.getBookieDriver((URI) null);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testClientDriverUnknownScheme() {
+ MetadataDrivers.getClientDriver("unknown");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testBookieDriverUnknownScheme() {
+ MetadataDrivers.getBookieDriver("unknown");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testClientDriverUnknownSchemeURI() {
+ MetadataDrivers.getClientDriver(URI.create("unknown://"));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testBookieDriverUnknownSchemeURI() {
+ MetadataDrivers.getBookieDriver(URI.create("unknown://"));
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testClientDriverNullSchemeURI() {
+ MetadataDrivers.getClientDriver(URI.create("//127.0.0.1/ledgers"));
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testBookieDriverNullSchemeURI() {
+ MetadataDrivers.getBookieDriver(URI.create("//127.0.0.1/ledgers"));
+ }
+
+ @Test
+ public void testClientDriverLowerUpperCasedSchemes() {
+ String[] schemes = new String[] {
+ "zk", "Zk", "zK", "ZK"
+ };
+ for (String scheme : schemes) {
+ MetadataClientDriver clientDriver = MetadataDrivers.getClientDriver(scheme);
+ assertEquals(
+ ZK_CLIENT_DRIVER_CLASS,
+ clientDriver.getClass().getName());
+ }
+ }
+
+ @Test
+ public void testBookieDriverLowerUpperCasedSchemes() {
+ String[] schemes = new String[] {
+ "zk", "Zk", "zK", "ZK"
+ };
+ for (String scheme : schemes) {
+ MetadataBookieDriver bookieDriver = MetadataDrivers.getBookieDriver(scheme);
+ assertEquals(
+ ZK_BOOKIE_DRIVER_CLASS,
+ bookieDriver.getClass().getName());
+ }
+ }
+
+ @Test
+ public void testRegisterClientDriver() throws Exception {
+ MetadataClientDriver clientDriver = mock(MetadataClientDriver.class);
+ when(clientDriver.getScheme()).thenReturn("testdriver");
+
+ try {
+ MetadataDrivers.getClientDriver(clientDriver.getScheme());
+ fail("Should fail to get client driver if it is not registered");
+ } catch (IllegalArgumentException iae) {
+ // expected
+ }
+
+ MetadataDrivers.registerClientDriver(clientDriver.getScheme(), clientDriver.getClass());
+ MetadataClientDriver driver = MetadataDrivers.getClientDriver(clientDriver.getScheme());
+ assertEquals(clientDriver.getClass(), driver.getClass());
+ }
+
+ @Test
+ public void testRegisterBookieDriver() throws Exception {
+ MetadataBookieDriver bookieDriver = mock(MetadataBookieDriver.class);
+ when(bookieDriver.getScheme()).thenReturn("testdriver");
+
+ try {
+ MetadataDrivers.getBookieDriver(bookieDriver.getScheme());
+ fail("Should fail to get bookie driver if it is not registered");
+ } catch (IllegalArgumentException iae) {
+ // expected
+ }
+
+ MetadataDrivers.registerBookieDriver(bookieDriver.getScheme(), bookieDriver.getClass());
+ MetadataBookieDriver driver = MetadataDrivers.getBookieDriver(bookieDriver.getScheme());
+ assertEquals(bookieDriver.getClass(), driver.getClass());
+ }
+
+ @Test
+ public void testLoadClientDriverFromSystemProperty() throws Exception {
+ String saveDriversStr = System.getProperty(BK_METADATA_CLIENT_DRIVERS_PROPERTY);
+ try {
+ System.setProperty(
+ BK_METADATA_CLIENT_DRIVERS_PROPERTY,
+ StringUtils.join(new String[] {
+ ClientDriver1.class.getName(),
+ ClientDriver2.class.getName()
+ }, ':'));
+
+ MetadataDrivers.loadInitialDrivers();
+
+ MetadataClientDriver loadedDriver1 = MetadataDrivers.getClientDriver("driver1");
+ assertEquals(ClientDriver1.class, loadedDriver1.getClass());
+ MetadataClientDriver loadedDriver2 = MetadataDrivers.getClientDriver("driver2");
+ assertEquals(ClientDriver2.class, loadedDriver2.getClass());
+ } finally {
+ if (null != saveDriversStr) {
+ System.setProperty(BK_METADATA_CLIENT_DRIVERS_PROPERTY, saveDriversStr);
+ } else {
+ System.clearProperty(BK_METADATA_CLIENT_DRIVERS_PROPERTY);
+ }
+ }
+ }
+
+ @Test
+ public void testLoadBookieDriverFromSystemProperty() throws Exception {
+ String saveDriversStr = System.getProperty(BK_METADATA_BOOKIE_DRIVERS_PROPERTY);
+ try {
+ System.setProperty(
+ BK_METADATA_BOOKIE_DRIVERS_PROPERTY,
+ StringUtils.join(new String[] {
+ BookieDriver1.class.getName(),
+ BookieDriver2.class.getName()
+ }, ':'));
+
+ MetadataDrivers.loadInitialDrivers();
+
+ MetadataBookieDriver loadedDriver1 = MetadataDrivers.getBookieDriver("driver1");
+ assertEquals(BookieDriver1.class, loadedDriver1.getClass());
+ MetadataBookieDriver loadedDriver2 = MetadataDrivers.getBookieDriver("driver2");
+ assertEquals(BookieDriver2.class, loadedDriver2.getClass());
+ } finally {
+ if (null != saveDriversStr) {
+ System.setProperty(BK_METADATA_BOOKIE_DRIVERS_PROPERTY, saveDriversStr);
+ } else {
+ System.clearProperty(BK_METADATA_BOOKIE_DRIVERS_PROPERTY);
+ }
+ }
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataBookieDriverTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataBookieDriverTest.java
new file mode 100644
index 000000000..99504a0d6
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataBookieDriverTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.zk;
+
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.discover.RegistrationManager;
+import org.apache.bookkeeper.discover.RegistrationManager.RegistrationListener;
+import org.apache.bookkeeper.discover.ZKRegistrationManager;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+/**
+ * Unit test {@link ZKMetadataBookieDriver}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ ZKMetadataDriverBase.class, ZooKeeperClient.class, ZKMetadataBookieDriver.class })
+public class ZKMetadataBookieDriverTest extends ZKMetadataDriverTestBase {
+
+ private ZKMetadataBookieDriver driver;
+ private ServerConfiguration conf;
+
+ @Before
+ public void setup() throws Exception {
+ this.conf = new ServerConfiguration();
+ super.setup(conf);
+
+ driver = new ZKMetadataBookieDriver();
+ }
+
+ @Test
+ public void testGetRegManager() throws Exception {
+ RegistrationListener listener = mock(RegistrationListener.class);
+ driver.initialize(conf, listener, NullStatsLogger.INSTANCE);
+
+ assertSame(conf, driver.serverConf);
+ assertSame(listener, driver.listener);
+ assertNull(driver.regManager);
+
+ ZKRegistrationManager mockRegManager = PowerMockito.mock(ZKRegistrationManager.class);
+
+ PowerMockito.whenNew(ZKRegistrationManager.class)
+ .withNoArguments()
+ .thenReturn(mockRegManager);
+
+ RegistrationManager manager = driver.getRegistrationManager();
+ assertSame(mockRegManager, manager);
+ assertSame(mockRegManager, driver.regManager);
+
+ PowerMockito.verifyNew(ZKRegistrationManager.class, times(1))
+ .withNoArguments();
+ verify(mockRegManager, times(1))
+ .initialize(
+ same(conf),
+ same(listener),
+ same(NullStatsLogger.INSTANCE));
+
+ driver.close();
+ verify(mockRegManager, times(1)).close();
+ assertNull(driver.regManager);
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriverTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriverTest.java
new file mode 100644
index 000000000..4cdf11d1e
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataClientDriverTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.zk;
+
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.discover.RegistrationClient;
+import org.apache.bookkeeper.discover.ZKRegistrationClient;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+/**
+ * Unit test {@link ZKMetadataClientDriver}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ ZKMetadataDriverBase.class, ZooKeeperClient.class, ZKMetadataClientDriver.class })
+public class ZKMetadataClientDriverTest extends ZKMetadataDriverTestBase {
+
+ private ZKMetadataClientDriver driver;
+ private ClientConfiguration conf;
+
+ @Before
+ public void setup() throws Exception {
+ this.conf = new ClientConfiguration();
+ super.setup(conf);
+
+ driver = new ZKMetadataClientDriver();
+ }
+
+ @Test
+ public void testGetRegClient() throws Exception {
+ ScheduledExecutorService mockExecutor = mock(ScheduledExecutorService.class);
+ driver.initialize(conf, mockExecutor, NullStatsLogger.INSTANCE, Optional.empty());
+
+ assertSame(conf, driver.clientConf);
+ assertSame(mockExecutor, driver.scheduler);
+ assertNull(driver.regClient);
+
+ ZKRegistrationClient mockRegClient = PowerMockito.mock(ZKRegistrationClient.class);
+
+ PowerMockito.whenNew(ZKRegistrationClient.class)
+ .withNoArguments()
+ .thenReturn(mockRegClient);
+
+ RegistrationClient client = driver.getRegistrationClient();
+ assertSame(mockRegClient, client);
+ assertSame(mockRegClient, driver.regClient);
+
+ PowerMockito.verifyNew(ZKRegistrationClient.class, times(1))
+ .withNoArguments();
+ verify(mockRegClient, times(1))
+ .initialize(
+ same(conf),
+ same(mockExecutor),
+ same(NullStatsLogger.INSTANCE),
+ eq(Optional.of(driver.zk)));
+
+ driver.close();
+ verify(mockRegClient, times(1)).close();
+ assertNull(driver.regClient);
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBaseStaticTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBaseStaticTest.java
new file mode 100644
index 000000000..2ce4e34fd
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBaseStaticTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.zk;
+
+import static org.junit.Assert.assertEquals;
+
+import java.net.URI;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
+import org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory;
+import org.junit.Test;
+
+/**
+ * Unit test the static methods of {@link ZKMetadataDriverBase}.
+ */
+public class ZKMetadataDriverBaseStaticTest {
+
+ @Test
+ public void testGetZKServersFromServiceUri() {
+ String uriStr = "zk://server1;server2;server3/ledgers";
+ URI uri = URI.create(uriStr);
+
+ String zkServers = ZKMetadataDriverBase.getZKServersFromServiceUri(uri);
+ assertEquals(
+ "server1,server2,server3",
+ zkServers);
+
+ uriStr = "zk://server1,server2,server3/ledgers";
+ uri = URI.create(uriStr);
+ zkServers = ZKMetadataDriverBase.getZKServersFromServiceUri(uri);
+ assertEquals(
+ "server1,server2,server3",
+ zkServers);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testResolveLedgerManagerFactoryNullUri() {
+ ZKMetadataDriverBase.resolveLedgerManagerFactory(null);
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testResolveLedgerManagerFactoryNullScheme() {
+ ZKMetadataDriverBase.resolveLedgerManagerFactory(URI.create("//127.0.0.1/ledgers"));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testResolveLedgerManagerFactoryUnknownScheme() {
+ ZKMetadataDriverBase.resolveLedgerManagerFactory(URI.create("unknown://127.0.0.1/ledgers"));
+ }
+
+ @Test
+ public void testResolveLedgerManagerFactoryDefaultValue() {
+ assertEquals(
+ HierarchicalLedgerManagerFactory.class,
+ ZKMetadataDriverBase.resolveLedgerManagerFactory(
+ URI.create("zk://127.0.0.1/ledgers"))
+ );
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testResolveLedgerManagerFactoryFlat() {
+ assertEquals(
+ org.apache.bookkeeper.meta.FlatLedgerManagerFactory.class,
+ ZKMetadataDriverBase.resolveLedgerManagerFactory(
+ URI.create("zk+flat://127.0.0.1/ledgers"))
+ );
+ }
+
+ @Test
+ public void testResolveLedgerManagerFactoryHierarchical() {
+ assertEquals(
+ HierarchicalLedgerManagerFactory.class,
+ ZKMetadataDriverBase.resolveLedgerManagerFactory(
+ URI.create("zk+hierarchical://127.0.0.1/ledgers"))
+ );
+ }
+
+ @Test
+ public void testResolveLedgerManagerFactoryLongHierarchical() {
+ assertEquals(
+ LongHierarchicalLedgerManagerFactory.class,
+ ZKMetadataDriverBase.resolveLedgerManagerFactory(
+ URI.create("zk+longhierarchical://127.0.0.1/ledgers"))
+ );
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testResolveLedgerManagerFactoryUnknownLedgerManagerFactory() {
+ ZKMetadataDriverBase.resolveLedgerManagerFactory(
+ URI.create("zk+unknown://127.0.0.1/ledgers"));
+ }
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBaseTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBaseTest.java
new file mode 100644
index 000000000..60e768e2f
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverBaseTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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.zk;
+
+import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+import static org.powermock.api.mockito.PowerMockito.verifyStatic;
+
+import java.net.URI;
+import java.util.Optional;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
+import org.apache.bookkeeper.meta.LedgerManagerFactory;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+/**
+ * Unit test of {@link ZKMetadataDriverBase}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ ZKMetadataDriverBase.class, ZooKeeperClient.class, LedgerManagerFactory.class })
+public class ZKMetadataDriverBaseTest extends ZKMetadataDriverTestBase {
+
+ private ZKMetadataDriverBase driver;
+
+ @Before
+ public void setup() throws Exception {
+ super.setup(new ClientConfiguration());
+ driver = mock(ZKMetadataDriverBase.class, CALLS_REAL_METHODS);
+ }
+
+ @Test
+ public void testInitialize() throws Exception {
+ driver.initialize(conf, NullStatsLogger.INSTANCE, Optional.empty());
+
+ assertEquals(URI.create(metadataServiceUri), driver.metadataServiceUri);
+ assertEquals(
+ HierarchicalLedgerManagerFactory.class,
+ driver.ledgerManagerFactoryClass);
+ assertEquals(
+ "/path/to/ledgers",
+ driver.ledgersRootPath);
+ assertTrue(driver.ownZKHandle);
+
+ String readonlyPath = "/path/to/ledgers/" + AVAILABLE_NODE;
+ assertSame(mockZkc, driver.zk);
+ verifyStatic(ZooKeeperClient.class, times(1));
+ ZooKeeperClient.newBuilder();
+ verify(mockZkBuilder, times(1)).build();
+ verify(mockZkc, times(1))
+ .exists(eq(readonlyPath), eq(false));
+ assertNotNull(driver.layoutManager);
+ assertNull(driver.lmFactory);
+
+ driver.close();
+
+ verify(mockZkc, times(1)).close();
+ assertNull(driver.zk);
+ }
+
+ @Test
+ public void testInitializeExternalZooKeeper() throws Exception {
+ ZooKeeperClient anotherZk = mock(ZooKeeperClient.class);
+
+ driver.initialize(conf, NullStatsLogger.INSTANCE, Optional.of(anotherZk));
+
+ assertEquals(URI.create(metadataServiceUri), driver.metadataServiceUri);
+ assertEquals(
+ HierarchicalLedgerManagerFactory.class,
+ driver.ledgerManagerFactoryClass);
+ assertEquals(
+ "/path/to/ledgers",
+ driver.ledgersRootPath);
+ assertFalse(driver.ownZKHandle);
+
+ String readonlyPath = "/path/to/ledgers/" + AVAILABLE_NODE;
+ assertSame(anotherZk, driver.zk);
+ verifyStatic(ZooKeeperClient.class, times(0));
+ ZooKeeperClient.newBuilder();
+ verify(mockZkBuilder, times(0)).build();
+ verify(mockZkc, times(0))
+ .exists(eq(readonlyPath), eq(false));
+ assertNotNull(driver.layoutManager);
+ assertNull(driver.lmFactory);
+
+ driver.close();
+
+ verify(mockZkc, times(0)).close();
+ assertNotNull(driver.zk);
+ }
+
+ @Test
+ public void testGetLedgerManagerFactory() throws Exception {
+ driver.initialize(conf, NullStatsLogger.INSTANCE, Optional.empty());
+
+ mockStatic(LedgerManagerFactory.class);
+ LedgerManagerFactory factory = mock(LedgerManagerFactory.class);
+ PowerMockito.when(
+ LedgerManagerFactory.class,
+ "newLedgerManagerFactory",
+ same(conf),
+ same(driver.layoutManager))
+ .thenReturn(factory);
+
+ assertSame(factory, driver.getLedgerManagerFactory());
+ assertSame(factory, driver.lmFactory);
+ verifyStatic(LedgerManagerFactory.class, times(1));
+ LedgerManagerFactory.newLedgerManagerFactory(
+ same(conf),
+ same(driver.layoutManager));
+
+ driver.close();
+ verify(factory, times(1)).close();
+ assertNull(driver.lmFactory);
+ }
+
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverTestBase.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverTestBase.java
new file mode 100644
index 000000000..453769426
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/zk/ZKMetadataDriverTestBase.java
@@ -0,0 +1,68 @@
+/*
+ * 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.zk;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+
+import org.apache.bookkeeper.conf.AbstractConfiguration;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.zookeeper.RetryPolicy;
+import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
+import org.powermock.api.mockito.PowerMockito;
+
+/**
+ * Unit test of {@link ZKMetadataDriverBase}.
+ */
+public abstract class ZKMetadataDriverTestBase {
+
+ protected AbstractConfiguration<?> conf;
+ protected String metadataServiceUri;
+ protected ZooKeeperClient.Builder mockZkBuilder;
+ protected ZooKeeperClient mockZkc;
+
+ public void setup(AbstractConfiguration<?> conf) throws Exception {
+ metadataServiceUri = "zk://127.0.0.1/path/to/ledgers";
+ this.conf = conf;
+ conf.setMetadataServiceUri(metadataServiceUri);
+
+ this.mockZkBuilder = mock(ZooKeeperClient.Builder.class);
+ when(mockZkBuilder.connectString(eq("127.0.0.1"))).thenReturn(mockZkBuilder);
+ when(mockZkBuilder.sessionTimeoutMs(anyInt())).thenReturn(mockZkBuilder);
+ when(mockZkBuilder.operationRetryPolicy(any(RetryPolicy.class)))
+ .thenReturn(mockZkBuilder);
+ when(mockZkBuilder.statsLogger(any(StatsLogger.class))).thenReturn(mockZkBuilder);
+
+ this.mockZkc = mock(ZooKeeperClient.class);
+ when(mockZkc.exists(anyString(), eq(false)))
+ .thenReturn(null);
+
+ when(mockZkBuilder.build()).thenReturn(mockZkc);
+
+ mockStatic(ZooKeeperClient.class);
+ PowerMockito.when(ZooKeeperClient.class, "newBuilder")
+ .thenReturn(mockZkBuilder);
+ }
+
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services