Skip to content
Permalink
Browse files
HBASE-22694 Fallback to hbase.zookeeper.quorum if fs.hboss.sync.zk.co…
…nnectionString is undefined

Simple code change, but some refactoring to add testing for the change.

EmbeddedZK was made to be an object which we instantiate, rather than
static state and static methods. This lets us re-use the same code for
the contract tests without stomping on one-another.

Also adds in .gitignore entries for Eclipse metadata.

Closes #6

Signed-off-by: Sean Busbey <busbey@apache.org>
  • Loading branch information
joshelser committed Jul 15, 2019
1 parent 6fa0ad9 commit 27d36fd1d7c0a5b486999d260e0e2a2b2d2ccd8d
Showing 6 changed files with 205 additions and 71 deletions.
@@ -2,3 +2,6 @@ target
auth-keys.xml
.idea
*.iml
.settings
.project
.classpath
@@ -86,6 +86,10 @@ public void initialize(FileSystem fs) throws IOException {
// paths to constantly be resolved inside the root.

String zookeeperConnectionString = conf.get(Constants.ZK_CONN_STRING);
// Fallback to the HBase ZK quorum.
if (zookeeperConnectionString == null) {
zookeeperConnectionString = conf.get("hbase.zookeeper.quorum");
}
curator = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
curator.start();
waitForCuratorToConnect();
@@ -118,7 +122,9 @@ private void waitForCuratorToConnect() {

@Override
public void close() throws IOException {
curator.close();
if (curator != null) {
curator.close();
}
}

@Override

This file was deleted.

@@ -18,18 +18,15 @@

package org.apache.hadoop.hbase.oss;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.oss.sync.EmbeddedZK;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;
import org.junit.After;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Rule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@@ -42,6 +39,8 @@ public class TestUtils {
// This is defined by the Maven Surefire plugin configuration
private static final String TEST_UNIQUE_FORK_ID = "test.unique.fork.id";

private static EmbeddedZK zk = null;

public static final String S3A = "s3a";

public static String getScheme(Configuration conf) {
@@ -87,7 +86,12 @@ public static HBaseObjectStoreSemantics getFileSystem(Configuration conf) throws
}

EmbeddedS3.conditionalStart(conf);
EmbeddedZK.conditionalStart(conf);
synchronized (TestUtils.class) {
if (zk == null) {
zk = new EmbeddedZK();
}
}
zk.conditionalStart(conf);

try {
String dataURI = conf.get(Constants.DATA_URI);
@@ -107,6 +111,10 @@ public static void cleanup(HBaseObjectStoreSemantics hboss) throws Exception {
if (hboss != null) {
hboss.close();
}
EmbeddedZK.conditionalStop();
synchronized (TestUtils.class) {
if (zk != null) {
zk.conditionalStop();
}
}
}
}
@@ -0,0 +1,113 @@
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.hadoop.hbase.oss.sync;

import java.net.InetAddress;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.oss.Constants;
import org.apache.hadoop.hbase.oss.sync.TreeLockManager;
import org.apache.hadoop.hbase.oss.sync.ZKTreeLockManager;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;

@InterfaceAudience.Private
@InterfaceStability.Unstable
public class EmbeddedZK {

// Guards `testUtil` -- making sure we don't start mulitple ZKs.
private final ReentrantReadWriteLock LOCK = new ReentrantReadWriteLock();
private Object testUtil = null;
private String connectionString = null;

public void conditionalStart(Configuration conf) throws Exception {
Class<?> implementation = conf.getClass(Constants.SYNC_IMPL, TreeLockManager.class);
if (implementation != ZKTreeLockManager.class) {
return;
}
LOCK.readLock().lock();
if (testUtil == null) {
LOCK.readLock().unlock();
try {
LOCK.writeLock().lock();
// If the server is non-null after we took the lock, someone else just beat
// us here. Bail out.
if (testUtil != null) {
return;
}

start(conf);
} finally {
LOCK.writeLock().unlock();
}
} else {
// Set the ZK connection details into this conf. It might not have them.
setConfiguration(conf);
LOCK.readLock().unlock();
}
}

/**
* Requires external synchronization!
*/
protected void start(Configuration conf) throws Exception {
Class<?> testUtilImpl;
try {
testUtilImpl = Class.forName("org.apache.hadoop.hbase.HBaseZKTestingUtility");
} catch (ClassNotFoundException ex) {
testUtilImpl = Class.forName("org.apache.hadoop.hbase.HBaseTestingUtility");
}
testUtil = testUtilImpl.getDeclaredConstructor(Configuration.class).newInstance(conf);
testUtil.getClass().getDeclaredMethod("startMiniZKCluster").invoke(testUtil);

Object zkCluster = testUtil.getClass().getDeclaredMethod("getZkCluster").invoke(testUtil);
int port = (int) zkCluster.getClass().getDeclaredMethod("getClientPort").invoke(zkCluster);
String hostname = InetAddress.getLocalHost().getHostName();
connectionString = hostname + ":" + port;
setConfiguration(conf);
}

public void conditionalStop() throws Exception {
try {
LOCK.writeLock().lock();
if (testUtil != null) {
testUtil.getClass().getDeclaredMethod("shutdownMiniZKCluster").invoke(testUtil);
testUtil = null;
}
} finally {
LOCK.writeLock().unlock();
}
}

protected void setConfiguration(Configuration conf) {
conf.set(Constants.ZK_CONN_STRING, connectionString);
conf.set(HConstants.ZOOKEEPER_QUORUM, connectionString);
}

protected String getConnectionString() {
try {
LOCK.readLock().lock();
return connectionString;
} finally {
LOCK.readLock().unlock();
}
}
}
@@ -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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.hadoop.hbase.oss.sync;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.oss.Constants;
import org.junit.Test;

public class TestZKLockManagerConfig {

@Test
public void testLockManagerConfigFallback() throws Exception {
EmbeddedZK zk1 = new EmbeddedZK();
ZKTreeLockManager lockMgr = null;
Configuration conf = new Configuration(false);
conf.setClass(Constants.SYNC_IMPL, ZKTreeLockManager.class, TreeLockManager.class);
try {
zk1.conditionalStart(conf);

String expectedQuorum = zk1.getConnectionString();
assertNotNull(expectedQuorum);

// Validate that both config properties are set
assertEquals(conf.get(Constants.ZK_CONN_STRING), expectedQuorum);
assertEquals(conf.get(HConstants.ZOOKEEPER_QUORUM), expectedQuorum);

// Unset the HBoss-specific property to force the LockManager to use the HBase ZK quorum
conf.unset(Constants.ZK_CONN_STRING);
assertNull(conf.get(Constants.ZK_CONN_STRING));

// Get a LocalFS -- we don't really care about it, just passing it to the lockManager.
FileSystem fs = LocalFileSystem.get(conf);

// Initializing the ZKTreeLockManager should succeed, even when we only have
// the hbase.zookeeper.quorum config property set.
lockMgr = new ZKTreeLockManager();
lockMgr.initialize(fs);
} finally {
// Clean up everything.
if (lockMgr != null) {
lockMgr.close();
}
zk1.conditionalStop();
}
}
}

0 comments on commit 27d36fd

Please sign in to comment.